news archive: fix three bugs found reviewing 301bf90

1. A backfill pass silently wiped teacher identity from live rows.
   The provenance compare is `>=`, and backfill rows are llm-scored but
   carry no teacher_model/teacher_prompt_sha1 (neither source DB recorded
   one), so a bare assignment blanked what the live hook had stored.
   Re-running the backfill — e.g. after importing the live server's
   signals.db, which is the next planned step — would have erased it from
   every live row. COALESCE now keeps a stored identity when the incoming
   one is NULL, while a genuinely newer teacher still replaces it.
   Timestamps clamp for the same reason: backfill stamps historical
   values, so first_seen only moves earlier and last_seen only later.

2. A retried batch could be applied twice. executemany does not roll
   itself back, so a batch that half-landed and then hit a lock was
   re-applied whole by _with_lock_retry, double-counting seen_count on
   the rows that already landed. Writes now run in an explicit BEGIN
   IMMEDIATE with a rollback, which also takes the write lock up front so
   contention fails before any row is applied.

3. The archive hook could break a news fetch. archive_headlines swallows
   its own failures, but its ARGUMENTS are evaluated first, so a raising
   get_llm_client() escaped that guard. Guarded at the call site too:
   archiving is a side effect and must never cost us the headlines we
   just fetched.

Both 1 and 2 were verified to reproduce against 301bf90 before fixing
(teacher_model -> None; seen_count 3 instead of 2 on the rows that
landed pre-failure) and are now regression tests.

Also fixed two weak tests of my own: the disabled-by-config test never
pointed NEWS_ARCHIVE_PATH at its temp dir, so its assertion passed
regardless of behaviour and the write went to the real archive; and the
neutral-headline test asserted the headline was archived without showing
it is dropped downstream, which is the entire reason the hook sits where
it does. Positional binding of the provenance rank is now asserted rather
than trusted, since a miscount would shift every parameter silently.

483 -> 486 tests.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Bzv4ETVapFNauFDQht7QtA

app_config.py

@@ -85,7 +85,7 @@ _SCANNABLE_TFS = ("M5", "M15", "H1", "H4")
 
 _KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level",
                "oanda", "deepseek", "agent_panel", "market_outlook",
-               "violent_watch", "signal_log"}
+               "violent_watch", "signal_log", "news_archive"}
 _KNOWN_OANDA = {
     # "account_id" is TOLERATED but ignored: it belonged to the auto-
     # execution daemon removed 2026-09-03 (last commit 31002cb). The candle
@@ -106,6 +106,7 @@ _KNOWN_MARKET_OUTLOOK = {"enabled", "telegram", "max_per_day", "candles_bars"}
 _KNOWN_VIOLENT_WATCH = {"enabled", "alert_line", "headsup",
                         "max_headsup_per_day", "headsup_min_level"}
 _KNOWN_SIGNAL_LOG = {"enabled", "path", "log_blocked_scans"}
+_KNOWN_NEWS_ARCHIVE = {"enabled", "path"}
 _KNOWN_TELEGRAM = {"token", "chat_id", "channel_id", "alert_min_quality"}
 _KNOWN_SCANNING = {"pairs", "timeframes"}
 _KNOWN_PAIR_KEYS = {"blocked_hours_utc"}
@@ -153,6 +154,7 @@ class AppConfig:
     violent_watch_settings: Dict[str, Any] = field(
         default_factory=lambda: dict(_VIOLENT_WATCH_DEFAULTS))
     signal_log_settings: Dict[str, Any] = field(default_factory=dict)
+    news_archive_settings: Dict[str, Any] = field(default_factory=dict)
     log_level: str = "INFO"
 
 
@@ -387,6 +389,17 @@ def load_app_config(path: str = CONFIG_PATH) -> AppConfig:
             "config.json: signal_log.log_blocked_scans must be true or false")
     app.signal_log_settings = dict(slog)
 
+    narch = raw.get("news_archive", {})
+    _check_keys(narch, _KNOWN_NEWS_ARCHIVE, '"news_archive"')
+    if "enabled" in narch and not isinstance(narch["enabled"], bool):
+        raise AppConfigError(
+            "config.json: news_archive.enabled must be true or false")
+    if "path" in narch and (not isinstance(narch["path"], str)
+                            or not narch["path"]):
+        raise AppConfigError(
+            "config.json: news_archive.path must be a non-empty string")
+    app.news_archive_settings = dict(narch)
+
     level = raw.get("log_level", app.log_level)
     if level not in _VALID_LOG_LEVELS:
         raise AppConfigError(
@@ -538,6 +551,34 @@ def signal_log_path(app: Optional[AppConfig] = None) -> str:
     return path
 
 
+# ── News-archive resolution helpers (env → config.json → default) ────
+
+def news_archive_enabled(app: Optional[AppConfig] = None) -> bool:
+    """Default ON — the headline corpus is training data for a local
+    scorer and cannot be backfilled once a fetch window has passed. It
+    lives in its own DB so a signals.db rotation cannot take it with it
+    (which is how three weeks of news were lost on 2026-09-04)."""
+    env = os.environ.get("NEWS_ARCHIVE_ENABLED")
+    if env:
+        return env.strip().lower() in ("1", "true", "yes", "on")
+    app = app or get_app_config()
+    return bool(app.news_archive_settings.get("enabled", True))
+
+
+def news_archive_path(app: Optional[AppConfig] = None) -> str:
+    """Absolute path to the SQLite headline corpus. A relative config
+    value resolves against the repo root so CWD never matters."""
+    env = os.environ.get("NEWS_ARCHIVE_PATH")
+    if env:
+        path = env
+    else:
+        app = app or get_app_config()
+        path = str(app.news_archive_settings.get("path", "logs/newsdata.db"))
+    if not os.path.isabs(path):
+        path = os.path.join(os.path.dirname(os.path.abspath(__file__)), path)
+    return path
+
+
 # ── PlatformConfig overlay ────────────────────────────────────────────
 
 def apply_pair_overrides(

config.json.example

@@ -66,5 +66,9 @@
     "path": "logs/signals.db",
     "log_blocked_scans": true
   },
+  "news_archive": {
+    "enabled": true,
+    "path": "logs/newsdata.db"
+  },
   "log_level": "INFO"
 }

news_archive.py

