From 301bf90f863a03067b41fd7910277ded8ef40044 Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Tue, 8 Sep 2026 12:35:57 -0400 Subject: [PATCH] news archive: keep every headline + its DeepSeek score in its own DB MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Headlines only ever survived as a side effect of signal logging, and logs/signals.db was rotated around 2026-09-04 — destroying the 3-week, 4,054-headline corpus news_sentiment_value_2026-09-04.md was written on. logs/ is gitignored, so git could not recover it and nobody noticed for four days. There is no way to backfill a fetch window once it passes. news_archive.py writes to its OWN database (logs/newsdata.db) so a signal-log rotation cannot take the news with it. One row per distinct headline (sha256 key) with the text, description, link, publisher, publish time, the score, and — new — which scorer produced that score. Provenance is the load-bearing part. DeepSeek scores 78/78 on the 2026-08-06 golden set; the keyword scorer it overrides scores 24/78 (31%), and that is what silently stands on no key, disabled, daily cap, timeout, HTTP/parse failure, or the >25-new-headlines-per-cycle batch overflow. Until now both landed in the same column, indistinguishable, so a model trained on the mix would inherit the keyword scorer's errors. NewsSentiment.score_source now records it, mirrored into signals.db via signal_log._headline_to_dict, and teacher_model/teacher_prompt_sha1 keep labels made under a changed prompt from being pooled with older ones. Writes are one atomic UPSERT per row, upgrade-only: an llm score always replaces a keyword one (the batch-overflow tail gets labelled on the next window in a long-running process — verified 13/13), a keyword score never overwrites an llm one, and re-seeing a headline always counts. A read-then-write pair lost increments when the bot, the dashboard publisher and the tests first saw a headline together, and the WAL switch itself needed _with_lock_retry; both are regression-tested with three concurrent writers. The hook sits at the end of _fetch_news_sentiment_raw, BEFORE _is_neutral_headline runs in get_fundamental_context. Archiving downstream of that ±0.1 filter is why signals.db holds almost no neutral headlines and would have taught a trained scorer that "neutral" barely exists. scripts/backfill_news_archive.py seeds it from both DB shapes on disk, read-only and idempotent, recomputing provenance by comparing each stored score against the keyword scorer (neither source recorded it): 1,016 distinct headlines, 846 DeepSeek-labelled, growing ~1,400/week. Nothing in the signal core changed. Both news gating flags stay False, no gate reads any of this, and the backtester never builds a NewsEngine so replays never archive (test-locked). 468 -> 483 tests, including a new guard that AST-walks live code for sklearn/scipy/lightgbm imports — that check did not exist before. Do not train yet: ~800 labels gives a holdout too small to grade (SE +/-0.032). Target is ~4,000 deduplicated headlines over >=3 weeks. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Bzv4ETVapFNauFDQht7QtA --- CLAUDE.md | 6 +- app_config.py | 43 +++- config.json.example | 4 + news_archive.py | 325 ++++++++++++++++++++++++++ news_engine.py | 25 ++ scripts/backfill_news_archive.py | 230 ++++++++++++++++++ signal_log.py | 4 + test_all.py | 385 +++++++++++++++++++++++++++++++ 8 files changed, 1019 insertions(+), 3 deletions(-) create mode 100644 news_archive.py create mode 100755 scripts/backfill_news_archive.py diff --git a/CLAUDE.md b/CLAUDE.md index ef914df..d9007dd 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -12,7 +12,7 @@ Every numeric threshold in `config.py` and `pair_signals/xauusd.py` was backtest ```bash python test_all.py # QA suite, no network (mocked); prints its own count - # (468 tests / 31 modules as of 2026-09-08) + # (483 tests / 32 modules as of 2026-09-08) python main.py --pair XAUUSD --timeframe H1 --detail # CLI signal report (--json for full dump, # --panel for the advisory LLM agent panel, --outlook for the # hourly AI market brief; the violent-move watch line appears @@ -41,7 +41,7 @@ Python 3.12; a `venv/` exists in-repo. Dependencies in `requirements.txt` (numpy ## Configuration & credentials -Precedence for every credential/deployment setting: **env var → config.json → refuse to run** (no hardcoded fallbacks). A gitignored repo-root `.env` is loaded at `app_config` import via `os.environ.setdefault`. `config.json` (gitignored; schema in `config.json.example`) holds secrets plus deployment settings: `data_providers.active` (the live-feed switch), `data_providers.streaming` (optional tick-stream provider for the level-breach monitor), `scanning.pairs`/`timeframes`, per-pair `blocked_hours_utc`, `log_level`. Key env vars: `TELEGRAM_TOKEN`, `TELEGRAM_CHAT_ID`, `TELEGRAM_CHANNEL_ID`, `TWELVE_DATA_API_KEY`, `TIINGO_API_KEY`, `OANDA_API_TOKEN`/`OANDA_BASE_URL` (the data feed needs no account id; a legacy `oanda.account_id` key is tolerated and ignored), `DEEPSEEK_API_KEY` (news LLM; the `deepseek` config.json block holds model/enabled/timeout/daily cap). The `agent_panel` config.json block (all-defaults = feature dark) controls the advisory LLM agent panel: `enabled`, per-surface `telegram`/`dashboard` toggles, `min_quality`, `max_panels_per_day`, `reuse_ttl_seconds`, plus the technical-payload enrichment knobs `candles_enabled`/`candle_bars_primary`/`candle_bars_confirm`. The dark-by-default `market_outlook` block controls the hourly advisory market brief: `enabled`, `telegram`, `max_per_day`, `candles_bars`; `MARKET_OUTLOOK_ENABLED` overrides its enabled setting. The dark-by-default `violent_watch` block controls the display-only violent-move watch: `enabled`, `alert_line`, `headsup`, `max_headsup_per_day`, `headsup_min_level` (`"watch"`/`"hot"`); `VIOLENT_WATCH_ENABLED` overrides its enabled setting. +Precedence for every credential/deployment setting: **env var → config.json → refuse to run** (no hardcoded fallbacks). A gitignored repo-root `.env` is loaded at `app_config` import via `os.environ.setdefault`. `config.json` (gitignored; schema in `config.json.example`) holds secrets plus deployment settings: `data_providers.active` (the live-feed switch), `data_providers.streaming` (optional tick-stream provider for the level-breach monitor), `scanning.pairs`/`timeframes`, per-pair `blocked_hours_utc`, `log_level`. Key env vars: `TELEGRAM_TOKEN`, `TELEGRAM_CHAT_ID`, `TELEGRAM_CHANNEL_ID`, `TWELVE_DATA_API_KEY`, `TIINGO_API_KEY`, `OANDA_API_TOKEN`/`OANDA_BASE_URL` (the data feed needs no account id; a legacy `oanda.account_id` key is tolerated and ignored), `DEEPSEEK_API_KEY` (news LLM; the `deepseek` config.json block holds model/enabled/timeout/daily cap). The `news_archive` block (`enabled` default TRUE, `path` default `logs/newsdata.db`; env `NEWS_ARCHIVE_ENABLED`/`NEWS_ARCHIVE_PATH`) controls the durable headline corpus. The `agent_panel` config.json block (all-defaults = feature dark) controls the advisory LLM agent panel: `enabled`, per-surface `telegram`/`dashboard` toggles, `min_quality`, `max_panels_per_day`, `reuse_ttl_seconds`, plus the technical-payload enrichment knobs `candles_enabled`/`candle_bars_primary`/`candle_bars_confirm`. The dark-by-default `market_outlook` block controls the hourly advisory market brief: `enabled`, `telegram`, `max_per_day`, `candles_bars`; `MARKET_OUTLOOK_ENABLED` overrides its enabled setting. The dark-by-default `violent_watch` block controls the display-only violent-move watch: `enabled`, `alert_line`, `headsup`, `max_headsup_per_day`, `headsup_min_level` (`"watch"`/`"hot"`); `VIOLENT_WATCH_ENABLED` overrides its enabled setting. ## Architecture @@ -67,6 +67,7 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz - **`pair_signals/`** — per-pair config factories. **Always use the factory (`produce_signal` / `make_config`), never bare `PlatformConfig()`** — the factory applies the config.json overlay first, then the validated per-TF deltas (M5 hour-9 block, M15 trail arm 0.70, M15 SELL-hour block, M5 SELL-in-transitioning block, H1 all-shorts block (2026-09-03), etc.; evidence in the docstring). A test enforces factory ≡ inline-config equivalence. Confirmation map: M5→[M15,H1], M15→[M5,H1], H1→[M15,H4], H4→[H1,D1]. - **`signal_log.py` / `advisory_outcome.py`** — SQLite signal-history logger (stdlib only). Hooked into `SignalAggregator.analyze_pair` only (the live entry point — the backtester never writes here, preserving parity), so every live scan from every surface (Telegram `telegram`, CLI `cli`, dashboard `dashboard`) lands one row in `logs/signals.db`: signal/grade/price, MTF + trade-setup + reasons JSON, full per-TF indicator snapshots, the last 14 closed OHLCV candles per fetched TF, and the news at scan time (event risk + calendar events + scored headlines, re-fetched via the NewsEngine class-level caches). Config.json `signal_log` block (`enabled`, `path`, `log_blocked_scans`; env `SIGNAL_LOG_ENABLED`/`SIGNAL_LOG_PATH`/`SIGNAL_LOG_BLOCKED_SCANS`). **Fail-safe by contract** — a logging failure can never break a scan. Panel fields arrive post-hoc: `log_signal` stashes the rowid and `agent_panel.attach_panel` writes verdict + creation/cache provenance + payload hash onto the same row. **Advisory outcomes (2026-09):** EVERY row with a trade setup is seeded at log time (not only panel rows); same-setup re-logs within 60 min from any surface (`setup_key` = tf|direction|entry|stop|strategy) are `DUPLICATE` rows pointing at the canonical row via `advisory_event_id`, so sum `advisory_r_multiple` over `RESOLVED` rows only. The versioned model advances from later stored closed-candle snapshots (next-open reanchoring, intrabar peak with close-only ratchet, SL-first, scheduled weekend/21:00-UTC closures allowed, 300-bar timeout, $0.30/$0.65 views); when a snapshot no longer overlaps the outcome's last bar it rebuilds the stream from every stored snapshot of that TF before giving up. It is a cleaned-mid evidence model, not execution P&L. **Post-gate truth columns:** `gate_reason` (the warning that removed the setup), `actionable` (grade ≥ C AND setup — the `action` text is written BEFORE the gate chain and can say "STRONG SELL — High-quality setup" on a D row), `cleaned_warnings_json` (warnings a shipped setup superseded — the XAU_CP acceptance's stripped vetoes and, since 2026-09-08, a stale core spread-gate veto; non-empty is NOT a CP marker, use `setup_json.strategy`), `bar_range_median` (= the misnamed `spread` column: median High-Low, NOT a broker spread), `scan_kind` (`blocked` weekend/day early returns carry no news/candles; `failed` = the primary fetch failed — price 0, no signal — readers skip it like `blocked`), `config_fingerprint` (gate config hash — surfaces config drift between processes). **News is content-addressed:** `headlines_json`/`events_json` are interned into `news_snapshots` (they were 73% of a 667 MB DB); read through the `signals_with_news` view or `signal_log.read_news`. Readers must open with `mode=ro`, never `immutable=1` (blind to WAL rows). **Weight-research columns (2026-09):** `bar_ts` (the primary bar LABEL the scan closed on — same string as `indicators_json`, indexed with `timeframe`), `engine_fingerprint` (hash of the SCORING config: indicator params, thresholds, the regime weight table in force, TFs — `SignalAggregator.engine_config_fingerprint`, the gate fingerprint's twin), `code_version` (short git SHA), `penalty_tags_json`/`penalty_total`/`momentum_fading`/`exhaustion_count`, and the post-hoc `watch_level`/`watch_kinds`/`watch_line` written by `signal_log.update_watch` from `violent_watch.attach_watch` (level and kinds are stored even when the display line is off). Every `indicators_json` entry also carries `eff_weight` (base weight × regime multiplier = the weight the composite actually used; `engine.regime_weight_table` is the single source of truth) and `wscore` (its contribution) — null for skipped indicators (weight 0, `_`-prefixed); **divergence and candlestick entries exist only when they fire, so absence is "no vote", not a zero vote**. **Table `scan_outcomes`:** model-free forward moves per scan (`fwd_1h/4h/24h` in $ and ATRs, `mfe_24h_atr`/`mae_24h_atr`, `burst_1h` = a ≥3×ATR bar within the hour, `bars_used`, `complete`), measured from the row's own last closed bar on the stored candle stream; a horizon stays NULL until it has elapsed, and `complete` needs the full 24 h with no gap over 3 bars beyond the scheduled weekend/17:00-NY closures (`advisory_outcome._has_continuity`). `SignalLogger.label_scan_outcomes` runs opportunistically at the end of `log()` (at most once per 60 s per process, 150 rows and ~2 s per pass, idempotent — a row is rewritten only when its numbers change), so no daemon is needed; `scripts/label_scan_outcomes.py` is the backlog driver (`--dry-run` counts read-only). Readers must dedup by the FIRST row per (`timeframe`, `bar_ts`) — several surfaces log the same bar. Tools: `analyze_signal_log.py` (`--stored-candles` DB-only replay with event/opportunity/one-position books + persona scorecard, `--news` news-sentiment lead/lag + extremes + outcome buckets, `--parity` telegram-vs-dashboard divergence report, `--scan-outcomes` forward-move + hit-rate table by TF×signal/grade with a NEUTRAL baseline and burst rate by watch level), `scripts/backfill_signal_outcomes.py` (seed/dedup/advance legacy rows), `scripts/label_scan_outcomes.py` (label/relabel scan outcomes), `scripts/migrate_signal_log_news.py` (intern legacy blobs; `--vacuum` with both writers stopped). - **`news_engine.py`** — Forex Factory calendar (MyFXBook RSS fallback) + RSS keyword sentiment with gold-inverse keywords. Class-level caches (fresh engine per scan). Optional DeepSeek enrichment via `llm_client.py` (requests-only, no SDK): batched headline sentiment overriding keyword scores, gold-direction on release surprises, and a market-summary blurb (`SignalReport.news_summary`) — **display/warning-only, keyword fallback on any failure, inert without a key**; prompts are static module constants (DeepSeek prefix-cache pricing) and requests send `thinking: disabled` (V4-Flash reasoning otherwise blows the timeout). By XAU-tuned default, event risk and sentiment **warn but do not gate signals** (`event_risk_caps_signals=False`, `sentiment_affects_signals=False`) — the LLM does not change that. **Headline pool (2026-09-04):** only headlines ≤ 4 h old with a parsable publish time (`headline_max_age_seconds=14400`, undated items dropped), and `get_fundamental_context` drops headlines whose every relevant score is inside ±0.1 (`_is_neutral_headline`, `NEUTRAL_HEADLINE_THRESHOLD`) before the per-currency mean — so `news_sentiment` from that date is a mean over directional headlines only (bolder values; log comparisons must split at the deploy date). The market summary prompt writes for someone who has never traded (no acronyms), sends headlines newest-first with the platform's own gold direction, and caps at 400 chars. **Verdict 2026-09-04** (`backtest/research/news_sentiment_value_2026-09-04.md`): `news_sentiment` is a trailing-24h echo of price (r≈0.67 with the past day, ≈0.03 with the next; flips 44 h after tops / 16 h after bottoms) — both gating flags stay False; re-check with `analyze_signal_log.py --stored-candles --news`. +- **`news_archive.py`** — durable headline corpus in its OWN database (`logs/newsdata.db`), separate from `signals.db` so a signal-log rotation cannot take the news with it (that is exactly how the 3-week corpus behind `news_sentiment_value_2026-09-04.md` was lost — 4,054 distinct headlines, unrecoverable, `logs/` is gitignored). One row per distinct headline keyed on `sha256(headline)`, carrying the text, description, link, publisher, publish time, the score, `score_source` (`keyword`/`llm`/`model` — a new `NewsSentiment` field, also mirrored into `signal_log._headline_to_dict`), and `teacher_model`/`teacher_prompt_sha1` so labels made under a changed prompt are never pooled with older ones. **Writes are one atomic UPSERT per row and upgrade-only**: an `llm` score always replaces a `keyword` one (the >25/cycle batch-overflow tail gets its DeepSeek label on the next fetch window in a long-running process — verified: 13/13 upgraded) but a `keyword` score can never overwrite an `llm` one; re-seeing a headline always bumps `seen_count`. A read-then-write pair loses increments when the bot, the dashboard publisher and the tests first see a headline together, and the WAL switch itself needs `_with_lock_retry` — both are regression-tested with 3 concurrent writers. Hooked at the END of `news_engine._fetch_news_sentiment_raw`, **before** `_is_neutral_headline` runs in `get_fundamental_context`: archiving downstream of that ±0.1 filter is why `signals.db` holds almost no neutral headlines and would teach a trained scorer that "neutral" barely exists. Never raises (fail-safe by contract, like `signal_log`). The backtester never builds a `NewsEngine`, so replays never archive (test-locked). Seeded by `scripts/backfill_news_archive.py`, which reads both DB shapes read-only and **recomputes** provenance (stored score vs the keyword scorer: differs → `llm`, exact non-zero match → `keyword`, both zero → `unknown`) because neither source recorded it. Purpose: training data for a local numpy-only headline scorer to replace the 24/78 keyword fallback — see the note below. - **`agent_panel.py`** — ai-hedge-fund-style advisory panel: 4 DeepSeek persona calls (macro/technical/risk/contrarian, concurrent on a panel-private pool) + 1 synthesizer producing an **ADVISORY-ONLY** consensus (`AGREE/NEUTRAL/OPPOSE` + confidence + rationale) written onto `SignalReport.panel_*` fields. Attached **post-hoc** by consumers (`telegram_bot` alert path, `main --panel`, `dashboard_publisher`) — never runs inside `analyze_from_dataframes` (backtest parity, test-locked) and no gate reads its fields. Persona payloads are role-filtered: macro gets high-impact events plus only code-scored directional news; technical gets strategy/pattern, primary indicators, MTF, levels, and ATR-normalized candles; risk gets setup strategy, strategy cost assumption, trailing semantics, volatility/events/warnings (never the median candle range as spread); contrarian gets directional news, setup, timestamped latest-closed price, indicators, reasons, and warnings. Spend is bounded by exact cache, full-evidence-identity TTL reuse, and `max_panels_per_day`; changed evidence never inherits a direction-only cached verdict (test-locked — which also means TTL reuse effectively never fires live because price/candles change every bar; the daily cap is the real spend bound). Verdict provenance logs creation time, `MISS`/`EXACT_HIT`/`TTL_REUSE`, and a SHA-256 payload hash. Prompts forbid outside knowledge and outputs remain short. **OPPOSE guard (2026-09):** the 3-week signal-log scorecard showed the contrarian/risk/technical personas oppose almost every trend entry (chair said OPPOSE 73% of the time; opposed trades made +30R, PF 2.6) while the macro persona was the only informative voice (macro=support → 88% win). `_apply_oppose_guard` therefore downgrades a chair OPPOSE to NEUTRAL unless macro opposes, or technical opposes while macro is not supportive; the synthesizer prompt states the same rule. Re-score with `analyze_signal_log.py --stored-candles` (persona scorecard) before changing it again. **Verdict source (2026-09-03, owner decision):** all four personas still run and are logged (`panel_takes_json`), but `agent_panel.verdict_source` defaults to `"macro"` — the alert verdict is the macro persona's stance (support/neutral/oppose → AGREE/NEUTRAL/OPPOSE) and the synthesizer call is skipped (4 LLM calls per panel instead of 5); `"chair"` restores the four-voice synthesizer + OPPOSE guard. Re-score personas monthly with `analyze_signal_log.py --stored-candles`; weights are not to be tuned before ~300 graded outcomes with a held-out half. - **`market_outlook.py`** — hourly DeepSeek market brief for XAU/USD, using ATR-normalized M15/H1/H4 candles, visible indicator states, levels, news, and recent signal history (read `mode=ro`). It returns a written **ADVISORY-ONLY** bias/setup-forming assessment and persists it independently in `outlooks` within `logs/signals.db` together with the news, indicator/regime/level, and candle evidence sent to the model. It may be sent post-scan by Telegram or requested with `main.py --outlook`; it never imports or runs inside the signal path. Hourly dedup is by UTC hour BUCKET (the old `now − last < 3600` rule dropped every other hour because the H1 loop fires at boundary+10 s+scan time) plus its own UTC-day cap (default 12); every silent early return logs `market outlook: skipped (...)` and an empty LLM reply logs a warning — a dark feature must be visible in `logs/telegram.log`. Provider/news failures leave a safe partial brief while LLM/database failures degrade to no outlook. - **`violent_watch.py`** — **display-only** violent-move watch, dark by default (`violent_watch` config block: `enabled`, `alert_line`, `headsup`, `max_headsup_per_day`, `headsup_min_level`; env `VIOLENT_WATCH_ENABLED`). Flags when a "violent bar" (range ≥ 3× prior ATR14) is unusually likely and stamps `SignalReport.watch_line` **post-hoc** via `attach_watch` (telegram alert path, `main.py`, `dashboard_publisher`) — same contract as the agent panel: never imported by `signal_aggregator`/`engine`/`backtest/`, never raises, never read by a gate (test-locked, Module 30). Triggers: a high-impact **USD** calendar release within ±30 min (hot), hot clock slots (weekday 12–14 UTC, weekday 01 UTC, Sunday reopen 22–23 UTC), and **price-extreme** indicator readings (outside the Bollinger bands, RSI <30 / ≥70, outside the Keltner channel). Level = hot when the calendar fires or a clock slot plus ≥1 price extreme; watch on a clock slot alone or ≥2 price extremes. Context readings (ADX slope >3, volatility "extreme", regime "transitioning", MACD STRONG_BUY/STRONG_SELL) are printed but **never raise the level** — the first draft that let them count flagged 27% of bars "hot" at only ~2× odds; the shipped rule flags 8.6% at 3–5× (evidence: `backtest/research/sizing_campaign_2026-09.md` addenda). Direction is called DOWN on bearish readings (70–78% on M5 and M15) and UP only on M5. Telegram also sends a standalone heads-up on the fastest scanned TF (`_HEADSUP_TF`), rate-limited per UTC hour+trigger-set and per day; it explicitly tells the owner **not** to touch open trades — every protection tested lost money, the stop is the protection. @@ -106,3 +107,4 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz - 2026-09-07 signal-log inspection (fresh 2-day `logs/signals.db` import: NFP Friday + Monday; 5 setups, all BUY, 4 stopped out at local tops — two days is not evidence). Three sample-size-independent fixes shipped: (1) the signal-side 10σ outlier filter now drops only bars whose NEXT close snaps back ≥80% (it had been deleting ~36 real M5 news bars/yr in the backtester and a different ~24/yr live, e.g. the 2026-09-04 12:30 UTC NFP bar; 3-yr drops are now M5 2 / M15 1 / H1 0). Book impact at $0.65, 12mo: M5 557/+112.21R/PF 1.56 → 560/+108.81R/PF 1.54 (halves +55.6/+53.2), M15 397/+73.93/1.53 → 398/+74.61/1.53 (max DD 10.8→9.1R), H1 byte-identical; 3-yr (2023-09-05→2026-09-05): M5 857/+161.3/1.53 → 870/+153.6/1.50, M15 1086/+148.7/1.37 → 1092/+151.9/1.38, H1 255/+70.5/1.98 → 253/+70.0/1.98, H4 113/+22.7/1.63 → 114/+21.7/1.59 — only trades on the flagged event days move (M5 gives back a few Jan–Mar 2026 crash-period winners); the pinned M5/M15 CSVs are therefore no longer byte-reproducible (regenerate with the parity recipe when re-pinning). (2) Oanda candle fetches retry along a 2/5/10 s ladder under a 45 s budget (11 boundary scans in two days had lost their primary frame). (3) A scan that still loses its primary frame logs `scan_kind='failed'` and readers skip it. The live dashboard publisher was found dead (0 rows Friday, stopped Mon 02:20 UTC): install `deploy/gold-dashboard.service` on the live server. - 2026-09-08 delivery-side fixes from the 3-day signal-log + telegram-log review (nothing in the signal core changed; 468 tests): (1) `_drop_stale_spread_gate_vetoes` (after every setup archetype, so the XAU_PB path is covered too) moves a core generator's stale `SPREAD GATE:` veto to `cleaned_warnings` when a later generator ships a setup — the 3-yr H1 and M15 books were regenerated before/after and are identical on every trade column (H1 253/+70.02R/PF 1.98, M15 1092/+151.90R/PF 1.38); only the exported `warnings` text differs, on 50/253 H1 and 358/1092 M15 trades, i.e. a fifth to a third of all A-grade alerts had been carrying "costs would eat the edge" beside a valid target; (2) Telegram alerts now require a trade setup (`_is_alert_candidate`, scan + breach paths; no-setup candidates log `alert skipped: … (gate reason)`); (3) every delivered alert / breach alert / violent-move heads-up logs an `… sent:` INFO line with the signal-log row id (the review had to reconstruct sends from panel rows); (4) `telegram.log` rotates daily × 30 instead of 3-hourly × 240 (old `_HH` files on the live box need one manual delete). Review facts (not evidence — three days): 8 setups, 1 right; strong readings (|score| ≥ 20) were followed by reversals in 16 of 18 episodes; hot watch level 0/25 bursts on M5 (calendar trigger never fired, 15 of 33 hot stamps were the 12–14 UTC slot on Labor Day — no holiday awareness anywhere); Oanda published no XAU bars 18:30–22:00 UTC on Labor Day and the bot scanned the stale bar 43×; the retry ladder recovered all 25 post-deploy 401 bursts (41 of 58 incidents sit on the :00 boundary); the live dashboard publisher ran only 01:00–02:20 UTC Sep 7 and is dead again. - 2026-09-06 stale-code cleanup (behaviour-neutral; suite green, M5/H1 12-month books byte-identical after every signal-side commit): `main.py --timeframe` now accepts M5/M15/H1/H4 and takes its confirmation TFs from the pair factory (the CLI's private table had paired M5 with M1+M15); `PlatformConfig.pip_value` is gone (`PAIR_INFO` is the source); removed as dead or superseded: `_get_affected_pairs`/`event_impact_on_pairs`, `SignalLogger.advance_with_bars`/`path`/`news_for`, `reset_llm_client`, the Telegram scan-summary/keyboard helpers and duplicate confirmation map, the 2026-07 Oanda-vs-TD research pager+study, `burst_profile.py`, `hour_analysis.py`, `scripts/compare_oanda_tiingo.py`, the never-implemented LLM-provider plan and the merged config.json spec. Second pass the same day (owner directive: keep only what the system needs): every research script and data file under `backtest/research/` (book_report, book_diff, the burst tools, sizing_model, commit_ladder, llm_news_eval + golden CSV, ff_calendar_scraper, calendar_impact, watch_rule_eval, the Dukascopy spread study + CSVs, the panel dry-test JSONL), the two completed study scripts in `scripts/`, the architecture diagram + generator and the data-source test plan were deleted; the campaign `.md` reports stay as the decision ledger. Git history before `953c5ad` has all of it. Third pass (owner: "remove dead functions as well"): the write-only pb/cp rejection counters, the unreachable `_TP_QUALITY_SCALE` B/C entries, the engine's unused DataFrame argument and M1 arm, the providers' generic-symbol guesses (unknown symbols now raise), the sub-$10 price format, the `=X` pair parsing and the unread `minutes_until`/`is_high_impact` news.json keys are gone. +- 2026-09-08 news archive shipped (`news_archive.py`, 483 tests; nothing in the signal core changed, both news gating flags stay `False`, books untouched). Motivation: headlines only ever survived as a side effect of signal logging and `signals.db` was rotated around 2026-09-04, destroying the 3-week / 4,054-headline corpus the news study was written on — with `logs/` gitignored, git could not recover it and nobody noticed for four days. Day-one corpus after backfilling both DBs on disk: **1,016 distinct headlines, 846 of them DeepSeek-labelled**, growing ~200/day (~1,400/week). The point of collecting it: distil DeepSeek's headline scoring (78/78 on the 2026-08-06 golden set) into a numpy-only local model so the fallback stops being the keyword scorer (**24/78 = 31%**, which is what silently runs on no key / disabled / daily cap / timeout / HTTP or parse failure / batch overflow). **Do not train yet** — 800-odd labels gives a holdout too small to grade (SE ±0.032); the pre-registered plan is ~4,000 deduplicated headlines over ≥3 weeks, time-split with near-duplicates clustered first (the same story is re-titled by several outlets, so a random split scores memorisation), headline text ONLY (the teacher never sees `description` — `_llm_headline_scores` sends titles), graded against the golden set recoverable with `git show 1cebebf^:backtest/research/llm_news_golden.csv`. When it ships it slots in as DeepSeek → model → keyword and stays display-only. Training happens offline in `venv/`; a Module 32 test now AST-walks the root, `providers/` and `pair_signals/` asserting no `sklearn`/`scipy`/`lightgbm`/`torch` import and that none of them reach `requirements.txt` — that guard did not exist before. diff --git a/app_config.py b/app_config.py index ed3fe62..21acd6d 100644 --- a/app_config.py +++ b/app_config.py @@ -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( diff --git a/config.json.example b/config.json.example index 1b7fff5..41e409f 100644 --- a/config.json.example +++ b/config.json.example @@ -66,5 +66,9 @@ "path": "logs/signals.db", "log_blocked_scans": true }, + "news_archive": { + "enabled": true, + "path": "logs/newsdata.db" + }, "log_level": "INFO" } diff --git a/news_archive.py b/news_archive.py new file mode 100644 index 0000000..1965e1c --- /dev/null +++ b/news_archive.py @@ -0,0 +1,325 @@ +""" +ForexBot2 — News archive (durable headline corpus) +================================================== + +Every headline the news engine fetches is written here, exactly as it +arrived, together with the score that was attached to it and **which +scorer produced that score**. One row per distinct headline, kept +forever. + +Why this exists +--------------- +Headlines used to survive only as a side effect of signal logging, inside +``logs/signals.db``. That database was rotated/rebuilt around 2026-09-04 +and three weeks of news — 4,054 distinct headlines, the corpus the +``news_sentiment_value_2026-09-04.md`` study was written on — was lost. +``logs/`` is gitignored, so git could not bring it back and nobody +noticed until 2026-09-08. + +This module writes to its own database (``logs/newsdata.db`` by default) +so signal logging can be wiped, rotated or rebuilt without touching the +news corpus. + +What the corpus is for +---------------------- +Training a small local scorer that reproduces DeepSeek's headline +labelling without calling it. On the 2026-08-06 golden set DeepSeek +scored 78/78 and the built-in keyword scorer it overrides scored 24/78 +(31%) — and the keyword score is what stands whenever DeepSeek is +unavailable (no key, disabled, daily cap, timeout, HTTP/parse failure, +or the >25-new-headlines-per-cycle batch overflow). ``score_source`` +records which one produced each row; training filters on ``'llm'``. + +Contract +-------- +* **Never raises.** ``archive_headlines`` swallows every exception. A + logging failure must not break a scan — the same contract + ``signal_log.log_signal`` holds. +* **Upgrade-only scores.** A row already marked ``llm`` is never + overwritten by a later ``keyword`` score. A headline first seen under + the batch-overflow cap is keyword-scored, then gets its DeepSeek score + on a later fetch window; the better label wins and is never demoted. +* **Dedup by headline text.** ``hash`` = sha256 of the headline. A + headline seen 50 times is one row with ``last_seen`` bumped. +* **stdlib only** (``sqlite3``), like ``signal_log``. + +Known corpus limitation: a headline mentioning no tracked currency is +never constructed by the news engine at all (``news_engine`` drops it +before scoring), so DeepSeek never labels it either and it cannot appear +here. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import sqlite3 +import threading +import time +from datetime import datetime, timezone +from typing import Any, Dict, Iterable, Optional + +import app_config + +logger = logging.getLogger(__name__) + + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS headlines ( + hash TEXT PRIMARY KEY, + headline TEXT NOT NULL, + description TEXT, + link TEXT, + source TEXT, + publisher TEXT, + published_at TEXT, + first_seen TEXT NOT NULL, + last_seen TEXT NOT NULL, + seen_count INTEGER NOT NULL DEFAULT 1, + score REAL, + score_by_currency_json TEXT, + currencies_json TEXT, + keywords_json TEXT, + score_source TEXT NOT NULL DEFAULT 'keyword', + teacher_model TEXT, + teacher_prompt_sha1 TEXT, + origin TEXT NOT NULL DEFAULT 'live' +); +CREATE INDEX IF NOT EXISTS idx_headlines_source ON headlines (score_source); +CREATE INDEX IF NOT EXISTS idx_headlines_first_seen ON headlines (first_seen); +""" + +#: Rank of a score's provenance. A write only lands when its rank is >= +#: the rank already stored, so a keyword score can never overwrite an LLM +#: one (the batch-overflow case) but an LLM score always upgrades. +_SOURCE_RANK = {"unknown": 0, "keyword": 1, "model": 2, "llm": 3} + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def headline_hash(headline: str) -> str: + """Primary key for a headline: sha256 of its exact text.""" + return hashlib.sha256(headline.encode("utf-8")).hexdigest() + + +def _json_or_none(value: Any) -> Optional[str]: + if value is None: + return None + try: + return json.dumps(value, default=str) + except Exception: + return None + + +def _iso_or_none(value: Any) -> Optional[str]: + if value is None: + return None + if isinstance(value, datetime): + return value.isoformat() + return str(value) + + +def _finite_score(value: Any) -> Optional[float]: + try: + out = float(value) + except (TypeError, ValueError): + return None + if out != out or out in (float("inf"), float("-inf")): + return None + return out + + +class NewsArchive: + """Thread-safe SQLite writer for the headline corpus. One instance + per database path; safe to share across the bot's executor threads + and the dashboard daemon's pool.""" + + def __init__(self, path: str): + parent = os.path.dirname(path) + if parent: + os.makedirs(parent, exist_ok=True) + self._conn = sqlite3.connect(path, check_same_thread=False) + # busy_timeout BEFORE the journal-mode switch: the bot, the + # dashboard publisher and the tests open this file at once and + # the WAL switch needs a brief exclusive lock. + self._conn.execute("PRAGMA busy_timeout=5000") + self._lock = threading.Lock() + with self._lock: + # busy_timeout does not cover the WAL switch, which needs a + # brief exclusive lock: three processes creating this file at + # once will race it. Same retry signal_log uses. + self._with_lock_retry( + lambda: self._conn.execute("PRAGMA journal_mode=WAL")) + self._with_lock_retry(lambda: self._conn.executescript(_SCHEMA)) + self._conn.commit() + + @staticmethod + def _with_lock_retry(fn, attempts: int = 8, delay: float = 0.05): + """Run ``fn`` again on 'database is locked' (concurrent openers + racing the same schema step); other errors propagate at once.""" + for attempt in range(attempts): + try: + return fn() + except sqlite3.OperationalError as exc: + if "locked" not in str(exc).lower() or attempt == attempts - 1: + raise + time.sleep(delay * (attempt + 1)) + + def close(self) -> None: + with self._lock: + try: + self._conn.close() + except Exception: + pass + + #: Rank of the score already stored, computed in SQL so the compare + #: happens inside the same atomic statement as the write. + _STORED_RANK = ("CASE headlines.score_source WHEN 'llm' THEN 3" + " WHEN 'model' THEN 2 WHEN 'keyword' THEN 1 ELSE 0 END") + + def store(self, rows: Iterable[Dict[str, Any]]) -> int: + """Insert-or-upgrade a batch of prepared rows. Returns the number + of rows written. Raises on a database failure — the module-level + wrapper is what guarantees a scan cannot break. + + One atomic UPSERT per row. A read-then-write pair would lose an + increment whenever two processes (bot, dashboard publisher, tests) + first see the same headline at the same moment: both would find it + absent, one INSERT would win and the other's count would vanish. + """ + rows = list(rows) + if not rows: + return 0 + now = _utc_now_iso() + wins = f"? >= {self._STORED_RANK}" # incoming rank vs stored rank + sql = ( + "INSERT INTO headlines (" + " hash, headline, description, link, source, publisher," + " published_at, first_seen, last_seen, seen_count," + " score, score_by_currency_json, currencies_json," + " keywords_json, score_source, teacher_model," + " teacher_prompt_sha1, origin" + ") VALUES (?,?,?,?,?,?,?,?,?,1,?,?,?,?,?,?,?,?)" + " ON CONFLICT(hash) DO UPDATE SET" + " last_seen = excluded.last_seen," + " seen_count = headlines.seen_count + 1," + # Upgrade-only: a weaker scorer never overwrites a stronger + # one's label, but re-seeing a headline always counts. + f" score = CASE WHEN {wins} THEN excluded.score" + " ELSE headlines.score END," + f" score_by_currency_json = CASE WHEN {wins}" + " THEN excluded.score_by_currency_json" + " ELSE headlines.score_by_currency_json END," + f" currencies_json = CASE WHEN {wins} THEN excluded.currencies_json" + " ELSE headlines.currencies_json END," + f" keywords_json = CASE WHEN {wins} THEN excluded.keywords_json" + " ELSE headlines.keywords_json END," + f" teacher_model = CASE WHEN {wins} THEN excluded.teacher_model" + " ELSE headlines.teacher_model END," + f" teacher_prompt_sha1 = CASE WHEN {wins}" + " THEN excluded.teacher_prompt_sha1" + " ELSE headlines.teacher_prompt_sha1 END," + f" score_source = CASE WHEN {wins} THEN excluded.score_source" + " ELSE headlines.score_source END," + # Enrichment fills gaps but never blanks what is already there + # (the legacy backfill shape carries no description or link). + " description = COALESCE(excluded.description, headlines.description)," + " link = COALESCE(excluded.link, headlines.link)," + " publisher = COALESCE(excluded.publisher, headlines.publisher)," + " published_at = COALESCE(excluded.published_at," + " headlines.published_at)") + + params = [] + for row in rows: + rank = _SOURCE_RANK.get(row["score_source"], 0) + params.append(( + row["hash"], row["headline"], row["description"], row["link"], + row["source"], row["publisher"], row["published_at"], + row.get("first_seen") or now, row.get("last_seen") or now, + row["score"], row["score_by_currency_json"], + row["currencies_json"], row["keywords_json"], + row["score_source"], row["teacher_model"], + row["teacher_prompt_sha1"], row.get("origin", "live"), + ) + (rank,) * sql.count("? >=")) + + with self._lock: + self._with_lock_retry(lambda: self._conn.executemany(sql, params)) + self._conn.commit() + return len(rows) + + +def build_row(headline: Any, *, teacher_model: Optional[str] = None, + teacher_prompt_sha1: Optional[str] = None, + origin: str = "live", + first_seen: Optional[str] = None, + last_seen: Optional[str] = None) -> Optional[Dict[str, Any]]: + """Flatten one ``NewsSentiment`` (or an equivalent object) into a row. + + Returns None when the object carries no usable headline text.""" + text = getattr(headline, "headline", None) + if not isinstance(text, str) or not text.strip(): + return None + source = getattr(headline, "score_source", None) or "keyword" + return { + "hash": headline_hash(text), + "headline": text, + "description": getattr(headline, "description", None), + "link": getattr(headline, "link", None), + "source": getattr(headline, "source", None), + "publisher": getattr(headline, "publisher", None), + "published_at": _iso_or_none(getattr(headline, "timestamp", None)), + "first_seen": first_seen, + "last_seen": last_seen, + "score": _finite_score(getattr(headline, "sentiment_score", None)), + "score_by_currency_json": _json_or_none( + getattr(headline, "sentiment_by_currency", None)), + "currencies_json": _json_or_none( + getattr(headline, "currencies_mentioned", None)), + "keywords_json": _json_or_none(getattr(headline, "keywords", None)), + "score_source": source, + "teacher_model": teacher_model if source == "llm" else None, + "teacher_prompt_sha1": ( + teacher_prompt_sha1 if source == "llm" else None), + "origin": origin, + } + + +#: path -> NewsArchive +_archives: Dict[str, NewsArchive] = {} +_archives_lock = threading.Lock() + + +def archive_headlines(headlines: Any, *, teacher_model: Optional[str] = None, + teacher_prompt_sha1: Optional[str] = None, + origin: str = "live") -> None: + """Persist a batch of scored headlines if the news_archive config + block enables it. NEVER raises — a logging failure must not break a + scan.""" + try: + if not headlines: + return + if not app_config.news_archive_enabled(): + return + rows = [] + for headline in headlines: + row = build_row(headline, teacher_model=teacher_model, + teacher_prompt_sha1=teacher_prompt_sha1, + origin=origin) + if row is not None: + rows.append(row) + if not rows: + return + path = app_config.news_archive_path() + with _archives_lock: + archive = _archives.get(path) + if archive is None: + archive = NewsArchive(path) + _archives[path] = archive + archive.store(rows) + except Exception: + logger.warning("news_archive: failed to persist headlines", + exc_info=True) diff --git a/news_engine.py b/news_engine.py index 286224b..60f173b 100755 --- a/news_engine.py +++ b/news_engine.py @@ -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 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,16 @@ 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. + news_archive.archive_headlines( + headlines, + teacher_model=getattr(get_llm_client(), "_model", None), + teacher_prompt_sha1=_LLM_SENTIMENT_PROMPT_SHA1) + return headlines def _analyze_headline_sentiment( @@ -1003,6 +1027,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 diff --git a/scripts/backfill_news_archive.py b/scripts/backfill_news_archive.py new file mode 100755 index 0000000..ed1f954 --- /dev/null +++ b/scripts/backfill_news_archive.py @@ -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()) diff --git a/signal_log.py b/signal_log.py index 0261b8b..5a01406 100644 --- a/signal_log.py +++ b/signal_log.py @@ -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), } diff --git a/test_all.py b/test_all.py index 99b3f5d..c4422ef 100644 --- a/test_all.py +++ b/test_all.py @@ -10739,6 +10739,391 @@ 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: 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 = os.environ.get("NEWS_ARCHIVE_ENABLED") + os.environ["NEWS_ARCHIVE_ENABLED"] = "0" + try: + assert not _ac.news_archive_enabled() + _na.archive_headlines([_m32_head()]) + finally: + if saved is None: + os.environ.pop("NEWS_ARCHIVE_ENABLED", None) + else: + os.environ["NEWS_ARCHIVE_ENABLED"] = saved + assert not os.path.exists(path), "nothing may be written when off" + + +@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"" + b"Gold steady as traders await US CPI" + b"" + _m19_pubdate(1) + b"" + b"") + + 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" + 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 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 # ══════════════════════════════════════════════════════════════════════