Harwinder Sekhon
44cadf7a9f
Resolves the Twelve Data API key through app_config.provider_api_key() (env → config.json → no default) instead of a hardcoded field on PlatformConfig. Adds data_provider.PROVIDER_REGISTRY + make_data_provider() so signal_aggregator and the backtester build the active provider through one seam; telegram_bot.py gets the same one-line WS-key fix now (per Task 4's already-planned edit for that line) so the suite doesn't crash on import between this commit and Task 4's full wiring pass. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VPNF37MP9Kuv62B1VBrZ1b
2276 lines
93 KiB
Python
2276 lines
93 KiB
Python
"""
|
||
ForexBot2 — GPU-Accelerated Backtester
|
||
======================================
|
||
Replays historical OHLCV bar-by-bar through the live SignalAggregator so
|
||
backtest results are identical to what the production scanner would have
|
||
generated. Pre-computes all indicator series with CuPy/NumPy on the full
|
||
dataset (causal, zero look-ahead) and looks up values per bar instead of
|
||
recomputing each time — giving a 100x+ speedup over a naive replay while
|
||
preserving the exact signal logic.
|
||
|
||
Key properties:
|
||
- Uses real SignalAggregator.analyze_from_dataframes (same gates,
|
||
same XAU strategy hooks, same trade setup math).
|
||
- News engine disabled (use_news=False) — backtests are pure technical.
|
||
- Session filter, MTF agreement, regime classification all flow from the
|
||
bar timestamp, never datetime.now().
|
||
- XAU weekend/weekday filter applied here (analyze_from_dataframes skips
|
||
it; only analyze_pair enforces it in live mode).
|
||
- Trade simulation is full candle walk on the primary timeframe using
|
||
High/Low; if both SL and TP1 hit in the same bar, SL is assumed first
|
||
(conservative).
|
||
- Ships with CuPy fallback to NumPy so the script runs anywhere.
|
||
|
||
Usage:
|
||
python backtest/gpu_backtest.py --pair XAUUSD --days 365
|
||
python backtest/gpu_backtest.py --pair XAUUSD --start 2025-01-01 --end 2025-06-01
|
||
python backtest/gpu_backtest.py --pair XAUUSD --days 180 --no-cache
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import logging
|
||
import multiprocessing as mp
|
||
import os
|
||
import sys
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
# Ensure project root on sys.path so this script can be invoked from anywhere.
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
if str(PROJECT_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
from config import ( # noqa: E402
|
||
PAIR_INFO,
|
||
PlatformConfig,
|
||
SignalStrength,
|
||
Timeframe,
|
||
XAUUSD,
|
||
)
|
||
from data_provider import ForexDataProvider, make_data_provider # noqa: E402
|
||
from engine import IndicatorSignal, TechnicalAnalysis, TechnicalEngine # noqa: E402
|
||
from signal_aggregator import SignalAggregator, SignalReport, TradeSetup # noqa: E402
|
||
|
||
# ─────────────────────────────────────────────────────────────────────
|
||
# GPU shim — CuPy if available AND CUDA runtime works, else NumPy.
|
||
# An importable CuPy can still fail at first kernel launch if CUDA libs
|
||
# are missing (containers, headless boxes), so verify with a tiny op.
|
||
# ─────────────────────────────────────────────────────────────────────
|
||
def _detect_gpu():
|
||
try:
|
||
import cupy as cp_mod # type: ignore
|
||
# Force runtime init: a kernel launch will surface missing libs.
|
||
_ = cp_mod.asarray(np.zeros(2, dtype=np.float64))
|
||
_ = cp_mod.cumsum(_).get()
|
||
return cp_mod, True
|
||
except Exception:
|
||
return np, False
|
||
|
||
|
||
cp, _GPU_AVAILABLE = _detect_gpu()
|
||
|
||
|
||
def _to_gpu(arr: np.ndarray):
|
||
"""Move a numpy array to GPU (or no-op if CuPy unavailable)."""
|
||
return cp.asarray(arr) if _GPU_AVAILABLE else arr
|
||
|
||
|
||
def _to_cpu(arr) -> np.ndarray:
|
||
"""Pull array back to host."""
|
||
if _GPU_AVAILABLE and hasattr(arr, "get"):
|
||
return arr.get()
|
||
return np.asarray(arr)
|
||
|
||
|
||
logger = logging.getLogger("gpu_backtest")
|
||
|
||
|
||
PAIR_ALIASES = {
|
||
"XAUUSD": XAUUSD,
|
||
"XAU/USD": XAUUSD,
|
||
"GOLD": XAUUSD,
|
||
}
|
||
|
||
# Bar duration in seconds — used to compute "fully-closed" cutoffs so
|
||
# confirmation timeframes can't see into the future at decision time.
|
||
BAR_SECONDS = {
|
||
Timeframe.M1: 60, Timeframe.M5: 300, Timeframe.M15: 900,
|
||
Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400,
|
||
Timeframe.D1: 86400, Timeframe.W1: 604800,
|
||
}
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# GPU INDICATOR PRE-COMPUTATION
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
def _gpu_ewm_alpha(x: np.ndarray, alpha: float, min_periods: int = 0) -> np.ndarray:
|
||
"""
|
||
Exponential weighted mean matching pandas
|
||
`ewm(alpha=alpha, min_periods=p, adjust=False).mean()` exactly.
|
||
|
||
Recursion: y[i] = alpha * x[i] + (1-alpha) * y[i-1], seeded at the
|
||
first non-NaN x. NaN inputs carry the previous y forward without
|
||
advancing the non-NaN counter (pandas' default `ignore_na=False`).
|
||
Output positions before min_periods non-NaN observations are NaN.
|
||
|
||
Sequential CPU pass — EMA is recursive so true GPU parallelism
|
||
requires a scan kernel, which is overkill for a one-shot pre-compute.
|
||
"""
|
||
n = x.shape[0]
|
||
out = np.full(n, np.nan, dtype=np.float64)
|
||
if n == 0:
|
||
return out
|
||
|
||
one_minus = 1.0 - alpha
|
||
seen = 0
|
||
last_y = np.nan
|
||
|
||
for i in range(n):
|
||
xi = x[i]
|
||
if np.isnan(xi):
|
||
# NaN input: y stays at last_y, don't advance counter.
|
||
if not np.isnan(last_y) and seen >= min_periods:
|
||
out[i] = last_y
|
||
continue
|
||
last_y = xi if np.isnan(last_y) else alpha * xi + one_minus * last_y
|
||
seen += 1
|
||
if seen >= min_periods:
|
||
out[i] = last_y
|
||
|
||
return out
|
||
|
||
|
||
def _gpu_ewm_span(x: np.ndarray, span: int) -> np.ndarray:
|
||
"""ewm with span= semantics (alpha = 2/(span+1), no min_periods, seeded at x[0])."""
|
||
alpha = 2.0 / (span + 1.0)
|
||
return _gpu_ewm_alpha(x, alpha, min_periods=0)
|
||
|
||
|
||
@dataclass
|
||
class IndicatorCache:
|
||
"""Pre-computed indicator series for one (symbol, timeframe).
|
||
|
||
All arrays are aligned to the source df's index (same length).
|
||
NaN at warmup positions; downstream lookup tolerates it.
|
||
"""
|
||
df: pd.DataFrame # full source df (cleaned)
|
||
index_pos: Dict[pd.Timestamp, int] # timestamp → position lookup
|
||
|
||
# EMAs (4-EMA mode keys; 2-EMA mode reuses ema_slow + ema_trend + ema_medium)
|
||
ema_fast: np.ndarray
|
||
ema_medium: np.ndarray
|
||
ema_slow: np.ndarray
|
||
ema_trend: Optional[np.ndarray] # None if len(df) < ema_trend period
|
||
|
||
# RSI
|
||
rsi: np.ndarray
|
||
rsi_series: pd.Series # pandas series for divergence
|
||
|
||
# MACD
|
||
macd_hist: np.ndarray
|
||
macd_hist_series: pd.Series
|
||
|
||
# Bollinger
|
||
bb_mid: np.ndarray
|
||
bb_upper: np.ndarray
|
||
bb_lower: np.ndarray
|
||
bb_position: np.ndarray
|
||
bb_band_width: np.ndarray
|
||
bb_avg_width: np.ndarray # 20-period rolling mean of band width
|
||
|
||
# ATR
|
||
atr: np.ndarray
|
||
|
||
# ADX
|
||
adx: np.ndarray
|
||
plus_di: np.ndarray
|
||
minus_di: np.ndarray
|
||
adx_slope: np.ndarray
|
||
|
||
# ROC
|
||
roc: np.ndarray
|
||
|
||
# Keltner Channels
|
||
kc_mid: np.ndarray
|
||
kc_upper: np.ndarray
|
||
kc_lower: np.ndarray
|
||
kc_squeeze: np.ndarray
|
||
|
||
|
||
def precompute_indicators(
|
||
df: pd.DataFrame, tech_cfg, gpu_log: bool = False
|
||
) -> IndicatorCache:
|
||
"""Compute every causal indicator series once on the full df."""
|
||
close = df["Close"].to_numpy(dtype=np.float64)
|
||
high = df["High"].to_numpy(dtype=np.float64)
|
||
low = df["Low"].to_numpy(dtype=np.float64)
|
||
n = len(close)
|
||
|
||
if gpu_log:
|
||
logger.info(
|
||
f"Pre-computing indicators on {n} bars "
|
||
f"({'GPU' if _GPU_AVAILABLE else 'CPU'} backend)"
|
||
)
|
||
|
||
# ── EMAs (use ewm span semantics, identical to pandas .ewm(span=, adjust=False).mean()) ─
|
||
ema_fast = _gpu_ewm_span(close, tech_cfg.ema_fast)
|
||
ema_medium = _gpu_ewm_span(close, tech_cfg.ema_medium)
|
||
ema_slow = _gpu_ewm_span(close, tech_cfg.ema_slow)
|
||
ema_trend = _gpu_ewm_span(close, tech_cfg.ema_trend) if n >= tech_cfg.ema_trend else None
|
||
|
||
# ── RSI (Wilder via ewm alpha=1/period, min_periods=period) ──
|
||
# pandas: delta = close.diff() → delta[0] is NaN; gain = delta.where(delta > 0, 0.0)
|
||
# evaluates NaN > 0 as False so gain[0] becomes 0.0 (NOT NaN). Same for loss.
|
||
delta = np.empty(n, dtype=np.float64)
|
||
delta[0] = np.nan
|
||
delta[1:] = close[1:] - close[:-1]
|
||
with np.errstate(invalid="ignore"):
|
||
gain = np.where(delta > 0, delta, 0.0)
|
||
loss = np.where(delta < 0, -delta, 0.0)
|
||
avg_gain = _gpu_ewm_alpha(gain, 1.0 / tech_cfg.rsi_period, min_periods=tech_cfg.rsi_period)
|
||
avg_loss = _gpu_ewm_alpha(loss, 1.0 / tech_cfg.rsi_period, min_periods=tech_cfg.rsi_period)
|
||
avg_loss_safe = np.where(avg_loss == 0, 1e-10, avg_loss)
|
||
rs = avg_gain / avg_loss_safe
|
||
rsi = 100.0 - (100.0 / (1.0 + rs))
|
||
rsi_series = pd.Series(rsi, index=df.index, name="rsi")
|
||
|
||
# ── MACD (12/26 EMA spans, signal 9 EMA span) ──
|
||
ema_macd_fast = _gpu_ewm_span(close, tech_cfg.macd_fast)
|
||
ema_macd_slow = _gpu_ewm_span(close, tech_cfg.macd_slow)
|
||
macd_line = ema_macd_fast - ema_macd_slow
|
||
signal_line = _gpu_ewm_span(macd_line, tech_cfg.macd_signal)
|
||
macd_hist = macd_line - signal_line
|
||
macd_hist_series = pd.Series(macd_hist, index=df.index, name="macd_hist")
|
||
|
||
# ── Bollinger ──
|
||
# Computed with the SAME pandas rolling calls as the live engine
|
||
# (engine._compute_bollinger). The previous cumsum E[x²]−E[x]² shortcut
|
||
# loses precision at gold's price level (bb_position error up to ~8e-3
|
||
# — enough to cross the 0.0/0.2/0.8/1.0 signal thresholds, worst during
|
||
# squeezes). Replay slices are prefixes of this df, so pandas rolling
|
||
# values here are bit-identical to what the live engine computes per
|
||
# slice. Rolling on the full series runs once — cost is negligible.
|
||
close_s = df["Close"]
|
||
bb_mid_s = close_s.rolling(tech_cfg.bb_period).mean()
|
||
bb_std_s = close_s.rolling(tech_cfg.bb_period).std(ddof=0)
|
||
bb_mid = bb_mid_s.to_numpy(dtype=np.float64)
|
||
bb_std = bb_std_s.to_numpy(dtype=np.float64)
|
||
bb_upper = bb_mid + tech_cfg.bb_std * bb_std
|
||
bb_lower = bb_mid - tech_cfg.bb_std * bb_std
|
||
bb_range = bb_upper - bb_lower
|
||
bb_range_safe = np.where(bb_range == 0, np.nan, bb_range)
|
||
bb_position = (close - bb_lower) / bb_range_safe
|
||
# pandas behavior: when range is 0 it returns 0.5 (handled by engine fallback)
|
||
bb_position = np.where(np.isnan(bb_position), 0.5, bb_position)
|
||
bb_mid_safe = np.where(bb_mid == 0, np.nan, bb_mid)
|
||
bb_band_width = (bb_upper - bb_lower) / bb_mid_safe
|
||
bb_avg_width = (
|
||
pd.Series(bb_band_width, index=df.index).rolling(20).mean()
|
||
.to_numpy(dtype=np.float64)
|
||
)
|
||
|
||
# ── ATR ──
|
||
prev_close = np.empty(n, dtype=np.float64)
|
||
prev_close[0] = np.nan
|
||
prev_close[1:] = close[:-1]
|
||
tr1 = high - low
|
||
tr2 = np.abs(high - prev_close)
|
||
tr3 = np.abs(low - prev_close)
|
||
# NaN handling at index 0: pandas concat+max returns max ignoring NaN
|
||
# but for our purposes the engine takes the latest value, so propagate NaN here
|
||
tr2[0] = np.nan
|
||
tr3[0] = np.nan
|
||
tr = np.fmax(np.fmax(tr1, tr2), tr3) # fmax ignores NaN — matches pandas behavior
|
||
# ewm: pandas uses the first non-NaN value to seed
|
||
atr = _gpu_ewm_alpha(tr, 1.0 / tech_cfg.atr_period, min_periods=tech_cfg.atr_period)
|
||
|
||
# ── ADX ──
|
||
# pandas: plus_dm = high.diff(); minus_dm = -low.diff(); both NaN at index 0.
|
||
# plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0) — NaN comparisons
|
||
# are False, so plus_dm[0] becomes 0.0 (NOT NaN). Same for minus_dm.
|
||
plus_dm_raw = np.empty(n, dtype=np.float64)
|
||
minus_dm_raw = np.empty(n, dtype=np.float64)
|
||
plus_dm_raw[0] = np.nan
|
||
minus_dm_raw[0] = np.nan
|
||
plus_dm_raw[1:] = high[1:] - high[:-1]
|
||
minus_dm_raw[1:] = -(low[1:] - low[:-1])
|
||
with np.errstate(invalid="ignore"):
|
||
cond_plus = (plus_dm_raw > minus_dm_raw) & (plus_dm_raw > 0)
|
||
plus_dm = np.where(cond_plus, plus_dm_raw, 0.0)
|
||
cond_minus = (minus_dm_raw > plus_dm_raw) & (minus_dm_raw > 0)
|
||
minus_dm = np.where(cond_minus, minus_dm_raw, 0.0)
|
||
|
||
atr_adx = _gpu_ewm_alpha(tr, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period)
|
||
plus_dm_smooth = _gpu_ewm_alpha(
|
||
plus_dm, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period
|
||
)
|
||
minus_dm_smooth = _gpu_ewm_alpha(
|
||
minus_dm, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period
|
||
)
|
||
atr_adx_safe = np.where(atr_adx == 0, np.nan, atr_adx)
|
||
plus_di = 100.0 * plus_dm_smooth / atr_adx_safe
|
||
minus_di = 100.0 * minus_dm_smooth / atr_adx_safe
|
||
di_sum = plus_di + minus_di
|
||
di_sum_safe = np.where(di_sum == 0, np.nan, di_sum)
|
||
dx = 100.0 * np.abs(plus_di - minus_di) / di_sum_safe
|
||
adx = _gpu_ewm_alpha(dx, 1.0 / tech_cfg.adx_period, min_periods=tech_cfg.adx_period)
|
||
|
||
# ── ADX Slope ──
|
||
adx_slope_lookback = getattr(tech_cfg, 'adx_slope_lookback', 5)
|
||
adx_slope = np.full(n, np.nan, dtype=np.float64)
|
||
if n > adx_slope_lookback:
|
||
adx_slope[adx_slope_lookback:] = adx[adx_slope_lookback:] - adx[:-adx_slope_lookback]
|
||
|
||
# ── ROC (Rate of Change) ──
|
||
roc_period = getattr(tech_cfg, 'roc_period', 10)
|
||
roc = np.full(n, np.nan, dtype=np.float64)
|
||
if n > roc_period:
|
||
prev_close = close[:-roc_period]
|
||
prev_safe = np.where(prev_close == 0, np.nan, prev_close)
|
||
roc[roc_period:] = ((close[roc_period:] - prev_close) / prev_safe) * 100.0
|
||
|
||
# ── Keltner Channels ──
|
||
kc_period = getattr(tech_cfg, 'kc_period', 20)
|
||
kc_atr_period = getattr(tech_cfg, 'kc_atr_period', 10)
|
||
kc_multiplier = getattr(tech_cfg, 'kc_multiplier', 1.5)
|
||
kc_mid = _gpu_ewm_span(close, kc_period)
|
||
kc_atr = _gpu_ewm_alpha(tr, 1.0 / kc_atr_period, min_periods=kc_atr_period)
|
||
kc_upper = kc_mid + kc_multiplier * kc_atr
|
||
kc_lower = kc_mid - kc_multiplier * kc_atr
|
||
with np.errstate(invalid="ignore"):
|
||
kc_squeeze = (bb_lower > kc_lower) & (bb_upper < kc_upper)
|
||
kc_squeeze = kc_squeeze.astype(np.float64)
|
||
|
||
index_pos = {ts: i for i, ts in enumerate(df.index)}
|
||
|
||
return IndicatorCache(
|
||
df=df,
|
||
index_pos=index_pos,
|
||
ema_fast=ema_fast,
|
||
ema_medium=ema_medium,
|
||
ema_slow=ema_slow,
|
||
ema_trend=ema_trend,
|
||
rsi=rsi,
|
||
rsi_series=rsi_series,
|
||
macd_hist=macd_hist,
|
||
macd_hist_series=macd_hist_series,
|
||
bb_mid=bb_mid,
|
||
bb_upper=bb_upper,
|
||
bb_lower=bb_lower,
|
||
bb_position=bb_position,
|
||
bb_band_width=bb_band_width,
|
||
bb_avg_width=bb_avg_width,
|
||
atr=atr,
|
||
adx=adx,
|
||
plus_di=plus_di,
|
||
minus_di=minus_di,
|
||
adx_slope=adx_slope,
|
||
roc=roc,
|
||
kc_mid=kc_mid,
|
||
kc_upper=kc_upper,
|
||
kc_lower=kc_lower,
|
||
kc_squeeze=kc_squeeze,
|
||
)
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# GPU-AWARE TECHNICAL ENGINE (subclasses live engine, lookup-only)
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
class GPUTechnicalEngine(TechnicalEngine):
|
||
"""
|
||
Drop-in TechnicalEngine that pulls indicator values from a pre-computed
|
||
cache instead of recomputing per call. The conditional/scoring/description
|
||
logic for each indicator is preserved verbatim by overriding only the
|
||
`_compute_*` methods — every downstream consumer (regime classifier,
|
||
composite scorer, divergence, structure_sr, candlestick) sees an
|
||
IndicatorSignal byte-identical to what the live engine would produce.
|
||
"""
|
||
|
||
def __init__(self, tech_cfg, caches: Dict[str, IndicatorCache]):
|
||
super().__init__(tech_cfg)
|
||
# cache key = timeframe.value (e.g. "15min", "1h", "4h")
|
||
self.caches = caches
|
||
self._current_tf: Optional[str] = None
|
||
|
||
def analyze(self, df: pd.DataFrame, symbol: str, timeframe: str) -> TechnicalAnalysis:
|
||
# Latch the timeframe so per-indicator overrides know which cache to use.
|
||
# Multiple TFs share aligned timestamps (e.g. M15 14:00 == H1 14:00),
|
||
# so timestamp-only lookup would match the wrong cache.
|
||
self._current_tf = timeframe
|
||
try:
|
||
return super().analyze(df, symbol, timeframe)
|
||
finally:
|
||
self._current_tf = None
|
||
|
||
def _resolve_pos(self, df: pd.DataFrame) -> Tuple[Optional[IndicatorCache], int]:
|
||
"""Find this slice's position in its pre-computed cache."""
|
||
if df is None or df.empty or self._current_tf is None:
|
||
return None, -1
|
||
cache = self.caches.get(self._current_tf)
|
||
if cache is None:
|
||
return None, -1
|
||
pos = cache.index_pos.get(df.index[-1])
|
||
return (cache, pos) if pos is not None else (None, -1)
|
||
|
||
# ── EMA overrides ──────────────────────────────────────────────
|
||
|
||
def _compute_ema_2(self, df: pd.DataFrame, result: TechnicalAnalysis):
|
||
cache, pos = self._resolve_pos(df)
|
||
if cache is None or pos < 0:
|
||
return super()._compute_ema_2(df, result)
|
||
|
||
c = self.config
|
||
latest = float(df["Close"].iloc[-1])
|
||
s = float(cache.ema_slow[pos])
|
||
m = float(cache.ema_medium[pos])
|
||
t = float(cache.ema_trend[pos]) if cache.ema_trend is not None and not np.isnan(cache.ema_trend[pos]) else None
|
||
|
||
if t is None or pos < c.ema_trend - 1:
|
||
result.indicators["ema_trend"] = IndicatorSignal(
|
||
name="EMA Trend", value=0, signal=SignalStrength.NEUTRAL,
|
||
confidence=30, description="EMA200 data insufficient", weight=1.0,
|
||
)
|
||
result.key_levels["ema_medium"] = m
|
||
result.key_levels["ema_slow"] = s
|
||
return
|
||
|
||
if latest > s > t:
|
||
signal, conf, desc = SignalStrength.STRONG_BUY, 85, "Full bullish: price > EMA50 > EMA200"
|
||
elif latest > s and s < t:
|
||
signal, conf, desc = SignalStrength.BUY, 60, "Emerging uptrend: price > EMA50, EMA50 < EMA200"
|
||
elif latest < s < t:
|
||
signal, conf, desc = SignalStrength.STRONG_SELL, 85, "Full bearish: price < EMA50 < EMA200"
|
||
elif latest < s and s > t:
|
||
signal, conf, desc = SignalStrength.SELL, 60, "Emerging downtrend: price < EMA50, EMA50 > EMA200"
|
||
else:
|
||
signal, conf, desc = SignalStrength.NEUTRAL, 40, "EMA alignment mixed"
|
||
|
||
# Crossover detection on prior 10 bars (live: `if len(ema50) >= 10`,
|
||
# `for lookback in range(2, min(11, len(ema50)))` — len = pos+1).
|
||
if pos >= 9:
|
||
for lookback in range(2, min(11, pos + 1)):
|
||
prev_s = float(cache.ema_slow[pos - lookback + 1])
|
||
prev_t = float(cache.ema_trend[pos - lookback + 1])
|
||
if prev_s <= prev_t and s > t:
|
||
desc += " | Golden cross (50/200)"
|
||
conf = min(conf + 10, 95)
|
||
break
|
||
if prev_s >= prev_t and s < t:
|
||
desc += " | Death cross (50/200)"
|
||
conf = min(conf + 10, 95)
|
||
break
|
||
|
||
normalized = {
|
||
SignalStrength.STRONG_BUY: 1.0, SignalStrength.BUY: 0.5,
|
||
SignalStrength.NEUTRAL: 0.0,
|
||
SignalStrength.SELL: -0.5, SignalStrength.STRONG_SELL: -1.0,
|
||
}[signal]
|
||
|
||
result.indicators["ema_trend"] = IndicatorSignal(
|
||
name="EMA Trend", value=normalized, signal=signal,
|
||
confidence=conf, description=desc, weight=1.0,
|
||
)
|
||
result.key_levels["ema_trend"] = t
|
||
result.key_levels["ema_medium"] = m
|
||
result.key_levels["ema_slow"] = s
|
||
|
||
def _compute_ema_4(self, df: pd.DataFrame, result: TechnicalAnalysis):
|
||
cache, pos = self._resolve_pos(df)
|
||
if cache is None or pos < 0:
|
||
return super()._compute_ema_4(df, result)
|
||
|
||
c = self.config
|
||
latest = float(df["Close"].iloc[-1])
|
||
f = float(cache.ema_fast[pos])
|
||
m = float(cache.ema_medium[pos])
|
||
s = float(cache.ema_slow[pos])
|
||
t_arr = cache.ema_trend
|
||
t = float(t_arr[pos]) if (t_arr is not None and not np.isnan(t_arr[pos])) else None
|
||
# Live engine guards `t` with `len(close) >= ema_trend`; replicate.
|
||
if t is not None and pos < c.ema_trend - 1:
|
||
t = None
|
||
|
||
alignment_score = 0
|
||
checks = [(latest, f), (f, m), (m, s)]
|
||
if t is not None:
|
||
checks.append((s, t))
|
||
for above, below in checks:
|
||
if above > below:
|
||
alignment_score += 1
|
||
elif above < below:
|
||
alignment_score -= 1
|
||
max_checks = len(checks)
|
||
normalized = alignment_score / max_checks if max_checks > 0 else 0
|
||
|
||
if normalized > 0.6:
|
||
signal = SignalStrength.STRONG_BUY
|
||
conf = min(90, 60 + normalized * 30)
|
||
elif normalized > 0.2:
|
||
signal = SignalStrength.BUY
|
||
conf = 55 + normalized * 20
|
||
elif normalized < -0.6:
|
||
signal = SignalStrength.STRONG_SELL
|
||
conf = min(90, 60 + abs(normalized) * 30)
|
||
elif normalized < -0.2:
|
||
signal = SignalStrength.SELL
|
||
conf = 55 + abs(normalized) * 20
|
||
else:
|
||
signal = SignalStrength.NEUTRAL
|
||
conf = 40
|
||
|
||
crossover_desc = ""
|
||
# Live: `if len(ema_fast) >= 3:` then iloc[-2] — needs len ≥ 2 functionally,
|
||
# but live's guard requires len ≥ 3 so we match that to stay byte-identical.
|
||
if pos >= 2:
|
||
prev_f = float(cache.ema_fast[pos - 1])
|
||
prev_m = float(cache.ema_medium[pos - 1])
|
||
if prev_f <= prev_m and f > m:
|
||
crossover_desc = " | Golden cross (fast/medium)"
|
||
conf = min(conf + 10, 95)
|
||
elif prev_f >= prev_m and f < m:
|
||
crossover_desc = " | Death cross (fast/medium)"
|
||
conf = min(conf + 10, 95)
|
||
|
||
# EMA ribbon compression/expansion detection — mirrors the live
|
||
# engine exactly (iloc[-5] ≡ pos-4, `len >= 6` ≡ pos >= 5). ATR is
|
||
# already in key_levels because analyze() computes ATR first.
|
||
ribbon_desc = ""
|
||
atr_val = result.key_levels.get("atr", 0)
|
||
ema_values = [f, m, s]
|
||
if atr_val > 0:
|
||
ribbon_width = (max(ema_values) - min(ema_values)) / atr_val
|
||
if ribbon_width < 1.0:
|
||
ribbon_desc = " | Ribbon COMPRESSED (trend change imminent)"
|
||
if pos >= 5:
|
||
prev_vals = [float(cache.ema_fast[pos - 4]),
|
||
float(cache.ema_medium[pos - 4]),
|
||
float(cache.ema_slow[pos - 4])]
|
||
prev_width = (max(prev_vals) - min(prev_vals)) / atr_val if atr_val > 0 else 0
|
||
if prev_width < 1.0 and ribbon_width >= 1.0:
|
||
ribbon_desc = " | Ribbon EXPANDING (new trend starting)"
|
||
conf = min(conf + 10, 95)
|
||
|
||
desc = f"EMA alignment: {alignment_score}/{max_checks}{crossover_desc}{ribbon_desc}"
|
||
result.indicators["ema_trend"] = IndicatorSignal(
|
||
name="EMA Trend", value=normalized, signal=signal,
|
||
confidence=conf, description=desc, weight=1.0,
|
||
)
|
||
if t is not None:
|
||
result.key_levels["ema_trend"] = t
|
||
result.key_levels["ema_fast"] = f
|
||
result.key_levels["ema_medium"] = m
|
||
result.key_levels["ema_slow"] = s
|
||
|
||
# ── RSI override ──────────────────────────────────────────────
|
||
|
||
def _compute_rsi(self, df: pd.DataFrame, result: TechnicalAnalysis):
|
||
cache, pos = self._resolve_pos(df)
|
||
if cache is None or pos < 0:
|
||
return super()._compute_rsi(df, result)
|
||
|
||
c = self.config
|
||
rsi_val = float(cache.rsi[pos])
|
||
if np.isnan(rsi_val):
|
||
rsi_val = 50.0
|
||
|
||
if rsi_val >= c.rsi_overbought:
|
||
signal = SignalStrength.SELL
|
||
conf = min(80, 50 + (rsi_val - c.rsi_overbought))
|
||
desc = f"RSI {rsi_val:.1f} — Overbought"
|
||
elif rsi_val <= c.rsi_oversold:
|
||
signal = SignalStrength.BUY
|
||
conf = min(80, 50 + (c.rsi_oversold - rsi_val))
|
||
desc = f"RSI {rsi_val:.1f} — Oversold"
|
||
elif rsi_val >= c.rsi_neutral_upper:
|
||
signal = SignalStrength.BUY
|
||
conf = 45
|
||
desc = f"RSI {rsi_val:.1f} — Bullish momentum zone"
|
||
elif rsi_val <= c.rsi_neutral_lower:
|
||
signal = SignalStrength.SELL
|
||
conf = 45
|
||
desc = f"RSI {rsi_val:.1f} — Bearish momentum zone"
|
||
else:
|
||
signal = SignalStrength.NEUTRAL
|
||
conf = 30
|
||
desc = f"RSI {rsi_val:.1f} — Neutral"
|
||
|
||
result.indicators["rsi"] = IndicatorSignal(
|
||
name="RSI", value=rsi_val, signal=signal,
|
||
confidence=conf, description=desc, weight=1.5,
|
||
)
|
||
# Slice the pre-computed series so divergence detector sees only
|
||
# bars up to the current decision point (no look-ahead).
|
||
result._rsi_series = cache.rsi_series.iloc[: pos + 1] # type: ignore[attr-defined]
|
||
|
||
# ── MACD override ─────────────────────────────────────────────
|
||
|
||
def _compute_macd(self, df: pd.DataFrame, result: TechnicalAnalysis):
|
||
cache, pos = self._resolve_pos(df)
|
||
if cache is None or pos < 0:
|
||
return super()._compute_macd(df, result)
|
||
|
||
c = self.config
|
||
result._macd_histogram = cache.macd_hist_series.iloc[: pos + 1] # type: ignore[attr-defined]
|
||
|
||
if not c.macd_as_pillar or pos < 2:
|
||
return
|
||
|
||
hist_now = float(cache.macd_hist[pos])
|
||
hist_prev = float(cache.macd_hist[pos - 1])
|
||
hist_prev2 = float(cache.macd_hist[pos - 2])
|
||
if np.isnan(hist_now) or np.isnan(hist_prev):
|
||
return
|
||
|
||
crossed_up = hist_prev <= 0 < hist_now
|
||
crossed_down = hist_prev >= 0 > hist_now
|
||
accel_up = hist_now > 0 and hist_now > hist_prev > 0
|
||
accel_down = hist_now < 0 and hist_now < hist_prev < 0
|
||
fading_up = hist_now > 0 and hist_now < hist_prev
|
||
fading_down = hist_now < 0 and hist_now > hist_prev
|
||
|
||
if pos >= 19:
|
||
window = np.abs(cache.macd_hist[pos - 19: pos + 1])
|
||
mean_abs = float(np.nanmean(window))
|
||
hist_pct = abs(hist_now) / mean_abs if mean_abs > 0 else 1.0
|
||
else:
|
||
hist_pct = 1.0
|
||
|
||
if crossed_up or accel_up:
|
||
signal = SignalStrength.STRONG_BUY if accel_up else SignalStrength.BUY
|
||
conf = min(80, 55 + hist_pct * 10)
|
||
desc = "MACD histogram" + (" accelerating ↑" if accel_up else " crossed above zero")
|
||
elif crossed_down or accel_down:
|
||
signal = SignalStrength.STRONG_SELL if accel_down else SignalStrength.SELL
|
||
conf = min(80, 55 + hist_pct * 10)
|
||
desc = "MACD histogram" + (" accelerating ↓" if accel_down else " crossed below zero")
|
||
elif fading_up:
|
||
signal, conf, desc = SignalStrength.NEUTRAL, 35, "MACD histogram positive but fading"
|
||
elif fading_down:
|
||
signal, conf, desc = SignalStrength.NEUTRAL, 35, "MACD histogram negative but fading"
|
||
else:
|
||
signal, conf, desc = SignalStrength.NEUTRAL, 30, "MACD histogram flat"
|
||
|
||
result.indicators["macd"] = IndicatorSignal(
|
||
name="MACD", value=hist_now, signal=signal,
|
||
confidence=conf, description=desc, weight=1.5,
|
||
)
|
||
|
||
# ── Bollinger override ────────────────────────────────────────
|
||
|
||
def _compute_bollinger(self, df: pd.DataFrame, result: TechnicalAnalysis):
|
||
cache, pos = self._resolve_pos(df)
|
||
if cache is None or pos < 0:
|
||
return super()._compute_bollinger(df, result)
|
||
|
||
c = self.config
|
||
latest = float(df["Close"].iloc[-1])
|
||
upper_val = float(cache.bb_upper[pos])
|
||
lower_val = float(cache.bb_lower[pos])
|
||
sma_val = float(cache.bb_mid[pos])
|
||
if np.isnan(upper_val) or np.isnan(lower_val) or np.isnan(sma_val) or sma_val == 0:
|
||
return
|
||
|
||
bb_range = upper_val - lower_val
|
||
bb_position = (latest - lower_val) / bb_range if bb_range != 0 else 0.5
|
||
band_width = float(cache.bb_band_width[pos])
|
||
|
||
is_squeeze = False
|
||
if pos >= 19:
|
||
avg_width = float(cache.bb_avg_width[pos])
|
||
if not np.isnan(avg_width) and avg_width > 0:
|
||
is_squeeze = band_width < avg_width * 0.75
|
||
|
||
squeeze_note = " | SQUEEZE detected (expect breakout)" if is_squeeze else ""
|
||
if bb_position > 1.0:
|
||
signal, conf = SignalStrength.SELL, 65
|
||
desc = f"Price above upper BB ({bb_position:.2f}){squeeze_note}"
|
||
elif bb_position < 0.0:
|
||
signal, conf = SignalStrength.BUY, 65
|
||
desc = f"Price below lower BB ({bb_position:.2f}){squeeze_note}"
|
||
elif bb_position > 0.8:
|
||
signal, conf = SignalStrength.SELL, 45
|
||
desc = f"Price near upper BB ({bb_position:.2f}){squeeze_note}"
|
||
elif bb_position < 0.2:
|
||
signal, conf = SignalStrength.BUY, 45
|
||
desc = f"Price near lower BB ({bb_position:.2f}){squeeze_note}"
|
||
else:
|
||
signal, conf = SignalStrength.NEUTRAL, 30
|
||
desc = f"Price mid-band ({bb_position:.2f}){squeeze_note}"
|
||
|
||
if is_squeeze:
|
||
conf = min(conf + 10, 85)
|
||
|
||
result.indicators["bollinger"] = IndicatorSignal(
|
||
name="Bollinger Bands", value=bb_position, signal=signal,
|
||
confidence=conf, description=desc, weight=1.5,
|
||
)
|
||
result.key_levels["bb_upper"] = upper_val
|
||
result.key_levels["bb_lower"] = lower_val
|
||
result.key_levels["bb_mid"] = sma_val
|
||
|
||
# ── ADX override ──────────────────────────────────────────────
|
||
|
||
def _compute_adx(self, df: pd.DataFrame, result: TechnicalAnalysis):
|
||
cache, pos = self._resolve_pos(df)
|
||
if cache is None or pos < 0:
|
||
return super()._compute_adx(df, result)
|
||
|
||
c = self.config
|
||
adx_val = float(cache.adx[pos])
|
||
if np.isnan(adx_val):
|
||
adx_val = 0.0
|
||
plus_di_val = float(cache.plus_di[pos])
|
||
minus_di_val = float(cache.minus_di[pos])
|
||
if np.isnan(plus_di_val):
|
||
plus_di_val = 0.0
|
||
if np.isnan(minus_di_val):
|
||
minus_di_val = 0.0
|
||
|
||
if adx_val >= c.adx_very_strong:
|
||
trend_strength, conf = "Very strong", 80
|
||
elif adx_val >= c.adx_strong_trend:
|
||
trend_strength, conf = "Strong", 65
|
||
else:
|
||
trend_strength, conf = "Weak/Ranging", 40
|
||
|
||
if plus_di_val > minus_di_val:
|
||
signal = SignalStrength.BUY if adx_val >= c.adx_strong_trend else SignalStrength.NEUTRAL
|
||
direction = "bullish"
|
||
else:
|
||
signal = SignalStrength.SELL if adx_val >= c.adx_strong_trend else SignalStrength.NEUTRAL
|
||
direction = "bearish"
|
||
|
||
adx_slope = float(cache.adx_slope[pos]) if not np.isnan(cache.adx_slope[pos]) else 0.0
|
||
|
||
slope_desc = ""
|
||
emerging_threshold = c.adx_slope_emerging_threshold
|
||
if adx_val < c.adx_strong_trend and adx_slope > emerging_threshold:
|
||
slope_desc = " | TREND EMERGING (ADX rising fast)"
|
||
conf = min(conf + 15, 85)
|
||
elif adx_val >= c.adx_strong_trend and adx_slope > 0:
|
||
slope_desc = f" | Trend strengthening (slope +{adx_slope:.1f})"
|
||
conf = min(conf + 5, 85)
|
||
elif adx_val >= c.adx_strong_trend and adx_slope < -emerging_threshold:
|
||
slope_desc = f" | Trend weakening (slope {adx_slope:.1f})"
|
||
conf = max(conf - 10, 30)
|
||
|
||
desc = f"ADX {adx_val:.1f} ({trend_strength} {direction}) | +DI={plus_di_val:.1f} -DI={minus_di_val:.1f}{slope_desc}"
|
||
result.indicators["adx"] = IndicatorSignal(
|
||
name="ADX", value=adx_val, signal=signal, confidence=conf,
|
||
description=desc, weight=0.5,
|
||
)
|
||
result._adx_slope = adx_slope
|
||
|
||
# ── ATR override ──────────────────────────────────────────────
|
||
|
||
def _compute_atr(self, df: pd.DataFrame, result: TechnicalAnalysis):
|
||
cache, pos = self._resolve_pos(df)
|
||
if cache is None or pos < 0:
|
||
return super()._compute_atr(df, result)
|
||
|
||
latest_close = float(df["Close"].iloc[-1])
|
||
atr_val = float(cache.atr[pos])
|
||
if np.isnan(atr_val):
|
||
atr_val = 0.0
|
||
atr_pct = (atr_val / latest_close) * 100 if latest_close != 0 else 0
|
||
|
||
# Volatility regime: replicate live behavior on the 50-bar window.
|
||
if pos >= 49:
|
||
window = cache.atr[pos - 49: pos + 1]
|
||
atr_percentile = float(np.mean(window < atr_val)) * 100 # bars strictly below current
|
||
# Live code: (atr.iloc[-1] > atr.iloc[-50:]).mean() * 100 → fraction of window strictly less than current
|
||
else:
|
||
atr_percentile = 50.0
|
||
|
||
if atr_percentile > 80:
|
||
vol_regime = "extreme"
|
||
elif atr_percentile > 60:
|
||
vol_regime = "high"
|
||
elif atr_percentile > 30:
|
||
vol_regime = "normal"
|
||
else:
|
||
vol_regime = "low"
|
||
|
||
result.volatility_regime = vol_regime
|
||
result.key_levels["atr"] = atr_val
|
||
result.key_levels["atr_pct"] = atr_pct
|
||
result.indicators["atr"] = IndicatorSignal(
|
||
name="ATR", value=atr_val, signal=SignalStrength.NEUTRAL,
|
||
confidence=50,
|
||
description=f"ATR {atr_val:.5f} ({atr_pct:.3f}%) | Volatility: {vol_regime} (P{atr_percentile:.0f})",
|
||
weight=0.0,
|
||
)
|
||
|
||
|
||
def _compute_roc(self, df: pd.DataFrame, result: TechnicalAnalysis):
|
||
cache, pos = self._resolve_pos(df)
|
||
if cache is None or pos < 0:
|
||
return super()._compute_roc(df, result)
|
||
|
||
c = self.config
|
||
n = c.roc_period
|
||
|
||
if pos < n + 1:
|
||
return
|
||
|
||
roc_val = float(cache.roc[pos])
|
||
roc_prev = float(cache.roc[pos - 1]) if pos > 0 else np.nan
|
||
|
||
if np.isnan(roc_val) or np.isnan(roc_prev):
|
||
return
|
||
|
||
roc_rising = roc_val > roc_prev
|
||
roc_abs = abs(roc_val)
|
||
threshold = c.roc_threshold
|
||
|
||
if roc_val > 0 and roc_rising:
|
||
if roc_abs > threshold * 2:
|
||
signal = SignalStrength.STRONG_BUY
|
||
conf = min(80, 60 + roc_abs * 5)
|
||
desc = f"ROC +{roc_val:.2f}% — Strong momentum acceleration"
|
||
else:
|
||
signal = SignalStrength.BUY
|
||
conf = min(70, 50 + roc_abs * 10)
|
||
desc = f"ROC +{roc_val:.2f}% — Rising momentum"
|
||
elif roc_val < 0 and not roc_rising:
|
||
if roc_abs > threshold * 2:
|
||
signal = SignalStrength.STRONG_SELL
|
||
conf = min(80, 60 + roc_abs * 5)
|
||
desc = f"ROC {roc_val:.2f}% — Strong bearish momentum"
|
||
else:
|
||
signal = SignalStrength.SELL
|
||
conf = min(70, 50 + roc_abs * 10)
|
||
desc = f"ROC {roc_val:.2f}% — Falling momentum"
|
||
elif roc_prev <= 0 < roc_val:
|
||
signal = SignalStrength.BUY
|
||
conf = 65
|
||
desc = f"ROC crossed above zero (+{roc_val:.2f}%)"
|
||
elif roc_prev >= 0 > roc_val:
|
||
signal = SignalStrength.SELL
|
||
conf = 65
|
||
desc = f"ROC crossed below zero ({roc_val:.2f}%)"
|
||
else:
|
||
signal = SignalStrength.NEUTRAL
|
||
conf = 30
|
||
desc = f"ROC {roc_val:+.2f}% — No clear momentum"
|
||
|
||
result.indicators["roc"] = IndicatorSignal(
|
||
name="ROC", value=roc_val, signal=signal,
|
||
confidence=conf, description=desc, weight=1.5,
|
||
)
|
||
|
||
def _compute_keltner(self, df: pd.DataFrame, result: TechnicalAnalysis):
|
||
cache, pos = self._resolve_pos(df)
|
||
if cache is None or pos < 0:
|
||
return super()._compute_keltner(df, result)
|
||
|
||
c = self.config
|
||
latest = float(df["Close"].iloc[-1])
|
||
kc_upper_val = float(cache.kc_upper[pos])
|
||
kc_lower_val = float(cache.kc_lower[pos])
|
||
kc_mid_val = float(cache.kc_mid[pos])
|
||
|
||
if np.isnan(kc_upper_val) or np.isnan(kc_lower_val):
|
||
return
|
||
|
||
is_squeeze = bool(cache.kc_squeeze[pos])
|
||
squeeze_released = False
|
||
if pos >= 1:
|
||
was_squeeze = bool(cache.kc_squeeze[pos - 1])
|
||
squeeze_released = was_squeeze and not is_squeeze
|
||
|
||
macd_direction = 0
|
||
if not np.isnan(cache.macd_hist[pos]):
|
||
macd_direction = 1 if cache.macd_hist[pos] > 0 else -1
|
||
|
||
if squeeze_released:
|
||
if macd_direction > 0:
|
||
signal = SignalStrength.STRONG_BUY
|
||
desc = "Keltner SQUEEZE RELEASED — bullish breakout"
|
||
elif macd_direction < 0:
|
||
signal = SignalStrength.STRONG_SELL
|
||
desc = "Keltner SQUEEZE RELEASED — bearish breakout"
|
||
else:
|
||
signal = SignalStrength.NEUTRAL
|
||
desc = "Keltner squeeze released — direction unclear"
|
||
conf = 80
|
||
elif is_squeeze:
|
||
signal = SignalStrength.NEUTRAL
|
||
conf = 60
|
||
desc = "Keltner SQUEEZE active — breakout imminent"
|
||
elif latest > kc_upper_val:
|
||
signal = SignalStrength.BUY
|
||
conf = 65
|
||
desc = "Price above Keltner upper — strong uptrend continuation"
|
||
elif latest < kc_lower_val:
|
||
signal = SignalStrength.SELL
|
||
conf = 65
|
||
desc = "Price below Keltner lower — strong downtrend continuation"
|
||
else:
|
||
signal = SignalStrength.NEUTRAL
|
||
conf = 30
|
||
desc = "Price within Keltner Channels"
|
||
|
||
result.indicators["keltner"] = IndicatorSignal(
|
||
name="Keltner", value=float(latest - kc_mid_val),
|
||
signal=signal, confidence=conf,
|
||
description=desc, weight=1.5,
|
||
)
|
||
result.key_levels["kc_upper"] = kc_upper_val
|
||
result.key_levels["kc_lower"] = kc_lower_val
|
||
result.key_levels["kc_mid"] = kc_mid_val
|
||
result._kc_squeeze = is_squeeze
|
||
result._kc_squeeze_released = squeeze_released
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# DATA LOADING (API + parquet cache)
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
CACHE_DIR = Path(__file__).resolve().parent / "cache"
|
||
|
||
|
||
def _cache_path(symbol: str, tf: Timeframe, start: datetime, end: datetime) -> Path:
|
||
safe_sym = symbol.replace("/", "_").replace("=X", "")
|
||
return CACHE_DIR / f"{safe_sym}_{tf.value}_{start:%Y%m%d}_{end:%Y%m%d}.parquet"
|
||
|
||
|
||
def fetch_history(
|
||
provider: ForexDataProvider,
|
||
symbol: str,
|
||
timeframe: Timeframe,
|
||
start: datetime,
|
||
end: datetime,
|
||
use_cache: bool = True,
|
||
clean: bool = True,
|
||
) -> pd.DataFrame:
|
||
"""Fetch OHLCV between start/end with parquet caching and pagination.
|
||
|
||
Twelve Data caps responses at 5000 bars; we paginate by date range so
|
||
multi-month M5/M15 backtests work.
|
||
|
||
The parquet cache stores the RAW series; `_validate_and_clean` runs on
|
||
load. Pass clean=False to get the raw price path — trade resolution must
|
||
use it, because cleaning drops >10σ spike bars (news candles) that hit
|
||
stops in reality. (Caches written before this change already contain
|
||
cleaned data; a refetch is needed to restore dropped spike bars.)
|
||
"""
|
||
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||
path = _cache_path(symbol, timeframe, start, end)
|
||
if use_cache and path.exists():
|
||
logger.info(f"Loading cached {symbol} {timeframe.value}: {path.name}")
|
||
df = pd.read_parquet(path)
|
||
if df.index.tz is None:
|
||
df.index = df.index.tz_localize("UTC")
|
||
return provider._validate_and_clean(df, symbol, timeframe) if clean else df
|
||
|
||
logger.info(f"Fetching {symbol} {timeframe.value} from {start.date()} to {end.date()}")
|
||
|
||
# Bars-per-call estimate to size each window safely under the 5000 cap.
|
||
bars_per_day = ForexDataProvider.BARS_PER_DAY[timeframe]
|
||
days_per_chunk = max(1, int(4500 / bars_per_day))
|
||
|
||
chunks: List[pd.DataFrame] = []
|
||
cursor_end = end
|
||
empty_streak = 0
|
||
while cursor_end > start:
|
||
cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk))
|
||
try:
|
||
chunk = provider._fetch_twelve_data(
|
||
symbol=symbol,
|
||
timeframe=timeframe,
|
||
outputsize=5000,
|
||
start_date=cursor_start.strftime("%Y-%m-%d %H:%M:%S"),
|
||
end_date=cursor_end.strftime("%Y-%m-%d %H:%M:%S"),
|
||
)
|
||
except Exception as e:
|
||
# Twelve Data raises BadRequest for windows with no data (weekends,
|
||
# holidays, asset gaps). Skip the window, step back, retry — but
|
||
# bail out after a few consecutive empties to avoid spinning.
|
||
logger.warning(
|
||
f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}"
|
||
)
|
||
empty_streak += 1
|
||
if empty_streak >= 3:
|
||
break
|
||
cursor_end = cursor_start - timedelta(seconds=1)
|
||
continue
|
||
|
||
if chunk is None or chunk.empty:
|
||
empty_streak += 1
|
||
if empty_streak >= 3:
|
||
break
|
||
cursor_end = cursor_start - timedelta(seconds=1)
|
||
continue
|
||
|
||
empty_streak = 0
|
||
chunks.append(chunk)
|
||
# Step back; dedup on concat.
|
||
cursor_end = chunk.index[0] - timedelta(seconds=1)
|
||
if len(chunk) < 2:
|
||
break
|
||
|
||
if not chunks:
|
||
raise RuntimeError(f"No data fetched for {symbol} {timeframe.value}")
|
||
|
||
df = pd.concat(chunks).sort_index()
|
||
df = df[~df.index.duplicated(keep="last")]
|
||
# Fetches ending "now" include the still-forming candle — drop it so the
|
||
# cache only ever contains closed bars.
|
||
df = provider._drop_forming_candle(df, timeframe)
|
||
# Trim to requested window. `start`/`end` are already tz-aware UTC
|
||
# datetimes — wrapping with pd.Timestamp(..., tz=...) would raise on
|
||
# already-localized inputs.
|
||
end_ts = pd.Timestamp(end)
|
||
start_ts = pd.Timestamp(start)
|
||
df = df[(df.index >= start_ts) & (df.index <= end_ts)]
|
||
|
||
if use_cache:
|
||
df.to_parquet(path)
|
||
logger.info(f"Cached {len(df)} raw bars to {path.name}")
|
||
return provider._validate_and_clean(df, symbol, timeframe) if clean else df
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# TRADE SIMULATION (vectorized SL/TP detection)
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
@dataclass
|
||
class Trade:
|
||
"""One backtested trade — entry through exit + full reasoning."""
|
||
pair: str
|
||
entry_time: pd.Timestamp
|
||
entry_price: float
|
||
sl: float
|
||
tp1: float
|
||
tp2: float
|
||
direction: str # "BUY" or "SELL"
|
||
pip_value: float
|
||
risk_pips: float
|
||
reward_pips: float
|
||
rr_planned: float
|
||
quality: str # signal grade at entry
|
||
final_score: float
|
||
final_confidence: float
|
||
market_regime: str
|
||
volatility_regime: str = ""
|
||
strategy_tag: str = "" # "XAU", "XAU_PB", "GENERIC"
|
||
# Per-trade trailing overrides (0 = use the run-level params).
|
||
trail_arm_r: float = 0.0
|
||
trail_dist_r: float = 0.0
|
||
|
||
# Per-instructions: entry reasoning and indicator breakdown ─────
|
||
entry_reasons: List[str] = field(default_factory=list)
|
||
entry_warnings: List[str] = field(default_factory=list)
|
||
# name -> {value, signal_name, confidence, weight, agrees_with_dir}
|
||
indicator_breakdown: Dict[str, Dict] = field(default_factory=dict)
|
||
indicators_agreeing: int = 0 # count agreeing with trade direction
|
||
indicators_opposing: int = 0 # count opposing
|
||
indicators_neutral: int = 0
|
||
indicators_total: int = 0 # total scored indicators
|
||
|
||
# Filled on exit:
|
||
exit_time: Optional[pd.Timestamp] = None
|
||
exit_price: Optional[float] = None
|
||
exit_reason: str = "" # "TP1", "SL", "TIMEOUT"
|
||
bars_held: int = 0
|
||
r_multiple: float = 0.0 # realized R (negative if loss)
|
||
mae_r: float = 0.0 # max adverse excursion in R
|
||
mfe_r: float = 0.0 # max favorable excursion in R
|
||
|
||
|
||
def _build_indicator_breakdown(
|
||
report: SignalReport, direction: str
|
||
) -> Tuple[Dict[str, Dict], int, int, int, int]:
|
||
"""Build per-indicator breakdown and tally direction agreement.
|
||
|
||
For each scored indicator on the primary timeframe:
|
||
- record value/signal/confidence/weight
|
||
- flag whether its signal agrees with the trade direction
|
||
"""
|
||
breakdown: Dict[str, Dict] = {}
|
||
agree = oppose = neutral = total = 0
|
||
if report.primary_analysis is None:
|
||
return breakdown, 0, 0, 0, 0
|
||
|
||
is_buy = direction == "BUY"
|
||
bull = (SignalStrength.BUY, SignalStrength.STRONG_BUY)
|
||
bear = (SignalStrength.SELL, SignalStrength.STRONG_SELL)
|
||
|
||
for name, ind in report.primary_analysis.indicators.items():
|
||
if name.startswith("_") or ind.weight == 0:
|
||
continue # skip private series + context-only indicators (atr)
|
||
agrees = "neutral"
|
||
if ind.signal in bull:
|
||
agrees = "agree" if is_buy else "oppose"
|
||
elif ind.signal in bear:
|
||
agrees = "agree" if not is_buy else "oppose"
|
||
|
||
breakdown[name] = {
|
||
"value": float(ind.value) if ind.value is not None else None,
|
||
"signal": ind.signal.name,
|
||
"confidence": float(ind.confidence),
|
||
"weight": float(ind.weight),
|
||
"agrees": agrees,
|
||
"description": ind.description,
|
||
}
|
||
total += 1
|
||
if agrees == "agree":
|
||
agree += 1
|
||
elif agrees == "oppose":
|
||
oppose += 1
|
||
else:
|
||
neutral += 1
|
||
return breakdown, agree, oppose, neutral, total
|
||
|
||
|
||
def _simulate_trailing(
|
||
trades: List[Trade],
|
||
opens_arr: np.ndarray,
|
||
highs_arr: np.ndarray,
|
||
lows_arr: np.ndarray,
|
||
closes_arr: np.ndarray,
|
||
valid_lens: np.ndarray,
|
||
exit_times: List[List[pd.Timestamp]],
|
||
spread: float,
|
||
trail_arm_r: float,
|
||
trail_dist_r: float,
|
||
) -> None:
|
||
"""Resolve trades with a trailing stop (per-trade CPU loop).
|
||
|
||
Exit reasons: "SL" (original stop, full loss), "TRAIL" (trailed stop
|
||
filled at/above entry region), "TIMEOUT".
|
||
"""
|
||
for i, t in enumerate(trades):
|
||
n = int(valid_lens[i])
|
||
if n == 0:
|
||
t.exit_reason = "NO_DATA"
|
||
continue
|
||
|
||
o = opens_arr[i, :n]
|
||
h = highs_arr[i, :n]
|
||
l = lows_arr[i, :n]
|
||
c = closes_arr[i, :n]
|
||
buy = t.direction == "BUY"
|
||
risk = abs(t.entry_price - t.sl)
|
||
if risk <= 0:
|
||
t.exit_reason = "NO_DATA"
|
||
continue
|
||
|
||
arm = t.trail_arm_r if t.trail_arm_r > 0 else trail_arm_r
|
||
dist = t.trail_dist_r if t.trail_dist_r > 0 else trail_dist_r
|
||
|
||
cur_sl = t.sl
|
||
peak = t.entry_price
|
||
exit_price = None
|
||
exit_idx = n - 1
|
||
reason = "TIMEOUT"
|
||
for j in range(n):
|
||
if np.isnan(h[j]):
|
||
break
|
||
hit = (l[j] <= cur_sl) if buy else (h[j] >= cur_sl)
|
||
if hit:
|
||
exit_price = cur_sl
|
||
# Gap through the stop → filled at the open.
|
||
if (buy and o[j] < cur_sl) or (not buy and o[j] > cur_sl):
|
||
exit_price = o[j]
|
||
reason = "SL" if cur_sl == t.sl else "TRAIL"
|
||
exit_idx = j
|
||
break
|
||
# Bar closed: advance peak, then ratchet the stop for the
|
||
# NEXT bar.
|
||
fav = h[j] if buy else l[j]
|
||
peak = max(peak, fav) if buy else min(peak, fav)
|
||
exc = ((peak - t.entry_price) if buy
|
||
else (t.entry_price - peak)) / risk
|
||
if exc >= arm:
|
||
lvl = (peak - dist * risk if buy
|
||
else peak + dist * risk)
|
||
cur_sl = max(cur_sl, lvl) if buy else min(cur_sl, lvl)
|
||
|
||
if exit_price is None:
|
||
exit_price = float(c[exit_idx])
|
||
t.exit_reason = reason
|
||
t.bars_held = exit_idx + 1
|
||
t.exit_price = float(exit_price)
|
||
t.exit_time = exit_times[i][exit_idx]
|
||
|
||
pnl = ((t.exit_price - t.entry_price) if buy
|
||
else (t.entry_price - t.exit_price))
|
||
t.r_multiple = (pnl - spread) / risk
|
||
|
||
h_win = h[:t.bars_held]
|
||
l_win = l[:t.bars_held]
|
||
if buy:
|
||
mfe = (np.nanmax(h_win) - t.entry_price) / risk
|
||
mae = (t.entry_price - np.nanmin(l_win)) / risk
|
||
else:
|
||
mfe = (t.entry_price - np.nanmin(l_win)) / risk
|
||
mae = (np.nanmax(h_win) - t.entry_price) / risk
|
||
t.mfe_r = float(max(0.0, mfe))
|
||
t.mae_r = float(max(0.0, mae))
|
||
|
||
|
||
def simulate_trades(
|
||
trades: List[Trade],
|
||
sim_df: pd.DataFrame,
|
||
max_bars: int = 200,
|
||
spread: float = 0.0,
|
||
trail_arm_r: float = 0.0,
|
||
trail_dist_r: float = 0.75,
|
||
) -> None:
|
||
"""Walk forward bar-by-bar after each entry and resolve the exit.
|
||
|
||
Two exit modes:
|
||
- Static (trail_arm_r <= 0): resolve SL / TP1, vectorized on GPU when
|
||
CuPy is available.
|
||
- Trailing (trail_arm_r > 0): TP1 is advisory. Once favorable
|
||
excursion reaches trail_arm_r × risk, the stop trails
|
||
trail_dist_r × risk behind the best price. Stop levels update at
|
||
bar CLOSE (intra-bar sequence is unknowable), so a new stop can
|
||
only fill from the next bar on — conservative.
|
||
|
||
SL/TP-same-bar tie: SL assumed first (conservative).
|
||
Gap-through fills: a bar *opening* beyond SL/TP fills at the open (worse
|
||
than SL on gaps against, better than TP on gaps in favor).
|
||
`spread` is a round-trip cost in price units subtracted from every
|
||
trade's PnL before computing R.
|
||
Times out (no hit) → marked TIMEOUT, exit at the last simulated bar.
|
||
"""
|
||
if not trades:
|
||
return
|
||
|
||
# Build per-trade post-entry windows (clipped to max_bars).
|
||
opens_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
|
||
highs_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
|
||
lows_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
|
||
closes_arr = np.full((len(trades), max_bars), np.nan, dtype=np.float64)
|
||
valid_lens = np.zeros(len(trades), dtype=np.int64)
|
||
exit_times: List[List[pd.Timestamp]] = []
|
||
|
||
sim_index = sim_df.index
|
||
sim_open = sim_df["Open"].to_numpy(dtype=np.float64)
|
||
sim_high = sim_df["High"].to_numpy(dtype=np.float64)
|
||
sim_low = sim_df["Low"].to_numpy(dtype=np.float64)
|
||
sim_close = sim_df["Close"].to_numpy(dtype=np.float64)
|
||
|
||
for i, t in enumerate(trades):
|
||
# entry_time is an exact bar timestamp from sim_df (we choose the
|
||
# first bar AFTER the signal bar as entry — see backtester loop).
|
||
try:
|
||
entry_pos = sim_index.get_loc(t.entry_time)
|
||
except KeyError:
|
||
# Snap to next available bar (rare timezone/edge case).
|
||
entry_pos = sim_index.searchsorted(t.entry_time)
|
||
if entry_pos >= len(sim_index):
|
||
exit_times.append([])
|
||
continue
|
||
|
||
end_pos = min(entry_pos + max_bars, len(sim_index))
|
||
n = end_pos - entry_pos
|
||
valid_lens[i] = n
|
||
opens_arr[i, :n] = sim_open[entry_pos:end_pos]
|
||
highs_arr[i, :n] = sim_high[entry_pos:end_pos]
|
||
lows_arr[i, :n] = sim_low[entry_pos:end_pos]
|
||
closes_arr[i, :n] = sim_close[entry_pos:end_pos]
|
||
exit_times.append(list(sim_index[entry_pos:end_pos]))
|
||
|
||
if trail_arm_r > 0:
|
||
_simulate_trailing(
|
||
trades, opens_arr, highs_arr, lows_arr, closes_arr,
|
||
valid_lens, exit_times, spread, trail_arm_r, trail_dist_r,
|
||
)
|
||
return
|
||
|
||
# GPU vectorized: for each trade, scan its window for first SL/TP hit.
|
||
sls = np.array([t.sl for t in trades], dtype=np.float64)[:, None]
|
||
tps = np.array([t.tp1 for t in trades], dtype=np.float64)[:, None]
|
||
is_buy = np.array([t.direction == "BUY" for t in trades], dtype=bool)[:, None]
|
||
|
||
h_g = _to_gpu(highs_arr)
|
||
l_g = _to_gpu(lows_arr)
|
||
c_g = _to_gpu(closes_arr)
|
||
sls_g = _to_gpu(sls)
|
||
tps_g = _to_gpu(tps)
|
||
is_buy_g = _to_gpu(is_buy)
|
||
|
||
# Hit masks. For BUY: SL when low<=sl, TP when high>=tp.
|
||
# For SELL: SL when high>=sl, TP when low<=tp.
|
||
sl_hit_buy = l_g <= sls_g
|
||
tp_hit_buy = h_g >= tps_g
|
||
sl_hit_sell = h_g >= sls_g
|
||
tp_hit_sell = l_g <= tps_g
|
||
|
||
sl_hit = cp.where(is_buy_g, sl_hit_buy, sl_hit_sell)
|
||
tp_hit = cp.where(is_buy_g, tp_hit_buy, tp_hit_sell)
|
||
|
||
# Mask out NaN bars (no data past valid_lens).
|
||
valid_mask = ~cp.isnan(h_g)
|
||
sl_hit = sl_hit & valid_mask
|
||
tp_hit = tp_hit & valid_mask
|
||
|
||
# First-hit indices (or max_bars if never).
|
||
idx_template = cp.arange(max_bars)[None, :]
|
||
sl_idx = cp.where(sl_hit, idx_template, max_bars)
|
||
tp_idx = cp.where(tp_hit, idx_template, max_bars)
|
||
first_sl = cp.min(sl_idx, axis=1)
|
||
first_tp = cp.min(tp_idx, axis=1)
|
||
|
||
sl_first_arr = _to_cpu(first_sl)
|
||
tp_first_arr = _to_cpu(first_tp)
|
||
|
||
# Resolve each trade and compute realized MFE/MAE *up to exit bar*
|
||
# (a trade closed at bar 5 cannot have meaningful excursions at bar 12).
|
||
for i, t in enumerate(trades):
|
||
n_valid = valid_lens[i]
|
||
if n_valid == 0:
|
||
t.exit_reason = "NO_DATA"
|
||
continue
|
||
|
||
sl_hit_idx = int(sl_first_arr[i])
|
||
tp_hit_idx = int(tp_first_arr[i])
|
||
|
||
# SL wins ties (conservative same-bar handling).
|
||
if sl_hit_idx < n_valid and sl_hit_idx <= tp_hit_idx:
|
||
t.exit_reason = "SL"
|
||
t.bars_held = sl_hit_idx + 1
|
||
exit_price = t.sl
|
||
bar_open = opens_arr[i, sl_hit_idx]
|
||
# Gap through the stop: the bar opened beyond SL → filled at open.
|
||
if not np.isnan(bar_open):
|
||
if (t.direction == "BUY" and bar_open < t.sl) or \
|
||
(t.direction == "SELL" and bar_open > t.sl):
|
||
exit_price = bar_open
|
||
t.exit_price = exit_price
|
||
t.exit_time = exit_times[i][sl_hit_idx]
|
||
elif tp_hit_idx < n_valid:
|
||
t.exit_reason = "TP1"
|
||
t.bars_held = tp_hit_idx + 1
|
||
exit_price = t.tp1
|
||
bar_open = opens_arr[i, tp_hit_idx]
|
||
# Gap beyond the target in our favor → filled at open.
|
||
if not np.isnan(bar_open):
|
||
if (t.direction == "BUY" and bar_open > t.tp1) or \
|
||
(t.direction == "SELL" and bar_open < t.tp1):
|
||
exit_price = bar_open
|
||
t.exit_price = exit_price
|
||
t.exit_time = exit_times[i][tp_hit_idx]
|
||
else:
|
||
# Time out at last simulated bar.
|
||
t.exit_reason = "TIMEOUT"
|
||
t.bars_held = int(n_valid)
|
||
last_idx = int(n_valid) - 1
|
||
t.exit_price = float(closes_arr[i, last_idx])
|
||
t.exit_time = exit_times[i][last_idx]
|
||
|
||
# Realized R from actual fill prices, net of round-trip spread.
|
||
risk = abs(t.entry_price - t.sl)
|
||
if risk > 0 and t.exit_price is not None:
|
||
pnl = ((t.exit_price - t.entry_price) if t.direction == "BUY"
|
||
else (t.entry_price - t.exit_price))
|
||
t.r_multiple = (pnl - spread) / risk
|
||
|
||
# MFE / MAE measured only over bars [0 .. bars_held).
|
||
bars = t.bars_held
|
||
risk = abs(t.entry_price - t.sl)
|
||
if bars > 0 and risk > 0:
|
||
h_win = highs_arr[i, :bars]
|
||
l_win = lows_arr[i, :bars]
|
||
if t.direction == "BUY":
|
||
mfe = (np.nanmax(h_win) - t.entry_price) / risk
|
||
mae = (t.entry_price - np.nanmin(l_win)) / risk
|
||
else:
|
||
mfe = (t.entry_price - np.nanmin(l_win)) / risk
|
||
mae = (np.nanmax(h_win) - t.entry_price) / risk
|
||
t.mfe_r = float(max(0.0, mfe))
|
||
t.mae_r = float(max(0.0, mae))
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# BACKTEST RUNNER
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
@dataclass
|
||
class BacktestConfig:
|
||
pair: str
|
||
start: datetime
|
||
end: datetime
|
||
primary_tf: Timeframe = Timeframe.M15
|
||
confirmation_tfs: List[Timeframe] = field(
|
||
default_factory=lambda: [Timeframe.H1, Timeframe.H4]
|
||
)
|
||
warmup_bars: int = 250 # min bars before first signal (EMA200 warmup)
|
||
cooldown_bars: int = 5 # min bars between trades on same pair
|
||
max_hold_bars: int = 300 # cap simulation length per trade
|
||
# (300: trailing exits need room to run;
|
||
# 200→300 adds ~10R on 12mo M5)
|
||
use_cache: bool = True
|
||
quality_filter: List[str] = field(
|
||
default_factory=lambda: ["A+", "A", "B", "C"]
|
||
)
|
||
# Restrict to specific strategy tags (XAU / XAU_PB / GENERIC).
|
||
# Empty = take all. Use to A/B test whether GENERIC trades add or destroy edge.
|
||
strategies_only: List[str] = field(default_factory=list)
|
||
# Round-trip transaction cost in price units, charged on every trade
|
||
# (XAU/USD retail spread ≈ $0.30). 0 disables.
|
||
spread: float = 0.30
|
||
# Replay worker processes for the bar loop. 0 = auto (cpu_count - 2,
|
||
# capped at 12), 1 = sequential in-process. Signal evaluation per bar
|
||
# is pure (past data only), so it fans out across forked workers; the
|
||
# one piece of sequential state — the entry cooldown — is applied
|
||
# afterwards in a fast ordered pass. Results are identical to the
|
||
# sequential path by construction.
|
||
workers: int = 0
|
||
|
||
|
||
def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optional[str]:
|
||
"""Apply XAU weekend + allowed-weekday filter matching live `analyze_pair`.
|
||
|
||
Forex market closed: Friday 22:00 UTC -> Sunday 22:00 UTC (Sydney open).
|
||
Must match `_is_forex_weekend` in signal_aggregator.py exactly.
|
||
"""
|
||
xau = cfg.xau_strategy
|
||
if xau.weekend_filter:
|
||
wd = ts.weekday()
|
||
hour = ts.hour
|
||
if wd == 4 and hour >= 22:
|
||
return "WEEKEND_FILTER"
|
||
if wd == 5:
|
||
return "WEEKEND_FILTER"
|
||
if wd == 6 and hour < 22:
|
||
return "WEEKEND_FILTER"
|
||
if xau.allowed_weekdays and ts.weekday() not in xau.allowed_weekdays:
|
||
return "DAY_FILTER"
|
||
return None
|
||
|
||
|
||
# Shared context for replay workers. Populated in run_backtest before the
|
||
# fork; child processes inherit it copy-on-write, so nothing large is pickled.
|
||
_REPLAY_CTX: Dict = {}
|
||
|
||
|
||
def _replay_span(span: Tuple[int, int]):
|
||
"""Evaluate primary bars [start, stop) — everything except the cooldown.
|
||
|
||
Per-bar signal evaluation is pure: it sees only data up to the bar's
|
||
close, never prior trades. The single stateful rule (entry cooldown)
|
||
is applied by the caller in a sequential pass over the returned
|
||
candidates, so parallel and sequential runs produce identical books.
|
||
Must stay behaviorally identical to the original sequential loop.
|
||
"""
|
||
ctx = _REPLAY_CTX
|
||
primary_df = ctx["primary_df"]
|
||
dfs_full = ctx["dfs_full"]
|
||
timeframes = ctx["timeframes"]
|
||
aggregator = ctx["aggregator"]
|
||
platform_cfg = ctx["platform_cfg"]
|
||
bt_cfg = ctx["bt_cfg"]
|
||
primary_dur = ctx["primary_dur"]
|
||
|
||
skipped: Dict[str, int] = {}
|
||
candidates: List[Tuple[int, Trade]] = []
|
||
is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD"
|
||
|
||
for pos in range(span[0], span[1]):
|
||
ts = primary_df.index[pos]
|
||
# Decision time = close of the current primary bar.
|
||
decision_time = ts + primary_dur
|
||
|
||
# XAU calendar filter (matches live analyze_pair behavior).
|
||
if is_xau:
|
||
block = _is_xau_blocked_by_calendar(decision_time, platform_cfg)
|
||
if block:
|
||
skipped[block] = skipped.get(block, 0) + 1
|
||
continue
|
||
|
||
# Session window filter (matches the live session check in
|
||
# analyze_pair, which evaluates the signal bar's close time).
|
||
bt_filter = platform_cfg.backtest
|
||
if bt_filter.session_filter:
|
||
bar_hour = decision_time.hour
|
||
if not (bt_filter.session_start_hour <= bar_hour < bt_filter.session_end_hour):
|
||
skipped["SESSION_WINDOW"] = skipped.get("SESSION_WINDOW", 0) + 1
|
||
continue
|
||
_blocked = (platform_cfg.xau_strategy.blocked_hours_utc
|
||
if is_xau else bt_filter.blocked_hours_utc)
|
||
if bar_hour in _blocked:
|
||
skipped["SESSION_WINDOW"] = skipped.get("SESSION_WINDOW", 0) + 1
|
||
continue
|
||
|
||
# (Cooldown is applied by the caller — it is the only cross-bar state.)
|
||
|
||
# Build sliced dfs per TF — only bars fully closed by decision_time.
|
||
slice_dfs: Dict[Timeframe, pd.DataFrame] = {}
|
||
for tf in timeframes:
|
||
full = dfs_full[tf]
|
||
bar_dur = pd.Timedelta(seconds=BAR_SECONDS[tf])
|
||
cutoff = decision_time - bar_dur
|
||
cut = int(full.index.searchsorted(cutoff, side="right"))
|
||
if cut == 0:
|
||
continue
|
||
slice_dfs[tf] = full.iloc[:cut]
|
||
|
||
if bt_cfg.primary_tf not in slice_dfs:
|
||
continue
|
||
|
||
report: SignalReport = aggregator.analyze_from_dataframes(
|
||
symbol=bt_cfg.pair,
|
||
dfs=slice_dfs,
|
||
context=None,
|
||
use_news=False,
|
||
)
|
||
|
||
if report.trade_setup is None:
|
||
continue
|
||
if report.signal_quality not in bt_cfg.quality_filter:
|
||
continue
|
||
|
||
# Strategy tag (from reasons[0] which the aggregator pre-pends).
|
||
tag = "GENERIC"
|
||
first_bull = report.bullish_reasons[0] if report.bullish_reasons else ""
|
||
first_bear = report.bearish_reasons[0] if report.bearish_reasons else ""
|
||
if "XAU PULLBACK" in first_bull or "XAU PULLBACK" in first_bear:
|
||
tag = "XAU_PB"
|
||
elif "XAU CHART PATTERN" in first_bull or "XAU CHART PATTERN" in first_bear:
|
||
tag = "XAU_CP"
|
||
elif "XAU STRATEGY" in first_bull:
|
||
tag = "XAU"
|
||
|
||
if bt_cfg.strategies_only and tag not in bt_cfg.strategies_only:
|
||
continue
|
||
|
||
ts_setup: TradeSetup = report.trade_setup
|
||
# Entry at the OPEN of the next bar (realistic fill).
|
||
entry_pos = pos + 1
|
||
if entry_pos >= len(primary_df):
|
||
continue
|
||
entry_time = primary_df.index[entry_pos]
|
||
entry_price = float(primary_df["Open"].iloc[entry_pos])
|
||
|
||
breakdown, agree, oppose, neutral, total = _build_indicator_breakdown(
|
||
report, ts_setup.direction
|
||
)
|
||
reasons = (report.bullish_reasons if ts_setup.direction == "BUY"
|
||
else report.bearish_reasons)
|
||
|
||
# Adjust SL/TP relative to actual entry (preserve setup distances).
|
||
sl_distance = abs(ts_setup.entry_price - ts_setup.stop_loss)
|
||
tp1_distance = abs(ts_setup.take_profit_1 - ts_setup.entry_price)
|
||
tp2_distance = abs(ts_setup.take_profit_2 - ts_setup.entry_price)
|
||
if ts_setup.direction == "BUY":
|
||
adj_sl = entry_price - sl_distance
|
||
adj_tp1 = entry_price + tp1_distance
|
||
adj_tp2 = entry_price + tp2_distance
|
||
else:
|
||
adj_sl = entry_price + sl_distance
|
||
adj_tp1 = entry_price - tp1_distance
|
||
adj_tp2 = entry_price - tp2_distance
|
||
|
||
candidates.append((pos, Trade(
|
||
pair=bt_cfg.pair,
|
||
entry_time=entry_time,
|
||
entry_price=entry_price,
|
||
sl=adj_sl,
|
||
tp1=adj_tp1,
|
||
tp2=adj_tp2,
|
||
direction=ts_setup.direction,
|
||
pip_value=ts_setup.pip_value,
|
||
risk_pips=sl_distance / ts_setup.pip_value,
|
||
reward_pips=tp1_distance / ts_setup.pip_value,
|
||
rr_planned=ts_setup.risk_reward,
|
||
quality=report.signal_quality,
|
||
final_score=report.final_score,
|
||
final_confidence=report.final_confidence,
|
||
market_regime=report.market_regime,
|
||
volatility_regime=report.volatility_regime,
|
||
strategy_tag=tag,
|
||
entry_reasons=list(reasons),
|
||
entry_warnings=list(report.warnings),
|
||
indicator_breakdown=breakdown,
|
||
indicators_agreeing=agree,
|
||
indicators_opposing=oppose,
|
||
indicators_neutral=neutral,
|
||
indicators_total=total,
|
||
trail_arm_r=(platform_cfg.signals.pb_trail_arm_r
|
||
if tag == "XAU_PB" else 0.0),
|
||
trail_dist_r=(platform_cfg.signals.pb_trail_dist_r
|
||
if tag == "XAU_PB" else 0.0),
|
||
)))
|
||
|
||
return skipped, candidates
|
||
|
||
|
||
def _reanchor_entries_to_raw(
|
||
trades: List[Trade],
|
||
signal_ts_list: List[pd.Timestamp],
|
||
sim_df: pd.DataFrame,
|
||
) -> None:
|
||
"""Entry = the first RAW bar strictly after the signal bar.
|
||
|
||
The replay picks pos+1 in the CLEANED frame. When cleaning dropped a
|
||
>10σ spike bar right after the signal bar, that entry skipped the spike
|
||
entirely — a fill (and an adverse bar) the live trade would have taken.
|
||
SL/TP distances are preserved relative to the corrected entry price,
|
||
matching the next-open adjustment in the replay loop.
|
||
"""
|
||
sim_index = sim_df.index
|
||
sim_open = sim_df["Open"].to_numpy(dtype=np.float64)
|
||
moved = 0
|
||
for t, sig_ts in zip(trades, signal_ts_list):
|
||
raw_pos = int(sim_index.searchsorted(sig_ts, side="right"))
|
||
if raw_pos >= len(sim_index):
|
||
continue # no raw bar after signal — resolution reports NO_DATA
|
||
new_time = sim_index[raw_pos]
|
||
if new_time == t.entry_time:
|
||
continue
|
||
new_price = float(sim_open[raw_pos])
|
||
if np.isnan(new_price):
|
||
continue
|
||
sl_dist = abs(t.entry_price - t.sl)
|
||
tp1_dist = abs(t.tp1 - t.entry_price)
|
||
tp2_dist = abs(t.tp2 - t.entry_price)
|
||
if t.direction == "BUY":
|
||
t.sl = new_price - sl_dist
|
||
t.tp1 = new_price + tp1_dist
|
||
t.tp2 = new_price + tp2_dist
|
||
else:
|
||
t.sl = new_price + sl_dist
|
||
t.tp1 = new_price - tp1_dist
|
||
t.tp2 = new_price - tp2_dist
|
||
t.entry_time = new_time
|
||
t.entry_price = new_price
|
||
moved += 1
|
||
if moved:
|
||
logger.info(
|
||
f"Re-anchored {moved} entries to the raw price path "
|
||
f"(cleaned frame was missing their next bar — usually a >10σ spike)"
|
||
)
|
||
|
||
|
||
def run_backtest(
|
||
bt_cfg: BacktestConfig,
|
||
platform_cfg: Optional[PlatformConfig] = None,
|
||
) -> Tuple[List[Trade], Dict[Timeframe, pd.DataFrame]]:
|
||
"""Replay history and collect trades."""
|
||
if platform_cfg is None:
|
||
platform_cfg = PlatformConfig(
|
||
symbol=bt_cfg.pair,
|
||
display_name=PAIR_INFO[bt_cfg.pair]["display"],
|
||
pip_value=PAIR_INFO[bt_cfg.pair]["pip_value"],
|
||
primary_timeframe=bt_cfg.primary_tf,
|
||
confirmation_timeframes=list(bt_cfg.confirmation_tfs),
|
||
)
|
||
elif platform_cfg.primary_timeframe != bt_cfg.primary_tf:
|
||
# A silent mismatch yields 0 trades: the replayer walks
|
||
# bt_cfg.primary_tf bars while the aggregator looks for
|
||
# platform_cfg.primary_timeframe in the slices and never finds it.
|
||
raise ValueError(
|
||
f"primary_tf mismatch: BacktestConfig={bt_cfg.primary_tf.value} "
|
||
f"vs PlatformConfig={platform_cfg.primary_timeframe.value}"
|
||
)
|
||
|
||
provider = make_data_provider(platform_cfg)
|
||
|
||
# ── Load all required timeframes ──
|
||
timeframes = [bt_cfg.primary_tf] + list(bt_cfg.confirmation_tfs)
|
||
# Pad lookback so earliest TF has enough warmup for EMA200, ADX, etc.
|
||
pad_days = max(60, bt_cfg.warmup_bars // 4)
|
||
fetch_start = bt_cfg.start - timedelta(days=pad_days)
|
||
|
||
dfs_full: Dict[Timeframe, pd.DataFrame] = {}
|
||
for tf in timeframes:
|
||
# Slow TFs need a longer pad: 60 days of D1 is only ~44 bars — no
|
||
# EMA200, truncated swing/divergence windows — while live always has
|
||
# ≥210 bars (outputsize floor). Target data_provider's min_required
|
||
# (210 bars); ×7/5 converts trading days to calendar days. Intraday
|
||
# TFs stay on the base pad, so their books are unchanged.
|
||
bars_per_day = ForexDataProvider.BARS_PER_DAY[tf]
|
||
min_pad_days = int(np.ceil(210 / bars_per_day * 7 / 5)) + 5
|
||
tf_start = bt_cfg.start - timedelta(days=max(pad_days, min_pad_days))
|
||
df = fetch_history(
|
||
provider, bt_cfg.pair, tf,
|
||
start=tf_start, end=bt_cfg.end,
|
||
use_cache=bt_cfg.use_cache,
|
||
)
|
||
if df.empty:
|
||
raise RuntimeError(f"No data for {bt_cfg.pair} {tf.value}")
|
||
dfs_full[tf] = df
|
||
|
||
# ── Pre-compute indicators on each TF ──
|
||
t0 = time.perf_counter()
|
||
caches: Dict[str, IndicatorCache] = {
|
||
tf.value: precompute_indicators(dfs_full[tf], platform_cfg.technical, gpu_log=True)
|
||
for tf in timeframes
|
||
}
|
||
logger.info(f"Indicator pre-compute took {time.perf_counter() - t0:.2f}s")
|
||
|
||
# ── Inject GPU engine into a fresh aggregator ──
|
||
aggregator = SignalAggregator(platform_cfg)
|
||
aggregator.tech_engine = GPUTechnicalEngine(platform_cfg.technical, caches)
|
||
|
||
# ── Walk primary TF bar-by-bar ──
|
||
primary_df = dfs_full[bt_cfg.primary_tf]
|
||
# bt_cfg.start/end are already tz-aware UTC datetimes; specifying tz=
|
||
# again would raise. Just wrap.
|
||
bt_start_ts = pd.Timestamp(bt_cfg.start)
|
||
bt_end_ts = pd.Timestamp(bt_cfg.end)
|
||
|
||
# Start at warmup_bars OR first bar inside the bt window (whichever later).
|
||
first_eligible_pos = max(
|
||
bt_cfg.warmup_bars,
|
||
int(primary_df.index.searchsorted(bt_start_ts)),
|
||
)
|
||
last_pos = int(primary_df.index.searchsorted(bt_end_ts, side="right"))
|
||
last_pos = min(last_pos, len(primary_df))
|
||
|
||
logger.info(
|
||
f"Replaying {bt_cfg.pair} {bt_cfg.primary_tf.value}: "
|
||
f"bars {first_eligible_pos}–{last_pos} of {len(primary_df)}"
|
||
)
|
||
|
||
primary_dur = pd.Timedelta(seconds=BAR_SECONDS[bt_cfg.primary_tf])
|
||
|
||
# ── Phase 1: evaluate every bar (parallelizable — no cross-bar state).
|
||
_REPLAY_CTX.update(
|
||
primary_df=primary_df, dfs_full=dfs_full, timeframes=timeframes,
|
||
aggregator=aggregator, platform_cfg=platform_cfg, bt_cfg=bt_cfg,
|
||
primary_dur=primary_dur,
|
||
)
|
||
n_bars = last_pos - first_eligible_pos
|
||
workers = (bt_cfg.workers if bt_cfg.workers > 0
|
||
else max(1, min((os.cpu_count() or 2) - 2, 12)))
|
||
if n_bars < 5_000:
|
||
workers = 1 # fork + merge overhead outweighs the gain
|
||
try:
|
||
fork_ctx = mp.get_context("fork") # COW-shares dfs/caches with workers
|
||
except ValueError:
|
||
workers = 1
|
||
fork_ctx = None
|
||
|
||
t_loop = time.perf_counter()
|
||
if workers > 1:
|
||
n_chunks = workers * 4
|
||
bounds = np.linspace(first_eligible_pos, last_pos, n_chunks + 1, dtype=int)
|
||
spans = [(int(bounds[i]), int(bounds[i + 1]))
|
||
for i in range(n_chunks) if bounds[i] < bounds[i + 1]]
|
||
logger.info(f"Replay: {workers} worker processes, {len(spans)} spans")
|
||
with fork_ctx.Pool(processes=workers) as pool:
|
||
results = pool.map(_replay_span, spans)
|
||
else:
|
||
results = [_replay_span((first_eligible_pos, last_pos))]
|
||
|
||
skipped_reasons: Dict[str, int] = {}
|
||
candidates: List[Tuple[int, Trade]] = []
|
||
for skips, cands in results:
|
||
for reason, count in skips.items():
|
||
skipped_reasons[reason] = skipped_reasons.get(reason, 0) + count
|
||
candidates.extend(cands)
|
||
candidates.sort(key=lambda pc: pc[0])
|
||
|
||
# ── Phase 2: entry cooldown — the only sequential rule. A candidate
|
||
# inside the cooldown window of the previously ACCEPTED entry is
|
||
# dropped, exactly as the sequential loop skipped those bars.
|
||
trades: List[Trade] = []
|
||
signal_ts_list: List[pd.Timestamp] = []
|
||
last_entry_pos = -10_000
|
||
for pos, trade in candidates:
|
||
if pos - last_entry_pos < bt_cfg.cooldown_bars:
|
||
continue
|
||
trades.append(trade)
|
||
signal_ts_list.append(primary_df.index[pos])
|
||
last_entry_pos = pos
|
||
|
||
logger.info(
|
||
f"Replay complete: {len(trades)} entries in "
|
||
f"{time.perf_counter() - t_loop:.1f}s "
|
||
f"({(last_pos - first_eligible_pos) / max(time.perf_counter() - t_loop, 1e-6):.0f} bars/s, "
|
||
f"{workers} worker{'s' if workers != 1 else ''})"
|
||
)
|
||
if skipped_reasons:
|
||
logger.info(f"Skipped bars by reason: {skipped_reasons}")
|
||
|
||
# ── Resolve trade outcomes ──
|
||
# Resolve against the RAW price path: cleaning removes >10σ spike bars
|
||
# (news candles) that hit stops in reality. Cache hit → no API call.
|
||
sim_df = fetch_history(
|
||
provider, bt_cfg.pair, bt_cfg.primary_tf,
|
||
start=fetch_start, end=bt_cfg.end,
|
||
use_cache=bt_cfg.use_cache, clean=False,
|
||
)
|
||
_reanchor_entries_to_raw(trades, signal_ts_list, sim_df)
|
||
t_sim = time.perf_counter()
|
||
simulate_trades(trades, sim_df, max_bars=bt_cfg.max_hold_bars,
|
||
spread=bt_cfg.spread,
|
||
trail_arm_r=platform_cfg.signals.trail_arm_r,
|
||
trail_dist_r=platform_cfg.signals.trail_dist_r)
|
||
logger.info(
|
||
f"Trade simulation took {time.perf_counter() - t_sim:.2f}s "
|
||
f"(GPU={'yes' if _GPU_AVAILABLE else 'no'})"
|
||
)
|
||
|
||
return trades, dfs_full
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# METRICS & REPORT
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
@dataclass
|
||
class Metrics:
|
||
n_trades: int
|
||
n_wins: int
|
||
n_losses: int
|
||
n_timeout: int
|
||
win_rate: float
|
||
avg_r: float
|
||
median_r: float
|
||
total_r: float
|
||
profit_factor: float
|
||
sharpe: float
|
||
max_drawdown_r: float
|
||
avg_bars_held: float
|
||
avg_mfe_r: float
|
||
avg_mae_r: float
|
||
by_quality: Dict[str, Dict]
|
||
by_strategy: Dict[str, Dict]
|
||
by_regime: Dict[str, Dict]
|
||
|
||
|
||
def _summarize(rs: List[float]) -> Dict:
|
||
if not rs:
|
||
return {"n": 0, "win_rate": 0.0, "avg_r": 0.0, "total_r": 0.0}
|
||
arr = np.array(rs)
|
||
wins = (arr > 0).sum()
|
||
return {
|
||
"n": len(rs),
|
||
"win_rate": float(wins / len(rs)) * 100,
|
||
"avg_r": float(arr.mean()),
|
||
"total_r": float(arr.sum()),
|
||
}
|
||
|
||
|
||
def compute_metrics(trades: List[Trade]) -> Metrics:
|
||
resolved = [t for t in trades if t.exit_reason and t.exit_reason != "NO_DATA"]
|
||
if not resolved:
|
||
return Metrics(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, {}, {}, {})
|
||
|
||
rs = np.array([t.r_multiple for t in resolved])
|
||
wins_mask = rs > 0
|
||
losses_mask = rs < 0
|
||
|
||
gross_profit = float(rs[wins_mask].sum()) if wins_mask.any() else 0.0
|
||
gross_loss = float(-rs[losses_mask].sum()) if losses_mask.any() else 0.0
|
||
pf = gross_profit / gross_loss if gross_loss > 0 else float("inf")
|
||
|
||
cum = np.cumsum(rs)
|
||
high_water = np.maximum.accumulate(cum)
|
||
drawdown = high_water - cum
|
||
max_dd = float(drawdown.max()) if len(drawdown) else 0.0
|
||
|
||
sharpe = float(rs.mean() / rs.std()) if rs.std() > 0 else 0.0
|
||
|
||
by_quality = {q: _summarize([t.r_multiple for t in resolved if t.quality == q])
|
||
for q in sorted({t.quality for t in resolved})}
|
||
by_strategy = {s: _summarize([t.r_multiple for t in resolved if t.strategy_tag == s])
|
||
for s in sorted({t.strategy_tag for t in resolved})}
|
||
by_regime = {r: _summarize([t.r_multiple for t in resolved if t.market_regime == r])
|
||
for r in sorted({t.market_regime for t in resolved})}
|
||
|
||
return Metrics(
|
||
n_trades=len(resolved),
|
||
n_wins=int(wins_mask.sum()),
|
||
n_losses=int(losses_mask.sum()),
|
||
n_timeout=sum(1 for t in resolved if t.exit_reason == "TIMEOUT"),
|
||
win_rate=float(wins_mask.mean()) * 100,
|
||
avg_r=float(rs.mean()),
|
||
median_r=float(np.median(rs)),
|
||
total_r=float(rs.sum()),
|
||
profit_factor=pf,
|
||
sharpe=sharpe,
|
||
max_drawdown_r=max_dd,
|
||
avg_bars_held=float(np.mean([t.bars_held for t in resolved])),
|
||
avg_mfe_r=float(np.mean([t.mfe_r for t in resolved])),
|
||
avg_mae_r=float(np.mean([t.mae_r for t in resolved])),
|
||
by_quality=by_quality,
|
||
by_strategy=by_strategy,
|
||
by_regime=by_regime,
|
||
)
|
||
|
||
|
||
_DAY_NAMES = {0: "Mon", 1: "Tue", 2: "Wed", 3: "Thu",
|
||
4: "Fri", 5: "Sat", 6: "Sun"}
|
||
|
||
|
||
def _fmt_table(headers: List[str], rows: List[List[str]]) -> str:
|
||
"""Render a Markdown-style ASCII table with auto-sized columns."""
|
||
if not rows:
|
||
return ""
|
||
cols = len(headers)
|
||
widths = [len(h) for h in headers]
|
||
for r in rows:
|
||
for i in range(cols):
|
||
widths[i] = max(widths[i], len(str(r[i])))
|
||
|
||
def fmt_row(r):
|
||
return "│ " + " │ ".join(str(r[i]).ljust(widths[i]) for i in range(cols)) + " │"
|
||
|
||
sep = "├" + "┼".join("─" * (w + 2) for w in widths) + "┤"
|
||
top = "┌" + "┬".join("─" * (w + 2) for w in widths) + "┐"
|
||
bot = "└" + "┴".join("─" * (w + 2) for w in widths) + "┘"
|
||
out = [top, fmt_row(headers), sep]
|
||
out.extend(fmt_row(r) for r in rows)
|
||
out.append(bot)
|
||
return "\n".join(out)
|
||
|
||
|
||
def print_report(
|
||
pair: str, bt_cfg: BacktestConfig, m: Metrics,
|
||
trades: Optional[List[Trade]] = None, detail: bool = False,
|
||
) -> None:
|
||
bar = "═" * 78
|
||
print(f"\n{bar}")
|
||
print(f" BACKTEST REPORT — {PAIR_INFO[pair]['display']} "
|
||
f"({bt_cfg.start.date()} → {bt_cfg.end.date()})")
|
||
print(f" Primary: {bt_cfg.primary_tf.value} "
|
||
f"Confirms: {','.join(t.value for t in bt_cfg.confirmation_tfs)}")
|
||
print(f"{bar}")
|
||
|
||
if m.n_trades == 0:
|
||
print(" No trades produced — quality filter rejected all signals.")
|
||
print(f"{bar}\n")
|
||
return
|
||
|
||
# ── Summary table ──
|
||
summary_rows = [
|
||
["Trades", f"{m.n_trades}"],
|
||
["Wins / Losses / Timeouts", f"{m.n_wins} / {m.n_losses} / {m.n_timeout}"],
|
||
["Win rate", f"{m.win_rate:.1f}%"],
|
||
["Total R", f"{m.total_r:+.2f}"],
|
||
["Avg R / trade", f"{m.avg_r:+.3f}"],
|
||
["Median R", f"{m.median_r:+.3f}"],
|
||
["Profit factor", f"{m.profit_factor:.2f}"],
|
||
["Sharpe (R)", f"{m.sharpe:.2f}"],
|
||
["Max DD (R)", f"{m.max_drawdown_r:.2f}"],
|
||
["Avg bars held", f"{m.avg_bars_held:.1f}"],
|
||
["Avg MFE / MAE", f"{m.avg_mfe_r:.2f}R / {m.avg_mae_r:.2f}R"],
|
||
]
|
||
print("\n ── Summary ──")
|
||
print(_fmt_table(["Metric", "Value"], summary_rows))
|
||
|
||
def _grouped_table(title: str, groups: Dict[str, Dict]):
|
||
if not groups:
|
||
return
|
||
rows = []
|
||
for k, s in groups.items():
|
||
rows.append([k, str(s["n"]),
|
||
f"{s['win_rate']:5.1f}%",
|
||
f"{s['avg_r']:+.3f}",
|
||
f"{s['total_r']:+.2f}"])
|
||
print(f"\n ── {title} ──")
|
||
print(_fmt_table([title.split(" by ")[-1], "n", "WR", "avgR", "totalR"], rows))
|
||
|
||
_grouped_table("Breakdown by Quality", m.by_quality)
|
||
_grouped_table("Breakdown by Strategy", m.by_strategy)
|
||
_grouped_table("Breakdown by Regime", m.by_regime)
|
||
|
||
# ── Per-trade table ──
|
||
if trades:
|
||
resolved = [t for t in trades if t.exit_reason and t.exit_reason != "NO_DATA"]
|
||
if detail or len(resolved) <= 30:
|
||
print("\n ── Trade Log ──")
|
||
rows = []
|
||
for i, t in enumerate(resolved, 1):
|
||
rows.append([
|
||
str(i),
|
||
t.entry_time.strftime("%m-%d %a %H:%M"),
|
||
t.direction,
|
||
t.strategy_tag,
|
||
t.quality,
|
||
f"{t.indicators_agreeing}/{t.indicators_total}",
|
||
f"{t.final_score:+.1f}",
|
||
f"{t.final_confidence:.0f}",
|
||
t.market_regime[:8],
|
||
t.exit_reason,
|
||
str(t.bars_held),
|
||
f"{t.r_multiple:+.2f}",
|
||
])
|
||
print(_fmt_table(
|
||
["#", "Entry (UTC)", "Dir", "Strat", "Q", "Agree",
|
||
"Score", "Conf", "Regime", "Exit", "Bars", "R"],
|
||
rows,
|
||
))
|
||
else:
|
||
print(f"\n Trade log suppressed ({len(resolved)} trades). "
|
||
f"Use --detail to print all, or open the CSV.")
|
||
|
||
print(f"{bar}\n")
|
||
|
||
|
||
def print_trade_reasoning(trades: List[Trade], limit: int = 5) -> None:
|
||
"""Per-instructions: dump the *why* for the first/last few trades —
|
||
indicators on each TF, agreement count, reasons, warnings."""
|
||
resolved = [t for t in trades if t.exit_reason and t.exit_reason != "NO_DATA"]
|
||
if not resolved:
|
||
return
|
||
bar = "─" * 78
|
||
print(f"\n{'═' * 78}")
|
||
print(f" TRADE REASONING (first {min(limit, len(resolved))} + last "
|
||
f"{min(limit, len(resolved))} of {len(resolved)})")
|
||
print('═' * 78)
|
||
sample = resolved[:limit] + ([] if len(resolved) <= 2 * limit else resolved[-limit:])
|
||
for t in sample:
|
||
print(f"\n{bar}")
|
||
weekday = _DAY_NAMES.get(t.entry_time.weekday(), "?")
|
||
print(f" {t.direction} {PAIR_INFO[t.pair]['display']} @ {t.entry_price:.5f} "
|
||
f"({weekday} {t.entry_time:%Y-%m-%d %H:%M UTC})")
|
||
print(f" Strategy={t.strategy_tag} Quality={t.quality} "
|
||
f"Score={t.final_score:+.1f} Confidence={t.final_confidence:.0f} "
|
||
f"Regime={t.market_regime}/{t.volatility_regime}")
|
||
print(f" SL={t.sl:.5f} TP1={t.tp1:.5f} RR={t.rr_planned:.2f} "
|
||
f"({t.risk_pips:.0f}pip risk → {t.reward_pips:.0f}pip target)")
|
||
print(f" Indicator agreement: {t.indicators_agreeing}/{t.indicators_total} agree, "
|
||
f"{t.indicators_opposing} oppose, {t.indicators_neutral} neutral")
|
||
if t.indicator_breakdown:
|
||
ind_rows = []
|
||
for name, b in t.indicator_breakdown.items():
|
||
tag = {"agree": "✓", "oppose": "✗", "neutral": "·"}.get(b["agrees"], "?")
|
||
ind_rows.append([
|
||
name, b["signal"],
|
||
f"{b['value']:.3f}" if b["value"] is not None else "—",
|
||
f"{b['confidence']:.0f}",
|
||
f"{b['weight']:.1f}",
|
||
tag,
|
||
b["description"][:40],
|
||
])
|
||
print(_fmt_table(
|
||
["Indicator", "Signal", "Value", "Conf", "Wt", "vs Dir", "Description"],
|
||
ind_rows,
|
||
))
|
||
if t.entry_reasons:
|
||
print(f" Entry reasons:")
|
||
for r in t.entry_reasons[:5]:
|
||
print(f" • {r[:120]}")
|
||
if t.entry_warnings:
|
||
print(f" Warnings at entry:")
|
||
for w in t.entry_warnings[:5]:
|
||
print(f" ⚠ {w[:120]}")
|
||
weekday_x = _DAY_NAMES.get(t.exit_time.weekday(), "?") if t.exit_time is not None else "?"
|
||
print(f" → EXIT: {t.exit_reason} @ {t.exit_price:.5f} after {t.bars_held} bars "
|
||
f"({weekday_x} {t.exit_time:%Y-%m-%d %H:%M UTC}) → R={t.r_multiple:+.2f}")
|
||
print(f" MFE / MAE: {t.mfe_r:.2f}R / {t.mae_r:.2f}R")
|
||
|
||
|
||
def trades_to_dataframe(trades: List[Trade]) -> pd.DataFrame:
|
||
"""Export trades for downstream analysis (e.g. parquet/CSV).
|
||
|
||
Includes per-instructions: entry reasoning, indicator-by-indicator
|
||
signal/value/agreement, and overall agreement count.
|
||
"""
|
||
rows = []
|
||
for t in trades:
|
||
row = {
|
||
"pair": t.pair,
|
||
"strategy": t.strategy_tag,
|
||
"quality": t.quality,
|
||
"regime": t.market_regime,
|
||
"vol_regime": t.volatility_regime,
|
||
"direction": t.direction,
|
||
"entry_time": t.entry_time,
|
||
"entry_weekday": _DAY_NAMES.get(t.entry_time.weekday(), "?"),
|
||
"entry": t.entry_price,
|
||
"sl": t.sl,
|
||
"tp1": t.tp1,
|
||
"rr_planned": t.rr_planned,
|
||
"score": t.final_score,
|
||
"confidence": t.final_confidence,
|
||
"indicators_agree": t.indicators_agreeing,
|
||
"indicators_oppose": t.indicators_opposing,
|
||
"indicators_neutral": t.indicators_neutral,
|
||
"indicators_total": t.indicators_total,
|
||
"entry_reasons": " || ".join(t.entry_reasons[:5]),
|
||
"warnings": " || ".join(t.entry_warnings[:5]),
|
||
"exit_time": t.exit_time,
|
||
"exit": t.exit_price,
|
||
"exit_reason": t.exit_reason,
|
||
"bars_held": t.bars_held,
|
||
"r_multiple": t.r_multiple,
|
||
"mfe_r": t.mfe_r,
|
||
"mae_r": t.mae_r,
|
||
}
|
||
# Flatten each indicator into columns so spreadsheets can sort/filter.
|
||
for name, b in t.indicator_breakdown.items():
|
||
row[f"ind_{name}_signal"] = b["signal"]
|
||
row[f"ind_{name}_value"] = b["value"]
|
||
row[f"ind_{name}_conf"] = b["confidence"]
|
||
row[f"ind_{name}_agrees"] = b["agrees"]
|
||
rows.append(row)
|
||
return pd.DataFrame(rows)
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# CLI
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
|
||
p = argparse.ArgumentParser(
|
||
description="GPU-accelerated backtester for ForexBot2",
|
||
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
||
)
|
||
p.add_argument("--pair", required=True, help="XAUUSD | GOLD")
|
||
p.add_argument("--days", type=int, default=180, help="Days of history (ignored if --start/--end set)")
|
||
p.add_argument("--start", help="ISO date YYYY-MM-DD (overrides --days)")
|
||
p.add_argument("--end", help="ISO date YYYY-MM-DD (defaults to today)")
|
||
p.add_argument("--primary-tf", default="15min", help="Primary TF (1min/5min/15min/1h/4h)")
|
||
p.add_argument("--confirm-tfs", default=None,
|
||
help="Comma-separated confirmation TFs. Default: the pair "
|
||
"factory's live map (M5→M15,1h | M15→5min,1h | "
|
||
"1h→15min,4h | 4h→1h,1day) so backtests replay the "
|
||
"exact live config.")
|
||
p.add_argument("--warmup-bars", type=int, default=250)
|
||
p.add_argument("--cooldown-bars", type=int, default=5)
|
||
p.add_argument("--max-hold-bars", type=int, default=300)
|
||
p.add_argument("--swing-sr-lookback", type=int, default=0,
|
||
help="Override TechnicalConfig.swing_sr_lookback (0 = use config default)")
|
||
p.add_argument("--quality", default="A+,A,B,C", help="Comma-separated allowed grades")
|
||
p.add_argument("--strategies", default="",
|
||
help="Comma-separated strategy tags to include (XAU,XAU_PB,XAU_CP,GENERIC). Empty = all.")
|
||
p.add_argument("--cp", action="store_true",
|
||
help="Force-enable the chart-pattern entry (XAU_CP) regardless "
|
||
"of the factory default — for validating the candidate.")
|
||
p.add_argument("--cp-patterns", default=None,
|
||
help="Comma-separated pattern subset for --cp (e.g. "
|
||
"double_bottom,bull_flag). Default: all detectors.")
|
||
p.add_argument("--no-cache", action="store_true", help="Skip parquet cache (force API fetch)")
|
||
p.add_argument("--spread", type=float, default=None,
|
||
help="Round-trip cost in price units per trade (default: pair's "
|
||
"default_spread from PAIR_INFO — XAU $0.30; 0 disables)")
|
||
p.add_argument("--export-csv", help="Optional path to dump per-trade CSV (incl. indicator breakdown)")
|
||
p.add_argument("--detail", action="store_true",
|
||
help="Print full per-trade reasoning (indicators, signals, agreement, why entered)")
|
||
p.add_argument("--workers", type=int, default=0,
|
||
help="Replay worker processes (0=auto: cpu-2 capped at 12, 1=sequential)")
|
||
p.add_argument("--reasoning-samples", type=int, default=3,
|
||
help="If --detail not set, print this many sample trades' full reasoning")
|
||
p.add_argument("--log-level", default="INFO")
|
||
return p.parse_args(argv)
|
||
|
||
|
||
def _resolve_pair(s: str) -> str:
|
||
s_up = s.upper().strip()
|
||
if s_up in PAIR_ALIASES:
|
||
return PAIR_ALIASES[s_up]
|
||
if s_up in PAIR_INFO:
|
||
return s_up
|
||
raise SystemExit(f"Unknown pair '{s}'. Try one of: {', '.join(sorted(PAIR_ALIASES))}")
|
||
|
||
|
||
def _resolve_tf(s: str) -> Timeframe:
|
||
for tf in Timeframe:
|
||
if tf.value == s:
|
||
return tf
|
||
raise SystemExit(f"Unknown timeframe '{s}'. Valid: {[t.value for t in Timeframe]}")
|
||
|
||
|
||
def main(argv: Optional[List[str]] = None) -> int:
|
||
args = _parse_args(argv)
|
||
|
||
logging.basicConfig(
|
||
level=args.log_level.upper(),
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
)
|
||
|
||
pair = _resolve_pair(args.pair)
|
||
primary_tf = _resolve_tf(args.primary_tf)
|
||
# None = defer to the pair factory's live confirmation map. A hardcoded
|
||
# default here would silently backtest a different MTF setup than the
|
||
# bot scans (e.g. M15 with 1h+4h instead of the live 5min+1h).
|
||
confirm_tfs = (
|
||
[_resolve_tf(s.strip()) for s in args.confirm_tfs.split(",") if s.strip()]
|
||
if args.confirm_tfs is not None else None
|
||
)
|
||
|
||
# Per-pair tuned configs come from the pair_signals factories (regime
|
||
# weights, strategy gates, sessions). Unregistered pairs fall back to
|
||
# the inline construction.
|
||
from pair_signals import PAIR_PRODUCERS # noqa: E402 (after sys.path fix)
|
||
factory = PAIR_PRODUCERS.get(pair)
|
||
if factory is not None:
|
||
platform_cfg = factory(primary_tf, confirm_tfs)
|
||
else:
|
||
platform_cfg = PlatformConfig(
|
||
symbol=pair,
|
||
display_name=PAIR_INFO[pair]["display"],
|
||
pip_value=PAIR_INFO[pair]["pip_value"],
|
||
primary_timeframe=primary_tf,
|
||
confirmation_timeframes=confirm_tfs or [Timeframe.H1, Timeframe.H4],
|
||
)
|
||
if args.swing_sr_lookback > 0:
|
||
platform_cfg.technical.swing_sr_lookback = args.swing_sr_lookback
|
||
if args.cp:
|
||
platform_cfg.xau_strategy.cp_enabled = True
|
||
if args.cp_patterns:
|
||
platform_cfg.xau_strategy.cp_patterns = [
|
||
s.strip() for s in args.cp_patterns.split(",") if s.strip()
|
||
]
|
||
confirm_tfs = list(platform_cfg.confirmation_timeframes)
|
||
|
||
end = (
|
||
datetime.fromisoformat(args.end).replace(tzinfo=timezone.utc)
|
||
if args.end else datetime.now(timezone.utc)
|
||
)
|
||
if args.start:
|
||
start = datetime.fromisoformat(args.start).replace(tzinfo=timezone.utc)
|
||
else:
|
||
start = end - timedelta(days=args.days)
|
||
|
||
spread = (args.spread if args.spread is not None
|
||
else PAIR_INFO[pair].get("default_spread", 0.30))
|
||
|
||
bt_cfg = BacktestConfig(
|
||
pair=pair,
|
||
start=start,
|
||
end=end,
|
||
primary_tf=primary_tf,
|
||
confirmation_tfs=confirm_tfs,
|
||
warmup_bars=args.warmup_bars,
|
||
cooldown_bars=args.cooldown_bars,
|
||
max_hold_bars=args.max_hold_bars,
|
||
use_cache=not args.no_cache,
|
||
quality_filter=[q.strip() for q in args.quality.split(",") if q.strip()],
|
||
strategies_only=[s.strip() for s in args.strategies.split(",") if s.strip()],
|
||
spread=spread,
|
||
workers=args.workers,
|
||
)
|
||
|
||
logger.info(
|
||
f"Backend: {'CuPy GPU' if _GPU_AVAILABLE else 'NumPy CPU fallback'} | "
|
||
f"Pair: {PAIR_INFO[pair]['display']} | Window: {start.date()} → {end.date()} | "
|
||
f"Confirm TFs: {','.join(tf.value for tf in confirm_tfs)} | Spread: {spread}"
|
||
)
|
||
|
||
trades, _ = run_backtest(bt_cfg, platform_cfg=platform_cfg)
|
||
metrics = compute_metrics(trades)
|
||
print_report(pair, bt_cfg, metrics, trades=trades, detail=args.detail)
|
||
print_trade_reasoning(trades, limit=args.reasoning_samples)
|
||
|
||
if args.export_csv:
|
||
df = trades_to_dataframe(trades)
|
||
df.to_csv(args.export_csv, index=False)
|
||
logger.info(f"Exported {len(df)} trades → {args.export_csv}")
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|