@@ -0,0 +1,361 @@
+"""
+ForexBot2 — News archive (durable headline corpus)
+==================================================
+
+Every headline the news engine fetches is written here, exactly as it
+arrived, together with the score that was attached to it and **which
+scorer produced that score**. One row per distinct headline, kept
+forever.
+
+Why this exists
+---------------
+Headlines used to survive only as a side effect of signal logging, inside
+``logs/signals.db``. That database was rotated/rebuilt around 2026-09-04
+and three weeks of news — 4,054 distinct headlines, the corpus the
+``news_sentiment_value_2026-09-04.md`` study was written on — was lost.
+``logs/`` is gitignored, so git could not bring it back and nobody
+noticed until 2026-09-08.
+
+This module writes to its own database (``logs/newsdata.db`` by default)
+so signal logging can be wiped, rotated or rebuilt without touching the
+news corpus.
+
+What the corpus is for
+----------------------
+Training a small local scorer that reproduces DeepSeek's headline
+labelling without calling it. On the 2026-08-06 golden set DeepSeek
+scored 78/78 and the built-in keyword scorer it overrides scored 24/78
+(31%) — and the keyword score is what stands whenever DeepSeek is
+unavailable (no key, disabled, daily cap, timeout, HTTP/parse failure,
+or the >25-new-headlines-per-cycle batch overflow). ``score_source``
+records which one produced each row; training filters on ``'llm'``.
+
+Contract
+--------
+* **Never raises.** ``archive_headlines`` swallows every exception. A
+  logging failure must not break a scan — the same contract
+  ``signal_log.log_signal`` holds.
+* **Upgrade-only scores.** A row already marked ``llm`` is never
+  overwritten by a later ``keyword`` score. A headline first seen under
+  the batch-overflow cap is keyword-scored, then gets its DeepSeek score
+  on a later fetch window; the better label wins and is never demoted.
+* **Dedup by headline text.** ``hash`` = sha256 of the headline. A
+  headline seen 50 times is one row with ``last_seen`` bumped.
+* **stdlib only** (``sqlite3``), like ``signal_log``.
+
+Known corpus limitation: a headline mentioning no tracked currency is
+never constructed by the news engine at all (``news_engine`` drops it
+before scoring), so DeepSeek never labels it either and it cannot appear
+here.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import sqlite3
+import threading
+import time
+from datetime import datetime, timezone
+from typing import Any, Dict, Iterable, Optional
+
+import app_config
+
+logger = logging.getLogger(__name__)
+
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS headlines (
+    hash TEXT PRIMARY KEY,
+    headline TEXT NOT NULL,
+    description TEXT,
+    link TEXT,
+    source TEXT,
+    publisher TEXT,
+    published_at TEXT,
+    first_seen TEXT NOT NULL,
+    last_seen TEXT NOT NULL,
+    seen_count INTEGER NOT NULL DEFAULT 1,
+    score REAL,
+    score_by_currency_json TEXT,
+    currencies_json TEXT,
+    keywords_json TEXT,
+    score_source TEXT NOT NULL DEFAULT 'keyword',
+    teacher_model TEXT,
+    teacher_prompt_sha1 TEXT,
+    origin TEXT NOT NULL DEFAULT 'live'
+);
+CREATE INDEX IF NOT EXISTS idx_headlines_source ON headlines (score_source);
+CREATE INDEX IF NOT EXISTS idx_headlines_first_seen ON headlines (first_seen);
+"""
+
+#: Rank of a score's provenance. A write only lands when its rank is >=
+#: the rank already stored, so a keyword score can never overwrite an LLM
+#: one (the batch-overflow case) but an LLM score always upgrades.
+_SOURCE_RANK = {"unknown": 0, "keyword": 1, "model": 2, "llm": 3}
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def headline_hash(headline: str) -> str:
+    """Primary key for a headline: sha256 of its exact text."""
+    return hashlib.sha256(headline.encode("utf-8")).hexdigest()
+
+
+def _json_or_none(value: Any) -> Optional[str]:
+    if value is None:
+        return None
+    try:
+        return json.dumps(value, default=str)
+    except Exception:
+        return None
+
+
+def _iso_or_none(value: Any) -> Optional[str]:
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        return value.isoformat()
+    return str(value)
+
+
+def _finite_score(value: Any) -> Optional[float]:
+    try:
+        out = float(value)
+    except (TypeError, ValueError):
+        return None
+    if out != out or out in (float("inf"), float("-inf")):
+        return None
+    return out
+
+
+class NewsArchive:
+    """Thread-safe SQLite writer for the headline corpus. One instance
+    per database path; safe to share across the bot's executor threads
+    and the dashboard daemon's pool."""
+
+    def __init__(self, path: str):
+        parent = os.path.dirname(path)
+        if parent:
+            os.makedirs(parent, exist_ok=True)
+        self._conn = sqlite3.connect(path, check_same_thread=False)
+        # busy_timeout BEFORE the journal-mode switch: the bot, the
+        # dashboard publisher and the tests open this file at once and
+        # the WAL switch needs a brief exclusive lock.
+        self._conn.execute("PRAGMA busy_timeout=5000")
+        self._lock = threading.Lock()
+        with self._lock:
+            # busy_timeout does not cover the WAL switch, which needs a
+            # brief exclusive lock: three processes creating this file at
+            # once will race it. Same retry signal_log uses.
+            self._with_lock_retry(
+                lambda: self._conn.execute("PRAGMA journal_mode=WAL"))
+            self._with_lock_retry(lambda: self._conn.executescript(_SCHEMA))
+            self._conn.commit()
+
+    @staticmethod
+    def _with_lock_retry(fn, attempts: int = 8, delay: float = 0.05):
+        """Run ``fn`` again on 'database is locked' (concurrent openers
+        racing the same schema step); other errors propagate at once."""
+        for attempt in range(attempts):
+            try:
+                return fn()
+            except sqlite3.OperationalError as exc:
+                if "locked" not in str(exc).lower() or attempt == attempts - 1:
+                    raise
+                time.sleep(delay * (attempt + 1))
+
+    def close(self) -> None:
+        with self._lock:
+            try:
+                self._conn.close()
+            except Exception:
+                pass
+
+    #: Rank of the score already stored, computed in SQL so the compare
+    #: happens inside the same atomic statement as the write.
+    _STORED_RANK = ("CASE headlines.score_source WHEN 'llm' THEN 3"
+                    " WHEN 'model' THEN 2 WHEN 'keyword' THEN 1 ELSE 0 END")
+
+    def store(self, rows: Iterable[Dict[str, Any]]) -> int:
+        """Insert-or-upgrade a batch of prepared rows. Returns the number
+        of rows written. Raises on a database failure — the module-level
+        wrapper is what guarantees a scan cannot break.
+
+        One atomic UPSERT per row. A read-then-write pair would lose an
+        increment whenever two processes (bot, dashboard publisher, tests)
+        first see the same headline at the same moment: both would find it
+        absent, one INSERT would win and the other's count would vanish.
+        """
+        rows = list(rows)
+        if not rows:
+            return 0
+        now = _utc_now_iso()
+        wins = f"? >= {self._STORED_RANK}"   # incoming rank vs stored rank
+        sql = (
+            "INSERT INTO headlines ("
+            " hash, headline, description, link, source, publisher,"
+            " published_at, first_seen, last_seen, seen_count,"
+            " score, score_by_currency_json, currencies_json,"
+            " keywords_json, score_source, teacher_model,"
+            " teacher_prompt_sha1, origin"
+            ") VALUES (?,?,?,?,?,?,?,?,?,1,?,?,?,?,?,?,?,?)"
+            " ON CONFLICT(hash) DO UPDATE SET"
+            # The backfill stamps historical timestamps, so clamp rather
+            # than assign: first_seen only ever moves earlier, last_seen
+            # only ever moves later.
+            "  first_seen = MIN(headlines.first_seen, excluded.first_seen),"
+            "  last_seen = MAX(headlines.last_seen, excluded.last_seen),"
+            "  seen_count = headlines.seen_count + 1,"
+            # Upgrade-only: a weaker scorer never overwrites a stronger
+            # one's label, but re-seeing a headline always counts.
+            f"  score = CASE WHEN {wins} THEN excluded.score"
+            "           ELSE headlines.score END,"
+            f"  score_by_currency_json = CASE WHEN {wins}"
+            "           THEN excluded.score_by_currency_json"
+            "           ELSE headlines.score_by_currency_json END,"
+            f"  currencies_json = CASE WHEN {wins} THEN excluded.currencies_json"
+            "           ELSE headlines.currencies_json END,"
+            f"  keywords_json = CASE WHEN {wins} THEN excluded.keywords_json"
+            "           ELSE headlines.keywords_json END,"
+            # COALESCE, not a bare overwrite: the backfill re-ingests
+            # llm-scored headlines with NO teacher identity (neither source
+            # DB recorded one), and rank >= rank is true, so a plain
+            # assignment blanks what the live hook already stored.
+            f"  teacher_model = CASE WHEN {wins}"
+            "           THEN COALESCE(excluded.teacher_model,"
+            "                         headlines.teacher_model)"
+            "           ELSE headlines.teacher_model END,"
+            f"  teacher_prompt_sha1 = CASE WHEN {wins}"
+            "           THEN COALESCE(excluded.teacher_prompt_sha1,"
+            "                         headlines.teacher_prompt_sha1)"
+            "           ELSE headlines.teacher_prompt_sha1 END,"
+            f"  score_source = CASE WHEN {wins} THEN excluded.score_source"
+            "           ELSE headlines.score_source END,"
+            # Enrichment fills gaps but never blanks what is already there
+            # (the legacy backfill shape carries no description or link).
+            "  description = COALESCE(excluded.description, headlines.description),"
+            "  link = COALESCE(excluded.link, headlines.link),"
+            "  publisher = COALESCE(excluded.publisher, headlines.publisher),"
+            "  published_at = COALESCE(excluded.published_at,"
+            "                          headlines.published_at)")
+
+        # The rank is bound once per CASE arm. Positional binding means a
+        # miscount silently shifts every parameter, so assert it instead of
+        # trusting the arithmetic: 17 INSERT values + one rank per arm.
+        n_rank = sql.count("? >= ")
+        n_values = 17
+        assert sql.count("?") == n_values + n_rank, (
+            f"placeholder count drifted: {sql.count('?')} "
+            f"!= {n_values} + {n_rank}")
+
+        params = []
+        for row in rows:
+            rank = _SOURCE_RANK.get(row["score_source"], 0)
+            params.append((
+                row["hash"], row["headline"], row["description"], row["link"],
+                row["source"], row["publisher"], row["published_at"],
+                row.get("first_seen") or now, row.get("last_seen") or now,
+                row["score"], row["score_by_currency_json"],
+                row["currencies_json"], row["keywords_json"],
+                row["score_source"], row["teacher_model"],
+                row["teacher_prompt_sha1"], row.get("origin", "live"),
+            ) + (rank,) * n_rank)
+
+        def _write() -> None:
+            # BEGIN IMMEDIATE takes the write lock up front, so contention
+            # fails here — before any row is applied — and the rollback
+            # makes the batch atomic if it fails later. Without this a
+            # partial executemany would be re-applied by the retry and
+            # double-count seen_count on the rows that already landed.
+            self._conn.execute("BEGIN IMMEDIATE")
+            try:
+                self._conn.executemany(sql, params)
+                self._conn.commit()
+            except Exception:
+                try:
+                    self._conn.rollback()
+                except Exception:
+                    pass
+                raise
+
+        with self._lock:
+            self._with_lock_retry(_write)
+        return len(rows)
+
+
+def build_row(headline: Any, *, teacher_model: Optional[str] = None,
+              teacher_prompt_sha1: Optional[str] = None,
+              origin: str = "live",
+              first_seen: Optional[str] = None,
+              last_seen: Optional[str] = None) -> Optional[Dict[str, Any]]:
+    """Flatten one ``NewsSentiment`` (or an equivalent object) into a row.
+
+    Returns None when the object carries no usable headline text."""
+    text = getattr(headline, "headline", None)
+    if not isinstance(text, str) or not text.strip():
+        return None
+    source = getattr(headline, "score_source", None) or "keyword"
+    return {
+        "hash": headline_hash(text),
+        "headline": text,
+        "description": getattr(headline, "description", None),
+        "link": getattr(headline, "link", None),
+        "source": getattr(headline, "source", None),
+        "publisher": getattr(headline, "publisher", None),
+        "published_at": _iso_or_none(getattr(headline, "timestamp", None)),
+        "first_seen": first_seen,
+        "last_seen": last_seen,
+        "score": _finite_score(getattr(headline, "sentiment_score", None)),
+        "score_by_currency_json": _json_or_none(
+            getattr(headline, "sentiment_by_currency", None)),
+        "currencies_json": _json_or_none(
+            getattr(headline, "currencies_mentioned", None)),
+        "keywords_json": _json_or_none(getattr(headline, "keywords", None)),
+        "score_source": source,
+        "teacher_model": teacher_model if source == "llm" else None,
+        "teacher_prompt_sha1": (
+            teacher_prompt_sha1 if source == "llm" else None),
+        "origin": origin,
+    }
+
+
+#: path -> NewsArchive
+_archives: Dict[str, NewsArchive] = {}
+_archives_lock = threading.Lock()
+
+
+def archive_headlines(headlines: Any, *, teacher_model: Optional[str] = None,
+                      teacher_prompt_sha1: Optional[str] = None,
+                      origin: str = "live") -> None:
+    """Persist a batch of scored headlines if the news_archive config
+    block enables it. NEVER raises — a logging failure must not break a
+    scan."""
+    try:
+        if not headlines:
+            return
+        if not app_config.news_archive_enabled():
+            return
+        rows = []
+        for headline in headlines:
+            row = build_row(headline, teacher_model=teacher_model,
+                            teacher_prompt_sha1=teacher_prompt_sha1,
+                            origin=origin)
+            if row is not None:
+                rows.append(row)
+        if not rows:
+            return
+        path = app_config.news_archive_path()
+        with _archives_lock:
+            archive = _archives.get(path)
+            if archive is None:
+                archive = NewsArchive(path)
+                _archives[path] = archive
+        archive.store(rows)
+    except Exception:
+        logger.warning("news_archive: failed to persist headlines",
+                       exc_info=True)

