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
This commit is contained in:
Harwinder Sekhon
@@ -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()
|
||||
Reference in New Issue
Block a user