""" Shared helpers for the ML-NEWS vs DeepSeek evaluation scripts. RESEARCH ONLY. This folder lives outside the ForexBot2 git tree on purpose (ML-NEWS is not a git repository). Nothing here is imported by live code. Runtime: the ForexBot2 venv (it has `requests`, `pandas` and the live news code) with ML-NEWS on sys.path (for `news_sentiment`). The model was pickled with scikit-learn 1.9.1; the ForexBot2 venv carries 1.9.0 and was verified on 2026-09-17 to return byte-identical probabilities on all 1,577 archived headlines, so the version warning is silenced. /home/harwinder/Development/ForexBot2/venv/bin/python eval_golden.py Three scorers are wrapped so every script measures the same things: * ModelScorer - the scikit-learn bundle (models/xauusd_sentiment.joblib) * KeywordScorer - ForexBot2's built-in keyword scorer, XAU-inverted (what the live system falls back to when DeepSeek is unavailable) * DeepSeekScorer - the LIVE DeepSeek path: the real system prompt, the real client, the real batch cap of 25, through NewsEngine._llm_headline_scores. Results are cached on disk (results/deepseek_cache.json) keyed by headline + prompt sha1 + model, so re-running a report never re-spends. """ from __future__ import annotations import csv import hashlib import json import logging import math import os import sqlite3 import sys import time import warnings from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple FOREXBOT2_DIR = Path("/home/harwinder/Development/ForexBot2") ML_NEWS_DIR = Path("/home/harwinder/Development/ML-NEWS") EVAL_DIR = Path(__file__).resolve().parent DATA_DIR = EVAL_DIR / "data" RESULTS_DIR = EVAL_DIR / "results" GOLDEN_CSV = DATA_DIR / "llm_news_golden.csv" MODEL_PATH = ML_NEWS_DIR / "models" / "xauusd_sentiment.joblib" TRANSFORMER_DIR = ML_NEWS_DIR / "models" / "xau_transformer" DEFAULT_DB = FOREXBOT2_DIR / "logs" / "newsdata.db" DEEPSEEK_CACHE = RESULTS_DIR / "deepseek_cache.json" #: |xau| below this = neutral. The 2026-08-06 golden-set grading used #: 0.15 (DeepSeek 78/78, keywords 24/78 under this rule). NEUTRAL_BAND_GOLDEN = 0.15 #: The ML-NEWS trainer maps a DeepSeek XAU score to a class with this #: threshold (news_sentiment.train_model neutral_threshold=0.05). LABEL_THRESHOLD_TRAIN = 0.05 #: Production batch cap in news_engine._apply_llm_sentiment. BATCH_CAP = 25 CLASSES = (-1, 0, 1) CLASS_NAMES = {-1: "bearish", 0: "neutral", 1: "bullish"} NAME_TO_CLASS = {v: k for k, v in CLASS_NAMES.items()} for _p in (str(FOREXBOT2_DIR), str(ML_NEWS_DIR)): if _p not in sys.path: sys.path.insert(0, _p) warnings.filterwarnings("ignore", message=".*InconsistentVersionWarning.*") try: # sklearn raises its own warning class for a version mismatch from sklearn.exceptions import InconsistentVersionWarning warnings.filterwarnings("ignore", category=InconsistentVersionWarning) except Exception: # pragma: no cover - sklearn missing pass logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(name)s: %(message)s") RESULTS_DIR.mkdir(parents=True, exist_ok=True) # ── small utilities ────────────────────────────────────────────────── def ts_tag() -> str: return datetime.now(timezone.utc).strftime("%Y%m%d_%H%M") def utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat() def to_class(x: Optional[float], band: float) -> Optional[int]: """-1 / 0 / +1 from a continuous score; None stays None.""" if x is None or (isinstance(x, float) and math.isnan(x)): return None if x >= band: return 1 if x <= -band: return -1 return 0 def headline_sha256(headline: str) -> str: """Same key news_archive uses (sha256 of the exact headline text).""" return hashlib.sha256(headline.encode("utf-8")).hexdigest() def fmt_pct(k: int, n: int) -> str: return f"{k}/{n} = {100.0 * k / n:.1f}%" if n else "n/a" def write_json(path: Path, obj: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(obj, indent=2, sort_keys=True, default=str) + "\n", encoding="utf-8") def write_csv(path: Path, rows: Sequence[Dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) if not rows: path.write_text("", encoding="utf-8") return keys: List[str] = [] for r in rows: for k in r: if k not in keys: keys.append(k) with path.open("w", newline="", encoding="utf-8") as fh: w = csv.DictWriter(fh, fieldnames=keys) w.writeheader() for r in rows: w.writerow(r) def hr(title: str = "") -> None: line = "═" * 72 print("\n" + line) if title: print(title) print(line) # ── metrics (numpy/pandas only, mirrors the 2026-08-06 harness) ───── def confusion(exp: Sequence[int], pred: Sequence[int], classes: Sequence[int] = CLASSES) -> List[List[int]]: idx = {c: i for i, c in enumerate(classes)} m = [[0] * len(classes) for _ in classes] for e, p in zip(exp, pred): m[idx[e]][idx[p]] += 1 return m def accuracy(exp: Sequence[int], pred: Sequence[int]) -> float: pairs = list(zip(exp, pred)) return sum(1 for e, p in pairs if e == p) / len(pairs) if pairs else float("nan") def macro_f1(exp: Sequence[int], pred: Sequence[int], classes: Sequence[int] = CLASSES) -> float: m = confusion(exp, pred, classes) f1s = [] for i in range(len(classes)): tp = m[i][i] fp = sum(m[r][i] for r in range(len(classes))) - tp fn = sum(m[i]) - tp prec = tp / (tp + fp) if tp + fp else 0.0 rec = tp / (tp + fn) if tp + fn else 0.0 f1s.append(2 * prec * rec / (prec + rec) if prec + rec else 0.0) return sum(f1s) / len(f1s) def directional_accuracy(exp: Sequence[int], pred: Sequence[int]) -> Tuple[int, int]: """Among rows whose expected class is not neutral: hits where the prediction equals it (a neutral prediction counts as a miss).""" pairs = [(e, p) for e, p in zip(exp, pred) if e != 0] return sum(1 for e, p in pairs if e == p), len(pairs) def sign_agreement(exp: Sequence[int], pred: Sequence[int]) -> Tuple[int, int]: """Among rows where BOTH sides lean one way: same sign.""" pairs = [(e, p) for e, p in zip(exp, pred) if e != 0 and p != 0] return sum(1 for e, p in pairs if e == p), len(pairs) def spearman(a: Sequence[float], b: Sequence[float]) -> float: import numpy as np import pandas as pd s = pd.DataFrame({"a": list(a), "b": list(b)}).dropna() if len(s) < 3: return float("nan") ra, rb = s["a"].rank(), s["b"].rank() if ra.std() == 0 or rb.std() == 0: return float("nan") return float(np.corrcoef(ra, rb)[0, 1]) def mae(a: Sequence[float], b: Sequence[float]) -> float: pairs = [(x, y) for x, y in zip(a, b) if x is not None and y is not None] return sum(abs(x - y) for x, y in pairs) / len(pairs) if pairs else float("nan") def mcnemar(hit_a: Sequence[bool], hit_b: Sequence[bool]) -> Tuple[int, int, float]: """Exact two-sided McNemar on the discordant pairs: b = A right and B wrong, c = A wrong and B right.""" 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) n = b + c if n == 0: return b, c, 1.0 k = min(b, c) p = 2.0 * sum(math.comb(n, i) for i in range(k + 1)) / 2 ** n return b, c, min(1.0, p) def print_confusion(cm: List[List[int]], title: str) -> None: print(f"\n{title} (rows = expected, cols = predicted)") print(" " + "".join(f"{CLASS_NAMES[c]:>9}" for c in CLASSES)) for c, row in zip(CLASSES, cm): print(f" {CLASS_NAMES[c]:<9} " + "".join(f"{v:>9}" for v in row)) def agreement_block(exp_cls: Sequence[int], pred_cls: Sequence[int], exp_score: Optional[Sequence[float]] = None, pred_score: Optional[Sequence[float]] = None) -> Dict[str, Any]: """Everything one comparison needs, as a dict (JSON-friendly).""" n = len(exp_cls) hits = sum(1 for e, p in zip(exp_cls, pred_cls) if e == p) d_hit, d_n = directional_accuracy(exp_cls, pred_cls) s_hit, s_n = sign_agreement(exp_cls, pred_cls) out: Dict[str, Any] = { "n": n, "accuracy": hits / n if n else None, "hits": hits, "macro_f1": macro_f1(exp_cls, pred_cls) if n else None, "directional_hits": d_hit, "directional_n": d_n, "directional_accuracy": d_hit / d_n if d_n else None, "sign_hits": s_hit, "sign_n": s_n, "sign_agreement": s_hit / s_n if s_n else None, "confusion": confusion(exp_cls, pred_cls) if n else None, } if exp_score is not None and pred_score is not None: out["spearman"] = spearman(exp_score, pred_score) out["mae"] = mae(exp_score, pred_score) return out def print_agreement(label: str, blk: Dict[str, Any]) -> None: if not blk["n"]: print(f" {label:<44} n=0") return extra = "" if "spearman" in blk and blk["spearman"] == blk["spearman"]: extra = f" rho {blk['spearman']:+.2f} MAE {blk['mae']:.2f}" print(f" {label:<44} n={blk['n']:<4} 3-class {100 * blk['accuracy']:5.1f}% " f"macro-F1 {blk['macro_f1']:.3f} dir {fmt_pct(blk['directional_hits'], blk['directional_n']):<18} " f"sign {fmt_pct(blk['sign_hits'], blk['sign_n']):<18}{extra}") # ── golden set ─────────────────────────────────────────────────────── def load_golden() -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] with GOLDEN_CSV.open(encoding="utf-8") as fh: for r in csv.DictReader(fh): r["expected_xau_sign"] = int(r["expected_xau_sign"]) rows.append(r) need = {"headline", "category", "expected_xau_sign", "rationale"} if rows and not need <= set(rows[0]): raise SystemExit(f"golden CSV missing columns {need - set(rows[0])}") if any(r["expected_xau_sign"] not in (-1, 0, 1) for r in rows): raise SystemExit("expected_xau_sign must be -1/0/+1") if len({r["headline"] for r in rows}) != len(rows): raise SystemExit("duplicate headlines in golden CSV") return rows # ── archive corpus ─────────────────────────────────────────────────── def _finite(value: Any) -> Optional[float]: try: value = float(value) if value is not None else None except (TypeError, ValueError): return None return value if value is not None and math.isfinite(value) else None def load_extra_labels(path: Optional[Path]) -> Dict[str, float]: """hash -> xau from a JSONL such as results/backfill_labels.jsonl.""" if path is None: return {} out: Dict[str, float] = {} with path.open(encoding="utf-8") as fh: for line in fh: if line.strip(): rec = json.loads(line) xau = _finite(rec.get("xau")) if rec.get("hash") and xau is not None: out[rec["hash"]] = xau return out def load_archive_rows(db_path: Path = DEFAULT_DB, extra_labels: Optional[Path] = None) -> List[Dict[str, Any]]: """Every row of the headlines table, read-only, with DeepSeek's XAU score as `xau` (None when absent). Same priority as the trainer: the xau_score column (every LLM row since 2026-09-18), then the XAU key of score_by_currency_json, then `extra_labels`. `xau_src` says which; `gold` is True when the JSON carries the XAU key (the rows the pre-2026-09-22 trainer used).""" if not db_path.is_file(): raise SystemExit(f"archive not found: {db_path}") extra = load_extra_labels(extra_labels) uri = f"{db_path.resolve().as_uri()}?mode=ro" out: List[Dict[str, Any]] = [] with sqlite3.connect(uri, uri=True) as conn: cols = {row[1] for row in conn.execute("PRAGMA table_info(headlines)")} xau_col = "xau_score" if "xau_score" in cols else "NULL" q = ("SELECT hash, headline, description, publisher, published_at, " "first_seen, score, score_by_currency_json, currencies_json, " f"score_source, teacher_model, teacher_prompt_sha1, {xau_col} " "FROM headlines") for (h, head, desc, pub, pat, fs, score, sbc, cur, src, tm, tp, xau_score) in conn.execute(q): by_cur: Dict[str, float] = {} try: by_cur = json.loads(sbc) if sbc else {} if not isinstance(by_cur, dict): by_cur = {} except ValueError: by_cur = {} xau, xau_src = _finite(xau_score), "xau_score" if xau is None: xau, xau_src = _finite(by_cur.get("XAU")), "json" if xau is None and h in extra: xau, xau_src = extra[h], "extra_labels" if xau is None: xau_src = None try: currencies = json.loads(cur) if cur else [] except ValueError: currencies = [] out.append({ "hash": h, "headline": head, "description": desc, "publisher": pub, "published_at": pat, "first_seen": fs, "day": str(pat or fs or "")[:10], "score": score, "by_currency": by_cur, "xau": xau, "xau_src": xau_src, "gold": "XAU" in by_cur, "currencies": currencies, "score_source": src, "teacher_model": tm, "teacher_prompt_sha1": tp, }) return out # ── scorers ────────────────────────────────────────────────────────── class ModelScorer: """The scikit-learn bundle. `score()` is one headline at a time so the measured latency is the live per-article cost.""" def __init__(self, model_path: Path = MODEL_PATH): from news_sentiment import load_model, score_news # noqa: WPS433 self.bundle = load_model(model_path) self._score_news = score_news tr = self.bundle.get("training", {}) self.meta = { "model_path": str(model_path), "trained_at": self.bundle.get("trained_at"), "sklearn_version": self.bundle.get("sklearn_version"), "model_type": self.bundle.get("model_type"), "training_examples": tr.get("examples"), "label_distribution": tr.get("label_distribution"), "neutral_threshold": tr.get("neutral_threshold"), # rows published after this were never trained on "oos_from": self.bundle.get("trained_at"), } self.total_ms = 0.0 self.calls = 0 def score(self, headline: str, description: Optional[str] = None) -> Dict[str, Any]: t0 = time.perf_counter() r = self._score_news(self.bundle, headline, description) ms = (time.perf_counter() - t0) * 1000.0 self.total_ms += ms self.calls += 1 probs = r["probabilities"] ordered = sorted(probs.values(), reverse=True) return { "label": r["label"], "cls": NAME_TO_CLASS[r["label"]], "score": float(r["score"]), "confidence": float(r["confidence"]), "margin": float(ordered[0] - ordered[1]), "p_bearish": probs["bearish"], "p_neutral": probs["neutral"], "p_bullish": probs["bullish"], "ms": ms, } @property def mean_ms(self) -> float: return self.total_ms / self.calls if self.calls else float("nan") class TransformerScorer: """The two-head ONNX model (news_transformer.py), same interface as ModelScorer plus p_relevant / p_bullish / relevant. Headline only: `description` is accepted and ignored.""" def __init__(self, model_dir: Path = TRANSFORMER_DIR, threads: Optional[int] = None): from news_transformer import load_model, score_news # noqa: WPS433 self.bundle = load_model(model_dir, threads) self._score_news = score_news meta = self.bundle["meta"] data = meta.get("data", {}) n_train = sum(data.get(k, {}).get("n", 0) for k in (("train", "dev", "val") if meta.get("final") else ("train",))) self.meta = { "model_path": str(model_dir), "trained_at": meta.get("trained_at"), "model_type": f"two_head_transformer:{meta.get('base_model')}", "training_examples": n_train, "relevance_threshold": meta.get("relevance_threshold"), "final": meta.get("final"), # a non-final model never saw newsdata rows from val_from on "oos_from": (meta.get("trained_at") if meta.get("final") else meta.get("hyperparameters", {}).get("val_from")), } self.total_ms = 0.0 self.calls = 0 def score(self, headline: str, description: Optional[str] = None) -> Dict[str, Any]: t0 = time.perf_counter() r = self._score_news(self.bundle, headline) ms = (time.perf_counter() - t0) * 1000.0 self.total_ms += ms self.calls += 1 pr, pb = r["p_relevant"], r["p_bullish"] return { "label": r["label"], "cls": NAME_TO_CLASS[r["label"]], "score": float(r["score"]), "confidence": float(r["confidence"]), "margin": float(abs(2 * pb - 1) if r["relevant"] else abs(2 * pr - 1)), "p_relevant": pr, "p_bullish": pb, "relevant": r["relevant"], "ms": ms, } @property def mean_ms(self) -> float: return self.total_ms / self.calls if self.calls else float("nan") def make_scorer(kind: str, path: Optional[Path] = None): """`--scorer tfidf|transformer` with an optional `--model` path.""" if kind == "transformer": return TransformerScorer(path or TRANSFORMER_DIR) return ModelScorer(path or MODEL_PATH) class KeywordScorer: """ForexBot2's keyword scorer with the XAU policy inversion — the score the live system shows when DeepSeek is unavailable.""" def __init__(self): from config import NewsConfig from news_engine import NewsEngine self._engine = NewsEngine(NewsConfig()) def score(self, headline: str) -> float: return float(self._engine._analyze_headline_sentiment( headline, for_currency="XAU")) def currencies(self, headline: str) -> List[str]: return list(self._engine._extract_currencies(headline)) def install_research_client(timeout: float, daily_cap: Optional[int] = None): """Replace the process-wide DeepSeek singleton with one that uses a longer timeout (cold prompt-cache batches ran past the live 12 s in the 2026-08-06 harness). Same key and model as live; the daily cap is live's unless `daily_cap` is given. The cap is counted in this process only, so it never consumes the live bot's allowance.""" import app_config import llm_client settings = app_config.get_app_config().deepseek_settings client = llm_client.DeepSeekClient( api_key=app_config.deepseek_api_key(), model=str(settings.get("model", llm_client.DEFAULT_MODEL)), timeout=timeout, daily_cap=daily_cap or int(settings.get("daily_request_cap", llm_client.DEFAULT_DAILY_CAP)), enabled=True, ) with llm_client._client_lock: llm_client._client = client return client class DeepSeekScorer: """The live DeepSeek headline path with a disk cache. Each uncached batch of <= 25 headlines is ONE real call through NewsEngine._llm_headline_scores (real prompt constant, real JSON contract, real clamping). A failed batch is retried `retries` times; headlines still unscored come back as None. """ def __init__(self, cache_path: Path = DEEPSEEK_CACHE, timeout: float = 30.0, retries: int = 3, no_network: bool = False, daily_cap: Optional[int] = None): from config import NewsConfig import news_engine self._engine = news_engine.NewsEngine(NewsConfig()) self.prompt_sha1 = news_engine._LLM_SENTIMENT_PROMPT_SHA1 self._client = install_research_client(timeout, daily_cap) self.model = getattr(self._client, "_model", "?") self.no_network = no_network self.retries = retries self.cache_path = cache_path self._cache: Dict[str, Dict[str, Any]] = {} if cache_path.is_file(): try: self._cache = json.loads(cache_path.read_text(encoding="utf-8")) except ValueError: self._cache = {} self.calls = 0 self.failed_calls = 0 self.headlines_sent = 0 self.cache_hits = 0 self.batch_ms: List[float] = [] @property def available(self) -> bool: return bool(getattr(self._client, "available", False)) and not self.no_network def key(self, headline: str) -> str: return (hashlib.sha1(headline.encode("utf-8")).hexdigest() + ":" + self.prompt_sha1[:8] + ":" + self.model) def _save(self) -> None: """Merge-on-save: re-read the file first so two scripts running at once (each writes after every batch) never drop each other's entries. Same key = same answer, so last-writer-wins is fine.""" self.cache_path.parent.mkdir(parents=True, exist_ok=True) merged: Dict[str, Dict[str, Any]] = {} if self.cache_path.is_file(): try: merged = json.loads(self.cache_path.read_text(encoding="utf-8")) except ValueError: merged = {} merged.update(self._cache) self._cache = merged tmp = self.cache_path.with_suffix(f".{os.getpid()}.tmp") tmp.write_text(json.dumps(merged, indent=0, sort_keys=True), encoding="utf-8") os.replace(tmp, self.cache_path) def _call(self, texts: List[str]) -> Optional[List[Tuple[float, float]]]: for attempt in range(self.retries): t0 = time.perf_counter() scores = self._engine._llm_headline_scores(texts) ms = (time.perf_counter() - t0) * 1000.0 self.calls += 1 if scores is not None: self.batch_ms.append(ms) self.headlines_sent += len(texts) return scores self.failed_calls += 1 err = getattr(self._client, "last_error", None) print(f" DeepSeek batch of {len(texts)} failed (attempt " f"{attempt + 1}/{self.retries}, {err}) — retrying") time.sleep(2.0) return None def score_many(self, headlines: Sequence[str], force: bool = False ) -> List[Optional[Dict[str, Any]]]: """Aligned list of {"s", "xau", "cached", "batch_ms"} or None.""" out: List[Optional[Dict[str, Any]]] = [None] * len(headlines) todo: List[int] = [] seen: Dict[str, int] = {} for i, h in enumerate(headlines): k = self.key(h) if not force and k in self._cache: c = self._cache[k] out[i] = {"s": c["s"], "xau": c["xau"], "cached": True, "batch_ms": c.get("batch_ms"), "ts": c.get("ts")} self.cache_hits += 1 elif h in seen: # duplicate text inside this request todo.append(i) else: seen[h] = i todo.append(i) uniq = [] for i in todo: if headlines[i] not in [headlines[j] for j in uniq]: uniq.append(i) if uniq and not self.available: why = "no network (--no-network)" if self.no_network else \ "DeepSeek client unavailable (no key / disabled)" print(f" {len(uniq)} headline(s) not in cache and {why}; " "they stay unscored") return out for start in range(0, len(uniq), BATCH_CAP): idxs = uniq[start:start + BATCH_CAP] texts = [headlines[i] for i in idxs] scores = self._call(texts) if scores is None: continue ms = self.batch_ms[-1] now = utc_now_iso() for i, (s, x) in zip(idxs, scores): rec = {"headline": headlines[i], "s": float(s), "xau": float(x), "ts": now, "batch_ms": ms, "batch_size": len(texts), "model": self.model, "prompt_sha1": self.prompt_sha1} self._cache[self.key(headlines[i])] = rec self._save() for i in todo: if out[i] is None: c = self._cache.get(self.key(headlines[i])) if c is not None: out[i] = {"s": c["s"], "xau": c["xau"], "cached": False, "batch_ms": c.get("batch_ms"), "ts": c.get("ts")} return out def stats(self) -> Dict[str, Any]: import statistics return { "model": self.model, "prompt_sha1": self.prompt_sha1, "calls": self.calls, "failed_calls": self.failed_calls, "headlines_sent": self.headlines_sent, "cache_hits": self.cache_hits, "batch_ms_mean": statistics.fmean(self.batch_ms) if self.batch_ms else None, "batch_ms_max": max(self.batch_ms) if self.batch_ms else None, "per_headline_ms": (sum(self.batch_ms) / self.headlines_sent if self.headlines_sent else None), }