news_engine.py

@@ -23,6 +23,7 @@ from typing import Dict, List, Optional, Tuple
 
 import requests
 
+import news_archive
 from config import NewsConfig
 from llm_client import get_llm_client
 
@@ -102,6 +103,13 @@ Examples:
 """
 
 
+#: Identity of the labelling prompt, stored on every archived LLM score.
+#: If the prompt changes, scores before and after are answers to different
+#: questions and must not be pooled into one training set.
+_LLM_SENTIMENT_PROMPT_SHA1 = hashlib.sha1(
+    _LLM_SENTIMENT_SYSTEM.encode("utf-8")).hexdigest()
+
+
 _RELEASE_VALUE_RE = re.compile(r"(-?\d+(?:\.\d+)?)\s*([kmbt])?", re.IGNORECASE)
 _RELEASE_MULTIPLIERS = {"k": 1e3, "m": 1e6, "b": 1e9, "t": 1e12}
 
@@ -212,6 +220,12 @@ class NewsSentiment:
     link: Optional[str] = None         # article URL (Google News = redirect)
     description: Optional[str] = None  # plain text, tags stripped, ≤500 chars
     publisher: Optional[str] = None    # Google News <source> text, else None
+    # Which scorer produced sentiment_score / sentiment_by_currency:
+    # "keyword" (the built-in scorer, ~31% on the 2026-08-06 golden set),
+    # "llm" (DeepSeek, 78/78 on the same set) or "model" (a trained local
+    # artifact). Without it the two are indistinguishable in the archive
+    # and a model trained on the mix inherits the keyword scorer's errors.
+    score_source: str = "keyword"
 
 
 @dataclass
@@ -902,6 +916,23 @@ class NewsEngine:
         except Exception as e:
             logger.debug(f"LLM sentiment failed, keyword scores stand: {e}")
 
+        # Archive BEFORE any downstream filtering. get_fundamental_context
+        # drops every headline inside ±NEUTRAL_HEADLINE_THRESHOLD, so a
+        # corpus built from what the context carries holds almost no
+        # genuinely neutral headlines and would teach a trained scorer
+        # that "neutral" barely exists. Never raises.
+        # Guarded at the call site as well as inside archive_headlines:
+        # arguments are evaluated BEFORE that function's own try/except, so
+        # a raising get_llm_client() would escape it. Archiving is a side
+        # effect — it must never cost us the headlines we just fetched.
+        try:
+            news_archive.archive_headlines(
+                headlines,
+                teacher_model=getattr(get_llm_client(), "_model", None),
+                teacher_prompt_sha1=_LLM_SENTIMENT_PROMPT_SHA1)
+        except Exception as e:
+            logger.debug(f"news archive skipped, headlines unaffected: {e}")
+
         return headlines
 
     def _analyze_headline_sentiment(
@@ -1003,6 +1034,7 @@ class NewsEngine:
                 cur: (xau if cur == "XAU" else overall)
                 for cur in h.currencies_mentioned
             }
+            h.score_source = "llm"
 
     def _llm_event_direction(self, event: EconomicEvent) -> Optional[str]:
         """Gold direction for a released surprise, e.g. "bearish: strong

