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.

Claude-Session: https://claude.ai/code/session_017mbXs9HRawZhD1uZ3ze7zQ

.gitignore

@@ -36,3 +36,4 @@ backtest/__pycache__/
 .env.*
 logs/
 backtest/research/commit_ladder_2026-08/
+backtest/research/llm_news_eval_2026-08/

backtest/research/llm_news_eval.py

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

backtest/research/llm_news_eval_2026-08-06.md

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

backtest/research/llm_news_golden.csv

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

llm_client.py

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

news_engine.py

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

telegram_bot.py

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

test_all.py

@@ -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
 # ══════════════════════════════════════════════════════════════════════