Audit fixes: mixed-regime bias, live==backtest session parity, parallel replay

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 <noreply@anthropic.com>
2026-07-11 22:57:37 -04:00
parent f3ba4d56a3
commit 53d8bc4390
5 changed files with 282 additions and 287 deletions
@@ -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

@@ -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(
@@ -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
@@ -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
@@ -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