scripts/backfill_news_archive.py

@@ -0,0 +1,230 @@
+#!/usr/bin/env python3
+"""
+Seed logs/newsdata.db from the headline blobs still on disk.
+==============================================================
+
+The durable headline corpus (``news_archive.py``) only started collecting
+when its hook shipped. Everything before that survives only inside signal
+logs, in two different shapes:
+
+* ``logs/signals.db`` — modern schema, headlines interned into
+  ``news_snapshots`` (kind ``headlines``) and referenced by hash.
+* an older orphan DB (default ``~/signals.db``) — pre-interning, blobs
+  inline in ``signals.headlines_json``, short-form item keys with no
+  ``description``, ``link``, ``publisher`` or ``score_by_currency``.
+
+Both are opened READ-ONLY. The script is idempotent: re-running it adds
+nothing new and never downgrades a score.
+
+Provenance is RECOMPUTED, not trusted
+-------------------------------------
+Neither source records which scorer produced its ``score`` — that flag
+only exists from this change onwards. So each headline is re-scored with
+the live keyword scorer and compared:
+
+* stored score differs from the keyword score  -> ``llm``   (DeepSeek wrote it)
+* exact match on a non-zero score              -> ``keyword``
+* both zero                                    -> ``unknown`` (indistinguishable)
+
+Rows are stamped ``origin='backfill'`` because they come from the
+post-filter pool: ``get_fundamental_context`` drops headlines inside
+±0.1 before they ever reach a signal row, so genuinely neutral headlines
+are under-represented in this history. Rows collected live by the hook
+in ``news_engine`` do not have that bias and are stamped ``'live'``.
+
+Usage
+-----
+    venv/bin/python scripts/backfill_news_archive.py --dry-run
+    venv/bin/python scripts/backfill_news_archive.py
+    venv/bin/python scripts/backfill_news_archive.py --legacy-db /path/to/old.db
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sqlite3
+import sys
+from collections import Counter
+from typing import Any, Dict, Iterator, Tuple
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import app_config          # noqa: E402
+import news_archive        # noqa: E402
+from config import NewsConfig  # noqa: E402
+from news_engine import NewsEngine  # noqa: E402
+
+
+def _open_ro(path: str) -> sqlite3.Connection:
+    """Read-only, but NOT immutable=1 — that flag is blind to WAL rows."""
+    return sqlite3.connect(f"file:{os.path.abspath(path)}?mode=ro", uri=True)
+
+
+def _has_table(con: sqlite3.Connection, name: str) -> bool:
+    row = con.execute(
+        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
+        (name,)).fetchone()
+    return row is not None
+
+
+def iter_blobs(path: str) -> Iterator[Tuple[str, str]]:
+    """Yield (headlines_json, seen_at) from either DB shape."""
+    con = _open_ro(path)
+    try:
+        if _has_table(con, "news_snapshots"):
+            yield from con.execute(
+                "SELECT body, first_seen FROM news_snapshots"
+                " WHERE kind='headlines' ORDER BY first_seen")
+        cols = {r[1] for r in con.execute("PRAGMA table_info(signals)")}
+        if "headlines_json" in cols:
+            order = "logged_at" if "logged_at" in cols else "id"
+            yield from con.execute(
+                "SELECT headlines_json, "
+                + ("logged_at" if "logged_at" in cols else "''")
+                + " FROM signals WHERE headlines_json IS NOT NULL"
+                  f" ORDER BY {order}")
+    finally:
+        con.close()
+
+
+class _Row:
+    """Duck-typed stand-in for NewsSentiment, so news_archive.build_row
+    flattens a stored dict exactly the way it flattens a live headline."""
+
+    __slots__ = ("headline", "source", "timestamp", "currencies_mentioned",
+                 "sentiment_score", "keywords", "sentiment_by_currency",
+                 "link", "description", "publisher", "score_source")
+
+    def __init__(self, item: Dict[str, Any], score_source: str):
+        self.headline = item.get("headline")
+        self.source = item.get("source")
+        self.timestamp = item.get("timestamp")
+        self.currencies_mentioned = item.get("currencies") or []
+        self.sentiment_score = item.get("score")
+        self.keywords = item.get("keywords") or []
+        self.sentiment_by_currency = item.get("score_by_currency") or {}
+        self.link = item.get("link")
+        self.description = item.get("description")
+        self.publisher = item.get("publisher")
+        self.score_source = score_source
+
+
+def classify(engine: NewsEngine, item: Dict[str, Any]) -> str:
+    """Recover which scorer produced this stored score."""
+    text = item.get("headline")
+    if not isinstance(text, str) or not text.strip():
+        return "unknown"
+    try:
+        stored = float(item.get("score") or 0.0)
+    except (TypeError, ValueError):
+        return "unknown"
+    keyword = engine._analyze_headline_sentiment(text)
+    if abs(stored - keyword) > 1e-9:
+        return "llm"
+    if abs(stored) < 1e-9:
+        return "unknown"   # both zero — the two scorers are indistinguishable
+    return "keyword"
+
+
+def collect(paths, engine: NewsEngine):
+    """Newest wins per headline: later blobs carry upgraded LLM scores."""
+    best: Dict[str, Tuple[str, Dict[str, Any], str]] = {}
+    stats = Counter()
+    for path in paths:
+        if not os.path.exists(path):
+            print(f"  skip (not found): {path}")
+            continue
+        blobs = 0
+        for body, seen_at in iter_blobs(path):
+            try:
+                items = json.loads(body)
+            except Exception:
+                stats["bad_blob"] += 1
+                continue
+            blobs += 1
+            for item in items:
+                if not isinstance(item, dict):
+                    continue
+                text = item.get("headline")
+                if not isinstance(text, str) or not text.strip():
+                    continue
+                source = classify(engine, item)
+                prior = best.get(text)
+                if prior is None or (
+                        news_archive._SOURCE_RANK.get(source, 0)
+                        >= news_archive._SOURCE_RANK.get(prior[0], 0)):
+                    best[text] = (source, item, seen_at or "")
+                stats["items"] += 1
+        print(f"  {path}: {blobs} blobs")
+    for source, _, _ in best.values():
+        stats[source] += 1
+    return best, stats
+
+
+def main() -> int:
+    default_legacy = os.path.join(os.path.expanduser("~"), "signals.db")
+    ap = argparse.ArgumentParser(description=__doc__.split("\n")[1])
+    ap.add_argument("--db", default=app_config.signal_log_path(),
+                    help="modern signal-log DB (default: config path)")
+    ap.add_argument("--legacy-db", default=default_legacy,
+                    help=f"older inline-blob DB (default: {default_legacy})")
+    ap.add_argument("--out", default=app_config.news_archive_path(),
+                    help="archive DB to write (default: config path)")
+    ap.add_argument("--dry-run", action="store_true",
+                    help="count only, write nothing")
+    args = ap.parse_args()
+
+    engine = NewsEngine(NewsConfig())
+    print("Reading headline blobs (read-only):")
+    best, stats = collect([args.db, args.legacy_db], engine)
+
+    print(f"\n  headline items scanned : {stats['items']}")
+    print(f"  distinct headlines     : {len(best)}")
+    print(f"    llm (DeepSeek)       : {stats['llm']}")
+    print(f"    keyword              : {stats['keyword']}")
+    print(f"    unknown (both zero)  : {stats['unknown']}")
+    if stats["bad_blob"]:
+        print(f"  unparseable blobs      : {stats['bad_blob']}")
+
+    if args.dry_run:
+        print("\n--dry-run: nothing written.")
+        return 0
+    if not best:
+        print("\nNothing to write.")
+        return 0
+
+    rows = []
+    for text, (source, item, seen_at) in best.items():
+        row = news_archive.build_row(
+            _Row(item, source),
+            teacher_model=None,          # not recorded by either source
+            teacher_prompt_sha1=None,
+            origin="backfill",
+            first_seen=seen_at or None,
+            last_seen=seen_at or None)
+        if row is not None:
+            rows.append(row)
+
+    archive = news_archive.NewsArchive(args.out)
+    try:
+        written = archive.store(rows)
+    finally:
+        archive.close()
+    print(f"\nWrote {written} rows to {args.out}")
+
+    con = _open_ro(args.out)
+    try:
+        print("\nArchive now holds:")
+        for source, count in con.execute(
+                "SELECT score_source, COUNT(*) FROM headlines"
+                " GROUP BY score_source ORDER BY COUNT(*) DESC"):
+            print(f"  {source:8s} {count}")
+    finally:
+        con.close()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

signal_log.py

@@ -573,6 +573,10 @@ def _headline_to_dict(h: Any) -> Dict[str, Any]:
         "link": getattr(h, "link", None),
         "description": getattr(h, "description", None),
         "publisher": getattr(h, "publisher", None),
+        # Which scorer produced "score": "keyword" | "llm" | "model".
+        # The durable corpus lives in news_archive.py; this mirrors the
+        # provenance here so signals.db rows are self-describing too.
+        "score_source": getattr(h, "score_source", None),
     }
 
 

test_all.py

@@ -10739,6 +10739,548 @@ def _():
     assert _ac.oanda_api_token(app) == payload["oanda"]["api_token"]
 
 
+# ══════════════════════════════════════════════════════════════════════
+#  MODULE 32: news archive (durable headline corpus for model training)
+# ══════════════════════════════════════════════════════════════════════
+
+print("\n═══ Module 32: news archive ═══")
+
+import ast
+import copy
+import sqlite3
+from pathlib import Path
+
+import news_archive as _na
+
+
+def _m32_head(headline="Gold surges as Fed signals rate cuts", score=0.5,
+              source="keyword", by=None, currencies=("XAU", "USD"),
+              description="lede", keywords=("surges",)):
+    """A NewsSentiment carrying an explicit score provenance."""
+    now = _dt_m19.datetime.now(_dt_m19.timezone.utc)
+    h = NewsSentiment(
+        headline=headline, source="https://feed.example/rss", timestamp=now,
+        currencies_mentioned=list(currencies), sentiment_score=score,
+        keywords=list(keywords),
+        sentiment_by_currency=(by if by is not None
+                               else {c: score for c in currencies}),
+        description=description, publisher="Example Wire",
+        link="https://example.test/a")
+    h.score_source = source
+    return h
+
+
+def _m32_rows(path):
+    con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
+    try:
+        return {r[0]: r for r in con.execute(
+            "SELECT headline, score, score_source, seen_count, description,"
+            " teacher_model, teacher_prompt_sha1, origin, currencies_json"
+            " FROM headlines")}
+    finally:
+        con.close()
+
+
+@test("news archive: creates its schema and stores a scored headline")
+def _():
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "newsdata.db")
+        arch = _na.NewsArchive(path)
+        try:
+            written = arch.store([_na.build_row(
+                _m32_head(), teacher_model="deepseek-v4-flash",
+                teacher_prompt_sha1="abc123")])
+        finally:
+            arch.close()
+        assert written == 1, written
+        rows = _m32_rows(path)
+        assert len(rows) == 1, rows
+        (head, score, src, seen, desc,
+         model, prompt, origin, currencies) = rows[
+            "Gold surges as Fed signals rate cuts"]
+        assert score == 0.5 and src == "keyword" and seen == 1
+        assert desc == "lede" and origin == "live"
+        assert json.loads(currencies) == ["XAU", "USD"]
+        # teacher identity is only meaningful for an LLM score
+        assert model is None and prompt is None, (model, prompt)
+
+
+@test("news archive: the same headline twice is one row with seen_count bumped")
+def _():
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "newsdata.db")
+        arch = _na.NewsArchive(path)
+        try:
+            arch.store([_na.build_row(_m32_head())])
+            arch.store([_na.build_row(_m32_head())])
+            arch.store([_na.build_row(_m32_head())])
+        finally:
+            arch.close()
+        rows = _m32_rows(path)
+        assert len(rows) == 1, "dedup is by headline text"
+        assert rows["Gold surges as Fed signals rate cuts"][3] == 3, rows
+
+
+@test("news archive: an LLM score upgrades a keyword one, never the reverse")
+def _():
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "newsdata.db")
+        arch = _na.NewsArchive(path)
+        try:
+            # Seen first under the >25/cycle batch-overflow cap: keyword only.
+            arch.store([_na.build_row(_m32_head(score=0.5, source="keyword"))])
+            # Next window DeepSeek labels it — the better label must win.
+            arch.store([_na.build_row(
+                _m32_head(score=-0.7, source="llm"),
+                teacher_model="deepseek-v4-flash", teacher_prompt_sha1="p1")])
+            row = _m32_rows(path)["Gold surges as Fed signals rate cuts"]
+            assert row[1] == -0.7 and row[2] == "llm", row
+            assert row[5] == "deepseek-v4-flash" and row[6] == "p1", row
+            # DeepSeek goes down; the keyword scorer must NOT clobber it.
+            arch.store([_na.build_row(_m32_head(score=0.5, source="keyword"))])
+            row = _m32_rows(path)["Gold surges as Fed signals rate cuts"]
+            assert row[1] == -0.7, "keyword must never overwrite an llm score"
+            assert row[2] == "llm", row
+            assert row[3] == 3, "but re-seeing it still counts"
+        finally:
+            arch.close()
+
+
+@test("news archive: a fresh llm score replaces an older llm score")
+def _():
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "newsdata.db")
+        arch = _na.NewsArchive(path)
+        try:
+            arch.store([_na.build_row(_m32_head(score=0.2, source="llm"))])
+            arch.store([_na.build_row(_m32_head(score=0.9, source="llm"))])
+            row = _m32_rows(path)["Gold surges as Fed signals rate cuts"]
+            assert row[1] == 0.9 and row[2] == "llm", row
+        finally:
+            arch.close()
+
+
+@test("news archive: a backfill pass never blanks a live row's teacher identity")
+def _():
+    """The backfill re-ingests llm-scored headlines carrying NO teacher
+    identity (neither source DB recorded one) and rank >= rank is true, so
+    a bare assignment wipes what the live hook stored."""
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "newsdata.db")
+        arch = _na.NewsArchive(path)
+        try:
+            arch.store([_na.build_row(
+                _m32_head(source="llm"), teacher_model="deepseek-v4-flash",
+                teacher_prompt_sha1="d32e04c6", origin="live",
+                first_seen="2026-09-08T10:00:00+00:00",
+                last_seen="2026-09-08T10:00:00+00:00")])
+            arch.store([_na.build_row(
+                _m32_head(source="llm"), teacher_model=None,
+                teacher_prompt_sha1=None, origin="backfill",
+                first_seen="2026-08-13T09:00:00+00:00",
+                last_seen="2026-08-13T09:00:00+00:00")])
+            con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
+            try:
+                model, sha, first, last = con.execute(
+                    "SELECT teacher_model, teacher_prompt_sha1, first_seen,"
+                    " last_seen FROM headlines").fetchone()
+            finally:
+                con.close()
+            assert model == "deepseek-v4-flash", f"teacher identity wiped: {model}"
+            assert sha == "d32e04c6", sha
+            # historical timestamps clamp outward, never overwrite
+            assert first.startswith("2026-08-13"), f"first_seen: {first}"
+            assert last.startswith("2026-09-08"), f"last_seen: {last}"
+
+            # A genuinely newer teacher still replaces the old identity.
+            arch.store([_na.build_row(
+                _m32_head(source="llm"), teacher_model="deepseek-v5",
+                teacher_prompt_sha1="newsha")])
+            con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
+            try:
+                assert con.execute(
+                    "SELECT teacher_model FROM headlines").fetchone()[0] \
+                    == "deepseek-v5"
+            finally:
+                con.close()
+        finally:
+            arch.close()
+
+
+@test("news archive: a retried batch is applied exactly once, never twice")
+def _():
+    """executemany does not roll itself back. Without an explicit
+    transaction a batch that half-lands and then hits a lock is re-applied
+    whole by the retry, double-counting the rows that already landed."""
+    class _Flaky:
+        """Real connection, but the first executemany lets 2 of 5 rows land
+        and then raises 'database is locked'."""
+
+        def __init__(self, con):
+            self._c = con
+            self.attempts = 0
+
+        def __getattr__(self, name):
+            return getattr(self._c, name)
+
+        def executemany(self, sql, params):
+            self.attempts += 1
+            if self.attempts == 1:
+                self._c.executemany(sql, list(params)[:2])
+                raise sqlite3.OperationalError("database is locked")
+            return self._c.executemany(sql, params)
+
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "newsdata.db")
+        arch = _na.NewsArchive(path)
+        try:
+            rows = [_na.build_row(_m32_head(headline=f"h{i}"))
+                    for i in range(5)]
+            arch.store(rows)                     # seed: seen_count == 1
+            arch._conn = _Flaky(arch._conn)
+            arch.store(rows)                     # half-lands, retries
+            assert arch._conn.attempts == 2, arch._conn.attempts
+        finally:
+            arch.close()
+        con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
+        try:
+            counts = dict(con.execute(
+                "SELECT headline, seen_count FROM headlines"))
+        finally:
+            con.close()
+        doubled = {k: v for k, v in counts.items() if v != 2}
+        assert not doubled, f"retry double-applied: {doubled}"
+
+
+@test("news archive: concurrent writers lose no rows and no seen_count")
+def _():
+    """The bot, the dashboard publisher and the tests all open this file at
+    once. A read-then-write pair loses an increment when two connections
+    first see the same headline together, and the WAL switch itself can
+    hit 'database is locked' while they race to create the file."""
+    import threading as _th
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "newsdata.db")
+        errors = []
+
+        def worker(tag):
+            try:
+                arch = _na.NewsArchive(path)      # its own connection
+                try:
+                    for i in range(40):
+                        arch.store([
+                            _na.build_row(_m32_head(
+                                headline=f"shared {i % 8}", score=0.3)),
+                            _na.build_row(_m32_head(
+                                headline=f"{tag} only {i}", score=-0.2)),
+                        ])
+                finally:
+                    arch.close()
+            except Exception as exc:            # noqa: BLE001
+                errors.append(f"{tag}: {exc!r}")
+
+        threads = [_th.Thread(target=worker, args=(t,))
+                   for t in ("bot", "dashboard", "tests")]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        assert not errors, errors
+
+        con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
+        try:
+            total = con.execute("SELECT COUNT(*) FROM headlines").fetchone()[0]
+            shared = con.execute(
+                "SELECT COUNT(*) FROM headlines"
+                " WHERE headline LIKE 'shared %'").fetchone()[0]
+            seen = con.execute(
+                "SELECT SUM(seen_count) FROM headlines"
+                " WHERE headline LIKE 'shared %'").fetchone()[0]
+            assert con.execute("PRAGMA integrity_check").fetchone()[0] == "ok"
+        finally:
+            con.close()
+        assert shared == 8, f"dedup broke under contention: {shared}"
+        assert total == 8 + 3 * 40, total
+        assert seen == 3 * 40, f"lost {3 * 40 - seen} increments"
+
+
+@test("news archive: build_row rejects objects with no usable headline text")
+def _():
+    assert _na.build_row(_m32_head(headline="   ")) is None
+    assert _na.build_row(object()) is None
+    row = _na.build_row(_m32_head(score=float("nan")))
+    assert row is not None and row["score"] is None, "NaN stored as NULL"
+
+
+@test("news archive: disabled by config writes nothing and does not raise")
+def _():
+    import app_config as _ac
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "newsdata.db")
+        saved_en = os.environ.get("NEWS_ARCHIVE_ENABLED")
+        saved_path = os.environ.get("NEWS_ARCHIVE_PATH")
+        saved_archives = dict(_na._archives)
+        _na._archives.clear()
+        # The path MUST be pointed at the temp dir, or the assertion below
+        # is vacuous: the write would land in the default logs/newsdata.db.
+        os.environ["NEWS_ARCHIVE_PATH"] = path
+        os.environ["NEWS_ARCHIVE_ENABLED"] = "0"
+        try:
+            assert not _ac.news_archive_enabled()
+            _na.archive_headlines([_m32_head()])
+            assert not os.path.exists(path), "nothing may be written when off"
+            # ...and the same call with it on DOES write, proving the check
+            # above is testing the switch and not a typo in the path.
+            os.environ["NEWS_ARCHIVE_ENABLED"] = "1"
+            _na.archive_headlines([_m32_head()])
+            assert os.path.exists(path), "the switch is what gates the write"
+        finally:
+            for arch in _na._archives.values():
+                arch.close()
+            _na._archives.clear()
+            _na._archives.update(saved_archives)
+            for key, val in (("NEWS_ARCHIVE_ENABLED", saved_en),
+                             ("NEWS_ARCHIVE_PATH", saved_path)):
+                if val is None:
+                    os.environ.pop(key, None)
+                else:
+                    os.environ[key] = val
+
+
+@test("news archive: a write failure never raises into a scan")
+def _():
+    saved_store = _na.NewsArchive.store
+    saved_archives = dict(_na._archives)
+    saved_env = os.environ.get("NEWS_ARCHIVE_PATH")
+    with tempfile.TemporaryDirectory() as d:
+        os.environ["NEWS_ARCHIVE_PATH"] = os.path.join(d, "newsdata.db")
+
+        def _boom(self, rows):
+            raise sqlite3.OperationalError("disk I/O error")
+
+        _na.NewsArchive.store = _boom
+        try:
+            _na.archive_headlines([_m32_head()])   # must NOT raise
+        finally:
+            _na.NewsArchive.store = saved_store
+            for arch in _na._archives.values():
+                arch.close()
+            _na._archives.clear()
+            _na._archives.update(saved_archives)
+            if saved_env is None:
+                os.environ.pop("NEWS_ARCHIVE_PATH", None)
+            else:
+                os.environ["NEWS_ARCHIVE_PATH"] = saved_env
+
+
+@test("news archive: config defaults are on, at logs/newsdata.db, env-overridable")
+def _():
+    import app_config as _ac
+    app = _ac.AppConfig()
+    assert _ac.news_archive_enabled(app) is True, "accumulating is the point"
+    assert _ac.news_archive_path(app).endswith("logs/newsdata.db")
+    assert os.path.isabs(_ac.news_archive_path(app)), "CWD must not matter"
+    saved = os.environ.get("NEWS_ARCHIVE_PATH")
+    os.environ["NEWS_ARCHIVE_PATH"] = "/tmp/other.db"
+    try:
+        assert _ac.news_archive_path(app) == "/tmp/other.db"
+    finally:
+        if saved is None:
+            os.environ.pop("NEWS_ARCHIVE_PATH", None)
+        else:
+            os.environ["NEWS_ARCHIVE_PATH"] = saved
+    # the corpus must live in its own file: a signals.db rotation took the
+    # previous three weeks of news with it (2026-09-04)
+    assert _ac.news_archive_path(app) != _ac.signal_log_path(app)
+
+
+@test("news archive: config.json rejects unknown keys in the block")
+def _():
+    payload = copy.deepcopy(_FULL_CFG)
+    payload["news_archive"] = {"enabled": True, "path": "logs/newsdata.db"}
+    app = load_app_config(_write_json(payload))        # must not raise
+    assert app.news_archive_settings["path"] == "logs/newsdata.db"
+    for bad in ({"enabled": "yes"}, {"path": ""}, {"nope": 1}):
+        payload["news_archive"] = bad
+        try:
+            load_app_config(_write_json(payload))
+        except AppConfigError:
+            continue
+        raise AssertionError(f"news_archive {bad} must be rejected")
+
+
+@test("news archive: score_source is 'keyword' at build and 'llm' after DeepSeek")
+def _():
+    import llm_client as lc
+    now = _dt_m19.datetime.now(_dt_m19.timezone.utc)
+    saved_cache = NewsEngine._llm_sent_cache
+    NewsEngine._llm_sent_cache = {}
+    lc._client = _StubLLM({})
+    try:
+        ne = NewsEngine(NewsConfig())
+        h = NewsSentiment(
+            headline="Fed signals rate hike", source="s", timestamp=now,
+            currencies_mentioned=["USD", "XAU"], sentiment_score=1.0,
+            keywords=["rate hike"],
+            sentiment_by_currency={"USD": 1.0, "XAU": -1.0})
+        assert h.score_source == "keyword", "the dataclass default"
+        ne._llm_headline_scores = lambda texts: [(0.4, -0.7)] * len(texts)
+        ne._apply_llm_sentiment([h])
+        assert h.score_source == "llm", "DeepSeek wrote this score"
+
+        # A failed batch changes nothing — including the provenance.
+        h2 = NewsSentiment(
+            headline="Dollar steady before CPI", source="s", timestamp=now,
+            currencies_mentioned=["USD"], sentiment_score=0.2, keywords=[],
+            sentiment_by_currency={"USD": 0.2})
+        ne._llm_headline_scores = lambda texts: None
+        ne._apply_llm_sentiment([h2])
+        assert h2.score_source == "keyword", "keyword score, keyword label"
+        assert h2.sentiment_score == 0.2
+    finally:
+        NewsEngine._llm_sent_cache = saved_cache
+        lc._client = None
+
+
+@test("news archive: signal_log records the score provenance too")
+def _():
+    from signal_log import _headline_to_dict
+    h = _m32_head(source="llm")
+    assert _headline_to_dict(h)["score_source"] == "llm"
+    assert _headline_to_dict(_m32_head())["score_source"] == "keyword"
+
+
+@test("news archive: neutral headlines reach the archive, not the context")
+def _():
+    """get_fundamental_context drops everything inside ±0.1. Archiving
+    downstream of that filter would build a corpus with almost no neutral
+    examples and teach a trained scorer that 'neutral' barely exists."""
+    from news_engine import _is_neutral_headline
+    saved_health = NewsEngine._source_health
+    saved_env = os.environ.get("NEWS_ARCHIVE_PATH")
+    saved_archives = dict(_na._archives)
+    _na._archives.clear()
+    NewsEngine._source_health = {}
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "newsdata.db")
+        os.environ["NEWS_ARCHIVE_PATH"] = path
+        rss = (b"<rss><channel>"
+               b"<item><title>Gold steady as traders await US CPI</title>"
+               b"<pubDate>" + _m19_pubdate(1) + b"</pubDate></item>"
+               b"</channel></rss>")
+
+        class _Sess:
+            def get(self, url, timeout=None):
+                return type("R", (), {"status_code": 200, "content": rss})()
+
+        cfg = NewsConfig()
+        cfg.rss_feeds = ["https://feed.example/rss"]
+        ne = NewsEngine(cfg)
+        ne._session = _Sess()
+        try:
+            heads = ne._fetch_news_sentiment_raw()
+            assert len(heads) == 1, heads
+            assert _is_neutral_headline(heads[0]), \
+                "fixture must be a neutral headline for this test to mean anything"
+            rows = _m32_rows(path)
+            assert "Gold steady as traders await US CPI" in rows, \
+                "the neutral headline must still be archived"
+            # ...and it really is dropped downstream — the filter
+            # get_fundamental_context applies before building the context.
+            kept = [h for h in heads if not _is_neutral_headline(h)]
+            assert kept == [], \
+                "the context keeps it, so archiving downstream would be fine"
+        finally:
+            NewsEngine._source_health = saved_health
+            for arch in _na._archives.values():
+                arch.close()
+            _na._archives.clear()
+            _na._archives.update(saved_archives)
+            if saved_env is None:
+                os.environ.pop("NEWS_ARCHIVE_PATH", None)
+            else:
+                os.environ["NEWS_ARCHIVE_PATH"] = saved_env
+
+
+@test("news archive: the hook cannot break a news fetch, whatever it hits")
+def _():
+    """archive_headlines swallows its own failures, but its ARGUMENTS are
+    evaluated first — a raising get_llm_client() would escape that guard."""
+    import news_engine as _ne_mod
+    saved_client = _ne_mod.get_llm_client
+    saved_archive = _na.archive_headlines
+    saved_health = NewsEngine._source_health
+    NewsEngine._source_health = {}
+
+    def _boom_client():
+        raise RuntimeError("config exploded")
+
+    def _boom_archive(*a, **k):
+        raise RuntimeError("archive exploded")
+
+    rss = (b"<rss><channel>"
+           b"<item><title>Gold surges as Fed signals rate cuts</title>"
+           b"<pubDate>" + _m19_pubdate(1) + b"</pubDate></item>"
+           b"</channel></rss>")
+
+    class _Sess:
+        def get(self, url, timeout=None):
+            return type("R", (), {"status_code": 200, "content": rss})()
+
+    try:
+        for label, patch in (("client", "client"), ("archive", "archive")):
+            _ne_mod.get_llm_client = (_boom_client if patch == "client"
+                                      else saved_client)
+            _na.archive_headlines = (_boom_archive if patch == "archive"
+                                     else saved_archive)
+            cfg = NewsConfig()
+            cfg.rss_feeds = ["https://feed.example/rss"]
+            ne = NewsEngine(cfg)
+            ne._session = _Sess()
+            heads = ne._fetch_news_sentiment_raw()   # must NOT raise
+            assert len(heads) == 1, f"{label}: fetch lost its headlines"
+    finally:
+        _ne_mod.get_llm_client = saved_client
+        _na.archive_headlines = saved_archive
+        NewsEngine._source_health = saved_health
+
+
+@test("news archive: the backtester never builds a NewsEngine, so replays never archive")
+def _():
+    src = open("backtest/gpu_backtest.py").read()
+    assert "NewsEngine(" not in src, "backtests must stay news-free (rule 4)"
+    assert "news_archive" not in src
+    assert "use_news=False" in src
+
+
+@test("news archive: shipped code imports no scikit-learn/scipy/lightgbm")
+def _():
+    """The corpus is trained on offline in venv/; anything the live signal
+    path can reach must stay numpy-only (requirements.txt is the contract)."""
+    banned = {"sklearn", "scipy", "lightgbm", "torch", "transformers"}
+    roots = [Path(".")] + [Path(p) for p in ("providers", "pair_signals")]
+    offenders = []
+    for root in roots:
+        for path in sorted(root.glob("*.py")):
+            if "venv" in path.parts:
+                continue
+            tree = ast.parse(path.read_text(encoding="utf-8"), str(path))
+            for node in ast.walk(tree):
+                if isinstance(node, ast.Import):
+                    names = [a.name for a in node.names]
+                elif isinstance(node, ast.ImportFrom):
+                    names = [node.module or ""]
+                else:
+                    continue
+                for name in names:
+                    if name.split(".")[0] in banned:
+                        offenders.append(f"{path}:{node.lineno} {name}")
+    assert not offenders, f"live code must stay numpy-only: {offenders}"
+    req = open("requirements.txt").read()
+    for name in banned:
+        assert name not in req, f"{name} must not enter requirements.txt"
+
+
 # ══════════════════════════════════════════════════════════════════════
 #  SUMMARY
 # ══════════════════════════════════════════════════════════════════════