Harwinder Sekhon
0c56b4e155
The owner's DeBERTa-v3-small two-head ONNX model (golden set 72/78, ~72% agreement with DeepSeek on unseen days) replaces DeepSeek for headline scoring. DeepSeek keeps event direction, the summary and the dashboard endpoint. - news_model.py: vendored ML-NEWS runtime (lazy onnxruntime/tokenizers, so the bot starts without them) + load-once, never-raising singleton. - news_engine: _enrich_headline_scores mode switch (off/fallback/primary), _apply_model_sentiment (gold head only: xau_score + XAU entry, never USD; not-relevant shows 0.0; never touches _llm_sent_cache), _audit_with_deepseek (archive-only labels, deepseek_audit_per_day). - Owner decision: primary mode writes nothing to logs/newsdata.db except audit copies — the model's output is not training data. - app_config: news_model section (path/threads/mode/deepseek_audit_per_day) + NEWS_MODEL_MODE env override; validated at boot. - news_archive: model rows carry teacher_model xau_transformer@<trained_at>. - deploy/README-news-model.md (ops for bot + dashboard daemon), unit gets a commented NEWS_MODEL_MODE=off line; requirements + onnxruntime, tokenizers. - test_all.py Module 38 (15 tests, one against the real bundle). Suite 581. - CLAUDE.md, spec status note, handoff marked implemented. Books unaffected: news never enters signals; no signal file touched. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
405 lines
17 KiB
Python
405 lines
17 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``.
|
|
|
|
Two columns exist for training only (2026-09-17):
|
|
|
|
* ``xau_score`` — DeepSeek's gold-specific score for the headline,
|
|
whatever currencies it names. ``score_by_currency_json["XAU"]``
|
|
carries it only for gold-keyword headlines; 930 of the first 1,546
|
|
DeepSeek rows had no gold label without this column. NULL for a
|
|
keyword-scored row and for rows written before the column existed
|
|
(an existing database is migrated in place, values are not
|
|
recoverable).
|
|
* ``currencies_json`` ``[]`` marks a headline naming no tracked currency.
|
|
The news engine used to drop those before scoring; now it scores and
|
|
archives them but never returns them to the platform. About 23% of a
|
|
day's feed; DeepSeek gave 10 of 24 a gold direction.
|
|
"""
|
|
|
|
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',
|
|
xau_score REAL
|
|
);
|
|
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._with_lock_retry(self._migrate)
|
|
self._conn.commit()
|
|
|
|
#: Columns added after the first databases were created, in order.
|
|
#: CREATE TABLE IF NOT EXISTS leaves an existing table untouched, so
|
|
#: each one is added in place on open; existing rows read NULL.
|
|
_ADDED_COLUMNS = (("xau_score", "REAL"),)
|
|
|
|
def _migrate(self) -> None:
|
|
present = {row[1] for row in
|
|
self._conn.execute("PRAGMA table_info(headlines)")}
|
|
for name, decl in self._ADDED_COLUMNS:
|
|
if name not in present:
|
|
self._conn.execute(
|
|
f"ALTER TABLE headlines ADD COLUMN {name} {decl}")
|
|
|
|
@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, xau_score"
|
|
") 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,"
|
|
# Same upgrade-only rule as score, and COALESCE on top: a
|
|
# keyword re-sighting carries no gold score and loses on
|
|
# rank, and even a same-rank write without one (a future
|
|
# backfill shape) must never blank a stored value.
|
|
f" xau_score = CASE WHEN {wins}"
|
|
" THEN COALESCE(excluded.xau_score, headlines.xau_score)"
|
|
" ELSE headlines.xau_score 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: 18 INSERT values + one rank per arm.
|
|
n_rank = sql.count("? >= ")
|
|
n_values = 18
|
|
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"),
|
|
row.get("xau_score"),
|
|
) + (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,
|
|
model_teacher: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
|
"""Flatten one ``NewsSentiment`` (or an equivalent object) into a row.
|
|
|
|
Provenance: an ``llm`` row stores ``teacher_model`` (the DeepSeek
|
|
model name); a ``model`` row stores ``model_teacher`` (the local
|
|
scorer's tag, ``xau_transformer@<trained_at>``, news_model.teacher_tag).
|
|
Both carry ``teacher_prompt_sha1``: for a model row it is the DeepSeek
|
|
prompt the model was distilled from, which is the truthful lineage.
|
|
Keyword rows carry neither.
|
|
|
|
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 model_teacher if source == "model" else None),
|
|
"teacher_prompt_sha1": (
|
|
teacher_prompt_sha1 if source in ("llm", "model") else None),
|
|
"origin": origin,
|
|
"xau_score": _finite_score(getattr(headline, "xau_score", None)),
|
|
}
|
|
|
|
|
|
#: 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",
|
|
model_teacher: Optional[str] = None) -> 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, model_teacher=model_teacher)
|
|
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)
|