news: DeepSeek validation, vs_forecast fix, /status source health

Validation (llm_news_eval.py harness + 78-headline golden CSV + verdict doc): headline scoring 78/78 vs keyword baseline 24/78 — validated; event gold-direction calls misread close K/M numeric comparisons ~11% of the time (both claims cases in sample, JOLTS 8.79M vs 8.84M). Fix: event payload now carries code-computed "vs_forecast" (from EconomicEvent.surprise) and the prompt gains an explicit inverted- metrics rule + worked claims example — iteration 1 (field alone) fixed the arithmetic but exposed a masked inversion error. 18/18 on live re-verification (4 misreads fixed, 11 regression cases clean, 3 fresh both-side probes). /status: new "News & AI" section — per-source health for the calendar chain (which of FF/MyFXBook served), each RSS feed, and DeepSeek (daily count, last error kind), written passively by the existing fetch paths under _cache_lock. Failures WARN once on transition (repeats DEBUG, recovery INFO) so the rotating log carries outage stories without 10-min spam. test_all.py: +5 Module 19 tests (351 total); also carries the Module 22 fixed-RR-TP tests from the prior research session (shared file). Replay artifacts dir gitignored like commit_ladder. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017mbXs9HRawZhD1uZ3ze7zQ
research: --fixed-rr-tp backtest flag + 2026-08-05 verdict (NOT shipped)
2026-08-06 22:21:02 -04:00 · 2026-08-06 22:20:43 -04:00
11 changed files with 1488 additions and 10 deletions
@@ -36,3 +36,4 @@ backtest/__pycache__/
 .env.*
 logs/
 backtest/research/commit_ladder_2026-08/
 backtest/research/llm_news_eval_2026-08/
