From 2836024e9d0915cb0e45c521571c36aaba85fccc Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Sun, 13 Sep 2026 21:42:37 -0400 Subject: [PATCH] audit 2026-09-13: eleven delivery/measurement fixes; books byte-identical MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signal path - A shipped setup always names a direction: grade C needs confidence >= 45 (gate 6c grants B at >= 45) while BUY/SELL needs >= 50, so C/B reports reached the alert gate as NEUTRAL with a full setup. The backtester traded them (the replay never reads final_signal) while Telegram rejected NEUTRAL first — 5/91 trades, +3.7R of +22.2R, in the H1 12-month book. _attach_core_setup stamps the setup's direction and repairs the pre-gate action text; no-setup reports stay NEUTRAL. - analyze_pair fails the scan when ANY required frame is missing ("DATA UNAVAILABLE:", scan_kind='failed'); a lost confirmation frame used to run with an unearned -15 MTF penalty. Outcomes and labels - advisory_outcome.resolve_locked_floor: a gap the database cannot fill resolves at the locked trailing stop (GAP_LOCKED, result.floor) when the stop had crossed the entry; the rebuild fires on both gap reasons (rows 915/920: +1.14R / +0.84R floors; resolved sum -2.46R -> -0.48R). - label_scan_outcomes: SCAN_LABEL_MAX_AGE_H = 72 h, stale rows never anchor the shared bar stream, labeled_at bumped on re-examination, _stream_since warns at its row cap (the Labor-Day pile would have blinded scan_outcomes around 2026-09-14). Providers and scheduling - The daily settlement pause (Mon-Thu 17:00 New York) is a scheduled closure for _has_latest_closed_bar (whole-span label walk-back) and _is_data_fresh (closed seconds are not age); bot and dashboard skip scans whose just-closed label sits in the pause (61 false stale rows in three nights). - fetch_range retries a failed page in place (three failures on one window raise) and logs an ERROR for interior gaps of >3 unscheduled bars. - pd.Timestamp.utcnow() replaced in providers/base.py and engine.py; requirements pin pandas<4, numpy<3. - telegram_bot: boundary loops share one scan per bar with the breach handler (_remember_scan keys on the analysed bar; breach defers 20 s after an M15 boundary); the rotating file handler attaches in main() only (the test suite used to write into logs/telegram.log). - dashboard_publisher: catch_up_timeframes re-scans boundaries missed while a cycle overran or the process was down. Backtester and engine guards - _replay_span returns the per-span GPU cache-miss count; run_backtest sums, logs it (must be 0) and writes cache_fallbacks to the manifest. - regime_weight_table raises on a partial regime_weights_override. Verification: 519 tests; M5/M15 2-yr, M15 OOS, H1/H4 3-yr books byte-identical before and after; live signals.db untouched (checked on a copy). CLAUDE.md carries the dated entry and the blocked-hours union semantics that every book was validated under. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_0139xz1J6YccndA2be6yFnEo --- CLAUDE.md | 7 +- advisory_outcome.py | 33 ++ backtest/gpu_backtest.py | 34 +- backtest/replay_record.py | 3 + dashboard_publisher.py | 98 +++- engine.py | 53 +- providers/base.py | 142 +++++- requirements.txt | 4 +- signal_aggregator.py | 70 ++- signal_log.py | 143 +++++- telegram_bot.py | 64 ++- test_all.py | 1021 ++++++++++++++++++++++++++++++++++++- 12 files changed, 1602 insertions(+), 70 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index ac1a957..062a3c1 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 - # (501 tests / 33 modules as of 2026-09-10) + # (519 tests / 34 modules as of 2026-09-13) 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 @@ -67,9 +67,9 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz - **`providers/`** — one class per provider subclassing `BaseDataProvider` (`providers/base.py`), which owns everything the pipeline depends on: TTL+boundary cache, forming-candle drop, freshness check, validate/clean, paginated `fetch_range` (feeds the backtester from any provider), optional tick streaming. Subclasses implement only `_fetch_raw` + symbol/interval maps. **Cleaning rule (2026-09-07):** the signal-side outlier filter is a bad-print filter, not a volatility filter — a >10σ close-to-close bar is dropped only when the NEXT close snaps back ≥80% of the way (`OUTLIER_SIGMA`/`OUTLIER_REVERT_FRACTION` on the base class); persisting moves (NFP/CPI/Sunday-gap bars) and the newest bar are always kept. The previous unconditional 10σ rule deleted ~36 real M5 bars/yr in the backtester and a *different* ~24/yr live (window-dependent σ), hiding e.g. the 2026-09-04 12:30 UTC NFP bar from ATR/exhaustion math. Oanda fetches retry 401/5xx/network faults along `RETRY_DELAYS_S` (2/5/10 s) under a 45 s `RETRY_BUDGET_S`; a scan that still loses its primary frame is logged with `scan_kind='failed'`. **Live fetch depth (2026-09-10):** every live fetch requests `max(MIN_LIVE_BARS=1000, min(60 days × bars/day + 20, 5000))` bars (`_live_outputsize`) — M5/M15 5000, H1 1460, H4 1000 (was 380), D1 1000 (was 300). The engine's EMAs use `ewm(adjust=False)`, so the first bar of the frame keeps weight (1−2/201)^n in the 200-bar trend EMA; at 380/300 bars the live H4/D1 trend reading flipped on ~0.3–0.75% of bars versus the backtester's converged history (and on the live D1 bar of 2026-09-10 itself), at ≥ 600 bars on none (`backtest/research/cycle_2026-09-10.md`). Registry + `make_data_provider(config, name=None)` in `providers/__init__.py`; config.json `data_providers.active` selects the live candle feed and the optional `data_providers.streaming` scalar selects a different provider for the tick stream. A test enforces that subclasses don't override shared base methods (`supports_streaming`/`iter_ticks` are the sanctioned streaming override points). Tiingo streams mid-price ticks (`wss://api.tiingo.com/fx`) — do not re-add spread display from its free FX feed (two-tier indicative quotes, removed 2026-07-23); the no-spread rule extends to the WS stream. - **`engine.py`** — `TechnicalEngine`: 9 scored indicators (ema_trend, rsi, macd, bollinger, adx, roc, keltner, divergence, candlestick) + ATR as context. Regime classification (trending_up/down, `trending` = mixed/direction-UNKNOWN, transitioning, ranging) drives per-indicator weight multipliers and regime-aware RSI reinterpretation. Compute order matters (ATR first; MACD always computed even when not a pillar — divergence needs the histogram). Private series ride on the result as `_underscore` attrs; scorers/formatters skip anything `_`-prefixed or weight-0. -- **`signal_aggregator.py`** — MTF agreement, diminishing-decay penalty system, exhaustion hard block, quality grading, then an ordered chain of post-grade gates (session, XAU SELL quality/vol/hour, regime blocks, M5 noise gates), then three setup generators: XAU ATR (A+/A), generic swing-based, plus the M5 pullback (`XAU_PB`) and chart-pattern (`XAU_CP`, `chart_patterns.py`) entry paths. Step 7 lives in `_attach_core_setup`; a final pass after the PB/CP archetypes (`_drop_stale_spread_gate_vetoes`, 2026-09-08) moves a core generator's `SPREAD GATE:` veto to `cleaned_warnings` whenever a later generator shipped a setup (metadata only — books unchanged; the exported `warnings` text column can differ; the prefix is the `_SPREAD_GATE_PREFIX` constant shared by the emitters and both cleanups). The PB/CP generators run AFTER the gate chain and force quality to "B"; the CP path deliberately accepts grade-F composites and re-checks only session/day/big-candle (via warning prefixes) plus its own SELL/vol/RSI gates — it does NOT re-check exhaustion or the M5 min-score/MTF gates. **This is validated behavior, not a bug** (2026-08-04 audit: 226/586 trades in the shipped M5 12mo book are XAU_CP, +38.9R, 104 of them exhaustion-flagged and net-positive) — changing it is a rule-1 revalidation event. `TradeSetup.strategy` selects trailing params; trail params are stamped on the setup from the *producing* config so per-TF overrides reach formatters. +- **`signal_aggregator.py`** — MTF agreement, diminishing-decay penalty system, exhaustion hard block, quality grading, then an ordered chain of post-grade gates (session, XAU SELL quality/vol/hour, regime blocks, M5 noise gates), then three setup generators: XAU ATR (A+/A), generic swing-based, plus the M5 pullback (`XAU_PB`) and chart-pattern (`XAU_CP`, `chart_patterns.py`) entry paths. Step 7 lives in `_attach_core_setup`, which also stamps BUY/SELL on any shipped setup whose `final_signal` was still NEUTRAL (grade C at confidence 45–50, or gate 6c's B — the signal thresholds need ≥ 50) and repairs the pre-gate action text: the backtester always traded those setups (the replay never reads `final_signal`) while Telegram's alert gate rejected NEUTRAL first (2026-09-13; 5/91 trades, +3.7R of +22.2R, in the H1 12-mo book); a final pass after the PB/CP archetypes (`_drop_stale_spread_gate_vetoes`, 2026-09-08) moves a core generator's `SPREAD GATE:` veto to `cleaned_warnings` whenever a later generator shipped a setup (metadata only — books unchanged; the exported `warnings` text column can differ; the prefix is the `_SPREAD_GATE_PREFIX` constant shared by the emitters and both cleanups). The PB/CP generators run AFTER the gate chain and force quality to "B"; the CP path deliberately accepts grade-F composites and re-checks only session/day/big-candle (via warning prefixes) plus its own SELL/vol/RSI gates — it does NOT re-check exhaustion or the M5 min-score/MTF gates. **This is validated behavior, not a bug** (2026-08-04 audit: 226/586 trades in the shipped M5 12mo book are XAU_CP, +38.9R, 104 of them exhaustion-flagged and net-positive) — changing it is a rule-1 revalidation event. `TradeSetup.strategy` selects trailing params; trail params are stamped on the setup from the *producing* config so per-TF overrides reach formatters. - **`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), M15 RSI regime-multiplier × 0.5 via `technical.regime_weights_override` (2026-09-09), 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). +- **`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 (either gap reason) it rebuilds the stream from every stored snapshot of that TF; a gap the database cannot fill resolves at the locked trailing stop as a FLOOR (`exit_reason=GAP_LOCKED`, `result.floor=true`, `advisory_outcome.resolve_locked_floor`) when the stop had already crossed the entry, and stays `UNRESOLVABLE` otherwise (2026-09-13 — rows 915/920 of the Sep-8 shorts were +1.14R/+0.84R floors the scorecard had been dropping). 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` = ANY required frame — primary, confirmation, or the regime-filter TF when that filter is on — failed to fetch: price 0, no signal, no analysis run (2026-09-13; was primary-only, and a lost confirmation frame used to run the scan with an unearned −15 MTF penalty) — 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, but its `labeled_at` is bumped so it rotates to the back; an incomplete row older than `SCAN_LABEL_MAX_AGE_H` = 72 h (measured from the newest logged row) is FINAL and never anchors the shared bar stream, and `_stream_since` warns when it hits its row cap — before 2026-09-13 the Labor-Day pile anchored the stream and the 4,000-row cap would have blinded every later row), 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. @@ -115,3 +115,4 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz - 2026-09-09 replay recording shipped (`backtest/replay_record.py`, `backtest/replay_scorecard.py`, Module 33, 498 tests; nothing in the signal core changed, books byte-identical — `backtest/research/replay_recording_2026-09-09.md`). `--record DIR` stores every ACCEPTED trade's decision evidence in Parquet: each indicator's vote / value / confidence / BASE weight / regime multiplier / EFFECTIVE weight / contribution on the primary AND both confirmation TFs (`indicators.parquet`, long; absence = the indicator did not fire, never a zero row), per-TF regime/composite, penalty tags, setup, outcome, and the Forex Factory calendar events from 4 h before entry to exit (`events.parquet`, all-day rows matched on the New-York date; scalar `ev_*`/`us_holiday` columns on `trades.parquet`); `--record-bars` adds one float32 row per ANALYZED bar with the same votes plus 1 h/4 h/24 h forward moves in ATRs (`bars.parquet`). The manifest carries the engine + gate fingerprints, the regime weight table in force and the book MD5. Calendar source is the gitignored `backtest/cache/calendar.parquet` (15,124 FF events 2023-09-01→2026-09-10, unix-UTC `dateline`; `timeLabel` is display-only); the scraper that built it is back at `scripts/ff_calendar_scraper.py` (resume + idempotent merge, ~1.6 s/request). The 2-year books (2024-09-05→2026-09-05, $0.65) live in `backtest/research/replay_2026-09/{m5,m15}/`: M5 781 / +146.07R / PF 1.52 (halves +28.0/+118.1), M15 793 / +117.62R / PF 1.41 (halves +38.2/+79.4); control-vs-recorded MD5 equal on both. First-look facts are descriptive only (Bollinger opposes 94% of M5 entries; divergence-opposed M5 trades did best again; every indicator's SELL votes were followed by moves against them on both TFs; the bar-level baseline drifts +1.9 ATR/24 h on M5 so BUY votes must be drift-adjusted first) — every candidate change stays a rule-1 rerun with a control book. - 2026-09-09 weight-change test (`backtest/research/weight_test_2026-09-09.md`, first use of the recorded replay files): a drift-adjusted per-bar screen ranked 108 regime-multiplier candidates per TF; the top 3 per TF (24 h-sign filter) went through the full 2-year replay against same-harness controls plus the 2024-06→2025-03 OOS window under a rule pre-registered before any run. M5: all three MACD candidates FAIL (−10 to −46R; the screen ranked them first — never trust the per-bar screen on M5 without a replay). M15: ema_trend × 2 FAILS (PF 1.32, DD 26R, +55% trades), divergence → 0 in trending FAILS (H1 −3R, DD); **RSI multiplier × 0.5 in every regime PASSES a–f** (2-yr 793/+117.6R/PF 1.41 → 931/+146.1R/PF 1.43, halves +27.9/+0.6, OOS +40.6 → +60.7R PF 1.49, 3-yr +151.9 → +186.6R PF 1.39 but max DD 15.5 → 19.3R and 2026 −4R; added trades are mostly B-grade BUYs). SHIPPED the same day on the owner's decision as an M15-only `technical.regime_weights_override` in `pair_signals/xauusd.py` (M15 override 5: the shipped table with RSI halved in every regime — 0.5 trending family / 0.75 ranging / 0.5 transitioning; derived from the default table, pinned literally by the Module 9 equivalence test) — the factory run without `--set` reproduces the validated candidate book byte-for-byte (parity MD5 in `weight_test_2026-09-09.md`); M15 canonical book re-pinned to `backtest/oanda_m15_2yr_065_20260909.csv`. Live M15 `engine_fingerprint` changes from that deploy; signal-log comparisons across it must split at the deploy time. Suite 499 tests. - 2026-09-10 improvement cycle 1 (`backtest/research/cycle_2026-09-10.md`; 501 tests; no signal-core change): HEAD baseline re-established on the 3-yr window at $0.65 (M5 870/+153.6R/PF 1.50, M15 1273/+186.6/1.39, H1 253/+70.0/1.98, H4 114/+21.7/1.59) and the three pinned MD5s (M5 2-yr, M15 2-yr, M15 OOS) reproduced exactly. Robustness: the cost ladder trips no red flag (every TF keeps PF ≥ 1.3 with positive halves at $0.80; at $1.30 M5 PF 1.36, M15 1.29, H1 1.89, H4 1.53); the intraday edge is a fat right tail (top 5% of trades carry 94% of M5 R and 96% of M15 R, versus 48% on H1; M5 2024 was flat at +1.0R over 102 trades). Pre-registered bucket screen: exactly one qualifying bucket — M15 grade-A SELLs (230 / −14.4R, negative in all three years and OOS; the 2026-09-03 5-yr run said the opposite) — parked as the only cycle-2 candidate. **C1 REJECTED:** H1 divergence multiplier → 0 in every regime (the parked 2026-09 candidate on its now-sanctioned vehicle): 3-yr ΔR +1.0, second half −1.5R, PF 1.98 → 1.89 (its OOS book 88/+21.9/PF 1.92 is trade-identical to the old `disabled_indicators` run, so a 0.0 regime multiplier ≡ disabling for the composite). **C2:** the M15 RSI × 0.5 ship sits on a stable plateau (k = 0.25 / 0.75 keep 78% / 71% of its 2-yr gain, both OOS-positive; R is monotone in k across 0…1.5) — no other k proposed. **Seeding parity SHIPPED:** live fetches floor at `MIN_LIVE_BARS = 1000` (H4 380 → 1000, D1 300 → 1000; measured ema_trend flips 2/267 H4 and 1/313 D1 bars at the old depths, none ≥ 600; the live D1 bar of 2026-09-10 itself read SELL at 300 bars and NEUTRAL at 1000/5000) and the backtester pad targets the same depth (`_history_pad_days`: 62 / 234 / 1400 days) — M5/M15 3-yr books byte-identical, H1/H4 trade-identical with only window-start float noise in the exported columns, so no re-pin. Also fixed in these notes: no PDF docs exist; `pyarrow` is a dev-venv-only dependency of the backtester/recorder. +- 2026-09-13 bug audit (plan file `~/.claude/plans/you-are-python-software-joyful-teacup.md`; 519 tests / 34 modules; the five reference books — M5 W2 `29120a1ee32979431d062f210d41a9d1`, M15 W2 `d185f099482a07f29502cb5a371a9bbf`, M15 OOS `6f8f24b49597ced28f50a19e08c3472f`, H1 W3 `3b45d12bade823178b267c13bc698458`, H4 W3 `e75adae13c839b15736f69c0114e6aa5` (the H1/H4 values are the post-pad books, `cycle_2026-09-10/w3_{1h,4h}_pad.csv`) — byte-identical before and after every change, 0 GPU cache fallbacks). Verified sound, no change: composite = Σwscore/Σeff_weight × 50 on 400/400 DB rows; scan-outcome labels reproduce later scan prices to 0.000 on 844 rows; outcome-model exits reproduced on rows 578/1372/1500; Oanda H4/D1 alignment is UTC; no look-ahead in engine/backtester/chart_patterns; no cross-bar state; parquet caches hold only holiday gaps. Fixed (all delivery/measurement side, books unchanged): (1) NEUTRAL-signal C/B setups now alert (see `_attach_core_setup`); (2) a missing confirmation frame fails the scan (`scan_kind='failed'`, warning `DATA UNAVAILABLE:`); (3) `pd.Timestamp.utcnow()` (deprecated, removed in pandas 4) replaced in `providers/base.py`/`engine.py`, `requirements.txt` pins `pandas<4`, `numpy<3`; (4) advisory outcomes keep locked-profit floors and the rebuild fires on both gap reasons; (5) scan-outcome labeler max-age/anchor bound; (6) the daily settlement pause (Mon–Thu 17:00 NY) is a scheduled closure for the freshness gates — `_is_daily_maintenance`/`_is_scheduled_closure`/`_scheduled_closed_seconds` in `providers/base.py`, the expected label walks back only past labels whose WHOLE span lies in a closure (a D1 bar labelled Sunday 00:00 UTC and an H4 bar labelled Sunday 20:00 exist), and the bot/dashboard skip scans whose just-closed label sits in the pause (61 false "stale" rows in 3 nights before); (7) `fetch_range` retries a failed page in place (3 failures on one window raise) and logs an ERROR for interior gaps of >3 unscheduled bars (holidays look identical — verify, never fatal); (8) `_replay_span` returns the per-span GPU cache-miss count, `run_backtest` sums/logs it (must be 0) and writes `cache_fallbacks` to the `--record` manifest; (9) `regime_weight_table` raises on a partial `regime_weights_override` (every regime × all nine indicators); (10) Telegram boundary loops go through `_cached_or_scan`, `_remember_scan` files a report under the bar it analysed (label + duration), a breach inside 20 s after an M15 boundary waits for the loop's scan (15 duplicate M15 rows in 5 days before), and the rotating file handler is attached in `main()` only (importing the module in tests wrote fixture alerts into `logs/telegram.log`); (11) dashboard `catch_up_timeframes` re-scans any TF whose newest closed boundary is newer than its last scanned one (missed while a cycle overran or the process was down). Semantics recorded, NOT changed (rule 1): `blocked_hours_utc` is applied to the bar LABEL hour in the core (6b) AND the bar CLOSE hour in the live wrapper/backtester, so H1 can only trade bars closing 04–09, 15, 18–20 UTC and H4 only bars closing 08:00/12:00 — parity holds and every book was validated that way. Post-deploy: `scripts/backfill_signal_outcomes.py --reseed-unresolvable` floors the two Sep-8 shorts on the live DB; the `--stored-candles` scorecard (`analyze_signal_log.replay_stored`) still calls `advance_outcome` without the floor step (owner decision pending). diff --git a/advisory_outcome.py b/advisory_outcome.py index d4407ab..ac30f70 100644 --- a/advisory_outcome.py +++ b/advisory_outcome.py @@ -348,6 +348,39 @@ def advance_outcome( return state +def resolve_locked_floor(state: Dict[str, Any]) -> Dict[str, Any]: + """If an UNRESOLVABLE gap outcome had already trailed its stop past the + entry, finish it at that stop as a FLOOR (the true outcome is at least + this good) instead of discarding it. Returns the state (mutated).""" + if state.get("status") != "UNRESOLVABLE": + return state + reason = str(state.get("reason", "")) + # Both gap reasons ("lost continuity" / "contain an internal gap") + # share this prefix; an invalid seed or evaluator error never floors. + if not reason.startswith("stored candle snapshots"): + return state + progress = state.get("progress") + setup = state.get("setup") + if not isinstance(progress, dict) or not isinstance(setup, dict): + return state + entry = _finite(progress.get("entry_price")) + stop = _finite(progress.get("current_stop")) + risk = _finite(setup.get("stop_distance")) + exit_at = progress.get("last_processed_bar") + if (entry is None or stop is None or risk is None or risk <= 0 + or not exit_at): + return state + buy = setup.get("direction") == "BUY" + locked = stop >= entry if buy else stop <= entry + if not locked: + return state + _finish(state, "GAP_LOCKED", str(exit_at), stop) + state["result"]["floor"] = True + state["reason"] = ( + f"{reason} — resolved at the locked trailing stop (floor)") + return state + + def canonical_r(state: Dict[str, Any]) -> Optional[float]: result = state.get("result") value = result.get("net_r_065") if isinstance(result, dict) else None diff --git a/backtest/gpu_backtest.py b/backtest/gpu_backtest.py index 9671606..f8979c1 100644 --- a/backtest/gpu_backtest.py +++ b/backtest/gpu_backtest.py @@ -1540,6 +1540,10 @@ def _replay_span(span: Tuple[int, int]): is applied by the caller in a sequential pass over the returned candidates, so parallel and sequential runs produce identical books. Must stay behaviorally identical to the original sequential loop. + + Returns (skipped, candidates, bar_rows, cache_fallbacks) — the last + element is this span's GPU-cache-miss count, which the caller sums so + a parity hazard inside a worker process cannot stay invisible. """ ctx = _REPLAY_CTX primary_df = ctx["primary_df"] @@ -1550,6 +1554,12 @@ def _replay_span(span: Tuple[int, int]): bt_cfg = ctx["bt_cfg"] primary_dur = ctx["primary_dur"] + # One worker process handles several spans, so the engine's cache-miss + # counter must be per span, not cumulative across them. + engine = aggregator.tech_engine + if hasattr(engine, "fallback_count"): + engine.fallback_count = 0 + skipped: Dict[str, int] = {} candidates: List[Tuple[int, Trade]] = [] is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD" @@ -1719,8 +1729,9 @@ def _replay_span(span: Tuple[int, int]): if tag == "XAU_PB" else 0.0), ))) - return skipped, candidates, (replay_record.finish_bar_rows(bar_buf) - if record_bars else None) + return (skipped, candidates, + (replay_record.finish_bar_rows(bar_buf) if record_bars else None), + int(getattr(engine, "fallback_count", 0))) def _reanchor_entries_to_raw( @@ -1927,14 +1938,28 @@ def run_backtest( skipped_reasons: Dict[str, int] = {} candidates: List[Tuple[int, Trade]] = [] bar_chunks: List = [] - for skips, cands, bars in results: + cache_fallbacks = 0 + for skips, cands, bars, fallbacks in results: for reason, count in skips.items(): skipped_reasons[reason] = skipped_reasons.get(reason, 0) + count candidates.extend(cands) if bars is not None: bar_chunks.append(bars) + cache_fallbacks += fallbacks candidates.sort(key=lambda pc: pc[0]) + # The workers' counters die with their processes, so surface the run + # total here and leave it on the parent engine for callers to read. + aggregator.tech_engine.fallback_count = cache_fallbacks + if cache_fallbacks > 0: + logger.warning( + f"GPU cache fallbacks: {cache_fallbacks} indicator recomputes " + f"(one per _compute_* override per missed bar) fell back to the " + f"live engine's pandas math (different EWM seeding) — parity " + f"hazard; see GPUTechnicalEngine._note_fallback") + else: + logger.info("GPU cache fallbacks: 0") + # ── Phase 2: entry cooldown — the only sequential rule. accepted = _apply_cooldown(candidates, bt_cfg.cooldown_bars) trades: List[Trade] = [t for _, t in accepted] @@ -1985,7 +2010,8 @@ def run_backtest( skipped=skipped_reasons, n_candidates=len(candidates), workers=workers, backend="CuPy GPU" if _GPU_AVAILABLE else "NumPy CPU fallback", - cache_files=cache_names, replay_seconds=replay_seconds), + cache_files=cache_names, replay_seconds=replay_seconds, + cache_fallbacks=cache_fallbacks), sim_df=sim_df, bar_chunks=bar_chunks or None) logger.info(f"--record: wrote run to {bt_cfg.record_dir}") diff --git a/backtest/replay_record.py b/backtest/replay_record.py index fd7b120..c0a1e00 100644 --- a/backtest/replay_record.py +++ b/backtest/replay_record.py @@ -1055,6 +1055,9 @@ def write_run(record_dir, *, trades, book_df, bt_cfg, platform_cfg, aggregator, "max_hold_bars": bt_cfg.max_hold_bars, "fixed_rr_tp": bt_cfg.fixed_rr_tp, "workers": run_info.get("workers"), + # >0 means some bar-analyses fell back to the live engine's pandas + # math instead of the pre-computed cache (a parity hazard). + "cache_fallbacks": run_info.get("cache_fallbacks"), "engine_fingerprint": engine_fp, "gate_fingerprint": gate_fp, "technical_config": dataclasses.asdict(platform_cfg.technical), diff --git a/dashboard_publisher.py b/dashboard_publisher.py index e65a3e8..5578c82 100644 --- a/dashboard_publisher.py +++ b/dashboard_publisher.py @@ -26,6 +26,11 @@ whichever timeframes just closed. Due-classification uses the RAW wake time — flooring 12:00:10 to the 5-minute grid yields the 12:00 boundary that just closed. Do NOT subtract the buffer first: 12:00:10 − 15 s floors to 11:55 and silently skips every H1/H4 scan (v1 bug, 2026-08-02). +A cycle can overrun its 5-minute slot (or the process can be down), so +each wake also catches up the timeframes whose latest closed boundary is +newer than their last scanned one (`catch_up_timeframes`) — without it the +page served an hours-old H1/H4 report behind a fresh countdown +(2026-09-13 audit). Scans run sequentially M5→M15→H1→H4 through one shared data provider so later timeframes hit the warm boundary cache — that ordering IS the @@ -46,10 +51,12 @@ import os import shutil import sys import time -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from logging.handlers import TimedRotatingFileHandler from typing import Dict, List, Optional +import pandas as pd + from agent_panel import attach_panel from app_config import get_app_config, violent_watch_enabled from config import PlatformConfig, Timeframe, XAUUSD @@ -230,10 +237,60 @@ def due_timeframes(wake_dt: datetime) -> List[Timeframe]: """Timeframes whose candle just closed at the boundary this wake belongs to. Takes the RAW wake time (boundary + buffer): flooring to the 5-minute grid yields the boundary that just passed. Never - subtract the buffer first — see module docstring (v1 bug).""" + subtract the buffer first — see module docstring (v1 bug). + + A timeframe whose just-closed bar would carry a label inside gold's + daily settlement pause is skipped: no such bar exists, so the scan + would only re-read the previous one (2026-09-13 audit).""" minutes = wake_dt.hour * 60 + wake_dt.minute floored = minutes - (minutes % 5) - return [tf for tf in DASHBOARD_TFS if floored % TF_INTERVAL_MINUTES[tf] == 0] + boundary = wake_dt.replace( + hour=floored // 60, minute=floored % 60, second=0, microsecond=0) + return [ + tf for tf in DASHBOARD_TFS + if floored % TF_INTERVAL_MINUTES[tf] == 0 + and not _providers_base._is_daily_maintenance( + boundary - timedelta(minutes=TF_INTERVAL_MINUTES[tf])) + ] + + +def catch_up_timeframes(wake_dt: datetime, + last_boundary: Dict[str, float]) -> List[Timeframe]: + """Timeframes whose most recent closed boundary (<= wake) is newer than + the boundary of their last successful scan — i.e. boundaries that were + missed while a cycle overran or the process was down. A TF never scanned + is due. Excludes bars whose label sits in the daily settlement pause + (same rule as due_timeframes). + + `due_timeframes` only ever names the timeframes whose boundary coincides + with THIS wake, so an overrunning cycle (scan + advisory panel + news can + take minutes) or a restart left the page serving the previous H1/H4 + report for up to 1 h / 4 h while advertising a fresh next_close_epoch + (2026-09-13 audit).""" + minutes = wake_dt.hour * 60 + wake_dt.minute + out = [] + for tf in DASHBOARD_TFS: + interval = TF_INTERVAL_MINUTES[tf] + floored = minutes - (minutes % interval) + last_closed = wake_dt.replace( + hour=floored // 60, minute=floored % 60, second=0, microsecond=0) + # No bar carries a label inside a scheduled closure (settlement + # pause or weekend), so nothing can have been "missed" there — and a + # weekend-blocked scan never advances the ledger, so without the + # weekend half of this test every weekend wake would re-scan + # M15/H1/H4 for nothing (~1,800 blocked rows per weekend). + # (Whole-span rule, as in providers.base._has_latest_closed_bar: an + # H4 bar labelled Sunday 20:00 UTC starts in the weekend but runs + # through the reopen, so it exists.) + label = last_closed - timedelta(minutes=interval) + if (_providers_base._is_scheduled_closure(label) + and _providers_base._is_scheduled_closure( + last_closed - timedelta(seconds=1))): + continue + scanned = last_boundary.get(tf.value) + if scanned is None or scanned < last_closed.timestamp(): + out.append(tf) + return out def seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float: @@ -324,6 +381,9 @@ class DashboardPublisher: self.out_dir = out_dir self.snapshots: Dict[str, dict] = {} self._reports = {} + # tf.value → epoch seconds of the boundary the last successful scan + # belonged to; feeds catch_up_timeframes (2026-09-13 audit). + self._last_boundary: Dict[str, float] = {} self.news_payload: Optional[dict] = None self._dp = None self._news_engine: Optional[NewsEngine] = None @@ -355,7 +415,14 @@ class DashboardPublisher: shutil.copyfile(src, dst + ".tmp") os.replace(dst + ".tmp", dst) - def scan_tf(self, tf: Timeframe) -> None: + def scan_tf(self, tf: Timeframe, + boundary_epoch: Optional[float] = None) -> None: + """Scan one timeframe and store its snapshot entry. + + `boundary_epoch` is the fallback boundary recorded when the report + carries no analysis (weekend/blocked or failed scan) or its bar + label will not parse; normally the boundary comes from the report + itself (bar label + interval).""" started = time.time() try: report = produce_signal(XAUUSD, tf, self._provider(), @@ -365,6 +432,19 @@ class DashboardPublisher: if violent_watch_enabled(): attach_watch(report, tf, get_app_config().violent_watch_settings) + # Which boundary this report belongs to: the bar LABEL it closed + # on plus one interval. Recorded so a boundary missed by an + # overrunning cycle or a restart is caught up on the next wake + # instead of sitting stale (2026-09-13 audit). + boundary = boundary_epoch + if report.primary_analysis is not None: + try: + boundary = (pd.Timestamp(report.timestamp).timestamp() + + TF_INTERVAL_MINUTES[tf] * 60) + except Exception: + boundary = boundary_epoch + if boundary is not None: + self._last_boundary[tf.value] = boundary self._reports[tf.value] = report self.snapshots[tf.value] = { "status": "ok", @@ -449,9 +529,15 @@ class DashboardPublisher: logger.info(f"next M5 boundary in {delay:.0f}s") time.sleep(delay) try: - due = due_timeframes(datetime.now(timezone.utc)) + now = datetime.now(timezone.utc) + due = due_timeframes(now) logger.info(f"boundary wake — scanning {[t.value for t in due]}") - self.run_cycle(due) + missed = catch_up_timeframes(now, self._last_boundary) + extra = [tf for tf in missed if tf not in due] + if extra: + logger.info(f"catch-up — scanning " + f"{[t.value for t in extra]} (missed boundary)") + self.run_cycle(due + extra) except Exception: logger.exception("cycle failed") diff --git a/engine.py b/engine.py index 8ef9760..3515952 100755 --- a/engine.py +++ b/engine.py @@ -12,6 +12,7 @@ Design philosophy: """ import logging +import math from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple @@ -84,15 +85,65 @@ _DEFAULT_REGIME_WEIGHTS = { } +def _validate_regime_override(override: Dict[str, Dict[str, float]]) -> None: + """Reject a PARTIAL ``regime_weights_override`` (raises ValueError). + + An override REPLACES the whole table, it does not patch it: whatever + it omits is multiplied by 1.0 in _compute_composite, not by the + shipped default. A half-filled research table (``--set + technical.regime_weights_override=``) would therefore silently + run 1.0 where the default said 0.3 or 2.5, with no warning — so every + regime and all nine scored indicators must be present, with a finite + multiplier >= 0. The shipped M15 override is derived from the default + table (pair_signals/xauusd._m15_regime_weights) and passes.""" + problems: List[str] = [] + want_regimes = set(_DEFAULT_REGIME_WEIGHTS) + want_indicators = set(_TRENDING_WEIGHTS) + got_regimes = set(override) + if got_regimes != want_regimes: + problems.append( + f"missing regimes {sorted(want_regimes - got_regimes)}, " + f"unexpected {sorted(got_regimes - want_regimes)}") + for regime in sorted(got_regimes & want_regimes): + weights = override[regime] + if not isinstance(weights, dict): + problems.append(f"regime '{regime}' is not an indicator -> " + f"multiplier mapping: {weights!r}") + continue + got = set(weights) + if got != want_indicators: + problems.append( + f"regime '{regime}' missing {sorted(want_indicators - got)}, " + f"unexpected {sorted(got - want_indicators)}") + for name in sorted(got & want_indicators): + mult = weights[name] + if (isinstance(mult, bool) or not isinstance(mult, (int, float)) + or not math.isfinite(mult) or mult < 0): + problems.append( + f"regime '{regime}' indicator '{name}' multiplier " + f"{mult!r} is not a finite number >= 0") + if problems: + raise ValueError( + "regime_weights_override is partial: " + "; ".join(problems) + + "; an override replaces the WHOLE table " + "(see engine._DEFAULT_REGIME_WEIGHTS)") + + def regime_weight_table(config: TechnicalConfig) -> Dict[str, Dict[str, float]]: """The regime -> indicator weight multipliers _compute_composite runs with (``regime_weights_override`` wins, else the gold-optimized defaults above). + A non-empty override must cover the WHOLE table (every regime, all + nine scored indicators) — a partial one raises ValueError instead of + silently multiplying the cells it omits by 1.0. + Pure and side-effect free: ONE source of truth shared by the composite and by signal_aggregator.engine_config_fingerprint. Returns a copy, so a caller can never mutate the shipped table.""" override = getattr(config, "regime_weights_override", None) + if override: + _validate_regime_override(override) table = override if override else _DEFAULT_REGIME_WEIGHTS return {regime: dict(weights) for regime, weights in table.items()} @@ -119,7 +170,7 @@ class TechnicalEngine: return TechnicalAnalysis( symbol=symbol, timeframe=timeframe, - timestamp=str(pd.Timestamp.utcnow()), + timestamp=str(pd.Timestamp.now(tz="UTC")), confidence=0.0, ) diff --git a/providers/base.py b/providers/base.py index 440798a..35cf441 100644 --- a/providers/base.py +++ b/providers/base.py @@ -65,6 +65,46 @@ def _is_forex_weekend(now=None) -> bool: return False +def _is_daily_maintenance(now=None) -> bool: + """Gold's daily settlement pause: Monday-Thursday 17:00-17:59 New York + (21:00 UTC in US summer, 22:00 in winter). No bar carries a label + inside this hour. Friday 17:00 is the weekend (see _is_forex_weekend). + Same rule as advisory_outcome._has_continuity's daily_maintenance.""" + if now is None: + now = pd.Timestamp.now(tz="UTC") + ny = now.astimezone(_NY_TZ) + return ny.weekday() < 4 and ny.hour == 17 + + +def _is_scheduled_closure(now=None) -> bool: + """The feed is legitimately silent: weekend or daily settlement pause.""" + return _is_forex_weekend(now) or _is_daily_maintenance(now) + + +def _scheduled_closed_seconds(start, end) -> float: + """Seconds of the [start, end) span that fall inside a scheduled + closure (weekend or the daily settlement pause). + + Bar age is measured from a bar LABEL, so at 21:30 UTC in summer the + 20:55 M5 bar looks 35 minutes old even though the feed could not have + printed anything since 21:00. Subtracting the closed span keeps the + 3-candle freshness limit meaningful instead of failing it — and + logging a false "feed lagging"/"STALE DATA" — every settlement hour + (2026-09-13 audit: 61 suppressed scans on three nights). + """ + hour = pd.Timedelta(hours=1) + closed = 0.0 + t = start.floor("h") + steps = 0 + while t < end and steps < 24 * 10: + if _is_scheduled_closure(t): + closed += max( + 0.0, (min(t + hour, end) - max(t, start)).total_seconds()) + t += hour + steps += 1 + return closed + + class DataIntegrityError(Exception): """Raised when data fails validation checks.""" pass @@ -276,7 +316,7 @@ class BaseDataProvider(ABC): chunks = [] cursor_end = end empty_streak = 0 - error_streak = 0 + window_errors = 0 while cursor_end > start: cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk)) try: @@ -286,23 +326,31 @@ class BaseDataProvider(ABC): ) except Exception as e: # A raising chunk is a PROVIDER ERROR (auth, 429, 5xx, - # malformed payload) — not "no data". Three in a row means - # the rest of the history is unreachable: raise instead of - # returning a silently truncated frame (the backtester - # would treat it as complete). + # malformed payload) — not "no data". RETRY THE SAME + # WINDOW: until 2026-09-13 a failure advanced the cursor + # past the window, so one or two NON-consecutive page + # failures (Oanda's sporadic 401/5xx, even after the + # provider's own retry ladder) silently dropped up to + # days_per_chunk of bars — 15 days of M5 — out of the + # middle of the frame, and only the HEAD was + # coverage-checked below, so the backtester cached the hole + # as complete history forever. Three failures on the SAME + # window means it is unreachable: raise instead of + # returning a frame with a hole in it. logger.warning( f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}" ) - error_streak += 1 - if error_streak >= 3: + window_errors += 1 + if window_errors >= 3: raise self.DATA_ERROR( - f"{symbol} {timeframe.value}: {error_streak} " - f"consecutive chunk failures walking back from " - f"{cursor_end.date()} — refusing to return " - f"truncated history (last error: {e})") from e - cursor_end = cursor_start - timedelta(seconds=1) - continue + f"{symbol} {timeframe.value}: {window_errors} " + f"consecutive failures fetching window " + f"{cursor_start.date()}–{cursor_end.date()} — " + f"refusing to return truncated history " + f"(last error: {e})") from e + continue # re-request the SAME window + window_errors = 0 if chunk is None or chunk.empty: # Genuinely empty windows (weekends/holidays/asset gaps) # are normal; several in a row = start of history. @@ -313,7 +361,6 @@ class BaseDataProvider(ABC): continue empty_streak = 0 - error_streak = 0 chunks.append(chunk) # Step back; dedup on concat. new_cursor = chunk.index[0] - timedelta(seconds=1) @@ -348,6 +395,38 @@ class BaseDataProvider(ABC): f"{df.index[0]} but {start_ts} was requested — " f"~{(df.index[0] - start_ts).days} days missing from the " f"head; treat downstream results accordingly") + # INTERIOR-gap audit (2026-09-13): the head check above is blind to + # a hole in the MIDDLE of the frame, which is exactly what a + # dropped page used to leave behind. Missing labels inside a + # scheduled closure (weekend, daily settlement pause) are normal, + # so only unscheduled runs are reported — and a real market holiday + # (Christmas, Good Friday, July 4) looks identical to a lost page, + # so this is LOUD BUT NOT FATAL and the frame is returned unchanged. + if len(df) > 400_000: + logger.info( + f"{symbol} {timeframe.value}: interior-gap audit skipped " + f"({len(df)} rows)") + elif len(df) > 1: + step = pd.Timedelta(seconds=self.CANDLE_SECONDS[timeframe]) + labels = df.index + # Missing labels between each consecutive pair; >3 is the floor + # worth a look (weekends/settlement hours leave 1-2 labels that + # the NY-anchored closure rule does not cover). + spans = ((labels[1:] - labels[:-1]) / step).astype("int64") - 1 + gaps = [] + for i in (spans > 3).nonzero()[0]: + missing = sum( + 1 for k in range(1, int(spans[i]) + 1) + if not _is_scheduled_closure(labels[i] + k * step)) + if missing > 3: + gaps.append((labels[i], labels[i + 1], missing)) + if gaps: + logger.error( + f"{symbol} {timeframe.value}: {len(gaps)} interior " + f"gap(s) of >3 unscheduled bars in fetched history — " + f"first: {gaps[0]}; a page fetch may have been dropped " + f"or the market was closed (holiday); verify before " + f"trusting this window") return df def is_data_fresh( @@ -398,7 +477,8 @@ class BaseDataProvider(ABC): # ── Shared internals ─────────────────────────────────────────── - def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe) -> bool: + def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe, + now=None) -> bool: """Cached data must include the most recent fully closed bar. The TTL alone is wall-clock based: a snapshot fetched mid-candle is @@ -408,12 +488,29 @@ class BaseDataProvider(ABC): """ if df is None or df.empty: return False - now = pd.Timestamp.now(tz="UTC") + if now is None: + now = pd.Timestamp.now(tz="UTC") if _is_forex_weekend(now): return True candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) dur = pd.Timedelta(seconds=candle_seconds) expected_label = now.floor(dur) - dur + # No bar ever carries a label inside a scheduled closure, so the + # newest bar that MUST be present is the last label before it. + # Without this walk-back every scan from 21:05 to 22:00 UTC in + # summer evicted the cache, refetched twice and logged a false + # "feed lagging" (2026-09-13 audit). + # A label counts as "inside" only when the closure covers the bar's + # whole span: a D1 bar labelled Sunday 00:00 UTC starts in the + # weekend but runs through the Sunday reopen, so it exists and + # must be present (68 such bars in the 3-year Oanda D1 cache). + steps = 0 + last_second = dur - pd.Timedelta(seconds=1) + while (_is_scheduled_closure(expected_label) + and _is_scheduled_closure(expected_label + last_second) + and steps < 2000): + expected_label -= dur + steps += 1 return df.index[-1] >= expected_label def _drop_forming_candle(self, df: pd.DataFrame, timeframe: Timeframe) -> pd.DataFrame: @@ -443,12 +540,14 @@ class BaseDataProvider(ABC): timeframe.value, dropped) return df - def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe) -> bool: + def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe, + now=None) -> bool: """Check if the last candle is within acceptable age for this timeframe.""" if df is None or df.empty: return False - now = pd.Timestamp.now(tz="UTC") + if now is None: + now = pd.Timestamp.now(tz="UTC") if _is_forex_weekend(now): return True @@ -456,7 +555,10 @@ class BaseDataProvider(ABC): last_bar = df.index[-1] candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600) max_age_seconds = candle_seconds * 3 - age_seconds = (now - last_bar).total_seconds() + # Scheduled closures (weekend, daily settlement pause) are dead + # time in which no bar can print — they must not count as age. + age_seconds = ((now - last_bar).total_seconds() + - _scheduled_closed_seconds(last_bar, now)) if age_seconds > max_age_seconds: logger.warning( @@ -482,7 +584,7 @@ class BaseDataProvider(ABC): df = df.sort_index() df = df[~df.index.duplicated(keep="last")] - now = pd.Timestamp.utcnow() + now = pd.Timestamp.now(tz="UTC") future_mask = df.index > now + pd.Timedelta(minutes=5) if future_mask.any(): logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps") diff --git a/requirements.txt b/requirements.txt index 19ed6a7..62f1e24 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ -numpy>=1.24 -pandas>=2.0 +numpy>=1.24,<3 +pandas>=2.0,<4 requests>=2.28 twelvedata>=1.2.0 python-telegram-bot>=20.0 diff --git a/signal_aggregator.py b/signal_aggregator.py index c80df90..5337c79 100755 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -242,6 +242,36 @@ class SignalAggregator: except Exception as e: logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}") + # A missing frame is a FAILED scan, never a degraded one. Losing a + # CONFIRMATION frame used to let the scan continue with fewer + # frames: _check_mtf_agreement needs >= 2 signals, so it returned + # False and the composite took a 15-point mtf_disagreement penalty + # the market never earned (A/A+ became impossible), while on + # M15/H1/H4 a B/C setup could still ship on one confirmation with + # no warning at all. The backtester never sees that state + # (run_backtest raises on an empty frame), so it was a live-only + # divergence. 2026-09-13 audit. + missing = [tf for tf in timeframes if tf not in dfs] + if missing: + report = SignalReport( + symbol=symbol, + display_name=pair_info["display"], + timestamp=str(now_utc), + ) + report.final_signal = SignalStrength.NEUTRAL + report.signal_quality = "F" + report.trade_setup = None + if self.config.primary_timeframe in missing: + # Legacy text signal_log.gate_reason already recognises. + report.warnings.append("Primary timeframe data unavailable") + report.warnings.append( + f"DATA UNAVAILABLE: {', '.join(tf.value for tf in missing)}" + " — scan failed" + ) + report.action_recommendation = "NO TRADE — market data unavailable" + self._log_signal(report, log_source, dfs, failed=True) + return report + report = self.analyze_from_dataframes( symbol=symbol, dfs=dfs, @@ -303,7 +333,7 @@ class SignalAggregator: def _log_signal( self, report: SignalReport, source: Optional[str], dataframes: Optional[Dict[Timeframe, pd.DataFrame]] = None, - blocked: bool = False, + blocked: bool = False, failed: bool = False, ) -> None: """Persist the scan to the signal-history DB (signal_log.py). @@ -314,9 +344,11 @@ class SignalAggregator: is still written, just without the news JSON blobs. ``blocked`` marks weekend/day early returns: no news fetch, no snapshots — a ~1 KB liveness row (or nothing, per signal_log.log_blocked_scans). - A scan whose PRIMARY fetch failed is logged as ``scan_kind`` - ``failed`` (price 0, no signal): liveness evidence that readers - skip like ``blocked``, so it never pollutes NEUTRAL statistics.""" + A scan missing ANY required frame (primary or confirmation, or + the regime-filter frame when that filter is on) is logged as + ``scan_kind`` ``failed`` (price 0, no signal): liveness evidence + that readers skip like ``blocked``, so it never pollutes NEUTRAL + statistics.""" context = None if not blocked: try: @@ -325,10 +357,10 @@ class SignalAggregator: logger.debug("signal log: news context unavailable", exc_info=True) kind = "blocked" if blocked else "scan" - if not blocked and dataframes is not None: - primary = dataframes.get(self.config.primary_timeframe) - if primary is None or primary.empty: - kind = "failed" + primary = (dataframes or {}).get(self.config.primary_timeframe) + if failed or (not blocked and dataframes is not None + and (primary is None or primary.empty)): + kind = "failed" extra = {"scan_kind": kind} for key, fingerprint in ( ("config_fingerprint", self.gate_config_fingerprint), @@ -1297,6 +1329,28 @@ class SignalAggregator: report.trade_setup = self._generate_xau_atr_trade_setup(report, atr) if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"): report.trade_setup = self._generate_trade_setup(report, pip_value) + # A shipped setup always names a direction. The signal thresholds + # (BUY/SELL need confidence >= 50) sit above the C grade floor + # (>= 45), and gate 6c grants B at >= 45 in "transitioning", so a + # C/B report could leave here as NEUTRAL with a full setup: the + # backtester traded it (the replay never reads final_signal) while + # the Telegram alert gate rejects NEUTRAL first (2026-09-13 audit: + # 5/91 H1 12-mo book trades, +3.7R of +22.2R, were never alerted). + # The generators derived the direction from the score sign — the + # same sign every direction gate above used — so this changes + # display and alert eligibility only; the trade books are unchanged. + # A report no generator served stays NEUTRAL, as before. + if (report.trade_setup is not None + and report.final_signal == SignalStrength.NEUTRAL): + report.final_signal = (SignalStrength.BUY + if report.trade_setup.direction == "BUY" + else SignalStrength.SELL) + # The action text was written before the gate chain with the + # old signal name; keep it consistent with the header. + if report.action_recommendation.startswith("NEUTRAL"): + report.action_recommendation = ( + report.final_signal.name.replace("_", " ") + + report.action_recommendation[len("NEUTRAL"):]) @staticmethod def _drop_stale_spread_gate_vetoes(report: SignalReport) -> None: diff --git a/signal_log.py b/signal_log.py index 5a01406..49d48cc 100644 --- a/signal_log.py +++ b/signal_log.py @@ -104,6 +104,7 @@ from advisory_outcome import ( canonical_r, initial_outcome, normalize_timeframe, + resolve_locked_floor, timeframe_value, ) # Private on purpose — the scan-outcome labeller must apply EXACTLY the @@ -258,6 +259,7 @@ _GATE_PREFIXES = ( "SESSION FILTER", "Primary timeframe stale", "Primary timeframe data unavailable", + "DATA UNAVAILABLE:", "REGIME FILTER", "EXHAUSTION BLOCK", "EXHAUSTED TREND FILTER", @@ -278,6 +280,10 @@ SCAN_OUTCOME_WINDOW_H = 24 # A 24 h window can only be judged once it has elapsed; the extra hour # leaves room for the last bar of the window to be stored. SCAN_LABEL_MIN_AGE_H = 25 +# An incomplete row older than this is FINAL: its window had a real hole +# (a feed outage), which no later bar can fill. It is neither re-examined +# nor allowed to anchor the shared bar stream for the rest of the batch. +SCAN_LABEL_MAX_AGE_H = 72 # Missing bars tolerated inside a window ON TOP of the scheduled closures # advisory_outcome._has_continuity already allows. SCAN_LABEL_MAX_GAP_BARS = 3 @@ -399,6 +405,20 @@ def setup_key(setup: Optional[Dict[str, Any]], timeframe: Any) -> Optional[str]: )) +def _is_gap_reason(reason: Any) -> bool: + """True when an outcome was left UNRESOLVABLE by a candle gap. + + advance_outcome writes TWO such reasons — "stored candle snapshots + lost continuity" (no overlap with the last processed bar) and "stored + candle snapshots contain an internal gap" — and each call site used to + match only one of them, so the stream rebuild and the locked-floor + rescue skipped half the gap rows. The substring clauses keep states + written by older code inside the same net.""" + text = str(reason or "") + return (text.startswith("stored candle snapshots") + or "continuity" in text or "gap" in text) + + def gate_reason( warnings: Iterable[str], cleaned_warnings: Iterable[str], @@ -1150,12 +1170,19 @@ class SignalLogger: before = _dumps(json.loads(hit[0])) advanced = advance_outcome(json.loads(before), bars) if (advanced.get("status") == "UNRESOLVABLE" - and "continuity" in str(advanced.get("reason", ""))): + and _is_gap_reason(advanced.get("reason"))): union = self._stream_since( symbol, timeframe, json.loads(before)["progress"]["last_processed_bar"]) if union: advanced = advance_outcome(json.loads(before), union) + # The database could not fill the hole. A trade + # whose trail had already locked profit still has a + # real answer — its floor — so record that instead + # of dropping the row out of the scorecard. + if (advanced.get("status") == "UNRESOLVABLE" + and _is_gap_reason(advanced.get("reason"))): + advanced = resolve_locked_floor(advanced) after = _dumps(advanced) if after == before: self._conn.rollback() @@ -1194,6 +1221,11 @@ class SignalLogger: " WHERE symbol=? AND candles_json IS NOT NULL" " AND logged_at >= ? ORDER BY id LIMIT ?", (f'$."{key}"', symbol, since, int(max_rows))).fetchall() + if len(rows) == int(max_rows): + logger.warning( + "signal_log: bar stream truncated for %s %s (key %s)" + " since %s — hit the %s-row cap, later bars are missing", + symbol, tf, key, since, int(max_rows)) for (blob,) in rows: if not blob: continue @@ -1212,42 +1244,82 @@ class SignalLogger: Candidates are rows that carry indicator snapshots (a real scan) and a price, whose signal bar closed more than 25 h ago and which - have no COMPLETE label yet; the least recently labelled go first, - so a permanently short stream can never starve the rest. Returns - the number of rows actually WRITTEN — a row whose numbers did not - change is skipped, which makes repeated passes idempotent and lets - a batch driver stop at 0. + have no COMPLETE label yet. An incomplete row is re-examined only + while it is younger than SCAN_LABEL_MAX_AGE_H (counted from the + newest logged row when the database is behind the clock) — beyond + that its window had a real hole that no later bar can fill, so the + label it already carries is final — and a stale row never anchors + the shared bar stream either (a key that has nothing else due + starts at the age bound plus one window). Both bounds are + load-bearing: a handful of permanently incomplete rows (a feed + outage) sort to the front of every pass, and anchoring the stream + at their bar pushes its row cap past today, which silently labels + every genuinely due row NULL and incomplete — the scan_outcomes + table then goes blind. Returns the number of rows actually WRITTEN + — a row whose numbers did not change is not rewritten (only its + labeled_at is bumped, so it rotates to the back of the queue), + which makes repeated passes idempotent and lets a batch driver + stop at 0. Bounded on purpose: at most ``budget`` candidates are examined and the pass stops after ``time_budget`` seconds (default 2 s) so the opportunistic call inside log() stays cheap. Per-row failures are logged and skipped.""" moment = now or datetime.now(timezone.utc) - cutoff = _utc(moment) - timedelta(hours=SCAN_LABEL_MIN_AGE_H) + moment_utc = _utc(moment) + cutoff = moment_utc - timedelta(hours=SCAN_LABEL_MIN_AGE_H) deadline = time.monotonic() + float( SCAN_LABEL_TIME_BUDGET if time_budget is None else time_budget) written = 0 with self._lock: + # Age is measured from the newest row when the database itself + # is behind the clock (an imported snapshot on the dev box), + # otherwise every row of a 3-day-old import would be declared + # final on the first pass. On a live logger the two coincide. + horizon = moment_utc + newest = self._conn.execute( + "SELECT logged_at FROM signals ORDER BY id DESC" + " LIMIT 1").fetchone() + try: + horizon = min(horizon, _utc(datetime.fromisoformat( + str(newest[0]).replace(" ", "T")))) + except (TypeError, ValueError, IndexError): + pass + stale_before = horizon - timedelta(hours=SCAN_LABEL_MAX_AGE_H) + # Deepest a stream may start when every row due for that key is + # stale: one window before the age bound, so such a row still + # gets its single final pass without dragging the stream back. + anchor_floor = stale_before - timedelta(hours=SCAN_LABEL_MIN_AGE_H) + # Row ids are monotonic in logged_at (a uniform ISO string), so + # an id bound is the cheap, index-friendly way to leave stale + # incomplete rows out of the re-examination set. bar_ts mixes + # separators and must never be compared as a string. + min_id = self._conn.execute( + "SELECT MIN(id) FROM signals WHERE logged_at >= ?", + (stale_before.isoformat(),)).fetchone()[0] + if min_id is None: + min_id = 2 ** 62 # nothing recent: re-examine none rows = self._conn.execute( "SELECT s.id, s.symbol, s.timeframe, s.price," " COALESCE(s.bar_ts, json_extract(s.indicators_json," " '$.\"' || s.timeframe || '\".timestamp')) AS bar," " json_extract(s.indicators_json," - " '$.\"' || s.timeframe || '\".key_levels.atr') AS atr" + " '$.\"' || s.timeframe || '\".key_levels.atr') AS atr," + " (o.signal_id IS NULL) AS unlabelled" " FROM signals s" " LEFT JOIN scan_outcomes o ON o.signal_id = s.id" " WHERE s.indicators_json IS NOT NULL AND s.price > 0" - " AND (o.signal_id IS NULL OR o.complete = 0)" + " AND (o.signal_id IS NULL OR (o.complete = 0 AND s.id >= ?))" " AND bar IS NOT NULL" " ORDER BY COALESCE(o.labeled_at, '') ASC, s.id ASC" - " LIMIT ?", (int(budget),)).fetchall() + " LIMIT ?", (int(min_id), int(budget))).fetchall() # One bar stream per (symbol, timeframe) for the whole batch, # built from the OLDEST bar in it — _stream_since is a scan # over every row logged since that bar, far too expensive to # repeat per candidate. due: List[Tuple[Any, ...]] = [] oldest: Dict[Tuple[str, str], Tuple[datetime, str]] = {} - for rowid, symbol, tf, price, bar, atr in rows: + for rowid, symbol, tf, price, bar, atr, unlabelled in rows: try: stamp = _utc(datetime.fromisoformat( str(bar).replace(" ", "T"))) @@ -1255,8 +1327,16 @@ class SignalLogger: continue if stamp > cutoff: continue # window has not elapsed yet + # Stale = old by the bar it closed on AND by when it was + # written (on a live database the two agree within minutes; + # requiring both keeps a freshly imported or backfilled DB, + # whose rows all carry today's logged_at, out of it). + stale = stamp < stale_before and rowid < min_id + if stale and not unlabelled: + continue # final: a hole no later bar can ever fill key = (str(symbol), str(tf)) - if key not in oldest or stamp < oldest[key][0]: + if not stale and (key not in oldest + or stamp < oldest[key][0]): oldest[key] = (stamp, str(bar)) due.append((rowid, symbol, tf, price, bar, atr, key)) streams: Dict[Tuple[str, str], List[Dict[str, Any]]] = {} @@ -1265,8 +1345,13 @@ class SignalLogger: break try: if key not in streams: + # 3 days of anchor is ~2,100 rows on the live DB; + # the cap is a safety net, not the working limit. + anchor = oldest.get(key) streams[key] = self._stream_since( - symbol, tf, oldest[key][1]) + symbol, tf, + anchor[1] if anchor else anchor_floor.isoformat(), + max_rows=10000) label = scan_outcome(bar, tf, streams[key], _finite(atr), price) if label is None: @@ -1277,6 +1362,14 @@ class SignalLogger: + " FROM scan_outcomes WHERE signal_id=?", (rowid,)).fetchone() if prior is not None and tuple(prior) == values: + # Unchanged, so NOT counted as written — but its + # queue position moves, or the same short-stream + # rows would head every pass for ever. + now_iso = datetime.now(timezone.utc).isoformat() + self._conn.execute( + "UPDATE scan_outcomes SET labeled_at=?" + " WHERE signal_id=?", (now_iso, rowid)) + self._conn.commit() continue self._conn.execute( _SCAN_OUTCOME_INSERT, @@ -1376,6 +1469,9 @@ class SignalLogger: continue before = _dumps(state) advanced = advance_outcome(json.loads(before), union) + if (advanced.get("status") == "UNRESOLVABLE" + and _is_gap_reason(advanced.get("reason"))): + advanced = resolve_locked_floor(advanced) after = _dumps(advanced) if after == before: continue @@ -1393,7 +1489,9 @@ class SignalLogger: """Bounded retry for continuity-broken outcomes: re-freeze the seed from the row's own setup/indicators (at most ``max_reseeds`` times per row) so the next resolver pass can walk the now-filled - stream. Invalid-seed rows stay terminal.""" + stream. Invalid-seed rows stay terminal. A gap row that had already + trailed its stop past the entry is finished at that locked floor + instead of re-seeded; the returned list holds re-seeded ids only.""" reseeded: List[int] = [] with self._lock: rows = self._conn.execute( @@ -1403,8 +1501,21 @@ class SignalLogger: " AND advisory_event_id = id").fetchall() for rowid, tf, setup_json, ind_json, state_json in rows: old = json.loads(state_json) if state_json else {} - if "continuity" not in str(old.get("reason", "")) and \ - "gap" not in str(old.get("reason", "")): + if not _is_gap_reason(old.get("reason")): + continue + # The locked trailing stop lives ONLY in this progress + # block, and a re-seed rebuilds the walk from scratch over + # a stream whose hole is still there (the continuity check + # is all-or-nothing), so the floor would be lost for good. + # Finish such a row here; re-seed only what has nothing. + floored = resolve_locked_floor(json.loads(_dumps(old))) + if floored.get("status") == "RESOLVED": + self._conn.execute( + "UPDATE signals SET advisory_outcome_status=?," + " advisory_r_multiple=?, advisory_outcome_json=?" + " WHERE id=?", + ("RESOLVED", canonical_r(floored), + _dumps(floored), rowid)) continue count = int((old.get("reseed") or {}).get("count", 0)) if count >= max_reseeds or not setup_json: diff --git a/telegram_bot.py b/telegram_bot.py index 56c9c74..2bab953 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -49,9 +49,10 @@ import sys import threading import time from concurrent.futures import ThreadPoolExecutor -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set +import pandas as pd from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.constants import ParseMode from telegram.error import BadRequest @@ -71,6 +72,7 @@ from app_config import ( telegram_channel_id, violent_watch_enabled, ) from providers import BaseDataProvider, make_data_provider +from providers.base import _is_daily_maintenance from agent_panel import attach_panel, panel_health from llm_client import get_llm_client from market_outlook import maybe_run_outlook @@ -110,8 +112,12 @@ _file_handler = logging.handlers.TimedRotatingFileHandler( when="midnight", backupCount=30, encoding="utf-8", ) _file_handler.setFormatter(_log_formatter) +# Console only at IMPORT: `python test_all.py` imports this module, and an +# import-time file handler wrote fixture alerts (2027 dates, 3300.00 prices) +# into logs/telegram.log on whatever box ran the suite. main() attaches +# _file_handler for real bot runs. logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO), - handlers=[_console_handler, _file_handler], + handlers=[_console_handler], force=True) logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) @@ -545,7 +551,21 @@ def _bar_epoch(tf: Timeframe, now: Optional[float] = None) -> int: def _remember_scan(symbol: str, tf: Timeframe, report: SignalReport) -> None: - _last_scan[f"{symbol}_{tf.value}"] = (_bar_epoch(tf), report) + # File the report under the bar it actually ANALYSED, not the wall + # clock at completion: a scan that straddles a boundary would land + # under the new bar, so the boundary loop's own scan of that bar + # found no cache entry and re-fetched/re-scored/re-logged the same + # closed bar (15 duplicate M15 rows in five days — 2026-09-13 audit). + epoch = _bar_epoch(tf) + if report.primary_analysis is not None: + try: + # report.timestamp is the primary bar's OPEN label; the bar + # closes (= the boundary this scan belongs to) one TF later. + epoch = (int(pd.Timestamp(report.timestamp).timestamp()) + + _TF_SECONDS_CACHE.get(tf, 900)) + except (ValueError, TypeError): + pass # unparseable label → wall-clock boundary, as before + _last_scan[f"{symbol}_{tf.value}"] = (epoch, report) async def _cached_or_scan( @@ -767,6 +787,16 @@ def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float: return remaining +def _closed_bar_in_settlement(interval_minutes: int, + now: Optional[datetime] = None) -> bool: + """True when the bar that just closed (label = now - interval) falls in + gold's daily settlement pause — no such bar exists, so a scan would only + re-read the previous bar (61 stale rows in the 2026-09 audit DB).""" + label_time = ((now or datetime.now(timezone.utc)) + - timedelta(minutes=interval_minutes)) + return _is_daily_maintenance(label_time) + + # ── Violent-move watch (display-only) ──────────────────────────────── # Per-process heads-up budget: {"day": date, "count": int, "sent": {...}}. _watch_state: Dict[str, Any] = {} @@ -800,7 +830,11 @@ async def _scan_timeframe( confirmation_tfs: List[Timeframe], ) -> None: """Scan all pairs for one timeframe and broadcast results.""" - tasks = [run_signal(symbol, tf, confirmation_tfs) for symbol in SCAN_PAIRS] + # _cached_or_scan, not run_signal: the WS breach handler may already + # have scanned this closed bar (and vice versa) — one fetch/score/log + # per bar per TF (2026-09-13 audit). + tasks = [_cached_or_scan(symbol, tf, confirmation_tfs) + for symbol in SCAN_PAIRS] results = await asyncio.gather(*tasks, return_exceptions=True) reports: List[SignalReport] = [] for i, result in enumerate(results): @@ -930,6 +964,9 @@ _pending_breach: Dict[str, tuple] = {} # A failed breach rescan shrinks the just-set cooldown to this retry # window instead of silencing the symbol for the full cooldown (H33a). _BREACH_RETRY_SECONDS = 30 +# A breach landing inside this many seconds after an M15 boundary waits for +# the candle loop's scan (boundary+10s) and reuses it — 2026-09-13 audit. +_BREACH_DEFER_SECONDS = 20.0 _cfg = PlatformConfig() _WS_COOLDOWN = _cfg.ws_cooldown_seconds @@ -1003,6 +1040,13 @@ async def _handle_breach(bot, pair_symbol: str, price: float, level_name: str) - logger.info(f"Level breach: {display} crossed {label} at {price} — rescanning M15") try: + # A tick a second after an M15 boundary would scan a bar the feed + # may still be lagging on, and the candle loop (boundary+10s) would + # then scan the same bar again. Wait for the loop and reuse it. + elapsed = time.time() - _bar_epoch(Timeframe.M15) + if elapsed < _BREACH_DEFER_SECONDS: + logger.debug("deferring breach rescan to the boundary scan") + await asyncio.sleep(_BREACH_DEFER_SECONDS - elapsed) report = await _cached_or_scan(pair_symbol, Timeframe.M15) _update_level_cache([report]) except Exception: @@ -1152,6 +1196,11 @@ async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None: await asyncio.sleep(wait) if not _feed_active(): break + if _closed_bar_in_settlement(interval_minutes): + logger.info( + f"{label} auto-scan: skipped — bar label inside the " + f"daily settlement pause") + continue logger.info(f"{label} auto-scan: firing") try: await _scan_timeframe( @@ -1824,6 +1873,13 @@ def main() -> None: "and telegram.chat_id in config.json, or the TELEGRAM_TOKEN and " "TELEGRAM_CHAT_ID environment variables.") + # Only a real bot run writes the rotating log file: tests import this + # module (an import-time handler logged their fixtures into + # logs/telegram.log) and one of them calls main() to assert the refusal + # above, so the attach sits after that guard — 2026-09-13 audit. + if _file_handler not in logging.getLogger().handlers: + logging.getLogger().addHandler(_file_handler) + _install_excepthooks() logger.info("Starting ForexBot2 Telegram bot…") diff --git a/test_all.py b/test_all.py index 7dc6717..2c53875 100644 --- a/test_all.py +++ b/test_all.py @@ -300,7 +300,7 @@ def _(): future = pd.Timestamp.now(tz="UTC") + pd.Timedelta(hours=10) df.loc[future] = df.iloc[-1] result = dp._validate_and_clean(df, "XAU/USD", Timeframe.H1) - assert (result.index <= pd.Timestamp.utcnow() + pd.Timedelta(minutes=5)).all() + assert (result.index <= pd.Timestamp.now(tz="UTC") + pd.Timedelta(minutes=5)).all() @test("TwelveDataProvider._validate_and_clean removes duplicates") @@ -9087,7 +9087,10 @@ def _(): try: lg = slog.SignalLogger(path) rowid = lg.log(_m27_report(), source="telegram", timeframe="5min") - bars = _m27_bars("2026-08-19T12:00:00+00:00", 60, up=0.2) # Wednesday + # Wednesday, and the drift stays under the 0.70 R trail arm so this + # row never locks a profit floor — a gap row that HAS one now + # resolves as a floor instead (Module 34's locked-floor tests). + bars = _m27_bars("2026-08-19T12:00:00+00:00", 60, up=0.05) # snapshot 1: bars 0..13 (own snapshot overlaps the signal bar) lg.log(_m27_report(setup=False, quality="D", ts=bars[13]["timestamp"]), @@ -9120,7 +9123,8 @@ def _(): (rowid,)).fetchone()[0]) assert st["status"] == "PENDING", st.get("reason") assert st["progress"]["bars_held"] == 55, st["progress"] - # a REAL hole (bars 56, 57 never logged anywhere) is still terminal + # a REAL hole (bars 56, 57 never logged anywhere) with no locked + # profit floor is still terminal lg.log(_m27_report(setup=False, quality="D", ts=bars[59]["timestamp"]), source="telegram", timeframe="5min", @@ -10522,7 +10526,11 @@ def _(): assert flipped_agg.gate_config_fingerprint() == m5.gate_config_fingerprint() weighted = make_config(Timeframe.M5) - weighted.technical.regime_weights_override = {"ranging": {"rsi": 9.0}} + # A WHOLE-table override (a partial one now raises — Module 34): the + # point here is only that the engine fingerprint sees the multipliers. + weighted.technical.regime_weights_override = { + regime: dict(weights, rsi=9.0) + for regime, weights in _DEFAULT_REGIME_WEIGHTS.items()} weighted_agg = SignalAggregator(weighted) assert weighted_agg.engine_config_fingerprint() != fingerprint assert weighted_agg.gate_config_fingerprint() == m5.gate_config_fingerprint() @@ -11821,9 +11829,9 @@ def _(): timeframes=[Timeframe.M5, Timeframe.M15, Timeframe.H1], aggregator=agg, platform_cfg=cfg, bt_cfg=bt_off, primary_dur=pd.Timedelta(minutes=5)) - skip_off, cands_off, bars_off = _gb33._replay_span(span) + skip_off, cands_off, bars_off, _fb_off = _gb33._replay_span(span) _gb33._REPLAY_CTX["bt_cfg"] = bt_on - skip_on, cands_on, bars_on = _gb33._replay_span(span) + skip_on, cands_on, bars_on, _fb_on = _gb33._replay_span(span) finally: _gb33._REPLAY_CTX.clear() @@ -11966,6 +11974,1007 @@ def _(): f"{tf} must carry no regime weight override" +# ══════════════════════════════════════════════════════════════════════ +# MODULE 34: 2026-09-13 audit fixes +# ══════════════════════════════════════════════════════════════════════ + +print("\n═══ Module 34: 2026-09-13 audit fixes ═══") + + +def _m34_report(quality="C", score=27.0, signal=SignalStrength.NEUTRAL, + price=4400.0, atr=8.0): + """Gold-realistic H1 report whose levels clear the $9 spread gate and + the RR floor, so `_generate_trade_setup` always returns a setup.""" + r = SignalReport(symbol=XAUUSD, display_name="XAU/USD", + timestamp="2026-09-10T12:00:00+00:00") + r.signal_quality = quality + r.final_signal = signal + r.final_score = score + r.final_confidence = 47.0 # above the C floor (45), below BUY/SELL (50) + r.current_price = price + r.primary_analysis = TechnicalAnalysis( + symbol=XAUUSD, timeframe=Timeframe.H1.value, timestamp=r.timestamp, + indicators={}, + composite_signal=SignalStrength.NEUTRAL, composite_score=score, + confidence=47.0, + market_regime="transitioning", volatility_regime="normal", + key_levels={"atr": atr, "swing_low_1": price - 20.0, + "swing_high_1": price + 30.0}, + ) + return r + + +@test("aggregator: a grade >= C setup is never NEUTRAL (audit 2026-09-13)") +def _(): + from pair_signals import PAIR_PRODUCERS + cfg = PAIR_PRODUCERS[XAUUSD](Timeframe.H1, [Timeframe.M15, Timeframe.H4]) + agg = SignalAggregator(cfg) + + # 1. C grade, NEUTRAL signal, positive score → BUY, with a BUY setup, + # and the pre-gate action text follows the header. + r = _m34_report(quality="C", score=27.0) + r.action_recommendation = "NEUTRAL — Decent setup, wait for confirmation" + agg._attach_core_setup(r, pip_value=0.10) + assert r.trade_setup is not None, f"expected a setup; warnings={r.warnings}" + assert r.final_signal == SignalStrength.BUY, \ + f"C grade with score +27 must name BUY, got {r.final_signal}" + assert r.trade_setup.direction == "BUY", r.trade_setup.direction + assert r.action_recommendation == "BUY — Decent setup, wait for confirmation", \ + r.action_recommendation + + # 2. Same, negative score → SELL. + r = _m34_report(quality="C", score=-27.0) + agg._attach_core_setup(r, pip_value=0.10) + assert r.trade_setup is not None, f"expected a setup; warnings={r.warnings}" + assert r.final_signal == SignalStrength.SELL, \ + f"C grade with score -27 must name SELL, got {r.final_signal}" + assert r.trade_setup.direction == "SELL", r.trade_setup.direction + + # 3. Below C (D) stays NEUTRAL and gets no setup — the fix must not + # give a direction to a report no generator will serve. + r = _m34_report(quality="D", score=27.0) + agg._attach_core_setup(r, pip_value=0.10) + assert r.trade_setup is None, "D grade must not reach a core generator" + assert r.final_signal == SignalStrength.NEUTRAL, \ + f"D grade must stay NEUTRAL, got {r.final_signal}" + + # 4. A signal that already names a direction is left alone. + r = _m34_report(quality="C", score=27.0, signal=SignalStrength.BUY) + agg._attach_core_setup(r, pip_value=0.10) + assert r.final_signal == SignalStrength.BUY, \ + f"an existing direction must be untouched, got {r.final_signal}" + assert r.trade_setup is not None and r.trade_setup.direction == "BUY" + + # 5. Grade C but every generator declines (ATR under the XAU floor → + # LOW ATR FILTER): no setup, so the report stays NEUTRAL exactly as + # before — the fix keys on a shipped setup, not on the grade. + r = _m34_report(quality="C", score=27.0, atr=0.5) + r.action_recommendation = "WATCH — Signal present but low confidence" + agg._attach_core_setup(r, pip_value=0.10) + assert r.trade_setup is None, "ATR 0.5 must trip the LOW ATR FILTER" + assert r.final_signal == SignalStrength.NEUTRAL, \ + f"no setup → no direction, got {r.final_signal}" + assert r.action_recommendation == "WATCH — Signal present but low confidence" + + +@test("providers/engine: no deprecated pandas utcnow under warnings-as-errors (audit 2026-09-13)") +def _(): + # pd.Timestamp.utcnow() is deprecated (pandas 3 warns, pandas 4 removes + # it). It sat in the future-timestamp guard of _validate_and_clean, i.e. + # on EVERY live candle fetch, and in the engine's short-frame early + # return — so an unpinned `pip install -U pandas` would have failed + # every scan. Both call sites now use Timestamp.now(tz="UTC"). + import warnings + from engine import TechnicalEngine + from config import TechnicalConfig + + df = make_xau_ohlcv(300) + with warnings.catch_warnings(): + warnings.filterwarnings("error", message=".*utcnow.*") + provider = _m21_stub_provider() + cleaned = provider.validate_and_clean(df, "XAU/USD", Timeframe.H1) + assert len(cleaned) > 0, "cleaning must still return bars" + result = TechnicalEngine(TechnicalConfig()).analyze(df, "XAU/USD", "1h") + assert result is not None + + # …and the requirement is capped so pandas 4 cannot arrive by surprise. + req = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), + "requirements.txt")).read().splitlines() + pandas_line = [ln for ln in req if ln.strip().startswith("pandas")] + assert pandas_line, "requirements.txt must pin pandas" + assert "<4" in pandas_line[0], \ + f"pandas requirement must exclude 4.x, got {pandas_line[0]!r}" + + +@test("aggregator: a missing confirmation frame fails the scan instead of " + "degrading it (audit 2026-09-13)") +def _(): + # Losing ONE confirmation frame used to let the scan continue: + # _check_mtf_agreement saw < 2 signals, charged a 15-point + # mtf_disagreement penalty the market never earned (A/A+ impossible), + # and on M15/H1/H4 a B/C setup could still ship on one confirmation + # with no warning at all. The backtester never has a missing frame + # (run_backtest raises on an empty one), so it was a live-only + # divergence. + import sqlite3 + import signal_log as slog + from pair_signals import PAIR_PRODUCERS + + class _QuietNews(_NoNews): + """_NoNews plus the two accessors the full analysis path calls.""" + def get_event_risk_level(self, symbol, context): + return "clear", "" + + def get_pair_sentiment(self, symbol, context): + return 0.0, "" + + def _run(raise_for): + """analyze_pair against a stub whose fetch raises for `raise_for`.""" + cfg = PAIR_PRODUCERS[XAUUSD](Timeframe.M15, + [Timeframe.M5, Timeframe.H1]) + cfg.xau_strategy.allowed_weekdays = [] # deterministic on any day + agg = SignalAggregator(cfg) + agg._get_news_engine = lambda: _QuietNews() + + class _Stub: + def get_pair_info(self, symbol): + return PAIR_INFO[XAUUSD] + + def fetch_ohlcv(self, symbol, timeframe, lookback_days=None): + if timeframe is raise_for: + raise RuntimeError("boom") + return make_xau_ohlcv(300, seed=42) + + def is_data_fresh(self, df, symbol, timeframe): + return True + + agg.data_provider = _Stub() + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + os.remove(path) + try: + os.environ["SIGNAL_LOG_PATH"] = path + os.environ.pop("SIGNAL_LOG_ENABLED", None) + with patch("signal_aggregator._is_forex_weekend", + return_value=False): + report = agg.analyze_pair(symbol=XAUUSD, + log_source="telegram") + conn = sqlite3.connect(path) + row = conn.execute( + "SELECT scan_kind, gate_reason FROM signals").fetchone() + conn.close() + return report, row + finally: + os.environ.pop("SIGNAL_LOG_PATH", None) + os.environ.pop("SIGNAL_LOG_ENABLED", None) + slog.reset_loggers_for_tests() + if os.path.exists(path): + os.remove(path) + + # 1. A missing CONFIRMATION frame fails the scan outright. + report, row = _run(Timeframe.H1) + assert report.final_signal == SignalStrength.NEUTRAL, report.final_signal + assert report.signal_quality == "F", report.signal_quality + assert report.trade_setup is None, "no setup may ship on one confirmation" + assert report.primary_analysis is None, \ + "no analysis may run on incomplete evidence" + assert len([w for w in report.warnings + if w.startswith("DATA UNAVAILABLE: 1h")]) == 1, report.warnings + assert "Primary timeframe data unavailable" not in report.warnings, \ + "the primary frame arrived — only the confirmation is missing" + assert row is not None and row[0] == "failed", row + assert row[1].startswith("DATA UNAVAILABLE:"), row[1] + + # 2. A missing PRIMARY frame also keeps the legacy warning text. + report, row = _run(Timeframe.M15) + assert "Primary timeframe data unavailable" in report.warnings, \ + report.warnings + assert any(w.startswith("DATA UNAVAILABLE: 15min") + for w in report.warnings), report.warnings + assert row[0] == "failed", row + + # 3. All frames present → the scan analyzes and logs as usual. + report, row = _run(None) + assert report.primary_analysis is not None, report.warnings + assert row[0] == "scan", row + + +@test("providers: daily settlement pause predicate matches the advisory " + "continuity rule across DST (audit 2026-09-13)") +def _(): + import providers.base as pbase + from zoneinfo import ZoneInfo + ny_tz = ZoneInfo("America/New_York") + # Summer week → the pause is 21:00 UTC; winter week → 22:00 UTC. + for start, end, utc_hour in [("2026-07-06", "2026-07-12", 21), + ("2026-12-07", "2026-12-13", 22)]: + for t in pd.date_range(f"{start} 00:00", f"{end} 23:00", + freq="h", tz="UTC"): + got = pbase._is_daily_maintenance(t) + ny = t.astimezone(ny_tz) + # advisory_outcome._has_continuity's daily_maintenance, inlined. + assert got is (ny.weekday() < 4 and ny.hour == 17), \ + f"{t} (NY {ny}) → {got}" + assert got is (t.weekday() < 4 and t.hour == utc_hour), \ + f"{t}: this week's pause must be {utc_hour}:00 UTC Mon-Thu" + if got: + assert not pbase._is_forex_weekend(t), \ + f"{t}: the pause is never the weekend rule" + assert pbase._is_scheduled_closure(t) is ( + got or pbase._is_forex_weekend(t)), t + + +@test("providers: freshness gates treat the settlement pause as a scheduled " + "closure (audit 2026-09-13)") +def _(): + p = _m21_stub_provider() + + def m5_frame(last_label): + idx = pd.date_range(end=pd.Timestamp(last_label, tz="UTC"), + periods=60, freq="5min", tz="UTC") + return pd.DataFrame({"Open": 3600.0, "High": 3601.0, "Low": 3599.0, + "Close": 3600.5, "Volume": 1.0}, + index=pd.DatetimeIndex(idx, name="Datetime")) + + # Summer (Tuesday): no label exists between 21:00 and 21:59 UTC, so + # 20:55 stays the newest closed bar until the 22:00 one prints. + df = m5_frame("2026-09-08 20:55") + for stamp in ("2026-09-08 21:30:10", "2026-09-08 22:00:10"): + now = pd.Timestamp(stamp, tz="UTC") + assert p._has_latest_closed_bar(df, Timeframe.M5, now=now) is True, \ + f"{stamp}: 20:55 IS the latest bar that can exist" + assert p._is_data_fresh(df, XAUUSD, Timeframe.M5, now=now) is True, \ + f"{stamp}: the closed hour must not count as age" + # Once the feed can print again the guard bites exactly as before. + assert p._has_latest_closed_bar( + df, Timeframe.M5, + now=pd.Timestamp("2026-09-08 22:05:10", tz="UTC")) is False, \ + "the 22:00 bar exists and is missing" + # A genuine gap still fails: 65m old − 30m closed = 35m > 15m limit. + assert p._is_data_fresh( + m5_frame("2026-09-08 20:25"), XAUUSD, Timeframe.M5, + now=pd.Timestamp("2026-09-08 21:30:10", tz="UTC")) is False + # Winter (Tuesday): the same NY hour is 22:00-22:59 UTC. + win = m5_frame("2026-12-08 21:55") + wnow = pd.Timestamp("2026-12-08 23:00:10", tz="UTC") + assert p._has_latest_closed_bar(win, Timeframe.M5, now=wnow) is True + assert p._is_data_fresh(win, XAUUSD, Timeframe.M5, now=wnow) is True + + +@test("telegram_bot/dashboard: no scan for a bar whose label sits in the " + "settlement pause (audit 2026-09-13)") +def _(): + def at(y, mo, d, h, m): + return datetime(y, mo, d, h, m, 10, tzinfo=timezone.utc) + + # Tuesday 2026-09-08 (summer): the pause is 21:00-21:59 UTC. + assert _tb21._closed_bar_in_settlement( + 5, now=at(2026, 9, 8, 22, 0)) is True, "the 21:55 M5 bar never printed" + assert _tb21._closed_bar_in_settlement( + 5, now=at(2026, 9, 8, 22, 5)) is False, "the 22:00 M5 bar is real" + assert _tb21._closed_bar_in_settlement( + 60, now=at(2026, 9, 8, 21, 0)) is False, \ + "the 20:00 H1 bar closed at 21:00 and must be scanned" + assert _tb21._closed_bar_in_settlement( + 60, now=at(2026, 9, 8, 22, 0)) is True, "no 21:00 H1 bar exists" + + due = [tf.value for tf in dpub.due_timeframes(at(2026, 9, 8, 22, 0))] + assert due == [], \ + f"M5/M15/H1 must all be skipped at 22:00 UTC in summer, got {due}" + later = [tf.value for tf in dpub.due_timeframes(at(2026, 9, 9, 0, 0))] + assert "4h" in later, f"the 20:00 H4 bar closed at 00:00 UTC: {later}" + + +@test("engine: a partial regime_weights_override raises instead of silently " + "defaulting to x1.0 (audit 2026-09-13)") +def _(): + # An override REPLACES the whole table: _compute_composite multiplies + # anything the table omits by 1.0, not by the shipped default. A + # research table from --set that covers one regime (or one regime + # minus one indicator) used to run silently at 1.0. + import copy + import engine as _eng + from pair_signals import PAIR_PRODUCERS + + # (a) The default path is untouched. + default = _eng.regime_weight_table(TechnicalConfig()) + assert default == copy.deepcopy(_eng._DEFAULT_REGIME_WEIGHTS), default + + # (b) The shipped M15 override validates — RSI exactly halved. + m15 = PAIR_PRODUCERS[XAUUSD](Timeframe.M15, [Timeframe.M5, Timeframe.H1]) + shipped = _eng.regime_weight_table(m15.technical) + assert set(shipped) == set(default), sorted(shipped) + for regime, weights in default.items(): + assert set(shipped[regime]) == set(weights), regime + for name, mult in weights.items(): + want = mult * 0.5 if name == "rsi" else mult + assert shipped[regime][name] == want, (regime, name) + + def _rejects(override, *must_mention): + try: + _eng.regime_weight_table( + TechnicalConfig(regime_weights_override=override)) + except ValueError as exc: + for token in must_mention: + assert token in str(exc), (token, str(exc)) + else: + raise AssertionError(f"must raise: {override}") + + # (c) One regime only, and that regime missing an indicator. + _rejects({"ranging": {name: mult for name, mult + in default["ranging"].items() if name != "rsi"}}, + "ranging", "rsi") + # (d) Full table, one non-numeric multiplier. + bad_value = copy.deepcopy(default) + bad_value["trending_up"]["macd"] = "2" + _rejects(bad_value, "trending_up", "macd") + # (e) Full table plus an unknown regime. + extra = copy.deepcopy(default) + extra["trending_sideways"] = dict(default["ranging"]) + _rejects(extra, "trending_sideways") + + # (f) A valid full override still analyzes (the per-call check costs + # a handful of set comparisons, nothing more). + valid = copy.deepcopy(default) + valid["ranging"]["rsi"] = 0.0 # zero is a legal multiplier + cfg = TechnicalConfig(regime_weights_override=valid) + result = TechnicalEngine(cfg).analyze(make_xau_ohlcv(300), "XAU/USD", "1h") + assert result.indicators, "analyze produced no indicators" + assert -100.0 <= result.composite_score <= 100.0, result.composite_score + assert _eng.regime_weight_table(cfg) == valid + + +@test("backtester: cache-miss fallbacks are counted per span and surfaced by run_backtest (audit 2026-09-13)") +def _(): + # A GPU cache miss recomputes the indicator with the live engine's + # pandas math on the sliced frame (different EWM seeding than the + # pre-computed cache) — a live-vs-backtest parity hazard. The counter + # existed but died inside the forked worker, so a run that fell back on + # thousands of bars reported nothing. _replay_span now returns its OWN + # span's count as the 4th element; run_backtest sums and logs it. + from pair_signals.xauusd import make_config as _mk34 + cfg = _mk34(Timeframe.M5, [Timeframe.M15, Timeframe.H1]) + dfs_full = _m33_span_frames() + caches = {tf.value: _gb33.precompute_indicators(df, cfg.technical) + for tf, df in dfs_full.items()} + agg = SignalAggregator(cfg) + agg.tech_engine = _gb33.GPUTechnicalEngine(cfg.technical, caches) + bt_cfg = _gb33.BacktestConfig( + pair=XAUUSD, start=datetime(2026, 7, 6), end=datetime(2026, 7, 9), + primary_tf=Timeframe.M5, + confirmation_tfs=[Timeframe.M15, Timeframe.H1]) + span = (300, 400) + try: + _gb33._REPLAY_CTX.update( + primary_df=dfs_full[Timeframe.M5], dfs_full=dfs_full, + timeframes=[Timeframe.M5, Timeframe.M15, Timeframe.H1], + aggregator=agg, platform_cfg=cfg, bt_cfg=bt_cfg, + primary_dur=pd.Timedelta(minutes=5)) + + # (a) Every slice end resolves in its cache → nothing recomputed. + _skips, cands, _bars, fb_clean = _gb33._replay_span(span) + assert len(cands) > 0, "the fixture must analyze (and accept) bars" + assert fb_clean == 0, fb_clean + assert _gb33._replay_span(span)[3] == 0, "a clean re-run stays 0" + + # (b) Drop ONE analyzed primary bar from the lookup table. That + # bar's analyze() then falls back on every override that runs: 8 + # of the 9 _compute_* overrides, because _compute_ema_2 and + # _compute_ema_4 are mutually exclusive (XAU runs 4ema) and each + # override calls _resolve_pos exactly once. Deterministic, so the + # count is asserted exactly. + n_overrides = len([m for m in vars(_gb33.GPUTechnicalEngine) + if m.startswith("_compute_")]) + assert n_overrides == 9, n_overrides + cache = caches[Timeframe.M5.value] + miss_ts = dfs_full[Timeframe.M5].index[cands[0][0]] + removed = cache.index_pos.pop(miss_ts) + try: + fb_miss = _gb33._replay_span(span)[3] + fb_again = _gb33._replay_span(span)[3] + finally: + cache.index_pos[miss_ts] = removed + assert fb_miss == n_overrides - 1, (fb_miss, n_overrides) + # (c) Per-span reset: one worker process reuses one engine across + # several spans, so the count must not accumulate. + assert fb_again == fb_miss, (fb_miss, fb_again) + finally: + _gb33._REPLAY_CTX.clear() + + # run_backtest hands its summed total to the recorder, and write_run + # assembles the manifest from an explicit key list — so the key has to + # survive that filter, not just sit in run_info. + m34_cfg, m34_agg, _r34, _s34 = _m33_report() + trades = [_m29_trade(r_multiple=1.5)] + book_df = _gb33.trades_to_dataframe(trades) + with tempfile.TemporaryDirectory() as d: + # NEVER read backtest/cache/calendar.parquet from a test. + cal_path = os.path.join(d, "cal.parquet") + pd.DataFrame([_m33_cal_row( + 1, pd.Timestamp("2026-01-05 13:30", tz="UTC"), + "Non-Farm Employment Change")]).to_parquet(cal_path, index=False) + rec_dir = os.path.join(d, "run") + rec_cfg = _gb33.BacktestConfig( + pair=XAUUSD, start=datetime(2026, 7, 1), end=datetime(2026, 7, 8), + primary_tf=Timeframe.M5, + confirmation_tfs=[Timeframe.M15, Timeframe.H1], + record_dir=rec_dir) + saved_cal = _rr.CALENDAR_PARQUET + try: + _rr.CALENDAR_PARQUET = cal_path + _rr.write_run(rec_dir, trades=trades, book_df=book_df, + bt_cfg=rec_cfg, platform_cfg=m34_cfg, + aggregator=m34_agg, + run_info=dict(workers=4, cache_fallbacks=37)) + finally: + _rr.CALENDAR_PARQUET = saved_cal + man = _rr.load_run(rec_dir)["manifest"] + assert man["cache_fallbacks"] == 37, man.get("cache_fallbacks") + + +def _m34_sell_seed(): + """The real 2026-09-08 07:30 M15 SELL setup (signal-log row 915).""" + from advisory_outcome import initial_outcome + setup = { + "direction": "SELL", "entry": 4395.66, "stop_loss": 4422.15, + "take_profit_1": 4369.17, "strategy": "GENERIC", + "setup_pattern": None, "trail_arm_r": 0.7, "trail_dist_r": 0.75, + } + indicators = {"15min": {"timestamp": "2026-09-08T07:30:00+00:00"}} + status, state = initial_outcome(setup, indicators, "M15") + assert status == "PENDING", state + return state + + +def _m34_bar(label, o, h, l, c): + return {"timestamp": f"2026-09-08T{label}:00+00:00", + "open": o, "high": h, "low": l, "close": c} + + +@test("advisory_outcome: a gap after the trail locked profit resolves as a " + "floor, not UNRESOLVABLE (audit 2026-09-13)") +def _(): + from advisory_outcome import (advance_outcome, canonical_r, + resolve_locked_floor) + state = _m34_sell_seed() + # Tuesday morning: entry at the next open, then a favourable run that + # takes the excursion past 0.70 R so the ratchet moves the stop BELOW + # the entry (a SELL's locked profit floor). + state = advance_outcome(state, [ + _m34_bar("07:30", 4396.0, 4397.0, 4394.0, 4395.5), # signal bar + _m34_bar("07:45", 4395.64, 4396.0, 4388.0, 4389.0), # entry + _m34_bar("08:00", 4389.0, 4390.0, 4345.06, 4350.0), # arms + ratchets + _m34_bar("08:15", 4350.0, 4356.0, 4348.0, 4352.0), + ]) + progress = state["progress"] + entry, stop = progress["entry_price"], progress["current_stop"] + risk = state["setup"]["stop_distance"] + assert state["status"] == "PENDING", state.get("reason") + assert entry == 4395.64 and stop < entry, progress + assert progress["last_processed_bar"] == "2026-09-08T08:15:00+00:00" + + # …then 8 tradable bars of that weekday morning are missing. + state = advance_outcome(state, [ + _m34_bar("08:15", 4350.0, 4356.0, 4348.0, 4352.0), + _m34_bar("10:30", 4380.0, 4381.0, 4379.0, 4380.5), + ]) + assert state["status"] == "UNRESOLVABLE", state + assert state["reason"] == \ + "stored candle snapshots contain an internal gap", state["reason"] + assert state["result"] is None and canonical_r(state) is None + + # The locked floor is a real, conservative outcome — keep it. + assert resolve_locked_floor(state) is state, "mutates in place" + assert state["status"] == "RESOLVED", state + result = state["result"] + assert result["exit_reason"] == "GAP_LOCKED", result + assert result["floor"] is True, result + assert result["exit_price"] == stop, (result["exit_price"], stop) + assert result["exit_at"] == "2026-09-08T08:15:00+00:00", result + expected = (entry - stop - 0.65) / risk + assert abs(result["net_r_065"] - expected) < 1e-9, (result, expected) + assert canonical_r(state) == result["net_r_065"], result + assert expected > 1.0, expected # row 915's ~+1.13 R was dropped + assert state["reason"].endswith( + "— resolved at the locked trailing stop (floor)"), state["reason"] + + +@test("advisory_outcome: a gap before the trail armed stays UNRESOLVABLE " + "(audit 2026-09-13)") +def _(): + from advisory_outcome import (advance_outcome, canonical_r, + resolve_locked_floor) + state = _m34_sell_seed() + state = advance_outcome(state, [ + _m34_bar("07:30", 4396.0, 4397.0, 4394.0, 4395.5), + _m34_bar("07:45", 4395.64, 4396.0, 4388.0, 4389.0), # 0.29 R < arm + ]) + progress = state["progress"] + assert state["status"] == "PENDING", state.get("reason") + assert progress["current_stop"] == progress["initial_stop"], progress + assert progress["current_stop"] > progress["entry_price"], progress + + state = advance_outcome(state, [ + _m34_bar("07:45", 4395.64, 4396.0, 4388.0, 4389.0), + _m34_bar("09:45", 4380.0, 4381.0, 4379.0, 4380.5), + ]) + assert state["status"] == "UNRESOLVABLE", state + assert "internal gap" in state["reason"], state["reason"] + + before = dict(state) + assert resolve_locked_floor(state)["status"] == "UNRESOLVABLE", state + assert state["result"] is None and canonical_r(state) is None, state + assert state["reason"] == before["reason"], "no floor, no rewrite" + + +@test("signal_log: the outcome resolver rebuilds the stream for both gap " + "reasons (audit 2026-09-13)") +def _(): + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + try: + lg = slog.SignalLogger(path) + calls = [] + + def _spy(symbol, timeframe, last_bar, max_rows=4000): + calls.append((symbol, str(getattr(timeframe, "value", timeframe)), + last_bar)) + return [] + + lg._stream_since = _spy + bars = _m27_bars("2026-08-19T12:00:00+00:00", 6, up=0.05) # Wednesday + + # 1. Overlaps the last processed bar but SKIPS one: the "internal + # gap" reason, which used to bypass the union rebuild entirely + # ("continuity" is not in its text). + a = lg.log(_m27_report(), source="telegram", timeframe="5min") + del calls[:] # log()'s scan-outcome labeller streams bars too + assert lg.resolve_pending( + "XAU/USD", slog._dumps({"5min": [bars[0], bars[2]]})) == 1 + assert len(calls) == 1, calls + assert calls[0][:2] == ("XAU/USD", "5min"), calls + assert calls[0][2] == "2026-08-19T12:00:00+00:00", calls + state = json.loads(lg._conn.execute( + "SELECT advisory_outcome_json FROM signals WHERE id=?", + (a,)).fetchone()[0]) + assert state["status"] == "UNRESOLVABLE", state + assert "internal gap" in state["reason"], state["reason"] + + # 2. The no-overlap ("lost continuity") reason rebuilds as before. + b = lg.log(_m27_report(entry=2460.0, sl=2455.0), source="telegram", + timeframe="5min") + del calls[:] + assert lg.resolve_pending( + "XAU/USD", slog._dumps({"5min": [bars[2], bars[3]]})) == 1 + assert len(calls) == 1, calls + state = json.loads(lg._conn.execute( + "SELECT advisory_outcome_json FROM signals WHERE id=?", + (b,)).fetchone()[0]) + assert "lost continuity" in state["reason"], state["reason"] + lg.close() + finally: + slog.reset_loggers_for_tests() + os.remove(path) + + +@test("dashboard: a missed H1/H4 boundary is caught up on the next wake " + "(audit 2026-09-13)") +def _(): + def ep(y, mo, d, h, m=0): + return datetime(y, mo, d, h, m, tzinfo=timezone.utc).timestamp() + + def names(wake, last): + return [tf.value for tf in dpub.catch_up_timeframes(wake, last)] + + wake = datetime(2026, 9, 10, 13, 5, 10, tzinfo=timezone.utc) # Thursday + + # Overrun/restart: the last H1 scan belonged to the 12:00 boundary (the + # 13:00 one was missed) and the last H4 scan to the 08:00 boundary (the + # 12:00 one was missed — the H4 grid is 00/04/08/12/16/20). M15's last + # closed boundary IS the one it scanned, so it is not due. + ledger = {"5min": ep(2026, 9, 10, 13, 5), "15min": ep(2026, 9, 10, 13), + "1h": ep(2026, 9, 10, 12), "4h": ep(2026, 9, 10, 8)} + assert names(wake, ledger) == ["1h", "4h"], names(wake, ledger) + + # Nothing missed → nothing to catch up; a TF never scanned is due. + current = dict(ledger, **{"1h": ep(2026, 9, 10, 13), + "4h": ep(2026, 9, 10, 12)}) + assert names(wake, current) == [], names(wake, current) + assert names(wake, {}) == ["5min", "15min", "1h", "4h"], names(wake, {}) + + # M5's own 13:05 boundary also counts as missed until it is scanned, but + # run_forever only scans the catch-up TFs that due_timeframes lacks. + stale = dict(ledger, **{"5min": ep(2026, 9, 10, 13)}) + assert names(wake, stale) == ["5min", "1h", "4h"], names(wake, stale) + due = dpub.due_timeframes(wake) + assert [tf.value for tf in due] == ["5min"], due + extra = [tf for tf in dpub.catch_up_timeframes(wake, stale) + if tf not in due] + assert [tf.value for tf in extra] == ["1h", "4h"], extra + + # Settlement pause (summer Tuesday = 21:00-21:59 UTC): the 22:00 H1 + # boundary would close a 21:00 bar that never printed, so a stale H1 + # ledger is NOT caught up there — the same exclusion due_timeframes + # makes — and is caught up an hour later, when a real bar exists. + paused = datetime(2026, 9, 8, 22, 5, 10, tzinfo=timezone.utc) + stale_h1 = {"5min": ep(2026, 9, 8, 22, 5), "15min": ep(2026, 9, 8, 22), + "1h": ep(2026, 9, 8, 20), "4h": ep(2026, 9, 8, 20)} + assert names(paused, stale_h1) == [], names(paused, stale_h1) + after = datetime(2026, 9, 8, 23, 5, 10, tzinfo=timezone.utc) + assert "1h" in names(after, stale_h1), names(after, stale_h1) + + # scan_tf records the boundary from the report's own bar LABEL, so the + # ledger tracks the bar the page is showing, not the wake clock. + pub = dpub.DashboardPublisher(out_dir="/nonexistent") + pub._provider = lambda: object() + report = make_fake_dashboard_report() + report.timestamp = "2026-09-10 12:00:00+00:00" + with patch("dashboard_publisher.produce_signal", return_value=report), \ + patch("dashboard_publisher.violent_watch_enabled", return_value=False): + pub.scan_tf(Timeframe.H1) + assert pub.snapshots["1h"]["status"] == "ok", pub.snapshots["1h"] + assert pub._last_boundary == {"1h": ep(2026, 9, 10, 13)}, pub._last_boundary + + # A blocked/failed scan carries no analysis: the caller's wake boundary + # is the fallback. + blocked = SignalReport(symbol=XAUUSD, display_name="XAU/USD", + timestamp="now") + with patch("dashboard_publisher.produce_signal", return_value=blocked), \ + patch("dashboard_publisher.violent_watch_enabled", return_value=False): + pub.scan_tf(Timeframe.H4, boundary_epoch=ep(2026, 9, 10, 12)) + assert pub._last_boundary["4h"] == ep(2026, 9, 10, 12), pub._last_boundary + + +@test("providers: fetch_range retries a failed page in place and audits " + "interior gaps (audit 2026-09-13)") +def _(): + import logging as _lg34 + import providers.base as pbase + + def _hourly(first, last, drop=None): + """Continuous weekday hourly UTC frame (the feed's real shape).""" + idx = pd.date_range(first, last, freq="1h", tz="UTC") + idx = idx[idx.dayofweek < 5] + if drop is not None: + idx = idx[~idx.isin(pd.DatetimeIndex(drop))] + v = 3000.0 + np.arange(len(idx)) * 0.01 + return pd.DataFrame( + {"Open": v, "High": v + 1, "Low": v - 1, "Close": v, + "Volume": 1.0}, index=pd.DatetimeIndex(idx, name="Datetime")) + + def _stub(series, fails_on=None, fail_times=1): + """Provider serving `series` sliced to the requested window, which + raises the first `fail_times` requests of a chosen page.""" + class _Gap(pbase.BaseDataProvider): + name = "gap34" + + def __init__(self, config): + super().__init__(config) + self.windows = [] + self.failures = {} + + def _fetch_raw(self, symbol, timeframe, outputsize, + start=None, end=None): + key = (pd.Timestamp(start), pd.Timestamp(end)) + self.windows.append(key) + if (fails_on is not None and fails_on(len(self.windows)) + and self.failures.get(key, 0) < fail_times): + self.failures[key] = self.failures.get(key, 0) + 1 + raise RuntimeError("blip") + out = series[(series.index >= key[0]) + & (series.index <= key[1])] + return out.tail(min(len(out), outputsize)) + + def get_pair_info(self, symbol): + return PAIR_INFO[XAUUSD] + + return _Gap(PlatformConfig()) + + records = [] + + class _Collect(_lg34.Handler): + def emit(self, rec): + records.append(rec) + + handler = _Collect(level=_lg34.ERROR) + old_level = pbase.logger.level + pbase.logger.addHandler(handler) + pbase.logger.setLevel(_lg34.ERROR) # also silences the chunk warnings + try: + # 1. A page that fails ONCE is re-requested, not skipped. H1 pages + # in 187-day chunks, so a 400-day window walks three of them; + # before 2026-09-13 a failure advanced the cursor past the page + # and left the whole middle of the frame missing (15 days on M5), + # which the backtester then cached as complete history. + start = datetime(2025, 8, 1, tzinfo=timezone.utc) + end = datetime(2026, 9, 4, tzinfo=timezone.utc) + series = _hourly("2025-08-01", "2026-09-04") + p = _stub(series, fails_on=lambda n: n == 2) # 2nd = middle page + df = p.fetch_range(XAUUSD, Timeframe.H1, start, end) + expected = series[(series.index >= pd.Timestamp(start)) + & (series.index <= pd.Timestamp(end))] + holes = expected.index.difference(df.index) + assert len(holes) == 0, \ + f"{len(holes)} labels missing after one page failure (first " \ + f"{holes[:1]}) — the failed window was skipped, not retried" + assert p.windows.count(p.windows[1]) == 2, \ + f"the failed window must be re-requested in place: {p.windows}" + assert not [r for r in records if "interior gap" in r.getMessage()], \ + [r.getMessage() for r in records] + + # 2. Three failures on the SAME page still refuse to hand back + # history with a hole in it. + p3 = _stub(series, fails_on=lambda n: True, fail_times=99) + try: + p3.fetch_range(XAUUSD, Timeframe.H1, start, end) + assert False, "3 failures on one window must raise" + except pbase.DataIntegrityError as e: + assert "truncated" in str(e), str(e) + assert len(set(p3.windows)) == 1 and len(p3.windows) == 3, \ + f"all attempts must be the same window: {p3.windows}" + + # 3. Interior-gap audit: 6 missing WEEKDAY hours are reported at + # ERROR (loud, not fatal — the frame comes back untouched). + del records[:] + w0 = datetime(2026, 8, 3, tzinfo=timezone.utc) + w1 = datetime(2026, 9, 4, tzinfo=timezone.utc) + hole = pd.date_range("2026-08-19 10:00", "2026-08-19 15:00", + freq="1h", tz="UTC") + gapped = _hourly("2026-08-03", "2026-09-04", drop=hole) + dfg = _stub(gapped).fetch_range(XAUUSD, Timeframe.H1, w0, w1) + msgs = [r.getMessage() for r in records] + assert any("interior gap" in m for m in msgs), msgs + assert len(dfg) == len(gapped), "the audit must not touch the frame" + + # 4. Weekend hours are a SCHEDULED closure — never reported. + del records[:] + _stub(_hourly("2026-08-03", "2026-09-04")).fetch_range( + XAUUSD, Timeframe.H1, w0, w1) + assert not [r for r in records if "interior gap" in r.getMessage()], \ + [r.getMessage() for r in records] + finally: + pbase.logger.removeHandler(handler) + pbase.logger.setLevel(old_level) + + +@test("telegram_bot: the boundary scan and the breach rescan share one " + "report per bar (audit 2026-09-13)") +def _(): + async def main(): + _m21c_reset() + epoch = _tb21._bar_epoch(Timeframe.M15) # this bar's boundary + calls = {"n": 0} + + def fake_sync(symbol, timeframe, confirmation_tfs=None): + calls["n"] += 1 + r = _m21c_report() + r.final_signal = SignalStrength.NEUTRAL # keep the alert path out + # OPEN label of the bar that closed at `epoch`. + r.timestamp = str(pd.Timestamp(epoch - 900, unit="s", tz="UTC")) + return r + + with patch.object(_tb21, "_run_signal_sync", fake_sync), \ + patch.object(_tb21, "_bar_epoch", lambda tf, now=None: epoch): + await _tb21._scan_timeframe( + _M21cBot(), Timeframe.M15, [Timeframe.M5, Timeframe.H1]) + assert calls["n"] == 1, "the boundary loop scans the closed bar once" + await _tb21._cached_or_scan("XAU/USD", Timeframe.M15) + assert calls["n"] == 1, \ + "a breach in the same bar must reuse the boundary report, " \ + f"got {calls['n']} scans (15 duplicate M15 rows in five days)" + # Next bar: the cached epoch no longer matches, so it scans again. + with patch.object(_tb21, "_run_signal_sync", fake_sync), \ + patch.object(_tb21, "_bar_epoch", lambda tf, now=None: epoch + 900): + await _tb21._cached_or_scan("XAU/USD", Timeframe.M15) + assert calls["n"] == 2, "a new bar must scan exactly once" + _m21c_reset() + + _aio21.run(main()) + + +@test("telegram_bot: a scan that straddles a boundary is filed under the bar " + "it analysed (audit 2026-09-13)") +def _(): + _m21c_reset() + key = f"XAU/USD_{Timeframe.M15.value}" + # Wall clock 12:00:05 → M15 boundary 12:00:00 (pd.Timestamp reads these + # strings as UTC, so the epochs are exact). + wall = int(pd.Timestamp("2026-09-10 12:00:05+00:00").timestamp()) + noon = int(pd.Timestamp("2026-09-10 12:00:00+00:00").timestamp()) + secs = _tb21._TF_SECONDS_CACHE[Timeframe.M15] + try: + with patch.object(_tb21, "_bar_epoch", + lambda tf, now=None: wall - wall % secs): + r = _m21c_report() + r.timestamp = "2026-09-10 11:45:00+00:00" # bar OPEN label + _tb21._remember_scan("XAU/USD", Timeframe.M15, r) + assert _tb21._last_scan[key][0] == noon, \ + f"11:45 + 15 min = 12:00, got {_tb21._last_scan[key][0]}" + + # The straddle that matters: a breach scan that read the 11:30 + # bar (11:45's was still forming) but finished after 12:00 + # belongs to the 11:45 boundary — filing it under 12:00 made the + # 12:00:10 loop reuse a report one bar behind. + r2 = _m21c_report() + r2.timestamp = "2026-09-10 11:30:00+00:00" + _tb21._remember_scan("XAU/USD", Timeframe.M15, r2) + assert _tb21._last_scan[key][0] == noon - secs, \ + f"expected the 11:45 boundary, got {_tb21._last_scan[key][0]}" + + # No primary_analysis (failed/blocked scan) → wall-clock epoch. + r3 = _m21c_report() + r3.primary_analysis = None + r3.timestamp = "2026-09-10 11:30:00+00:00" + _tb21._remember_scan("XAU/USD", Timeframe.M15, r3) + assert _tb21._last_scan[key][0] == noon, \ + "a report with no analysed bar keeps the wall-clock boundary" + + r4 = _m21c_report() # timestamp "t" — unparseable + _tb21._remember_scan("XAU/USD", Timeframe.M15, r4) + assert _tb21._last_scan[key][0] == noon, \ + "an unparseable label must fall back, not raise" + finally: + _m21c_reset() + + +@test("telegram_bot: a breach just after an M15 boundary waits for the " + "loop's scan (audit 2026-09-13)") +def _(): + async def main(): + _m21c_reset() + calls = {"n": 0} + + async def rs(*a, **k): + calls["n"] += 1 + r = _m21c_report() + r.final_signal = SignalStrength.NEUTRAL # keep the alert path out + return r + + bot = _M21cBot() + with patch.object(_tb21, "_WS_COOLDOWN", 0.0), \ + patch.object(_tb21, "run_signal", rs), \ + patch.object(_tb21, "_BREACH_DEFER_SECONDS", 0.2): + # Well past the boundary window: rescan immediately. + with patch.object(_tb21, "_bar_epoch", + lambda tf, now=None: time.time() - 5.0): + t0 = time.time() + await _tb21._handle_breach(bot, "XAU/USD", 3301.0, + "swing_high_1") + assert time.time() - t0 < 0.1, "no deferral outside the window" + assert calls["n"] == 1 + # Inside the window: wait out the remainder so the candle loop's + # fresh scan is the one that gets reused. + with patch.object(_tb21, "_bar_epoch", + lambda tf, now=None: time.time() - 0.05): + t0 = time.time() + await _tb21._handle_breach(bot, "XAU/USD", 3302.0, + "swing_high_1") + waited = time.time() - t0 + assert waited >= 0.1, \ + f"must defer ~0.15s to the boundary scan, waited {waited:.3f}s" + assert calls["n"] == 2 + _m21c_reset() + + _aio21.run(main()) + + +@test("telegram_bot: importing the module does not attach the file handler; " + "main() does (audit 2026-09-13)") +def _(): + import inspect + import logging as _lg + assert _tb21._file_handler not in _lg.getLogger().handlers, \ + "importing telegram_bot must not write logs/telegram.log — " \ + "test runs left fixture alerts (2027 dates, 3300.00) in it" + # main() boots the bot, so the other half is asserted from its source. + src = inspect.getsource(_tb21.main) + assert "addHandler(_file_handler)" in src, \ + "main() must attach the rotating file handler for real bot runs" + + +@test("signal_log: the scan-outcome labeler ignores stale incomplete rows " + "and anchors its stream within the max age (audit 2026-09-13)") +def _(): + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + try: + # Friday noon: every bar below is a weekday, so a hole in a window + # is a real hole and not a scheduled closure. + now = datetime(2026, 9, 11, 12, 0, tzinfo=timezone.utc) + lg = slog.SignalLogger(path) + # No opportunistic pass inside log(): every labelling in this test + # is an explicit call with a fixed `now`. + lg._next_scan_label = time.monotonic() + 3600.0 + + # (i) The ancient row: a real hole in its window (the 2026-09-07 + # feed outage), labelled incomplete days ago, unimprovable. + old_bar = "2026-09-01 12:00:00+00:00" + old_labeled = "2026-09-01T13:00:00+00:00" + old_id = lg.log(_m31_report(old_bar), source="telegram", + timeframe="5min") + lg._conn.execute("UPDATE signals SET logged_at=? WHERE id=?", + ("2026-09-01T12:05:00+00:00", old_id)) + lg._conn.execute( + "INSERT OR REPLACE INTO scan_outcomes (signal_id, timeframe," + " bar_ts, complete, labeled_at) VALUES (?, ?, ?, ?, ?)", + (old_id, "5min", old_bar, 0, old_labeled)) + lg._conn.commit() + + # (ii) A row 30 h old whose window starts 4 h before the stored + # bars: a genuine hole, but young enough to keep re-examining. + hole_id = lg.log(_m31_report("2026-09-10 06:00:00+00:00"), + source="telegram", timeframe="5min") + # (iii) The fresh due row: bar 26 h back, never labelled, with its + # whole 24 h window stored the way live scans store it. + bars = _m31_bars("2026-09-10T10:00:00+00:00", 300, up=0.1) + fresh_id = lg.log(_m31_report(str(pd.Timestamp(bars[0]["timestamp"]))), + source="telegram", timeframe="5min", + dataframes={Timeframe.M5: _m27_df(bars[0:1])}) + _m31_log_stream(lg, bars) + + seen = [] + real_stream = lg._stream_since + + def _spy(symbol, timeframe, last_bar, max_rows=4000): + seen.append((last_bar, max_rows)) + return real_stream(symbol, timeframe, last_bar, max_rows) + + lg._stream_since = _spy + lg.label_scan_outcomes(budget=500, now=now, time_budget=30.0) + + assert lg._conn.execute( + "SELECT complete FROM scan_outcomes WHERE signal_id=?", + (fresh_id,)).fetchone() == (1,), "the fresh window is judgeable" + assert lg._conn.execute( + "SELECT labeled_at FROM scan_outcomes WHERE signal_id=?", + (old_id,)).fetchone()[0] == old_labeled, \ + "a stale incomplete row must not be re-examined at all" + assert seen, "the labeller built no bar stream" + floor = now - timedelta(hours=slog.SCAN_LABEL_MAX_AGE_H + + slog.SCAN_LABEL_MIN_AGE_H) + for last_bar, max_rows in seen: + stamp = datetime.fromisoformat(str(last_bar).replace(" ", "T")) + assert stamp >= floor, (last_bar, floor) + assert max_rows == 10000, max_rows + + # (b) The hole row stays incomplete; a further pass writes nothing + # but still moves it down the queue. + before = lg._conn.execute( + "SELECT complete, labeled_at FROM scan_outcomes" + " WHERE signal_id=?", (hole_id,)).fetchone() + assert before[0] == 0, before + assert lg.label_scan_outcomes(budget=500, now=now, + time_budget=30.0) == 0 + after = lg._conn.execute( + "SELECT complete, labeled_at FROM scan_outcomes" + " WHERE signal_id=?", (hole_id,)).fetchone() + assert after[0] == 0 and after[1] > before[1], (before, after) + + # (c) A truncated stream says so, instead of silently ending early. + records = [] + + class _Grab(logging.Handler): + def emit(self, record): + records.append(record) + + log = logging.getLogger("signal_log") + handler = _Grab() + log.addHandler(handler) + try: + real_stream("XAU/USD", "5min", "2026-09-10T06:00:00+00:00", 1) + finally: + log.removeHandler(handler) + assert any(r.levelno == logging.WARNING + and "truncated" in r.getMessage() for r in records), \ + [r.getMessage() for r in records] + lg.close() + finally: + slog.reset_loggers_for_tests() + os.remove(path) + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════