Harwinder Sekhon
1d72e4006a
TF-IDF baseline (news_sentiment.py, models/xauusd_sentiment.joblib) and the two-head DeBERTa-v3-small transformer (transformer/, news_transformer.py) with its Google News crawler and DeepSeek labeller (data_collect/), the golden-set and archive evals (eval/), and the 2026-09-22 and 2026-09-27 results and progress notes. Transformer weights (model.pt, model.onnx) and the crawl database are not tracked; rebuild with transformer/train.py and transformer/export_onnx.py. The DeepSeek label cache in data/ is tracked. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
254 lines
12 KiB
Python
254 lines
12 KiB
Python
#!/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())
|