harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix live/backtest parity, SELL-side losses, and indicator repairs (+42% total R)

Browse Source 
Live pipeline:
- Drop still-forming candle before analysis (signals no longer repaint)
- Direction-aware MTF agreement + opposition gate (no trade against 2-of-3 TF consensus)
- SELL volatility gate: block shorts in low/normal vol regimes (config sell_vol_regimes)
- Divergence: 40-bar lookback, price/indicator swings must align within ±2 bars
  (fired on 0.5% of trades before, 29% after), trending weight 2.0→1.0
- RSI regime_aware by default; transitioning regime only signals at extremes
- ADX ±DM tie no longer nudges bearish; TP2 clamped beyond TP1

Backtester realism:
- Charge $0.30 round-trip spread (--spread), gap-through SL/TP fills at bar open
- Resolve trades on RAW price path (cache stores raw bars, cleaning runs on load)
- Weekend/session filters use bar-close decision time to match live wall clock

Validated per-step on M15 2025-03-30→2026-05-31 under identical pricing:
+46.9R → +66.8R, win rate 69.9% → 73.0%, SELL side -2.3R → +9.6R.
Tested-and-reverted: ROC weight removal (-13R), BB fade suppression (-11.6R),
candlestick de-weight (-2R) — composite thresholds are co-tuned with weight mass.
This commit is contained in:
harwinder
2026-07-08 21:26:30 -04:00
parent baeee4997a
commit 1f076c0b54
7 changed files with 273 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backtest/gpu_backtest.py
+126 -32
View File
@@ -780,11 +780,26 @@ class GPUTechnicalEngine(TechnicalEngine):
signal = SignalStrength.SELL if adx_val >= c.adx_strong_trend else SignalStrength.NEUTRAL
direction = "bearish"
desc = f"ADX {adx_val:.1f} ({trend_strength} {direction}) | +DI={plus_di_val:.1f} -DI={minus_di_val:.1f}"
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 ──────────────────────────────────────────────
@@ -967,11 +982,18 @@ def fetch_history(
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)
@@ -980,7 +1002,7 @@ def fetch_history(
df = pd.read_parquet(path)
if df.index.tz is None:
df.index = df.index.tz_localize("UTC")
return df
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()}")
@@ -1033,18 +1055,20 @@ def fetch_history(
df = pd.concat(chunks).sort_index()
df = df[~df.index.duplicated(keep="last")]
df = provider._validate_and_clean(df, symbol, timeframe)
# Trim to requested window after cleaning. `start`/`end` are already
# tz-aware UTC datetimes — wrapping with pd.Timestamp(..., tz=...) would
# raise on already-localized inputs.
# 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)} bars to {path.name}")
return df
logger.info(f"Cached {len(df)} raw bars to {path.name}")
return provider._validate_and_clean(df, symbol, timeframe) if clean else df
# ═════════════════════════════════════════════════════════════════════
@@ -1141,6 +1165,7 @@ def simulate_trades(
trades: List[Trade],
sim_df: pd.DataFrame,
max_bars: int = 200,
spread: float = 0.0,
) -> None:
"""Walk forward bar-by-bar after each entry and resolve SL / TP1.
@@ -1148,12 +1173,17 @@ def simulate_trades(
matrices of post-entry highs/lows and finds the first hit per trade.
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)
@@ -1161,6 +1191,7 @@ def simulate_trades(
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)
@@ -1180,6 +1211,7 @@ def simulate_trades(
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]
@@ -1237,14 +1269,26 @@ def simulate_trades(
if sl_hit_idx < n_valid and sl_hit_idx <= tp_hit_idx:
t.exit_reason = "SL"
t.bars_held = sl_hit_idx + 1
t.exit_price = t.sl
t.r_multiple = -1.0
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
t.exit_price = t.tp1
t.r_multiple = t.rr_planned
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.
@@ -1252,12 +1296,15 @@ def simulate_trades(
t.bars_held = int(n_valid)
last_idx = int(n_valid) - 1
t.exit_price = float(closes_arr[i, last_idx])
risk = abs(t.entry_price - t.sl)
if risk > 0:
pnl = (t.exit_price - t.entry_price) if t.direction == "BUY" else (t.entry_price - t.exit_price)
t.r_multiple = pnl / risk
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)
@@ -1297,14 +1344,27 @@ class BacktestConfig:
# Restrict to specific strategy tags (XAU / EUR_PB / GBP_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
def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optional[str]:
"""Apply XAU weekend + allowed-weekday filter (live `analyze_pair` does this;
`analyze_from_dataframes` does not, so we replicate here)."""
"""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 and ts.weekday() >= 5:
return "WEEKEND_FILTER"
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
@@ -1388,12 +1448,23 @@ def run_backtest(
decision_time = ts + primary_dur
# XAU calendar filter (matches live analyze_pair behavior).
# Live evaluates wall-clock time ≈ bar close, so filter on
# decision_time, not the bar's open timestamp.
if PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD":
block = _is_xau_blocked_by_calendar(ts, platform_cfg)
block = _is_xau_blocked_by_calendar(decision_time, platform_cfg)
if block:
skipped_reasons[block] = skipped_reasons.get(block, 0) + 1
continue
# Session window filter (matches live _apply_session_filter_live,
# which checks datetime.now() ≈ bar close = decision_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_reasons["SESSION_WINDOW"] = skipped_reasons.get("SESSION_WINDOW", 0) + 1
continue
# Cooldown.
if pos - last_entry_pos < bt_cfg.cooldown_bars:
continue
@@ -1443,14 +1514,13 @@ def run_backtest(
continue
ts_setup: TradeSetup = report.trade_setup
# Entry on the NEXT bar's open price would be ideal, but live signals fire
# at bar close and we use that close as entry (matches setup.entry_price).
# For simulation realism we step entry to next bar so bar `pos` is not
# used to resolve the trade outcome (bar `pos` is the signal bar).
# Entry at the OPEN of the next bar — live signals fire at bar close,
# execution happens on the next candle open (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
@@ -1458,17 +1528,30 @@ def run_backtest(
reasons = (report.bullish_reasons if ts_setup.direction == "BUY"
else report.bearish_reasons)
# Adjust SL/TP relative to actual entry (preserve distance from setup).
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
trades.append(Trade(
pair=bt_cfg.pair,
entry_time=entry_time,
entry_price=ts_setup.entry_price,
sl=ts_setup.stop_loss,
tp1=ts_setup.take_profit_1,
tp2=ts_setup.take_profit_2,
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=ts_setup.risk_pips,
reward_pips=ts_setup.reward_pips,
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,
@@ -1495,8 +1578,16 @@ def run_backtest(
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,
)
t_sim = time.perf_counter()
simulate_trades(trades, primary_df, max_bars=bt_cfg.max_hold_bars)
simulate_trades(trades, sim_df, max_bars=bt_cfg.max_hold_bars,
spread=bt_cfg.spread)
logger.info(
f"Trade simulation took {time.perf_counter() - t_sim:.2f}s "
f"(GPU={'yes' if _GPU_AVAILABLE else 'no'})"
@@ -1825,6 +1916,8 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
p.add_argument("--strategies", default="",
help="Comma-separated strategy tags to include (XAU,EUR_PB,GBP_PB,GENERIC). Empty = all.")
p.add_argument("--no-cache", action="store_true", help="Skip parquet cache (force API fetch)")
p.add_argument("--spread", type=float, default=0.30,
help="Round-trip cost in price units per trade (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)")
@@ -1883,6 +1976,7 @@ def main(argv: Optional[List[str]] = None) -> int:
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=args.spread,
)
logger.info(