#!/usr/bin/env python3 """ Live shadow run: fetch today's headlines through the real ForexBot2 news engine, score every one with DeepSeek (live API) and with the scikit model, and measure how often they agree. Re-run it on later days: rows accumulate in results/live_shadow.jsonl (deduplicated by headline hash) and the report always covers the whole log, so the out-of-sample sample grows with time. This is the "shadow mode" the ML-NEWS WORKFLOW doc asks for before any routing goes live. What is real here * The fetch is NewsEngine._fetch_news_sentiment_raw: same four feeds, same age window (overridable with --max-age-hours to get a bigger sample than the live 4 h), same dedup, same currency filter, same description cleaning. The engine's own DeepSeek call and the archive write are switched off for the fetch so this run neither spends inside the engine nor pollutes logs/newsdata.db. * DeepSeek is then called through the same _llm_headline_scores path in batches of 25, once per headline ever (disk cache). * The model sees headline + description (the shipped recipe) and, as a second column, the headline alone. Truth here is DeepSeek, so this measures how well the student copies the teacher on headlines it never trained on ("not in training" = hash absent from the archive's DeepSeek-XAU rows). Independent truth is eval_golden.py. Usage: venv/bin/python eval_live.py --max-age-hours 24 venv/bin/python eval_live.py --report-only """ from __future__ import annotations import argparse import json import statistics from pathlib import Path from typing import Any, Dict, List, Optional from eval_common import (DEFAULT_DB, LABEL_THRESHOLD_TRAIN, RESULTS_DIR, DeepSeekScorer, ModelScorer, agreement_block, fmt_pct, headline_sha256, hr, load_archive_rows, print_agreement, to_class, ts_tag, utc_now_iso, write_json) SHADOW_LOG = RESULTS_DIR / "live_shadow.jsonl" def fetch_headlines(max_age_hours: float): """Run the real RSS fetch with the engine's LLM step and the archive hook disabled. Returns the NewsSentiment list (keyword-scored).""" import news_engine from config import NewsConfig cfg = NewsConfig(headline_max_age_seconds=int(max_age_hours * 3600)) engine = news_engine.NewsEngine(cfg) engine._apply_llm_sentiment = lambda headlines: None # instance-level saved = news_engine.news_archive.archive_headlines news_engine.news_archive.archive_headlines = lambda *a, **k: None try: return engine, engine._fetch_news_sentiment_raw() finally: news_engine.news_archive.archive_headlines = saved def load_log() -> List[Dict[str, Any]]: if not SHADOW_LOG.is_file(): return [] out = [] with SHADOW_LOG.open(encoding="utf-8") as fh: for line in fh: line = line.strip() if line: out.append(json.loads(line)) return out def report(log: List[Dict[str, Any]], thr: float) -> Dict[str, Any]: rows = [r for r in log if r.get("ds_xau") is not None] hr(f"LIVE SHADOW LOG — {len(rows)} DeepSeek-scored headlines over " f"{len({r['run_ts'][:10] for r in rows})} run day(s); labels at ±{thr}") if not rows: print(" nothing to report yet") return {} def block(sub: List[Dict[str, Any]], pred_key: str, score_key: str) -> Dict[str, Any]: return agreement_block([to_class(r["ds_xau"], thr) for r in sub], [to_class(r[score_key], thr) if pred_key is None else r[pred_key] for r in sub], [r["ds_xau"] for r in sub], [r[score_key] for r in sub]) subsets = { "all headlines": rows, "mentions gold (XAU keyword)": [r for r in rows if r["mentions_gold"]], "not in training set": [r for r in rows if not r["in_training"]], "not in training & mentions gold": [r for r in rows if not r["in_training"] and r["mentions_gold"]], } out: Dict[str, Any] = {"n": len(rows), "subsets": {}} for name, sub in subsets.items(): print(f"\n [{name}] n={len(sub)}") if not sub: continue b_full = block(sub, "model_cls", "model_score") b_head = block(sub, "model_h_cls", "model_h_score") b_kw = agreement_block([to_class(r["ds_xau"], thr) for r in sub], [to_class(r["kw_xau"], thr) for r in sub]) print_agreement("model (headline+description) vs DeepSeek", b_full) print_agreement("model (headline only) vs DeepSeek", b_head) print_agreement("keyword fallback vs DeepSeek", b_kw) out["subsets"][name] = {"n": len(sub), "model_full": b_full, "model_head": b_head, "keyword": b_kw} # DeepSeek's own stability on headlines it scored before (archive) same = [r for r in rows if r.get("archived_xau") is not None] if same: b = agreement_block([to_class(r["archived_xau"], thr) for r in same], [to_class(r["ds_xau"], thr) for r in same], [r["archived_xau"] for r in same], [r["ds_xau"] for r in same]) exact = sum(1 for r in same if abs(r["archived_xau"] - r["ds_xau"]) < 1e-9) print(f"\n [DeepSeek today vs DeepSeek archived, same headline] n={len(same)}") print_agreement("teacher self-agreement (xau)", b) print(f" identical value {fmt_pct(exact, len(same))}") out["deepseek_self_agreement"] = {**b, "identical": exact} # What the live platform displays for USD-only headlines is -overall, # not DeepSeek's own xau (news_engine._gold_direction_label). Side note. usd_only = [r for r in rows if not r["mentions_gold"] and "USD" in r["currencies"]] if usd_only: b = agreement_block([to_class(r["ds_xau"], thr) for r in usd_only], [to_class(-r["ds_s"], thr) for r in usd_only]) print(f"\n [side note] USD-only headlines n={len(usd_only)}: the live gold " f"reading (-overall) matches DeepSeek's own xau on " f"{fmt_pct(b['hits'], b['n'])} (sign {fmt_pct(b['sign_hits'], b['sign_n'])})") out["usd_only_display_vs_xau"] = b hr("Largest disagreements (model headline+description vs DeepSeek xau)") worst = sorted(rows, key=lambda r: -abs(r["model_score"] - r["ds_xau"])) shown = 0 for r in worst: if to_class(r["model_score"], thr) == to_class(r["ds_xau"], thr): continue print(f" DS {r['ds_xau']:+.2f} model {r['model_score']:+.2f} c={r['model_conf']:.2f} " f"{'gold ' if r['mentions_gold'] else ' '}" f"{'TRAIN ' if r['in_training'] else ' '}{r['headline'][:78]}") shown += 1 if shown >= 15: break hr("Latency") m_ms = [r["model_ms"] for r in rows if r.get("model_ms") is not None] d_ms = [r["ds_batch_ms"] / r["ds_batch_size"] for r in rows if r.get("ds_batch_ms") and r.get("ds_batch_size")] if m_ms: print(f" model: mean {statistics.fmean(m_ms):.2f} ms, median " f"{statistics.median(m_ms):.2f} ms per headline") if d_ms: print(f" DeepSeek: ≈ {statistics.fmean(d_ms):.0f} ms per headline inside " f"batches of ≤25 (mean batch " f"{statistics.fmean(r['ds_batch_ms'] for r in rows if r.get('ds_batch_ms')):.0f} ms)") out["latency"] = {"model_ms_mean": statistics.fmean(m_ms) if m_ms else None, "deepseek_ms_per_headline": statistics.fmean(d_ms) if d_ms else None} return out def main() -> int: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("--max-age-hours", type=float, default=24.0, help="RSS age window for the fetch (live uses 4 h)") ap.add_argument("--threshold", type=float, default=LABEL_THRESHOLD_TRAIN) ap.add_argument("--db", type=Path, default=DEFAULT_DB, help="archive used to flag training-set membership") ap.add_argument("--timeout", type=float, default=30.0) ap.add_argument("--no-network", action="store_true", help="no DeepSeek calls (cache only); the RSS fetch still runs") ap.add_argument("--report-only", action="store_true", help="skip the fetch; report the accumulated shadow log") args = ap.parse_args() if not args.report_only: archive = {r["hash"]: r for r in load_archive_rows(args.db)} training_hashes = {h for h, r in archive.items() if r["score_source"] == "llm" and r["xau"] is not None} existing = {r["hash"] for r in load_log()} engine, headlines = fetch_headlines(args.max_age_hours) print(f"fetched {len(headlines)} headlines within {args.max_age_hours:g} h " f"(currency-filtered, deduplicated); {len(existing)} already in the shadow log") fresh = [h for h in headlines if headline_sha256(h.headline) not in existing] print(f" {len(fresh)} new to the shadow log") if fresh: model = ModelScorer() ds = DeepSeekScorer(timeout=args.timeout, no_network=args.no_network) d_res = ds.score_many([h.headline for h in fresh]) run_ts = utc_now_iso() appended = 0 with SHADOW_LOG.open("a", encoding="utf-8") as fh: for h, d in zip(fresh, d_res): if d is None: continue # retried on a later run if still fresh hsh = headline_sha256(h.headline) arch = archive.get(hsh) # Only a DeepSeek-scored archive row is a valid # "DeepSeek before" for the self-agreement check; the # batch-overflow rows carry the keyword score under # the same XAU key until a later cycle upgrades them. arch_llm = arch if arch and arch["score_source"] == "llm" else None m_full = model.score(h.headline, h.description) m_head = model.score(h.headline, None) kw_xau = engine._analyze_headline_sentiment(h.headline, "XAU") row = { "run_ts": run_ts, "hash": hsh, "headline": h.headline, "description": h.description, "publisher": h.publisher, "published_at": h.timestamp.isoformat() if h.timestamp else None, "currencies": list(h.currencies_mentioned), "mentions_gold": "XAU" in h.currencies_mentioned, "seen_in_archive": arch is not None, "in_training": hsh in training_hashes, "archived_source": arch["score_source"] if arch else None, "archived_xau": arch_llm["xau"] if arch_llm else None, "archived_s": arch_llm["score"] if arch_llm else None, "ds_s": d["s"], "ds_xau": d["xau"], "ds_cached": d["cached"], "ds_batch_ms": d.get("batch_ms"), "ds_batch_size": (ds._cache.get(ds.key(h.headline)) or {}).get("batch_size"), "model_label": m_full["label"], "model_cls": m_full["cls"], "model_score": round(m_full["score"], 4), "model_conf": round(m_full["confidence"], 4), "model_margin": round(m_full["margin"], 4), "model_ms": round(m_full["ms"], 2), "model_h_label": m_head["label"], "model_h_cls": m_head["cls"], "model_h_score": round(m_head["score"], 4), "model_h_conf": round(m_head["confidence"], 4), "kw_xau": float(kw_xau), } fh.write(json.dumps(row, ensure_ascii=False) + "\n") appended += 1 st = ds.stats() print(f" appended {appended} rows; DeepSeek {st['calls']} call(s), " f"{st['headlines_sent']} sent, {st['cache_hits']} cache hits, " f"{st['failed_calls']} failed") log = load_log() summary = report(log, args.threshold) tag = ts_tag() write_json(RESULTS_DIR / f"live_{tag}.json", {"generated_at": tag, "threshold": args.threshold, "log_rows": len(log), **summary}) print(f"\nshadow log: {SHADOW_LOG} ({len(log)} rows); wrote results/live_{tag}.json") return 0 if __name__ == "__main__": raise SystemExit(main())