ForexBot2/news_archive.py

"""
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"
            "  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,"
            f"  teacher_model = CASE WHEN {wins} THEN excluded.teacher_model"
            "           ELSE headlines.teacher_model END,"
            f"  teacher_prompt_sha1 = CASE WHEN {wins}"
            "           THEN excluded.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)")

        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,) * sql.count("? >="))

        with self._lock:
            self._with_lock_retry(lambda: self._conn.executemany(sql, params))
            self._conn.commit()
        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)