Harwinder Sekhon
d05348d1e4
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.
Claude-Session: https://claude.ai/code/session_01Bzv4ETVapFNauFDQht7QtA
362 lines
15 KiB
Python
362 lines
15 KiB
Python
"""
|
|
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)
|