ForexBot2/scripts/backfill_news_archive.py

#!/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())