Compare commits
2 Commits
3afc62c847
...
28b591d786
| Author | SHA1 | Date | |
|---|---|---|---|
| 28b591d786 | |||
| 301bf90f86 |
+42
-1
@@ -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(
|
||||
|
||||
@@ -66,5 +66,9 @@
|
||||
"path": "logs/signals.db",
|
||||
"log_blocked_scans": true
|
||||
},
|
||||
"news_archive": {
|
||||
"enabled": true,
|
||||
"path": "logs/newsdata.db"
|
||||
},
|
||||
"log_level": "INFO"
|
||||
}
|
||||
|
||||
+361
@@ -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)
|
||||
@@ -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
|
||||
|
||||
Executable
+230
@@ -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())
|
||||
@@ -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),
|
||||
}
|
||||
|
||||
|
||||
|
||||
+542
@@ -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
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
|
||||
Reference in New Issue
Block a user