From 53d8bc4390c0878b49e657eb11e031b0ec461818 Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Sat, 11 Jul 2026 22:57:37 -0400 Subject: [PATCH] Audit fixes: mixed-regime bias, live==backtest session parity, parallel replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix 1 — mixed-"trending" regime (ADX strong, EMAs unaligned) was hard-coded bullish in 5 sites: RSI reinterpretation used the uptrend playbook (RSI 38-55 -> STRONG_BUY conf 75), BUY-only confidence bonuses, BUY-only extension penalties, no counter-trend penalty. Now: extremes-only RSI (as transitioning), no directional bonuses, symmetric extension penalties. 6yr validation: H1 +67.2->+61.9R (mixed skew 16B/11S -> 4B/7S), H4 +31.6->+27.5R (the 4 bias BUYs), M15 +152.6->+154.7R, M5 +207.1->+206.1R with maxDD 28.7->24.3R. The ~8R aggregate give-back is bull-market subsidy from the bug. Fix 2 — live session filter now applies the EXACT replay-loop predicate (session window + blocked hours on the signal bar's CLOSE time, from the bar timestamp not wall clock). Replaces a wall-clock filter gated on a "XAU STRATEGY" marker that no code writes, which made live diverge from every validated book. Fix 3 — delete dead _check_xau_oversold_bounce and _generate_xau_trade_setup (unreachable; the latter hardcoded BUY and skipped the spread/RR gates). Backtester: parallel bar replay via forked workers (--workers, default auto = cores-2 capped 12). Per-bar evaluation is pure; the entry cooldown is applied in an ordered phase-2 pass. Proven byte-identical to sequential on the same window; ~300 -> 1030-1680 bars/s (6yr M5: ~26min -> 5-8min). --workers 1 = sequential path. Tests: 179 passed (new mixed-regime RSI regression test). Co-Authored-By: Claude Fable 5 --- CLAUDE.md | 2 +- backtest/gpu_backtest.py | 354 +++++++++++++++++++++++---------------- engine.py | 12 +- signal_aggregator.py | 181 +++++--------------- test_all.py | 20 +++ 5 files changed, 282 insertions(+), 287 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 7831fb7..0da5b1a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -116,7 +116,7 @@ The `backtest/` directory contains `gpu_backtest.py` — a GPU-accelerated backt Fill realism: charges a round-trip spread per trade (`--spread`; defaults to the pair's `default_spread` in PAIR_INFO), fills gap-through SL/TP at the bar open, and resolves trades against the RAW price path (the parquet cache stores raw bars; validation/cleaning runs on load — signals see cleaned data, trade resolution sees real spikes). Entry is at the next bar's open after the signal bar. Weekend/session filters evaluate bar-close (decision) time to match the live bot's wall clock. -GPU note: CuPy needs the NVIDIA runtime pip wheels (`nvidia-cuda-runtime-cu12`, `nvidia-cuda-nvrtc-cu12`, `nvidia-curand-cu12`, `nvidia-nvjitlink-cu12`) or it silently falls back to NumPy — check the "Backend:" line in the run log. The bar-replay loop itself is single-threaded Python either way (~340 bars/s); GPU only accelerates indicator precompute and SL/TP scanning. +GPU note: CuPy needs the NVIDIA runtime pip wheels (`nvidia-cuda-runtime-cu12`, `nvidia-cuda-nvrtc-cu12`, `nvidia-curand-cu12`, `nvidia-nvjitlink-cu12`) or it silently falls back to NumPy — check the "Backend:" line in the run log. GPU accelerates indicator precompute and SL/TP scanning; the bar-replay loop is Python (~300 bars/s/core) but fans out across forked worker processes (`--workers`, default auto = cores−2 capped at 12, ~1700 bars/s at 10). Per-bar signal evaluation is pure, so parallel and sequential books are identical (the entry cooldown is applied in an ordered pass afterwards); `--workers 1` forces the sequential path. ## Code Conventions diff --git a/backtest/gpu_backtest.py b/backtest/gpu_backtest.py index 6a8df43..1388698 100644 --- a/backtest/gpu_backtest.py +++ b/backtest/gpu_backtest.py @@ -31,6 +31,7 @@ from __future__ import annotations import argparse import logging +import multiprocessing as mp import os import sys import time @@ -1458,6 +1459,13 @@ class BacktestConfig: # 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]: @@ -1481,6 +1489,164 @@ def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Option 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 STRATEGY" in first_bull: + tag = "XAU" + elif "EUR STRATEGY" in first_bull or "EUR STRATEGY" in first_bear: + tag = "EUR" + elif "GBP STRATEGY" in first_bull: + tag = "GBP_PB" + + 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 run_backtest( bt_cfg: BacktestConfig, platform_cfg: Optional[PlatformConfig] = None, @@ -1549,10 +1715,6 @@ def run_backtest( last_pos = int(primary_df.index.searchsorted(bt_end_ts, side="right")) last_pos = min(last_pos, len(primary_df)) - trades: List[Trade] = [] - last_entry_pos = -10_000 - skipped_reasons: Dict[str, int] = {} - logger.info( f"Replaying {bt_cfg.pair} {bt_cfg.primary_tf.value}: " f"bars {first_eligible_pos}–{last_pos} of {len(primary_df)}" @@ -1560,154 +1722,59 @@ def run_backtest( 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() - for pos in range(first_eligible_pos, last_pos): - ts = primary_df.index[pos] - # Decision time = close of the current primary bar. - decision_time = ts + primary_dur + 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))] - # 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(decision_time, platform_cfg) - if block: - skipped_reasons[block] = skipped_reasons.get(block, 0) + 1 - continue + 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]) - # 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 - # Blocked hours (live checks datetime.now().hour ≈ bar close). - # analyze_from_dataframes step 6b checks bar-open timestamp, - # which is one bar behind — misses the last 5min of each hour. - _is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD" - _blocked = (platform_cfg.xau_strategy.blocked_hours_utc - if _is_xau else bt_filter.blocked_hours_utc) - if bar_hour in _blocked: - skipped_reasons["SESSION_WINDOW"] = skipped_reasons.get("SESSION_WINDOW", 0) + 1 - continue - - # Cooldown. + # ── 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] = [] + last_entry_pos = -10_000 + for pos, trade in candidates: if pos - last_entry_pos < bt_cfg.cooldown_bars: continue - - # Build sliced dfs per TF — only include bars fully closed by - # decision_time. A bar timestamped T closes at T + bar_dur, so we - # keep bars where T + bar_dur <= decision_time → T <= decision_time - bar_dur. - # Without this, the H1 bar containing the current M15 close would be - # included with future high/low/close values: a look-ahead leak. - 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). - # Pullback SELLs mark bearish_reasons, so check both lists. - 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 STRATEGY" in first_bull: - tag = "XAU" - elif "EUR STRATEGY" in first_bull or "EUR STRATEGY" in first_bear: - tag = "EUR" - elif "GBP STRATEGY" in first_bull: - tag = "GBP_PB" - - 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 — 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 - ) - 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=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), - )) + trades.append(trade) 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"({(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}") @@ -2059,6 +2126,8 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: 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") @@ -2118,6 +2187,7 @@ def main(argv: Optional[List[str]] = None) -> int: 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( diff --git a/engine.py b/engine.py index c978570..095971e 100755 --- a/engine.py +++ b/engine.py @@ -955,7 +955,7 @@ class TechnicalEngine: rsi_val = rsi_ind.value regime = result.market_regime - if regime in ("trending_up", "trending"): + if regime == "trending_up": if 38 <= rsi_val <= 55: signal = SignalStrength.STRONG_BUY conf = 75 @@ -1001,10 +1001,12 @@ class TechnicalEngine: desc = f"RSI {rsi_val:.1f} — Overbought in downtrend, possible reversal" else: return - elif regime == "transitioning": - # Trend is being born — RSI can't tell direction yet, and - # range-fade rules would sell the newborn trend. Only extremes - # carry information here. + elif regime in ("transitioning", "trending"): + # Trend being born (transitioning) or strong ADX with EMAs + # unaligned (mixed "trending") — in both, RSI can't tell + # direction. Treating mixed-"trending" as an uptrend here was + # a systematic bullish bias (23 BUY / 1 SELL, -6.8R on the + # 6yr M15 book). Only extremes carry information. if rsi_val >= 70: signal = SignalStrength.SELL conf = 50 diff --git a/signal_aggregator.py b/signal_aggregator.py index e93ba58..ff97f2d 100755 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -227,38 +227,41 @@ class SignalAggregator: if bt.regime_filter: self._apply_regime_filter_live(report, dfs) - if bt.session_filter: - xau_active = ( - self.config.xau_strategy.enabled - and any("XAU STRATEGY" in r for r in report.bullish_reasons) - ) - if not xau_active: - self._apply_session_filter_live(report) + # Session window + blocked hours on the signal bar's CLOSE time — + # the exact predicate the backtester's replay loop applies before + # each bar, so live and backtest cannot diverge. Bar-close time + # (not wall clock) makes manual mid-candle scans behave like + # boundary auto-scans. This replaces an older wall-clock filter + # that was guarded by a marker string no code wrote. + if bt.session_filter and report.primary_analysis is not None: + try: + bar_close = ( + pd.Timestamp(report.timestamp) + + pd.Timedelta( + seconds=ForexDataProvider.CANDLE_SECONDS[ + self.config.primary_timeframe + ] + ) + ) + close_h = bar_close.hour + _is_xau = pair_info["display"] == "XAU/USD" + _blocked = (self.config.xau_strategy.blocked_hours_utc + if _is_xau else bt.blocked_hours_utc) + outside = not (bt.session_start_hour <= close_h < bt.session_end_hour) + if outside or close_h in _blocked: + report.warnings.append( + f"SESSION FILTER: bar closes {close_h:02}:00 UTC — " + + ("outside window" if outside else "blocked hour") + ) + report.final_signal = SignalStrength.NEUTRAL + report.signal_quality = "F" + report.action_recommendation = "NO TRADE — Outside tradeable session" + report.trade_setup = None + except Exception: + pass return report - def _apply_session_filter_live(self, report: SignalReport) -> None: - """Skip signals outside the configured UTC trading window.""" - bt = self.config.backtest - now_h = datetime.now(timezone.utc).hour - if not (bt.session_start_hour <= now_h < bt.session_end_hour): - report.warnings.append( - f"SESSION FILTER: {now_h:02}:00 UTC outside " - f"{bt.session_start_hour:02}:00-{bt.session_end_hour:02}:00 — skipping" - ) - report.final_signal = SignalStrength.NEUTRAL - report.signal_quality = "F" - report.action_recommendation = ( - f"NO TRADE — Outside session " - f"({bt.session_start_hour:02}-{bt.session_end_hour:02} UTC; now {now_h:02})" - ) - report.trade_setup = None - else: - report.warnings.append( - f"Session: {now_h:02}:00 UTC inside trading window " - f"{bt.session_start_hour:02}-{bt.session_end_hour:02}" - ) - def _apply_regime_filter_live( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> None: @@ -931,112 +934,6 @@ class SignalAggregator: # ── Internals ────────────────────────────────────────────────── - def _check_xau_oversold_bounce( - self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] - ) -> bool: - """Return True when XAU/USD oversold-bounce conditions are met.""" - cfg = self.config.xau_strategy - - try: - bar_hour = pd.Timestamp(report.timestamp).hour - if bar_hour in cfg.blocked_hours_utc: - return False - except Exception: - return False - - m15_analysis = None - h4_analysis = None - for ca in report.confirmation_analyses: - if ca.timeframe == Timeframe.M15.value: - m15_analysis = ca - elif ca.timeframe == Timeframe.H4.value: - h4_analysis = ca - - h1_analysis = report.primary_analysis - - if not m15_analysis or not h1_analysis: - return False - - m15_rsi = m15_analysis.indicators.get("rsi") - if not m15_rsi or m15_rsi.value >= cfg.rsi_entry_threshold: - return False - - ema200_count = 0 - - m15_ema200 = m15_analysis.key_levels.get("ema_trend") - m15_df = dfs.get(Timeframe.M15) - if m15_df is not None and m15_ema200 is not None: - m15_price = float(m15_df["Close"].iloc[-1]) - if m15_price > m15_ema200: - ema200_count += 1 - - h1_ema200 = h1_analysis.key_levels.get("ema_trend") - if h1_ema200 is not None and report.current_price > h1_ema200: - ema200_count += 1 - - if h4_analysis: - h4_ema200 = h4_analysis.key_levels.get("ema_trend") - h4_df = dfs.get(Timeframe.H4) - if h4_df is not None and h4_ema200 is not None: - h4_price = float(h4_df["Close"].iloc[-1]) - if h4_price > h4_ema200: - ema200_count += 1 - - if ema200_count < cfg.min_ema200_tfs: - return False - - return True - - def _generate_xau_trade_setup( - self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] - ) -> Optional[TradeSetup]: - """Generate XAU-specific trade setup with backtested TP/SL.""" - cfg = self.config.xau_strategy - - m15_analysis = None - for ca in report.confirmation_analyses: - if ca.timeframe == Timeframe.M15.value: - m15_analysis = ca - break - - if not m15_analysis: - return None - - atr = m15_analysis.key_levels.get("atr", 0) - if atr <= 0: - return None - - price = report.current_price - pip_value = 0.10 - - quality_scale = self._TP_QUALITY_SCALE.get(report.signal_quality, 0.75) - tp_mult = cfg.tp_atr_multiplier * quality_scale - tp1 = price + tp_mult * atr - sl = price - cfg.sl_atr_multiplier * atr - tp2 = price + tp_mult * 1.5 * atr - - tf = self.config.signals.tp_tightening_factor - tp1 = price + (tp1 - price) * tf - tp2 = price + (tp2 - price) * tf - - sl_distance = price - sl - tp_distance = tp1 - price - rr = tp_distance / sl_distance if sl_distance > 0 else 0 - - return TradeSetup( - pair=report.symbol, - display_name=report.display_name, - direction="BUY", - entry_price=price, - stop_loss=round(sl, 2), - take_profit_1=round(tp1, 2), - take_profit_2=round(tp2, 2), - risk_reward=round(rr, 2), - risk_pips=round(sl_distance / pip_value, 1), - reward_pips=round(tp_distance / pip_value, 1), - pip_value=pip_value, - ) - _TP_QUALITY_SCALE = {"A+": 2.0, "A": 1.5, "B": 1.0, "C": 1.0} def _generate_xau_atr_trade_setup( @@ -1143,15 +1040,17 @@ class SignalAggregator: if primary and report.market_regime.startswith("trending"): rsi_ind = primary.indicators.get("rsi") bb_ind = primary.indicators.get("bollinger") + # Pullback bonuses require a KNOWN trend direction — the mixed + # "trending" regime (EMAs unaligned) gets no directional bonus. if rsi_ind: rsi_val = rsi_ind.value - if report.market_regime in ("trending_up", "trending") and score > 0 and 38 <= rsi_val <= 55: + if report.market_regime == "trending_up" and score > 0 and 38 <= rsi_val <= 55: confidence += 5 elif report.market_regime == "trending_down" and score < 0 and 45 <= rsi_val <= 62: confidence += 5 if bb_ind: bb_pos = bb_ind.value - if report.market_regime in ("trending_up", "trending") and score > 0 and bb_pos < 0.4: + if report.market_regime == "trending_up" and score > 0 and bb_pos < 0.4: confidence += 5 elif report.market_regime == "trending_down" and score < 0 and bb_pos > 0.6: confidence += 5 @@ -1246,19 +1145,23 @@ class SignalAggregator: if report.market_regime.startswith("trending"): rsi_ind = primary.indicators.get("rsi") bb_ind = primary.indicators.get("bollinger") + # Extension penalties apply in ANY strong-ADX regime; the + # mixed "trending" regime penalizes BOTH directions (it + # previously penalized only BUYs — the SELL side of the + # same bullish-bias bug fixed in the RSI reinterpretation). if rsi_ind: rsi_val = rsi_ind.value if report.market_regime in ("trending_up", "trending") and score > 0 and rsi_val > 65: report.warnings.append(f"Extended — RSI {rsi_val:.0f} in uptrend") penalties.append(("rsi_extended", 5)) - elif report.market_regime == "trending_down" and score < 0 and rsi_val < 35: + elif report.market_regime in ("trending_down", "trending") and score < 0 and rsi_val < 35: report.warnings.append(f"Extended — RSI {rsi_val:.0f} in downtrend") penalties.append(("rsi_extended", 5)) if bb_ind: bb_pos = bb_ind.value if report.market_regime in ("trending_up", "trending") and score > 0 and bb_pos > 0.85: penalties.append(("bb_overextended", 8)) - elif report.market_regime == "trending_down" and score < 0 and bb_pos < 0.15: + elif report.market_regime in ("trending_down", "trending") and score < 0 and bb_pos < 0.15: penalties.append(("bb_overextended", 8)) # Hard block: static exhaustion + fading momentum → kill the trade diff --git a/test_all.py b/test_all.py index b132dda..40ce9f9 100644 --- a/test_all.py +++ b/test_all.py @@ -2025,6 +2025,26 @@ def _(): assert config.xau_strategy.enabled is True +@test("Mixed 'trending' regime: RSI mid-zone is neutral, extremes only") +def _(): + from engine import TechnicalEngine, TechnicalAnalysis, IndicatorSignal + from config import TechnicalConfig + eng = TechnicalEngine(TechnicalConfig()) + # Pre-fix, mixed "trending" used the uptrend playbook: RSI 45 became + # STRONG_BUY conf 75 in a regime whose EMAs disagree on direction. + for rsi_v, expected in ((45.0, SignalStrength.NEUTRAL), + (75.0, SignalStrength.SELL), + (25.0, SignalStrength.BUY)): + res = TechnicalAnalysis(symbol="XAU/USD", timeframe="1h", timestamp="t") + res.market_regime = "trending" + res.indicators["rsi"] = IndicatorSignal( + name="RSI", value=rsi_v, signal=SignalStrength.NEUTRAL, + confidence=30, description="", weight=1.5) + eng._reinterpret_rsi_for_regime(res) + got = res.indicators["rsi"].signal + assert got == expected, f"RSI {rsi_v} in mixed regime → {got}, want {expected}" + + @test("XAU regime filter: blocked regime never yields actionable quality") def _(): from pair_signals.xauusd import make_config