@@ -24,6 +24,8 @@ python backtest/gpu_backtest.py --pair XAUUSD --start 2025-01-01 --end 2025-06-0
    --primary-tf 15min --confirm-tfs 1h,4h
 # key flags: --spread --quality A+,A,B,C --strategies --workers --max-hold-bars
 #            --warmup-bars --cooldown-bars --no-cache --export-csv out.csv --detail
 #            --fixed-rr-tp N (research-only: hard TP at N×risk, trail off; verdict
 #            2026-08-05: NOT shipped — backtest/research/fixedtp_2026-08-05.md)
 ```
 **`test_all.py` is NOT pytest** — it uses a custom `@test("name")` decorator that runs each test immediately at import; don't grep for `def test_` or try `pytest test_all.py`. Run the whole file; it prints per-module results and exits non-zero on failure. There is no way to run a single test short of commenting modules out. Tests must stay green; every new gate/knob needs tests. When patching weekend logic in tests, patch `providers.base._is_forex_weekend` — but know there are THREE independent weekend implementations (`providers/base.py`, `signal_aggregator.py`, and an inline copy in `backtest/gpu_backtest.py:_is_xau_blocked_by_calendar`); patching the provider one does not reach the other two.
@@ -1441,6 +1441,12 @@ 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
    # RESEARCH (2026-08): >0 replaces the exit model with a hard TP at
    # this multiple of risk from entry (2.0 = fixed 1:2 RR); trailing is
    # disabled for ALL trades, including per-trade XAU_PB overrides.
    # Entries/stops/gates are untouched, so the signal book is identical
    # to the baseline — only the exit policy changes. Never shipped.
    fixed_rr_tp: float = 0.0
    # 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
@@ -1703,6 +1709,25 @@ def _apply_cooldown(
    return accepted
 def _apply_fixed_rr_tp(trades: List[Trade], mult: float) -> None:
    """RESEARCH: replace each trade's TP with a hard target at
    `mult` × |entry − sl| and zero the per-trade trail params so the
    XAU_PB overrides cannot resurrect trailing. Risk is measured off the
    final (re-anchored) entry/SL, so call this after
    `_reanchor_entries_to_raw`. The caller must also pass
    trail_arm_r=0 to `simulate_trades` — dispatch is on the run-level arg."""
    for t in trades:
        risk = abs(t.entry_price - t.sl)
        t.tp1 = (t.entry_price + mult * risk if t.direction == "BUY"
                 else t.entry_price - mult * risk)
        t.tp2 = t.tp1              # static branch never reads tp2; CSV hygiene
        t.trail_arm_r = 0.0
        t.trail_dist_r = 0.0
        t.rr_planned = mult
        if t.pip_value:
            t.reward_pips = mult * risk / t.pip_value
 def run_backtest(
    bt_cfg: BacktestConfig,
    platform_cfg: Optional[PlatformConfig] = None,
@@ -1853,10 +1878,17 @@ def run_backtest(
        use_cache=bt_cfg.use_cache, clean=False,
    )
    _reanchor_entries_to_raw(trades, signal_ts_list, sim_df)
    if bt_cfg.fixed_rr_tp > 0:
        _apply_fixed_rr_tp(trades, bt_cfg.fixed_rr_tp)
        logger.info(
            f"RESEARCH override: fixed-RR TP = {bt_cfg.fixed_rr_tp} x risk; "
            f"trailing exit DISABLED for all {len(trades)} trades (incl. XAU_PB)"
        )
    run_arm = 0.0 if bt_cfg.fixed_rr_tp > 0 else platform_cfg.signals.trail_arm_r
    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_arm_r=run_arm,
                    trail_dist_r=platform_cfg.signals.trail_dist_r)
    logger.info(
        f"Trade simulation took {time.perf_counter() - t_sim:.2f}s "
@@ -2200,6 +2232,10 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
                   help="Override signals.trail_arm_r for this run (research knob)")
    p.add_argument("--trail-dist", type=float, default=None,
                   help="Override signals.trail_dist_r for this run (research knob)")
    p.add_argument("--fixed-rr-tp", type=float, default=None,
                   help="RESEARCH: exit at a hard TP of MULT x risk (e.g. 2.0 = "
                        "fixed 1:2 RR) instead of the trailing stop; "
                        "entries/stops/gates unchanged.")
    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.")
@@ -2323,6 +2359,16 @@ def main(argv: Optional[List[str]] = None) -> int:
        # make_config; the fallback must match (live == backtest).
        from app_config import apply_pair_overrides  # noqa: E402
        apply_pair_overrides(platform_cfg)
    if args.fixed_rr_tp is not None:
        if args.fixed_rr_tp <= 0:
            raise SystemExit("--fixed-rr-tp must be > 0")
        if (args.trail_arm is not None or args.trail_dist is not None or
                any(s.split("=", 1)[0].strip().startswith(("signals.trail",
                                                           "signals.pb_trail"))
                    for s in (args.set or []))):
            raise SystemExit(
                "--fixed-rr-tp conflicts with --trail-arm/--trail-dist/"
                "--set signals.*trail* (the flag disables trailing entirely)")
    if args.swing_sr_lookback > 0:
        platform_cfg.technical.swing_sr_lookback = args.swing_sr_lookback
    if args.trail_arm is not None:
@@ -2369,6 +2415,7 @@ def main(argv: Optional[List[str]] = None) -> int:
        strategies_only=[s.strip() for s in args.strategies.split(",") if s.strip()],
        spread=spread,
        workers=args.workers,
        fixed_rr_tp=args.fixed_rr_tp or 0.0,
    )
    logger.info(
@@ -0,0 +1,123 @@
 # Fixed-RR hard-TP vs shipped trailing exit — research campaign 2026-08-05
 **Verdict: DO NOT SHIP. The trailing exit stays.** The fixed 2R target beats the trail on
 total R in every 2024+ window but the edge is a trend-regime artifact: it inverts in the
 2021–2023 ranging years, doubles drawdown, cuts win rate 10–19 points, and doubles hold
 times (amplifying the accepted live one-position divergence). Details below.
 ## Setup
 - Code: HEAD `fc5ce85` + the `--fixed-rr-tp` research flag (uncommitted at run time).
  The flag replaces every trade's TP with `mult × |entry − sl|` AFTER re-anchoring and
  disables trailing (run-level and per-trade PB params) — entries, stops, gates, and the
  cooldown pass are untouched, so both arms replay **trade-identical books** (verified:
  paired join on entry_time+direction matches 1:1 in every window). Only the exit policy
  differs → per-trade paired comparison.
 - Provider: Oanda (cache-warm; keys `oanda_XAU_USD_{5min,15min,1h}_{20250603_20260804,
  20240331_20250330,20210531_20260804}`). Charged cost `--spread 0.65` (real measured
  round trip); $0.30 letter-cost books derived analytically as `r + 0.35/risk` and
  verified exact against one real `--spread 0.30` run (max per-trade error 1.8e-15 —
  spread never enters trigger detection, per the owner-declined 2026-08-04 decision).
 - Control sanity gate: fresh M5 12mo control reproduced the pinned HEAD reference
  exactly — 555 trades, +98.64R, PF 1.483, maxDD −10.73R.
 - Commands (control = same without the flag):
  `venv/bin/python backtest/gpu_backtest.py --pair XAUUSD --primary-tf {5min,15min}
   --start 2025-08-04 --end 2026-08-04 --spread 0.65 [--fixed-rr-tp 2.0]
   --export-csv backtest/fixedtp_*.csv`
  OOS `--start 2024-06-01 --end 2025-03-30`; 5yr `--start 2021-08-01 --end 2026-08-04`.
 - Books pinned as `backtest/fixedtp_{m5,m15}_{12mo,oos,5yr}_{ctrl,rr20,...}.csv`
  (per-trade indicator breakdown, agreement counts, regime, vol-regime, weekday,
  MAE/MFE, bars-held — the full instructions_for_backtest record).
 ## Headline (@ $0.65)
 | Window | Arm | n | R | PF | WR | maxDD | recov | medBars | exits |
 |---|---|---|---|---|---|---|---|---|---|
 | M5 12mo | trail | 555 | +98.64 | 1.483 | 52.4% | −10.7 | 9.2 | 28 | TRAIL 361 / SL 183 / TO 11 |
 | M5 12mo | 2R hard | 555 | +124.85 | 1.378 | 42.9% | −24.5 | 5.1 | 50 | SL 312 / TP1 210 / TO 33 |
 | M15 12mo | trail | 397 | +73.93 | 1.525 | 62.2% | −10.8 | 6.8 | 28 | TRAIL 255 / SL 135 / TO 7 |
 | M15 12mo | 2R hard | 397 | +95.95 | 1.431 | 44.3% | −11.2 | 8.6 | 39 | SL 215 / TP1 151 / TO 31 |
 | M5 OOS | trail | 86 | +9.45 | 1.332 | 53.5% | −8.2 | 1.2 | 70 | — |
 | M5 OOS | 2R hard | 86 | +18.83 | 1.407 | 45.3% | −9.7 | 1.9 | 124 | — |
 | M15 OOS | trail | 299 | +38.77 | 1.360 | 59.2% | −15.5 | 2.5 | 38 | — |
 | M15 OOS | 2R hard | 299 | +72.14 | 1.435 | 45.2% | −13.9 | 5.2 | 82 | — |
 Halves (12mo, split 2026-02-03): M5 rr20 +67.99/+56.86 (both positive; H2 PF 1.325 vs
 ctrl 1.503), M15 rr20 +73.74/+22.21 (both positive; ctrl H2 +7.23 — the known H2-flat).
 At $0.30 all deltas keep the same sign (e.g. M5 12mo +109.0 vs +135.2).
 So the **letter** of rule 1 passes for rr20 on both TFs. The 5yr decomposition is why it
 still must not ship.
 ## Exit migration (paired, M5 12mo; same pattern everywhere)
 - SL→SL 183 trades: ΔR 0 (identical by construction — stops untouched).
 - TRAIL→TP1 207 trades: +248.9R → +424.5R (**+175.6R**) — winners the trail exits early.
 - TRAIL→SL 129 trades: +19.0R → −136.8R (**−155.7R**) — retraces the trail survives
  become full stop-outs.
 - TRAIL→TIMEOUT 25: +7.4R. Net = +26.2R. The whole effect is these two big flows nearly
  cancelling; which one wins is a regime property, not an exit-policy edge.
 - Trades with ctrl MFE ≥ 2R: 95; hard cap forfeits only +10.4R vs trail on M5 (the trail
  already gives ~0.75R back from peak, so a 2R cap costs little *in a trend*).
 ## Mini-grid (12mo @ $0.65) — the tell
 | mult | M5 R / PF / DD | M15 R / PF / DD |
 |---|---|---|
 | trail | +98.6 / 1.483 / −10.7 | +73.9 / 1.525 / −10.8 |
 | 1.0R | +52.5 / 1.205 / −12.8 | +67.1 / 1.410 / −7.3 |
 | 1.5R | +80.7 / 1.263 / −21.4 | +66.9 / 1.325 / −11.2 |
 | 2.0R | +124.9 / 1.378 / −24.5 | +96.0 / 1.431 / −11.2 |
 | 2.5R | +147.7 / 1.426 / −23.6 | +110.4 / 1.469 / −15.9 |
 | 3.0R | +155.8 / 1.432 / −21.6 | +135.8 / 1.562 / −17.5 |
 Monotone in the multiple on both TFs — "2.0" is not an optimum, it's a point on a curve
 that says *never take profit* (3R's M5 TIMEOUT bucket alone is +84.8R). That is the
 signature of a one-way trend sample, not of a superior exit rule. It also reconfirms the
 original finding that SHORT fixed targets (≤1.5R) are strictly bad (cf. RR-0.5 removal,
 `config.py:126-133`: +48R → +105R).
 ## 5yr regime decomposition (2021-08-01 → 2026-08-04 @ $0.65) — the verdict
 M15 (real sample in every year): ctrl +127.97 PF 1.209 DD −35.8; rr20 +158.97 PF 1.170
 DD **−59.8**; rr30 +204.03 PF 1.203 DD **−98.3**.
 | Year | n | trail | 2R hard | 3R hard |
 |---|---|---|---|---|
 | 2021 | 88 | **+4.00** | −16.09 | −17.41 |
 | 2022 | 284 | −9.54 | −7.63 | −26.74 |
 | 2023 | 227 | **+7.79** | −1.71 | −5.81 |
 | 2024 | 345 | +40.44 | +64.98 | +89.14 |
 | 2025 | 405 | +49.33 | +69.39 | +104.42 |
 | 2026 | 229 | +35.95 | +50.04 | +60.43 |
 M5 5yr mirrors it (ctrl +128.4 PF 1.387 DD −12.8 vs rr20 +160.2 PF 1.306 DD −24.7;
 2022: −1.1 vs −5.2), though pre-2024 M5 samples are thin (12–60 trades/yr).
 The hard target's entire outperformance lives in the 2024–2026 gold trend. In ranging
 years it flips sign exactly where the trail stays positive (M15 2021, 2023), and the
 further the target, the worse the ranging-year bleed and the deeper the DD. The 12mo,
 halves, and OOS windows all sit inside the trend era, which is why the rule-1 letter
 passes — this is precisely the within-regime-split trap rule 1's OOS clause exists to
 catch, one level up.
 ## Additional non-ship reasons
 - **Live divergence grows with hold time.** Median hold doubles (M5 28→50 bars; OOS
  70→124). The daemon holds one position total; the backtester doesn't model that
  blocking. Longer holds = more live signals skipped that the paper book banks.
 - **WR 52→43% (M5) / 62→44% (M15), max loss streak 6→11 (M15)** — materially harder to
  operate and size.
 - Aggregate tables (session/day/regime/vol, streaks, quarterly equity, rolling-20 WR)
  computed and consistent — no bucket reverses the conclusion; both arms' 12mo profit is
  concentrated in Q1-2026 (+87.7 ctrl / +102.0 rr20).
 ## Status
 - `--fixed-rr-tp` stays as a documented research-only flag (default off; conflicts with
  `--trail-arm/--trail-dist/--set signals.*trail*` are hard errors). Module 22 tests
  (5) pin the override math, TP fill + spread, SL-first tie, PB-trail-disable, and CLI
  guards. Full suite: 346 green.
 - Re-open only if gold's regime turns and a *regime-conditional* exit (trail in range,
  target in trend) is proposed with a pre-registered mechanism — a plain fixed ratio is
  now a closed hypothesis at every multiple in {0.5, 1.0, 1.5, 2.0, 2.5, 3.0}.
@@ -0,0 +1,697 @@
 """
 DeepSeek news-analysis validation harness — XAU/USD.
 ====================================================
 Research-only (no production code changes). Answers: is the DeepSeek
 layer's news analysis *correct*?
 Part A (events): replays every high-impact USD release with a genuine
 surprise (2024-01 → 2026-07, from backtest/cache/calendar.parquet)
 through the REAL `NewsEngine._llm_event_direction` — real prompt, real
 payload, real validation — and scores its "gold bullish/bearish" calls
 against gold's actual move (Oanda M15 mids) at +15m/30m/1h/4h/24h.
 Baselines: mechanical surprise-sign doctrine, FF actualBetterWorse,
 always-bullish/bearish (2024-26 is a gold bull regime), coin flip.
 Part B (headlines): a curated golden set (llm_news_golden.csv) through
 the REAL `NewsEngine._llm_headline_scores` in production-sized batches,
 vs the keyword scorer it overrides. Includes repeat-batch (determinism)
 and shuffled-order (cross-contamination) probes.
 Implementation notes that keep the replay faithful to production:
  * `_llm_event_direction`/`_llm_headline_scores` resolve the client via
    `get_llm_client()`, so we seed the module singleton with a research
    client (raised daily cap: 586 events > production cap 500).
  * The event cache key omits the forecast, so a repeat `actual` would
    silently re-serve a direction computed for a different forecast —
    the cache is cleared before every call.
  * A "neutral" reply and a hard failure both return None; they are
    disambiguated by reading the event cache back after the call
    (absent = failure, "" = neutral, "bullish:/bearish: …" = call).
  * Raw request/response bytes are captured with a requests response
    hook — the request-building code path is untouched.
 Usage (from repo root):
  venv/bin/python backtest/research/llm_news_eval.py events --limit 5
  venv/bin/python backtest/research/llm_news_eval.py events --determinism 20
  venv/bin/python backtest/research/llm_news_eval.py headlines
  venv/bin/python backtest/research/llm_news_eval.py report
 Artifacts land in backtest/research/llm_news_eval_2026-08/ (JSONL replay
 logs are resume-safe: completed events are never re-paid). `report` is a
 pure offline function of the artifacts.
 """
 import argparse
 import json
 import math
 import random
 import sys
 import time
 from datetime import datetime, timezone
 from pathlib import Path
 import numpy as np
 import pandas as pd
 REPO_ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(REPO_ROOT))
 import llm_client
 from app_config import deepseek_api_key, get_app_config
 from config import NewsConfig
 from llm_client import DEFAULT_MODEL, DeepSeekClient
 from news_engine import EconomicEvent, NewsEngine
 RESEARCH_DIR = Path(__file__).parent
 ART_DIR = RESEARCH_DIR / "llm_news_eval_2026-08"
 CALENDAR = REPO_ROOT / "backtest" / "cache" / "calendar.parquet"
 BARS = (REPO_ROOT / "backtest" / "cache" /
        "oanda_XAU_USD_15min_20210531_20260804.parquet")
 GOLDEN = RESEARCH_DIR / "llm_news_golden.csv"
 EVENTS_JSONL = ART_DIR / "events_replay.jsonl"
 HEADLINES_JSONL = ART_DIR / "headlines_replay.jsonl"
 SCORED_CSV = ART_DIR / "events_scored.csv"
 HORIZONS = [("15m", 15), ("30m", 30), ("1h", 60), ("4h", 240),
            ("24h", 1440)]
 NEUTRAL_BAND = 0.15  # |xau| below this = neutral prediction (Part B)
 BATCH_CAP = 25       # production batch cap (news_engine.py:783)
 EXPECTED_CORPUS = 586
 # ── shared plumbing ──────────────────────────────────────────────────
 class _Recorder:
    """requests response hook capturing the exact wire traffic."""
    def __init__(self):
        self.last = None
    def hook(self, resp, **kwargs):
        body = resp.request.body
        if isinstance(body, bytes):
            body = body.decode("utf-8", "replace")
        self.last = {
            "request_body": body,
            "http_status": resp.status_code,
            "response_body": resp.text,
            "elapsed_s": round(resp.elapsed.total_seconds(), 3),
        }
        return resp
    def pop(self):
        cap, self.last = self.last, None
        return cap or {}
 def install_research_client(timeout: float):
    """Build a research DeepSeekClient (raised cap) and seed the module
    singleton so the production code paths use it. Returns (client,
    recorder)."""
    app = get_app_config()
    key = deepseek_api_key(app)
    ds = app.deepseek_settings
    if not key:
        sys.exit("No DeepSeek API key (env DEEPSEEK_API_KEY or "
                 "config.json deepseek.api_key) — aborting before spend.")
    if ds.get("enabled") is False:
        sys.exit("deepseek.enabled is false in config.json — aborting.")
    client = DeepSeekClient(
        api_key=key,
        model=str(ds.get("model", DEFAULT_MODEL)),
        timeout=timeout,
        daily_cap=10 ** 6,  # research: corpus 586 > production cap 500
    )
    rec = _Recorder()
    client._session.hooks["response"].append(rec.hook)
    with llm_client._client_lock:
        llm_client._client = client
    return client, rec
 def _s(x):
    """Parquet str columns carry float NaN — EconomicEvent needs None."""
    return None if pd.isna(x) else str(x)
 def load_corpus():
    """High-impact USD releases with a genuine surprise, through the
    production `EconomicEvent.surprise` gate. Asserts run before any
    API spend."""
    df = pd.read_parquet(CALENDAR)
    df = df[(df["impactName"] == "high") & (df["currency"] == "USD")]
    rows = []
    for r in df.itertuples():
        ev = EconomicEvent(
            time=datetime.fromtimestamp(int(r.dateline), tz=timezone.utc),
            currency="USD", impact="high", event=_s(r.name) or "",
            forecast=_s(r.forecast), previous=_s(r.previous),
            actual=_s(r.actual),
        )
        if ev.surprise not in ("positive", "negative"):
            continue
        rows.append({
            "ffid": int(r.id), "dateline": int(r.dateline),
            "utc_time": ev.time.isoformat(), "name": ev.event,
            "actual": ev.actual, "forecast": ev.forecast,
            "previous": ev.previous, "surprise": ev.surprise,
            "better_worse": int(r.actualBetterWorse), "ev": ev,
        })
    ids = [d["ffid"] for d in rows]
    if len(rows) != EXPECTED_CORPUS:
        sys.exit(f"corpus size {len(rows)} != {EXPECTED_CORPUS} — "
                 "calendar.parquet changed; re-verify before spending.")
    if len(set(ids)) != len(ids):
        sys.exit("FF ids not unique — resume keying would break.")
    # Pin the actualBetterWorse mapping on two unambiguous NFP prints.
    nfp = {d["utc_time"][:10]: d["better_worse"] for d in rows
           if d["name"] == "Non-Farm Employment Change"}
    if nfp.get("2024-01-05") != 1 or nfp.get("2024-05-03") != 2:
        sys.exit("actualBetterWorse mapping pins failed (expected "
                 "1=better-for-USD on 2024-01-05 beat, 2=worse on "
                 "2024-05-03 miss).")
    rows.sort(key=lambda d: (d["dateline"], d["ffid"]))
    return rows
 def _load_jsonl(path: Path):
    recs = []
    if path.exists():
        with open(path) as f:
            for line in f:
                line = line.strip()
                if line:
                    recs.append(json.loads(line))
    return recs
 # ── Part A: event replay ─────────────────────────────────────────────
 def _extract_reason(cap):
    try:
        content = json.loads(cap["response_body"])[
            "choices"][0]["message"]["content"]
        return json.loads(content).get("reason")
    except Exception:
        return None
 def cmd_events(args):
    ART_DIR.mkdir(exist_ok=True)
    corpus = load_corpus()
    if args.limit:
        corpus = corpus[:args.limit]
    engine = NewsEngine(NewsConfig())
    existing = _load_jsonl(EVENTS_JSONL)
    done1 = {r["ffid"] for r in existing
             if r.get("run_pass") == 1 and r.get("llm_direction")}
    done2 = {r["ffid"] for r in existing
             if r.get("run_pass") == 2 and r.get("llm_direction")}
    pending = [d for d in corpus if d["ffid"] not in done1]
    print(f"corpus {len(corpus)} | already recorded {len(done1)} | "
          f"pending {len(pending)} | determinism re-calls "
          f"{args.determinism} | model calls are ~100 tokens each "
          f"(deepseek-v4-flash, static-prefix cached) → cost ≈ cents")
    if not pending and not args.determinism:
        print("nothing to do")
        return
    _, rec = install_research_client(args.timeout)
    out = open(EVENTS_JSONL, "a")
    def call_one(d, run_pass):
        # Cache key omits forecast → cleared per call so 196 colliding
        # corpus rows can't be served a stale direction.
        with NewsEngine._cache_lock:
            NewsEngine._llm_event_cache.clear()
        ev = d["ev"]
        engine._llm_event_direction(ev)
        key = (ev.event, ev.currency, str(ev.actual or ""))
        with NewsEngine._cache_lock:
            cached = NewsEngine._llm_event_cache.get(key)
        cap = rec.pop()
        if cached is None:
            direction, text = None, None       # hard failure → retry
        elif cached == "":
            direction, text = "neutral", ""
        else:
            direction, text = cached.split(":", 1)[0], cached
        recd = {k: d[k] for k in (
            "ffid", "dateline", "utc_time", "name", "actual", "forecast",
            "previous", "surprise", "better_worse")}
        recd.update({
            "llm_direction": direction, "llm_text": text,
            "reason_raw": _extract_reason(cap), "run_pass": run_pass,
            "ts": datetime.now(timezone.utc).isoformat(), **cap,
        })
        out.write(json.dumps(recd) + "\n")
        out.flush()
        return direction
    remaining = pending
    for npass in range(1, args.retry_passes + 1):
        if not remaining:
            break
        failed = []
        for i, d in enumerate(remaining):
            if call_one(d, 1) is None:
                failed.append(d)
            if args.sleep:
                time.sleep(args.sleep)
            if (i + 1) % 25 == 0:
                print(f"  pass {npass}: {i + 1}/{len(remaining)} "
                      f"({len(failed)} failed)")
        print(f"pass {npass} done: {len(remaining) - len(failed)} ok, "
              f"{len(failed)} failed")
        remaining = failed
    if remaining:
        print(f"WARNING: {len(remaining)} events still unrecorded after "
              f"{args.retry_passes} passes — re-run `events` to resume.")
    if args.determinism:
        ok_ids = {x["ffid"] for x in remaining}
        done_now = [d for d in corpus if d["ffid"] not in ok_ids]
        done_now.sort(key=lambda d: (d["name"], d["dateline"]))
        stride = max(1, len(done_now) // args.determinism)
        sample = done_now[::stride][:args.determinism]
        todo = [d for d in sample if d["ffid"] not in done2]
        print(f"determinism probe: {len(todo)} re-calls "
              f"({len(sample) - len(todo)} already recorded)")
        for d in todo:
            call_one(d, 2)
            if args.sleep:
                time.sleep(args.sleep)
    out.close()
    print("events replay complete →", EVENTS_JSONL)
 # ── Part B: golden headlines ─────────────────────────────────────────
 def _load_golden():
    g = pd.read_csv(GOLDEN)
    need = {"headline", "category", "expected_xau_sign", "rationale"}
    if not need <= set(g.columns):
        sys.exit(f"golden CSV missing columns {need - set(g.columns)}")
    if not g["expected_xau_sign"].isin([-1, 0, 1]).all():
        sys.exit("expected_xau_sign must be -1/0/+1")
    if not g["headline"].is_unique:
        sys.exit("duplicate headlines in golden CSV")
    return g
 def cmd_headlines(args):
    ART_DIR.mkdir(exist_ok=True)
    g = _load_golden()
    heads = list(g["headline"])
    _, rec = install_research_client(args.timeout)
    engine = NewsEngine(NewsConfig())
    out = open(HEADLINES_JSONL, "w")  # tiny (~9 calls): fresh each run
    def run_batch(texts, tag, batch_idx):
        scores = None
        for attempt in range(3):
            scores = engine._llm_headline_scores(texts)
            cap = rec.pop()
            if scores is not None:
                break
            print(f"  {tag} batch {batch_idx} attempt {attempt + 1} "
                  f"failed (http {cap.get('http_status')})")
            time.sleep(2.0)
        out.write(json.dumps({
            "run_tag": tag, "batch": batch_idx, "headlines": texts,
            "scores": scores,
            "ts": datetime.now(timezone.utc).isoformat(), **cap,
        }) + "\n")
        out.flush()
        return scores
    batches = [heads[i:i + BATCH_CAP] for i in range(0, len(heads),
                                                     BATCH_CAP)]
    for bi, b in enumerate(batches):
        ok = run_batch(b, "main", bi) is not None
        print(f"main batch {bi} ({len(b)} headlines): "
              f"{'ok' if ok else 'FAILED'}")
        time.sleep(args.sleep)
    run_batch(batches[0], "repeat", 0)
    print("repeat probe done")
    shuffled = heads[:]
    random.Random(20260806).shuffle(shuffled)
    for bi in range(0, len(shuffled), BATCH_CAP):
        run_batch(shuffled[bi:bi + BATCH_CAP], "shuffled",
                  bi // BATCH_CAP)
    print("shuffled probe done →", HEADLINES_JSONL)
    out.close()
 # ── report: pure offline function of the artifacts ───────────────────
 def _binom_p(k, n):
    """Exact two-sided binomial p-value vs p=0.5 (no scipy)."""
    if n == 0:
        return float("nan")
    dist = [math.comb(n, i) for i in range(n + 1)]
    tot = float(2 ** n)
    pk = dist[k]
    return min(1.0, sum(c for c in dist if c <= pk) / tot)
 def _mcnemar_p(hit_a, hit_b):
    """Exact McNemar via binomial on discordant pairs. Inputs: aligned
    boolean lists."""
    b = sum(1 for x, y in zip(hit_a, hit_b) if x and not y)
    c = sum(1 for x, y in zip(hit_a, hit_b) if not x and y)
    return b, c, _binom_p(min(b, c), b + c)
 def _latest_events(recs):
    """Last pass-1 record per ffid (retries append; last wins)."""
    out = {}
    for r in recs:
        if r.get("run_pass") == 1:
            out[r["ffid"]] = r
    return list(out.values())
 def _fmt_rate(hits, n, label=""):
    if n == 0:
        return f"{label:<24} n=0"
    p = _binom_p(hits, n)
    return (f"{label:<24} {hits:>4}/{n:<4} = {hits / n * 100:5.1f}%  "
            f"(binom p={p:.3f})")
 def report_events():
    recs = _latest_events(_load_jsonl(EVENTS_JSONL))
    if not recs:
        print("no event replay records — run `events` first")
        return
    bars = pd.read_parquet(BARS)
    closes = bars["Close"].to_numpy()
    close_ns = (bars.index + pd.Timedelta(minutes=15)).asi8
    def anchor(t_ns, h_min):
        i = np.searchsorted(close_ns, t_ns, side="right") - 1
        if i < 0:
            return None
        stale_ns = max(45.0, h_min / 4.0) * 60e9
        if t_ns - close_ns[i] > stale_ns:
            return None
        return closes[i]
    rows, skip = [], {h: 0 for h, _ in HORIZONS}
    base_skipped = 0
    for r in recs:
        t_ns = int(r["dateline"]) * 10 ** 9
        base = anchor(t_ns, 0)
        if base is None:
            base_skipped += 1
            continue
        row = dict(r)
        row.pop("request_body", None)
        row.pop("response_body", None)
        row["year"] = r["utc_time"][:4]
        row["base"] = base
        for hname, hmin in HORIZONS:
            p = anchor(t_ns + hmin * 60 * 10 ** 9, hmin)
            if p is None:
                skip[hname] += 1
                row[f"d_{hname}"] = np.nan
            else:
                row[f"d_{hname}"] = p - base
        rows.append(row)
    df = pd.DataFrame(rows)
    df.to_csv(SCORED_CSV, index=False)
    n_fail = int((df["llm_direction"].isna()).sum())
    print("═" * 70)
    print("PART A — event gold-direction vs actual XAU move")
    print("═" * 70)
    print(f"events replayed: {len(df)} (baseline-price skipped: "
          f"{base_skipped}) | LLM hard failures: {n_fail}")
    print("horizon skip counts (stale/closed):",
          {k: v for k, v in skip.items() if v})
    dist = df["llm_direction"].value_counts(dropna=False).to_dict()
    print("LLM call distribution:", dist)
    # Unconditional per-horizon noise scale (null): median |Δ| over all
    # 2024+ bar times, position-shifted.
    bars24 = bars[bars.index >= "2024-01-01"]
    null_med = {}
    for hname, hmin in HORIZONS:
        k = hmin // 15
        d = (bars24["Close"].shift(-k) - bars24["Close"]).abs().dropna()
        null_med[hname] = float(d.median())
    print("unconditional null median |Δ| ($):",
          {k: round(v, 2) for k, v in null_med.items()})
    # Strategy call columns: +1 gold up, −1 gold down, 0/NaN excluded.
    df["c_llm"] = df["llm_direction"].map(
        {"bullish": 1, "bearish": -1}).fillna(0)
    df["c_doctrine"] = df["surprise"].map({"positive": -1, "negative": 1})
    df["c_ffbw"] = df["better_worse"].map({1: -1, 2: 1}).fillna(0)
    df["c_bull"] = 1
    df["c_bear"] = -1
    strategies = [("LLM", "c_llm"), ("doctrine(surprise)", "c_doctrine"),
                  ("FF betterWorse", "c_ffbw"), ("always-bullish", "c_bull"),
                  ("always-bearish", "c_bear")]
    for hname, _ in HORIZONS:
        dcol = f"d_{hname}"
        sub = df[df[dcol].notna()]
        print(f"\n── horizon {hname} (n scored bars={len(sub)}) ──")
        zeros = int((sub[dcol] == 0).sum())
        if zeros:
            print(f"  Δ==0 count (scored as miss): {zeros}")
        for label, col in strategies:
            s = sub[sub[col] != 0]
            hits = int((np.sign(s[dcol]) == s[col]).sum())
            print("  " + _fmt_rate(hits, len(s), label))
        # noise-filtered LLM hit rate
        s = sub[(sub["c_llm"] != 0) &
                (sub[dcol].abs() >= null_med[hname])]
        hits = int((np.sign(s[dcol]) == s["c_llm"]).sum())
        print("  " + _fmt_rate(hits, len(s),
                               f"LLM |Δ|≥null-med"))
        # event-window magnitude vs null
        mag = sub[dcol].abs().median()
        print(f"  event-window median |Δ| = ${mag:.2f} "
              f"(null ${null_med[hname]:.2f})")
        # paired McNemar: LLM vs each baseline on common directional set
        for label, col in [("doctrine", "c_doctrine"),
                           ("FF betterWorse", "c_ffbw"),
                           ("always-bullish", "c_bull")]:
            s = sub[(sub["c_llm"] != 0) & (sub[col] != 0)]
            ha = list(np.sign(s[dcol]) == s["c_llm"])
            hb = list(np.sign(s[dcol]) == s[col])
            b, c, p = _mcnemar_p(ha, hb)
            print(f"  McNemar LLM vs {label:<15} b={b:>3} c={c:>3} "
                  f"p={p:.3f}")
    # neutral-call flatness (15m)
    neu = df[(df["llm_direction"] == "neutral") & df["d_15m"].notna()]
    dire = df[(df["c_llm"] != 0) & df["d_15m"].notna()]
    if len(neu):
        print(f"\nneutral calls (n={len(neu)}): median |Δ_15m| "
              f"${neu['d_15m'].abs().median():.2f} vs directional "
              f"(n={len(dire)}) ${dire['d_15m'].abs().median():.2f} "
              f"(null ${null_med['15m']:.2f})")
    # breakdowns at 1h
    dcol = "d_1h"
    sub = df[df[dcol].notna() & (df["c_llm"] != 0)]
    print("\n── LLM hit rate by event name (1h, top 10 by count) ──")
    for name, grp in sorted(sub.groupby("name"), key=lambda kv: -len(kv[1]))[:10]:
        hits = int((np.sign(grp[dcol]) == grp["c_llm"]).sum())
        print("  " + _fmt_rate(hits, len(grp), name[:24]))
    print("── LLM hit rate by year (1h) ──")
    for yr, grp in sub.groupby("year"):
        hits = int((np.sign(grp[dcol]) == grp["c_llm"]).sum())
        print("  " + _fmt_rate(hits, len(grp), yr))
    # cluster view: shared datelines
    print("\n── same-timestamp clusters ──")
    clus = df.groupby("dateline").filter(lambda g: len(g) > 1)
    groups = list(clus.groupby("dateline"))
    consistent = 0
    net_hits, net_n = 0, 0
    for _, g in groups:
        signs = set(g.loc[g["c_llm"] != 0, "c_llm"])
        if len(signs) <= 1:
            consistent += 1
        net = int(g["c_llm"].sum())
        d = g["d_1h"].iloc[0]
        if net != 0 and pd.notna(d) and d != 0:
            net_n += 1
            if np.sign(d) == np.sign(net):
                net_hits += 1
    print(f"clusters: {len(groups)} datelines "
          f"({len(clus)} events, {len(clus) / max(len(df), 1) * 100:.0f}% of "
          f"corpus) | internally consistent: {consistent}/{len(groups)}")
    print("  " + _fmt_rate(net_hits, net_n, "cluster net-call (1h)"))
    # determinism probe
    p2 = {r["ffid"]: r for r in _load_jsonl(EVENTS_JSONL)
          if r.get("run_pass") == 2 and r.get("llm_direction")}
    if p2:
        p1 = {r["ffid"]: r for r in recs}
        agree = same_reason = n = 0
        for fid, r2 in p2.items():
            r1 = p1.get(fid)
            if not r1 or not r1.get("llm_direction"):
                continue
            n += 1
            if r1["llm_direction"] == r2["llm_direction"]:
                agree += 1
            if r1.get("reason_raw") == r2.get("reason_raw"):
                same_reason += 1
        print(f"\ndeterminism probe: direction agreement {agree}/{n}, "
              f"identical reason text {same_reason}/{n}")
    # stratified reason examples
    print("\n── reason samples (1h hit/miss) ──")
    sub = df[df[dcol].notna() & (df["c_llm"] != 0)].copy()
    sub["hit"] = np.sign(sub[dcol]) == sub["c_llm"]
    for verdict in (True, False):
        picked, seen = [], set()
        for _, r in sub[sub["hit"] == verdict].iterrows():
            if r["name"] in seen:
                continue
            seen.add(r["name"])
            picked.append(r)
            if len(picked) == 10:
                break
        tag = "HIT " if verdict else "MISS"
        for r in picked:
            print(f"  [{tag}] {r['utc_time'][:16]} {r['name'][:30]:<30} "
                  f"a={r['actual']} f={r['forecast']} → "
                  f"{r['llm_direction']:<7} Δ1h={r[dcol]:+7.2f}  "
                  f"“{(r['reason_raw'] or '')[:60]}”")
    print(f"\nscored table → {SCORED_CSV}")
 def report_headlines():
    recs = _load_jsonl(HEADLINES_JSONL)
    if not recs:
        print("\n(no headline replay records — run `headlines` first)")
        return
    g = _load_golden()
    engine = NewsEngine(NewsConfig())
    def collect(tag):
        m = {}
        for r in recs:
            if r["run_tag"] == tag and r["scores"]:
                for h, sc in zip(r["headlines"], r["scores"]):
                    m[h] = tuple(sc)
        return m
    main, shuf = collect("main"), collect("shuffled")
    print("\n" + "═" * 70)
    print("PART B — golden headline set: LLM vs keyword baseline")
    print("═" * 70)
    missing = [h for h in g["headline"] if h not in main]
    if missing:
        print(f"WARNING: {len(missing)} headlines missing from main run")
    def cls(x):
        if x >= NEUTRAL_BAND:
            return 1
        if x <= -NEUTRAL_BAND:
            return -1
        return 0
    rows = []
    for _, r in g.iterrows():
        h = r["headline"]
        if h not in main:
            continue
        llm_xau = main[h][1]
        kw = engine._analyze_headline_sentiment(h, for_currency="XAU")
        rows.append({
            "headline": h, "category": r["category"],
            "exp": int(r["expected_xau_sign"]),
            "llm_xau": llm_xau, "llm_cls": cls(llm_xau),
            "kw": kw, "kw_cls": cls(kw),
        })
    t = pd.DataFrame(rows)
    for who in ("llm", "kw"):
        acc = (t[f"{who}_cls"] == t["exp"]).mean()
        print(f"{who.upper():<4} 3-class accuracy (band ±{NEUTRAL_BAND}): "
              f"{(t[f'{who}_cls'] == t['exp']).sum()}/{len(t)} "
              f"= {acc * 100:.1f}%")
    ha = list(t["llm_cls"] == t["exp"])
    hb = list(t["kw_cls"] == t["exp"])
    b, c, p = _mcnemar_p(ha, hb)
    print(f"paired McNemar LLM vs keyword: b={b} c={c} p={p:.3f}")
    print("\nper-category accuracy (LLM | keyword):")
    for cat, grp in t.groupby("category"):
        a1 = (grp["llm_cls"] == grp["exp"]).sum()
        a2 = (grp["kw_cls"] == grp["exp"]).sum()
        print(f"  {cat:<18} {a1:>2}/{len(grp):<2} | {a2:>2}/{len(grp)}")
    print("\nLLM confusion (rows=expected, cols=predicted -1/0/+1):")
    cm = pd.crosstab(t["exp"], t["llm_cls"]).reindex(
        index=[-1, 0, 1], columns=[-1, 0, 1], fill_value=0)
    print(cm.to_string())
    print("\nLLM misclassified headlines:")
    for _, r in t[t["llm_cls"] != t["exp"]].iterrows():
        print(f"  exp {r['exp']:+d} got {r['llm_xau']:+.2f} "
              f"[{r['category']}] {r['headline'][:70]}")
    # repeat-batch determinism
    m_b0 = [r for r in recs if r["run_tag"] == "main" and r["batch"] == 0]
    r_b0 = [r for r in recs if r["run_tag"] == "repeat"]
    if m_b0 and r_b0 and m_b0[0]["scores"] and r_b0[0]["scores"]:
        a = np.array(m_b0[0]["scores"], dtype=float)
        b_ = np.array(r_b0[0]["scores"], dtype=float)
        exact = int((a == b_).all(axis=1).sum())
        print(f"\nrepeat-batch: {exact}/{len(a)} score pairs identical, "
              f"max |diff| = {np.abs(a - b_).max():.3f}")
    # shuffled-order drift
    if shuf:
        both = [h for h in t["headline"] if h in shuf]
        drift = [abs(main[h][1] - shuf[h][1]) for h in both]
        if drift:
            print(f"shuffled-order xau drift over {len(both)} headlines: "
                  f"mean {np.mean(drift):.3f}, max {np.max(drift):.3f}")
 def cmd_report(args):
    report_events()
    report_headlines()
 # ── CLI ──────────────────────────────────────────────────────────────
 def main():
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[1])
    sub = ap.add_subparsers(dest="cmd", required=True)
    e = sub.add_parser("events", help="replay release surprises")
    e.add_argument("--limit", type=int, default=0,
                   help="smoke-test on first N events")
    e.add_argument("--determinism", type=int, default=0,
                   help="re-call N stratified events a second time")
    e.add_argument("--sleep", type=float, default=0.3)
    e.add_argument("--retry-passes", type=int, default=3)
    e.add_argument("--timeout", type=float, default=30.0)
    e.set_defaults(fn=cmd_events)
    h = sub.add_parser("headlines", help="golden-set probes")
    h.add_argument("--sleep", type=float, default=0.3)
    h.add_argument("--timeout", type=float, default=30.0)
    h.set_defaults(fn=cmd_headlines)
    r = sub.add_parser("report", help="offline scoring report")
    r.set_defaults(fn=cmd_report)
    args = ap.parse_args()
    args.fn(args)
 if __name__ == "__main__":
    main()
@@ -0,0 +1,86 @@
 # DeepSeek news-analysis validation — 2026-08-06
 **Verdict: headline sentiment scoring is excellent; event-direction calls are
 directionally sound but ~11% of them misread the numeric comparison, and the
 failure concentrates on Unemployment Claims / close K-M-suffixed values.**
 Method (owner-directed, token-lean): independent expert re-analysis of a
 stratified sample of DeepSeek's recorded answers, not a full statistical
 backtest. Harness: `backtest/research/llm_news_eval.py` (replays releases
 through the REAL `_llm_event_direction` / `_llm_headline_scores` — real
 prompts, payloads, validation; raw wire traffic logged). 135 of the 586-event
 corpus were replayed before the full run was stopped by owner request
 (artifacts retained in `llm_news_eval_2026-08/`, resume-safe).
 ## Part A — event gold-direction (28-event stratified sample, Jan–Jun 2024)
 Reviewer agreement with DeepSeek: **24/28 (86%)**. All 4 disagreements are
 DeepSeek errors, in exactly two failure modes:
 1. **Numeric comparison misread (3/28 ≈ 11%)** — the doctrine logic is right
   but the premise is backwards:
   - JOLTS 8.79M vs f 8.84M (a miss) → DS: "Job openings **above** forecast" → bearish (should be bullish)
   - Claims 202K vs f 217K (fewer claims) → DS: "Claims **higher** than forecast" → bullish (should be bearish)
   - Claims 202K vs f 209K — same backwards read, same wrong call
   Both Unemployment Claims events in the sample were misread the same way —
   looks systematic on that event type, possibly an anchoring prior that
   "claims rose". All misreads involve close values with K/M suffixes.
 2. **Direction field contradicts its own reason (1/28)** — Empire State
   −43.7 vs f −4.9: reason text says "gold may rise", `gold_direction` says
   `bearish`.
 Where the numbers were read correctly, hawkish/dovish→gold mapping was
 consistently correct (CPI/PPI/NFP/ADP/GDP/PMI/wages/unemployment-rate all
 handled properly, including inverted metrics when read right).
 Price cross-check (1h horizon, cluster-confounded, indicative only):
 reviewer calls 22/28 with the subsequent move, DeepSeek 20/28.
 ## Part B — headline sentiment (78-headline golden set)
 DeepSeek matched the reviewer's expected gold-direction label on
 **78/78 (100%)** — including all 8 negation traps ("signals no rate hike",
 "no rate cuts until…") and all 6 inverted-metric phrasings ("claims hit
 two-year high"). The keyword scorer it overrides got **24/78 (31%)**.
 Repeat-batch and shuffled-order probes recorded (headlines_replay.jsonl).
 ## Implications
 - The headline-sentiment override (the main live LLM surface) is validated —
  a large, measured upgrade over keywords.
 - The event-direction enrichment ("— gold bullish/bearish" appended to
  surprise descriptions) is right ~85–90% of the time but can state a
  factually backwards comparison in the displayed reason. It is display-only
  (never gates signals), so exposure is a possibly-misleading Telegram/CLI
  line, not trades.
 - **Hardening SHIPPED same day (owner-approved):** the payload now carries a
  code-computed `vs_forecast` field ("above/below/in line with forecast",
  from `EconomicEvent.surprise`) and the prompt tells the model to trust it.
  First iteration fixed the arithmetic but exposed a masked doctrine error
  (model read "claims below forecast" correctly, then called it a *weaker*
  labor market — and regressed the unemployment-rate case the same way), so
  the prompt's inverted-metrics line was expanded to an explicit rule plus a
  worked claims example. Verification: **18/18** — all 4 original misreads
  fixed, 11/11 no-regression cases unchanged, 3/3 fresh probes on the side
  the prompt example does NOT cover (claims 245K vs f 220K → bullish,
  unemployment 4.1% vs 3.9% → bullish, JOLTS 9.10M vs 8.80M → bearish).
  Suite green at 346 tests, incl. a new payload assertion in Module 19.
 - Production observations found while building the harness (not fixed):
  `_llm_event_cache` key omits forecast → a repeated `actual` re-serves a
  direction computed for a different forecast; a "neutral" reply is
  indistinguishable from a failure in the return value.
 ## Reproduce
 ```
 venv/bin/python backtest/research/llm_news_eval.py events --limit 5   # smoke
 venv/bin/python backtest/research/llm_news_eval.py events --determinism 20
 venv/bin/python backtest/research/llm_news_eval.py headlines
 venv/bin/python backtest/research/llm_news_eval.py report             # offline
 ```
 Corpus: 586 high-impact USD surprises 2024-01→2026-07 (calendar.parquet,
 NFP-verified UTC); prices: Oanda M15 mids. Full-run cost ≈ cents. The
 `report` subcommand (hit rates vs always-bullish/doctrine/FF-betterWorse
 baselines, McNemar, cluster view) is built and untested at scale — run it
 after completing the replay if a statistical answer is ever wanted.
@@ -0,0 +1,79 @@
 headline,category,expected_xau_sign,rationale
 "Powell says restrictive policy needed for longer as inflation proves sticky",hawkish_fed,-1,"higher-for-longer rates lift real yields, gold-bearish"
 "Fed minutes show officials leaning toward two more rate hikes this year",hawkish_fed,-1,"more hikes = stronger dollar/yields"
 "Hot CPI print fuels bets on additional Fed tightening",hawkish_fed,-1,"tightening expectations weigh on gold"
 "Fed's Waller: higher-for-longer rates warranted by resilient economy",hawkish_fed,-1,"hawkish guidance"
 "FOMC raises rates 50bp, flags further increases ahead",hawkish_fed,-1,"outsized hike plus hawkish guidance"
 "Treasury yields spike as Fed pushes back on rate cut hopes",hawkish_fed,-1,"cut hopes dashed, yields up"
 "Core inflation accelerates, cementing case for hawkish Fed stance",hawkish_fed,-1,"hawkish repricing"
 "Fed dot plot shifts higher; markets price out easing for the year",hawkish_fed,-1,"easing priced out"
 "Fed cuts rates 25bp and signals more easing to come",dovish_fed,1,"easing cycle is gold-bullish"
 "Powell hints at rate cuts as inflation cools toward target",dovish_fed,1,"cut guidance"
 "FOMC minutes reveal growing support for policy easing",dovish_fed,1,"dovish minutes"
 "Fed's Williams says rate cuts appropriate later this year",dovish_fed,1,"cut guidance"
 "Soft inflation data boosts odds of a dovish Fed pivot",dovish_fed,1,"pivot odds up"
 "Fed pauses hikes, markets price aggressive easing cycle",dovish_fed,1,"easing priced in"
 "Central banks turn dovish as global growth falters",dovish_fed,1,"global easing backdrop"
 "Fed to restart bond purchases to support flagging economy",dovish_fed,1,"QE is gold-bullish"
 "Dollar index hits 12-month high on robust US economic data",strong_usd,-1,"strong dollar pressures gold"
 "US GDP grows 4.1% annualized, blowing past expectations",strong_usd,-1,"strong economy, hawkish implication"
 "Blockbuster jobs report sends dollar surging across the board",strong_usd,-1,"strong jobs lift dollar and yields"
 "US retail sales smash forecasts, underscoring consumer strength",strong_usd,-1,"strong data"
 "Greenback rallies as US data outshines Europe and Asia",strong_usd,-1,"dollar strength"
 "ISM services index jumps to two-year high on strong demand",strong_usd,-1,"strong data"
 "Consumer confidence surges to post-pandemic high in the US",strong_usd,-1,"strong data"
 "Dollar tumbles to eight-month low after dismal manufacturing data",weak_usd,1,"weak dollar lifts gold"
 "US economy adds far fewer jobs than expected; dollar slides",weak_usd,1,"weak jobs, dovish implication"
 "Greenback weakens as US retail sales unexpectedly contract",weak_usd,1,"weak data"
 "US GDP growth revised down sharply, stoking slowdown fears",weak_usd,1,"slowdown is dovish"
 "Dollar slumps as trade deficit balloons to record",weak_usd,1,"dollar weakness"
 "Soft services PMI adds to evidence of cooling US economy",weak_usd,1,"cooling economy"
 "US consumer sentiment plunges to two-year low",weak_usd,1,"weak data"
 "Real yields climb to multi-year highs as TIPS sell off",real_yields,-1,"rising real yields raise gold's opportunity cost"
 "10-year Treasury yield surges past 5% on supply glut",real_yields,-1,"yield spike"
 "Rising real interest rates dim appeal of non-yielding assets",real_yields,-1,"explicit opportunity-cost story"
 "Treasury yields sink as investors flock to bonds",real_yields,1,"falling yields support gold"
 "Real yields turn negative as inflation outpaces rates",real_yields,1,"negative real yields are gold's best backdrop"
 "Bond rally drives 10-year yield to lowest level in a year",real_yields,1,"falling yields"
 "Geopolitical tensions escalate in Middle East, investors seek safety",safe_haven,1,"safe-haven bid"
 "Stocks crater worldwide as banking crisis fears spread",safe_haven,1,"crisis flight to safety"
 "Escalating conflict drives flight to safe-haven assets",safe_haven,1,"explicit haven flow"
 "Investors dump risk assets after surprise sovereign default",safe_haven,1,"default shock"
 "Military escalation on NATO border rattles global markets",safe_haven,1,"geopolitical stress"
 "Wall Street plunges 4% in worst session since pandemic",safe_haven,1,"risk-off"
 "Currency crisis in emerging markets sparks contagion fears",safe_haven,1,"contagion, haven bid"
 "Nuclear rhetoric intensifies, sending investors to havens",safe_haven,1,"extreme geopolitical stress"
 "Ceasefire agreement lifts global markets, haven demand fades",risk_on,-1,"haven unwind"
 "Stocks roar to record highs as trade deal signed",risk_on,-1,"risk-on drains haven demand"
 "Risk appetite returns as banking fears subside",risk_on,-1,"risk-on"
 "Global equities rally on strong earnings and easing tensions",risk_on,-1,"risk-on"
 "Gold hits record high as central banks keep buying bullion",gold_direct,1,"direct demand"
 "Gold ETFs post biggest weekly inflows in three years",gold_direct,1,"investment inflows"
 "China's central bank extends gold buying spree to 18th month",gold_direct,1,"official-sector demand"
 "Miners struggle to meet gold demand as supply deficit widens",gold_direct,1,"supply deficit"
 "Gold slides 2% as investors book profits after record run",gold_direct,-1,"profit-taking pressure"
 "Gold tumbles below key support as ETF outflows accelerate",gold_direct,-1,"outflows"
 "Bullion demand weakens as Indian jewellery buying stalls",gold_direct,-1,"physical demand soft"
 "Speculators slash bullish gold bets as dollar strengthens",gold_direct,-1,"positioning unwind"
 "Fed holds rates steady, signals no rate hike on the horizon",negation_traps,1,"dovish hold; keyword layer sees the phrase rate hike and flips it gold-bearish"
 "Powell says no rate cuts until inflation shows sustained cooling",negation_traps,-1,"hawkish hold; keyword layer sees rate cut and calls it gold-bullish"
 "Recession fears fade as US data streak beats estimates",negation_traps,-1,"fading fear is risk-on; keywords see recession plus beats and cancel out"
 "Inflation fails to accelerate, easing pressure on the Fed",negation_traps,1,"disinflation opens door to cuts"
 "Fed's Bowman: not yet confident inflation is beaten, cuts premature",negation_traps,-1,"hawkish pushback phrased around negations"
 "Central bank stops short of hiking, but warns tightening may resume",negation_traps,-1,"hawkish hold with tightening bias"
 "No safe-haven bid for gold despite equity wobble, analysts say",negation_traps,-1,"explicitly negated haven demand"
 "Jobs report not as weak as feared, dollar recovers",negation_traps,-1,"less-bad data is dollar-positive"
 "US manufacturing output plunges as new orders dry up",inverted_metrics,1,"weak economy is dovish/gold-bullish; keyword layer reads plunges as bearish"
 "US factory activity slumps to three-year low",inverted_metrics,1,"weak data; keyword reads slumps as bearish"
 "Consumer spending falls for third straight month",inverted_metrics,1,"weak data; keyword reads falls as bearish"
 "Private payrolls growth slides to weakest pace since 2020",inverted_metrics,1,"weak jobs; keyword reads slides as bearish"
 "Jobless claims hit two-year high as layoffs spread",inverted_metrics,1,"higher claims = weaker economy = gold-bullish"
 "US unemployment rate hits 4.5%, highest since 2021",inverted_metrics,1,"higher unemployment = dovish = gold-bullish"
 "Lakers clinch playoff berth with overtime win over Suns",distractors,0,"irrelevant"
 "New Marvel film smashes box office records on opening weekend",distractors,0,"irrelevant"
 "Apple unveils thinner MacBook with faster chip lineup",distractors,0,"irrelevant"
 "Scientists discover new deep-sea species off Pacific coast",distractors,0,"irrelevant"
 "Heavy rain forecast for the Midwest through the weekend",distractors,0,"irrelevant"
 "Streaming service raises subscription prices in Europe",distractors,0,"irrelevant"
 "Electric vehicle startup recalls 12,000 cars over software glitch",distractors,0,"irrelevant"
 "Local elections see record turnout in several US states",distractors,0,"irrelevant"
@@ -24,7 +24,8 @@ import datetime as _dt
 import json
 import logging
 import threading
-from typing import Any, Dict, Optional
+import time
 from typing import Any, Dict, Optional, Tuple
 import requests
@@ -66,6 +67,10 @@ class DeepSeekClient:
        self._count_day: Optional[_dt.date] = None
        self._count = 0
        self._warned: set = set()
        # Health stamps for status displays: last successful call and
        # last failure (ts, kind). Read-only for consumers.
        self.last_ok: float = 0.0
        self.last_error: Optional[Tuple[float, str]] = None
    @property
    def available(self) -> bool:
@@ -125,9 +130,11 @@ class DeepSeekClient:
            resp = self._session.post(
                DEEPSEEK_URL, json=body, timeout=self._timeout)
        except requests.RequestException as e:
            self.last_error = (time.time(), "network")
            self._warn_once("network", "DeepSeek call failed: %s", e)
            return None
        if resp.status_code != 200:
            self.last_error = (time.time(), f"http_{resp.status_code}")
            self._warn_once(
                f"http_{resp.status_code}",
                "DeepSeek HTTP %s: %s", resp.status_code, resp.text[:200])
@@ -137,8 +144,11 @@ class DeepSeekClient:
        except ValueError:
            parsed = None
        if parsed is None:
            self.last_error = (time.time(), "parse")
            self._warn_once("parse", "DeepSeek response was not the "
                            "expected JSON object — ignoring")
        else:
            self.last_ok = time.time()
        return parsed
    @staticmethod
@@ -57,12 +57,22 @@ Examples:
 _LLM_EVENT_SYSTEM = """\
 You are a forex/gold analyst. You receive one released economic event as
-JSON: {"event", "currency", "actual", "forecast", "previous"}.
+JSON: {"event", "currency", "actual", "forecast", "previous",
-Decide the likely effect of this RELEASE (actual vs forecast) on the
+"vs_forecast"}. "vs_forecast" states whether the actual print came in
-GOLD (XAU/USD) price. A stronger-than-forecast USD economy or a hawkish
+above or below the forecast; it is computed programmatically — trust it
-implication is usually gold-bearish; weaker data or a dovish implication
+over your own comparison of the numbers. If it is "unknown", compare
-is usually gold-bullish. Mind inverted metrics: higher unemployment or
+actual and forecast yourself.
-jobless claims = weaker economy.
+Decide the likely effect of this RELEASE on the GOLD (XAU/USD) price.
 A stronger-than-forecast USD economy or a hawkish implication is usually
 gold-bearish; weaker data or a dovish implication is usually
 gold-bullish.
 Inverted metrics — for unemployment rate and unemployment/jobless
 claims a LOWER number means a STRONGER labor market: below forecast =
 strong economy = usually gold-bearish; above forecast = weak economy =
 usually gold-bullish.
 Example: {"event": "Unemployment Claims", "actual": "202K", "forecast":
 "217K", "vs_forecast": "below forecast"} ->
 {"gold_direction": "bearish", "reason": "fewer claims show strong labor market"}
 Reply with ONLY JSON:
 {"gold_direction": "bullish"|"bearish"|"neutral", "reason": "<max 10 words>"}
 """
@@ -250,6 +260,14 @@ class NewsEngine:
    _LLM_SENT_CACHE_MAX = 400
    _LLM_EVENT_CACHE_MAX = 100
    # Per-source fetch health for status displays (telegram /status).
    # name → {"ok": ts|None, "err": ts|None, "detail": str, "extra": str}.
    # All writers run under _cache_lock (fetches happen inside
    # _cached_fetch); readers render without the lock — worst case a
    # display line is one cycle stale. A source is "failing" when its
    # last error is newer than its last success.
    _source_health: Dict[str, dict] = {}
    def __init__(self, config: NewsConfig):
        self.config = config
        self._session = requests.Session()
@@ -453,6 +471,36 @@ class NewsEngine:
        "canada": "CAD", "new zealand": "NZD",
    }
    @classmethod
    def _record_source(cls, name: str, ok: bool, detail: str = "",
                       extra: str = "") -> None:
        """Track one dependency's fetch outcome for status displays.
        Logs state TRANSITIONS so the log file carries the story without
        a line every 10-min cycle: first failure WARNING (with detail),
        repeat failures DEBUG, recovery INFO."""
        h = cls._source_health.setdefault(
            name, {"ok": None, "err": None, "detail": "", "extra": ""})
        was_failing = h["err"] is not None and (
            h["ok"] is None or h["err"] > h["ok"])
        if ok:
            h["ok"] = time.time()
            h["extra"] = extra
            if was_failing:
                logger.info("Source recovered: %s", name)
        else:
            h["err"] = time.time()
            h["detail"] = detail
            if was_failing:
                logger.debug("Source still failing: %s (%s)", name, detail)
            else:
                logger.warning("Source failing: %s (%s)", name, detail)
    @staticmethod
    def source_is_failing(entry: Optional[dict]) -> bool:
        """True when the entry's last error is newer than its last ok."""
        return bool(entry and entry["err"] is not None and
                    (entry["ok"] is None or entry["err"] > entry["ok"]))
    def _cached_fetch(self, cache_attr: str, fetch_fn) -> list:
        """Return cached data for `cache_attr` or refetch when the TTL
        expired. Successful results live for context_cache_ttl_seconds;
@@ -485,14 +533,27 @@ class NewsEngine:
        Returns an empty list (with a log warning) if unavailable —
        we never generate fake events since that causes constant false alarms.
        """
-        for fetcher in (self._try_fetch_calendar_api, self._try_fetch_myfxbook_rss):
+        sources = (
            (self._try_fetch_calendar_api, "calendar:forexfactory",
             "ForexFactory"),
            (self._try_fetch_myfxbook_rss, "calendar:myfxbook", "MyFXBook"),
        )
        for fetcher, src, label in sources:
            try:
                events = fetcher()
                if events:
                    self._record_source(src, True)
                    self._record_source(
                        "calendar", True,
                        extra=f"{len(events)} events via {label}")
                    return events
                self._record_source(src, False, "no events returned")
            except Exception as e:
                self._record_source(
                    src, False, f"{type(e).__name__}: {e}")
                logger.debug(f"Calendar fetch exception ({fetcher.__name__}): {e}")
        self._record_source("calendar", False, "all sources failed")
        logger.info(
            "Economic calendar data unavailable. "
            "Event risk checks will be skipped for this cycle."
@@ -639,11 +700,16 @@ class NewsEngine:
        """Fetch and analyze news headlines for sentiment."""
        headlines = []
        seen_titles = set()
        ok_feeds = 0
        for feed_url in self.config.rss_feeds:
            host = feed_url.split("/")[2] if "//" in feed_url else feed_url
            kept_before = len(headlines)
            try:
                resp = self._session.get(feed_url, timeout=10)
                if resp.status_code != 200:
                    self._record_source(
                        f"rss:{host}", False, f"HTTP {resp.status_code}")
                    continue
                root = ET.fromstring(resp.content)
@@ -689,9 +755,21 @@ class NewsEngine:
                            },
                        ))
                ok_feeds += 1
                self._record_source(
                    f"rss:{host}", True,
                    extra=f"{len(headlines) - kept_before} kept")
            except Exception as e:
                self._record_source(
                    f"rss:{host}", False, f"{type(e).__name__}: {e}")
                logger.debug(f"Failed to fetch RSS from {feed_url}: {e}")
        self._record_source(
            "headlines", ok_feeds > 0,
            detail="all feeds failed" if ok_feeds == 0 else "",
            extra=(f"{len(headlines)} headlines from "
                   f"{ok_feeds}/{len(self.config.rss_feeds)} feeds"))
        try:
            self._apply_llm_sentiment(headlines)
        except Exception as e:
@@ -802,7 +880,13 @@ class NewsEngine:
    def _llm_event_direction(self, event: EconomicEvent) -> Optional[str]:
        """Gold direction for a released surprise, e.g. "bearish: strong
        jobs lift yields". None → caller keeps the numeric description.
-        Cached per release; failures are NOT cached (retry next scan)."""
+        Cached per release; failures are NOT cached (retry next scan).
        The actual-vs-forecast comparison is precomputed and sent as
        "vs_forecast": the 2026-08-06 validation found the model reads
        close K/M-suffixed pairs backwards ~11% of the time (e.g. claims
        202K vs 217K called "higher than forecast") when left to compare
        the raw numbers itself."""
        key = (event.event, event.currency, str(event.actual or ""))
        with NewsEngine._cache_lock:
            if key in NewsEngine._llm_event_cache:
@@ -810,12 +894,18 @@ class NewsEngine:
        client = get_llm_client()
        if not client.available:
            return None
        vs_forecast = {
            "positive": "above forecast",
            "negative": "below forecast",
            "inline": "in line with forecast",
        }.get(event.surprise or "", "unknown")
        result = client.chat_json(_LLM_EVENT_SYSTEM, {
            "event": event.event,
            "currency": event.currency,
            "actual": event.actual,
            "forecast": event.forecast,
            "previous": event.previous,
            "vs_forecast": vs_forecast,
        })
        if not result:
            return None
@@ -71,6 +71,7 @@ from app_config import (
    telegram_channel_id,
 )
 from providers import BaseDataProvider, make_data_provider
 from llm_client import get_llm_client
 from news_engine import FundamentalContext, NewsEngine
 from pair_signals import PAIR_PRODUCERS
 from signal_aggregator import SignalAggregator, SignalReport
@@ -1114,6 +1115,87 @@ async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(_HELP_TEXT, parse_mode=ParseMode.MARKDOWN)
 def _fmt_age(ts: Optional[float]) -> str:
    """Compact age like '45s', '12m', '3.2h' for status lines."""
    if not ts:
        return "?"
    s = time.time() - ts
    if s < 90:
        return f"{s:.0f}s"
    if s < 5400:
        return f"{s / 60:.0f}m"
    return f"{s / 3600:.1f}h"
 def _news_health_lines() -> List[str]:
    """News / calendar / DeepSeek source health for /status.
    Reads the class-level `NewsEngine._source_health` written by the
    scan-path fetches plus the LLM client's health stamps — display
    only, never triggers a fetch. ⚠️ marks a degraded dependency;
    details live in the rotating log (transition-logged WARNINGs)."""
    lines = ["\n*News & AI:*"]
    # Snapshot: a scan thread may insert a source's FIRST entry while we
    # iterate (dict-changed-size RuntimeError); entry mutation is safe.
    for _ in range(3):
        try:
            sh = dict(NewsEngine._source_health)
            break
        except RuntimeError:
            continue
    else:
        sh = {}
    if not sh:
        lines.append("  _no news cycle yet (fills after first scan)_")
    else:
        cal = sh.get("calendar")
        ff = sh.get("calendar:forexfactory")
        if cal:
            if NewsEngine.source_is_failing(cal):
                lines.append(f"  ⚠️ Calendar: unavailable — {cal['detail']}"
                             f" ({_fmt_age(cal['err'])})")
            elif NewsEngine.source_is_failing(ff):
                lines.append(f"  ⚠️ Calendar: {cal['extra']} — ForexFactory"
                             f" down: {ff['detail']} ({_fmt_age(ff['err'])})")
            else:
                lines.append(f"  ✅ Calendar: {cal['extra']}"
                             f" ({_fmt_age(cal['ok'])} ago)")
        heads = sh.get("headlines")
        if heads:
            failing = NewsEngine.source_is_failing(heads)
            mark = "⚠️" if failing else "✅"
            detail = f" — {heads['detail']}" if failing else ""
            ts = heads["err"] if failing else heads["ok"]
            lines.append(f"  {mark} Headlines: {heads['extra']}{detail}"
                         f" ({_fmt_age(ts)} ago)")
        for name, entry in sorted(sh.items()):
            if name.startswith("rss:") and NewsEngine.source_is_failing(entry):
                lines.append(f"  ⚠️ {name[4:]}: {entry['detail']}"
                             f" ({_fmt_age(entry['err'])})")
    client = get_llm_client()
    if not client.available:
        lines.append("  DeepSeek: off — keyword scoring only")
    else:
        # The client's counter only resets on the first call of a new
        # UTC day — show 0 (not yesterday's total, or a phantom
        # cap-exhausted warning) until that reset happens.
        today = datetime.now(timezone.utc).date()
        used = client._count if client._count_day == today else 0
        cap = client._daily_cap
        scored = len(NewsEngine._llm_sent_cache)
        err = client.last_error
        if used >= cap:
            lines.append(f"  ⚠️ DeepSeek: daily cap {cap} exhausted —"
                         " keyword fallback until UTC midnight")
        elif err and err[0] > client.last_ok:
            lines.append(f"  ⚠️ DeepSeek: last call failed ({err[1]},"
                         f" {_fmt_age(err[0])}) — keyword fallback")
        else:
            lines.append(f"  ✅ DeepSeek: {used}/{cap} calls today,"
                         f" {scored} headlines scored")
    return lines
 def _status_text() -> str:
    lines = ["📊 *ForexBot2 Status*\n"]
@@ -1172,6 +1254,8 @@ def _status_text() -> str:
    if active_breach:
        lines.append(f"*Breach tasks:* {active_breach} running")
    lines.extend(_news_health_lines())
    lines.append(f"\n*Alert quality:* {_alert_min_quality}+ "
                 f"| re-arm {_ALERT_REARM_SECONDS // 60}m")
    return "\n".join(lines)
@@ -4829,6 +4829,8 @@ def _():
 print("\n═══ Module 19: DeepSeek news enrichment ═══")
 import datetime as _dt_m19
 import logging
 import requests
 import llm_client as lc
 from llm_client import DeepSeekClient
 from news_engine import (EconomicEvent, FundamentalContext, NewsEngine,
@@ -4906,10 +4908,12 @@ class _StubLLM:
    def __init__(self, reply):
        self.reply = reply
        self.calls = 0
        self.last_payload = None
    available = True
    def chat_json(self, system_prompt, user_payload):
        self.calls += 1
        self.last_payload = user_payload
        return self.reply
@@ -5029,6 +5033,10 @@ def _():
        risk, desc = ne.get_event_risk_level("XAU/USD", ctx)
        assert risk == "caution"
        assert desc.endswith("— gold bearish: strong jobs lift yields"), desc
        # payload carries the precomputed comparison (300K > 200K) so the
        # model never re-compares the raw numbers (2026-08-06 validation)
        assert lc._client.last_payload["vs_forecast"] == "above forecast", \
            lc._client.last_payload
        # bogus direction → numeric fallback
        NewsEngine._llm_event_cache = {}
        lc._client = _StubLLM({"gold_direction": "sideways"})
@@ -5094,6 +5102,171 @@ def _():
    assert cfg.sentiment_affects_signals is False
@test("llm_client: health stamps track last success and failure")
 def _():
    client = DeepSeekClient(api_key="k", daily_cap=10)
    assert client.last_ok == 0.0 and client.last_error is None
    client._session = _counting_session(_chat_body({"a": 1}))
    assert client.chat_json("sys", {}) == {"a": 1}
    assert client.last_ok > 0 and client.last_error is None
    client._session = _counting_session({"detail": "quota"}, status=429)
    assert client.chat_json("sys", {}) is None
    assert client.last_error is not None
    ts, kind = client.last_error
    assert kind == "http_429" and ts >= client.last_ok
 class _LogCapture(logging.Handler):
    def __init__(self):
        super().__init__()
        self.records = []
    def emit(self, record):
        self.records.append(record)
@test("NewsEngine: _record_source logs transitions only, tracks state")
 def _():
    import news_engine as ne_mod
    cap = _LogCapture()
    ne_mod.logger.addHandler(cap)
    old_level = ne_mod.logger.level
    ne_mod.logger.setLevel(logging.INFO)
    saved = NewsEngine._source_health
    NewsEngine._source_health = {}
    try:
        NewsEngine._record_source("t:src", False, "HTTP 503")
        warns = [r for r in cap.records if r.levelno == logging.WARNING]
        assert len(warns) == 1 and "HTTP 503" in warns[0].getMessage()
        assert NewsEngine.source_is_failing(
            NewsEngine._source_health["t:src"])
        # repeat failure must NOT warn again (DEBUG only)
        NewsEngine._record_source("t:src", False, "HTTP 503")
        warns = [r for r in cap.records if r.levelno == logging.WARNING]
        assert len(warns) == 1
        # recovery logs INFO and clears the failing state
        NewsEngine._record_source("t:src", True, extra="42 kept")
        infos = [r for r in cap.records if r.levelno == logging.INFO]
        assert any("recovered" in r.getMessage() for r in infos)
        entry = NewsEngine._source_health["t:src"]
        assert not NewsEngine.source_is_failing(entry)
        assert entry["extra"] == "42 kept"
    finally:
        ne_mod.logger.removeHandler(cap)
        ne_mod.logger.setLevel(old_level)
        NewsEngine._source_health = saved
@test("NewsEngine: calendar fallback source is visible in health state")
 def _():
    ne = NewsEngine(NewsConfig())
    saved = NewsEngine._source_health
    NewsEngine._source_health = {}
    now = _dt_m19.datetime.now(_dt_m19.timezone.utc)
    ev = EconomicEvent(time=now, currency="USD", impact="high", event="X")
    try:
        ne._try_fetch_calendar_api = lambda: []
        ne._try_fetch_myfxbook_rss = lambda: [ev]
        assert ne._fetch_calendar_raw() == [ev]
        sh = NewsEngine._source_health
        assert NewsEngine.source_is_failing(sh["calendar:forexfactory"])
        assert not NewsEngine.source_is_failing(sh["calendar"])
        assert "MyFXBook" in sh["calendar"]["extra"]
        # both dead → calendar itself failing
        ne._try_fetch_myfxbook_rss = lambda: []
        assert ne._fetch_calendar_raw() == []
        assert NewsEngine.source_is_failing(sh["calendar"])
    finally:
        NewsEngine._source_health = saved
@test("NewsEngine: per-feed RSS health recorded, aggregate survives")
 def _():
    cfg = NewsConfig()
    cfg.rss_feeds = ["https://good.example/rss", "https://bad.example/rss"]
    ne = NewsEngine(cfg)
    saved = NewsEngine._source_health
    NewsEngine._source_health = {}
    rss = (b"<rss><channel><item>"
           b"<title>Fed signals rate cuts as dollar slides</title>"
           b"</item></channel></rss>")
    class _Sess:
        def get(self, url, timeout=None):
            if "bad.example" in url:
                raise requests.ConnectionError("refused")
            return type("R", (), {"status_code": 200, "content": rss})()
    ne._session = _Sess()
    try:
        heads = ne._fetch_news_sentiment_raw()
        assert len(heads) == 1, heads
        sh = NewsEngine._source_health
        assert not NewsEngine.source_is_failing(sh["rss:good.example"])
        assert NewsEngine.source_is_failing(sh["rss:bad.example"])
        assert "refused" in sh["rss:bad.example"]["detail"]
        agg = sh["headlines"]
        assert not NewsEngine.source_is_failing(agg)
        assert "1/2 feeds" in agg["extra"], agg
    finally:
        NewsEngine._source_health = saved
@test("/status renders news/calendar/AI health with ⚠️ degradations")
 def _():
    import telegram_bot as tb
    saved = NewsEngine._source_health
    t0 = time.time()
    NewsEngine._source_health = {
        "calendar": {"ok": t0, "err": None, "detail": "",
                     "extra": "42 events via MyFXBook"},
        "calendar:forexfactory": {"ok": None, "err": t0,
                                  "detail": "HTTP 503", "extra": ""},
        "headlines": {"ok": t0, "err": None, "detail": "",
                      "extra": "38 headlines from 3/4 feeds"},
        "rss:fxstreet.com": {"ok": None, "err": t0,
                             "detail": "timeout", "extra": ""},
    }
    stub = _StubLLM(None)
    stub.last_ok = 0.0
    stub.last_error = (t0, "http_429")
    stub._count = 87
    stub._daily_cap = 500
    stub._count_day = _dt_m19.datetime.now(_dt_m19.timezone.utc).date()
    lc._client = stub
    try:
        text = tb._status_text()
        assert "⚠️ Calendar: 42 events via MyFXBook — ForexFactory" in text
        assert "HTTP 503" in text
        assert "38 headlines from 3/4 feeds" in text
        assert "⚠️ fxstreet.com: timeout" in text
        assert "⚠️ DeepSeek: last call failed (http_429" in text
        # healthy path: FF serving, DeepSeek clean → no warnings
        NewsEngine._source_health = {
            "calendar": {"ok": t0, "err": None, "detail": "",
                         "extra": "42 events via ForexFactory"},
            "headlines": {"ok": t0, "err": None, "detail": "",
                          "extra": "38 headlines from 4/4 feeds"},
        }
        stub.last_ok = t0
        text = tb._status_text()
        assert "✅ Calendar: 42 events via ForexFactory" in text
        assert "✅ DeepSeek: 87/500 calls today" in text
        assert "⚠️" not in text.split("*News & AI:*")[1].split("*Alert")[0]
        # empty state (pre-first-scan) renders the neutral line
        NewsEngine._source_health = {}
        assert "no news cycle yet" in tb._status_text()
        # yesterday's exhausted counter must NOT show as cap-exhausted
        # after the UTC date roll (counter resets lazily on next call)
        stub._count = 500
        stub._count_day = stub._count_day - _dt_m19.timedelta(days=1)
        text = tb._status_text()
        assert "cap" not in text and "0/500" in text, text
    finally:
        NewsEngine._source_health = saved
        lc._client = None
 # ══════════════════════════════════════════════════════════════════════
 #  MODULE 20: dashboard_publisher.py (static web dashboard)
 # ══════════════════════════════════════════════════════════════════════
@@ -6456,6 +6629,92 @@ def _():
        os.environ.pop("FB2_EMA_LO", None)
 # ══════════════════════════════════════════════════════════════════════
 #  MODULE 22: fixed-RR TP research flag (--fixed-rr-tp)
 # ══════════════════════════════════════════════════════════════════════
 print("\n═══ Module 22: fixed-RR TP research flag ═══")
@test("fixed-RR TP: tp1 = entry ± mult×risk (BUY+SELL), tp2=tp1, trail zeroed")
 def _():
    import gpu_backtest as _gb
    buy = _m21_sim_trade()                                       # 2400 / sl 2390
    buy.trail_arm_r, buy.trail_dist_r = 0.8, 0.75                # PB-style stamp
    sell = _m21_sim_trade(direction="SELL", entry=2400.0, sl=2410.0, tp1=2390.0)
    _gb._apply_fixed_rr_tp([buy, sell], 2.0)
    assert abs(buy.tp1 - 2420.0) < 1e-9, f"BUY tp1 must be 2420, got {buy.tp1}"
    assert abs(sell.tp1 - 2380.0) < 1e-9, f"SELL tp1 must be 2380, got {sell.tp1}"
    for t in (buy, sell):
        assert t.tp2 == t.tp1, "tp2 must mirror the hard target"
        assert t.trail_arm_r == 0.0 and t.trail_dist_r == 0.0, \
            "per-trade trail params must be zeroed"
        assert t.rr_planned == 2.0
        assert abs(t.reward_pips - 200.0) < 1e-9   # 2 × $10 risk / 0.1 pip
@test("fixed-RR TP: 2R target fills on the static path, spread charged")
 def _():
    import gpu_backtest as _gb
    t = _m21_sim_trade()
    _gb._apply_fixed_rr_tp([t], 2.0)
    sim = _m21_sim_df([(2400, 2405, 2395, 2404), (2404, 2421, 2402, 2418)])
    _gb.simulate_trades([t], sim, max_bars=5, spread=0.30, trail_arm_r=0.0)
    assert t.exit_reason == "TP1", f"got {t.exit_reason}"
    assert abs(t.exit_price - 2420.0) < 1e-9
    assert abs(t.r_multiple - (20.0 - 0.30) / 10.0) < 1e-9, \
        "R must be (2R target - round-trip spread) / risk"
@test("fixed-RR TP: same-bar SL+TP tie still resolves SL-first (rule 6)")
 def _():
    import gpu_backtest as _gb
    t = _m21_sim_trade()
    _gb._apply_fixed_rr_tp([t], 2.0)
    sim = _m21_sim_df([(2400, 2421, 2389, 2400)])   # both barriers inside bar 0
    _gb.simulate_trades([t], sim, max_bars=5, spread=0.0, trail_arm_r=0.0)
    assert t.exit_reason == "SL", f"tie must stay SL-first, got {t.exit_reason}"
    assert abs(t.r_multiple - (-1.0)) < 1e-9
@test("fixed-RR TP: per-trade PB trail params cannot resurrect the trail")
 def _():
    import gpu_backtest as _gb
    t = _m21_sim_trade()
    t.trail_arm_r, t.trail_dist_r = 0.8, 0.75       # XAU_PB per-trade override
    _gb._apply_fixed_rr_tp([t], 2.0)
    # bar0 runs to +1.2R (a live 0.8R-arm trail would ratchet to 2404.5),
    # bar1 retraces through that would-be trail stop, bar2 reaches 2R.
    sim = _m21_sim_df([(2400, 2412, 2398, 2410),
                       (2410, 2411, 2402, 2403),
                       (2403, 2421, 2402, 2420)])
    _gb.simulate_trades([t], sim, max_bars=5, spread=0.0, trail_arm_r=0.0)
    assert t.exit_reason == "TP1", \
        f"trail must be fully disabled under the flag, got {t.exit_reason}"
    assert abs(t.exit_price - 2420.0) < 1e-9 and t.bars_held == 3
    assert abs(t.r_multiple - 2.0) < 1e-9
@test("fixed-RR TP: CLI parses the flag; trail-override combinations refused")
 def _():
    import gpu_backtest as _gb
    assert _gb._parse_args(
        ["--pair", "XAUUSD", "--fixed-rr-tp", "2.0"]).fixed_rr_tp == 2.0
    assert _gb._parse_args(["--pair", "XAUUSD"]).fixed_rr_tp is None
    for bad in (["--trail-arm", "0.5"], ["--trail-dist", "0.8"],
                ["--set", "signals.trail_arm_r=0.5"],
                ["--set", "signals.pb_trail_dist_r=0.5"]):
        try:
            _gb.main(["--pair", "XAUUSD", "--fixed-rr-tp", "2.0"] + bad)
            assert False, f"conflict {bad} must raise SystemExit"
        except SystemExit as e:
            assert "conflicts" in str(e)
    try:
        _gb.main(["--pair", "XAUUSD", "--fixed-rr-tp", "-1"])
        assert False, "non-positive multiple must raise SystemExit"
    except SystemExit as e:
        assert "must be > 0" in str(e)
 # ══════════════════════════════════════════════════════════════════════
 #  SUMMARY
 # ══════════════════════════════════════════════════════════════════════