From fc55da17db81bc534454404adf7b1f7d7e93b11c Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Sat, 5 Sep 2026 22:22:59 -0400 Subject: [PATCH] signal_log: weight-research data points, scan_outcomes forward labels, watch provenance New additive columns on every logged scan: bar_ts (primary closed-bar label, indexed with timeframe), engine_fingerprint (hash of the scoring config incl. the regime weight table in force - SignalAggregator.engine_config_fingerprint, twin of the gate fingerprint), code_version (short git SHA), penalty_tags_json / penalty_total / momentum_fading / exhaustion_count, and post-hoc watch_level / watch_kinds / watch_line written by violent_watch.attach_watch via signal_log.update_watch. Each indicators_json entry now also carries eff_weight (base weight x regime multiplier, the weight the composite actually used) and wscore. engine: IndicatorSignal.effective_weight / weighted_score are stamped inside _compute_composite (record-keeping only); regime_weight_table(config) is the single source of truth for the multiplier table (override honoured). Composite output byte-identical (120-report hash proof). New table scan_outcomes: model-free forward moves per scan (fwd_1h/4h/24h in $ and ATRs, mfe/mae over 24h, burst_1h, bars_used, complete) measured on the stored candle stream with the advisory model's continuity rule; labelled opportunistically at the end of log() (time-boxed, throttled, idempotent) - no daemon; scripts/label_scan_outcomes.py drives the backlog (--dry-run is read-only). analyze_signal_log --scan-outcomes: forward move + hit rate by TF x signal/grade with a NEUTRAL baseline and burst rate by watch level. Readers dedup by the first row per (timeframe, bar_ts). Module 31 (10 tests). Suite: 463 passed. Backtester untouched (test-locked). Claude-Session: https://claude.ai/code/session_01RWV2yLt3YXHHGVYHSQU2EE --- CLAUDE.md | 2 +- analyze_signal_log.py | 102 ++++++++ engine.py | 84 ++++--- scripts/label_scan_outcomes.py | 115 +++++++++ signal_aggregator.py | 37 ++- signal_log.py | 391 ++++++++++++++++++++++++++++++- test_all.py | 409 +++++++++++++++++++++++++++++++++ violent_watch.py | 16 ++ 8 files changed, 1117 insertions(+), 39 deletions(-) create mode 100755 scripts/label_scan_outcomes.py diff --git a/CLAUDE.md b/CLAUDE.md index cb65707..becc897 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -58,7 +58,7 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz - **`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. 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), 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` (what the XAU_CP acceptance stripped), `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), `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). 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), `scripts/backfill_signal_outcomes.py` (seed/dedup/advance legacy rows), `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 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` (what the XAU_CP acceptance stripped), `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), `config_fingerprint` (gate config hash — surfaces config drift between processes). **News is content-addressed:** `headlines_json`/`events_json` are interned into `news_snapshots` (they were 73% of a 667 MB DB); read through the `signals_with_news` view or `signal_log.read_news`. Readers must open with `mode=ro`, never `immutable=1` (blind to WAL rows). **Weight-research columns (2026-09):** `bar_ts` (the primary bar LABEL the scan closed on — same string as `indicators_json`, indexed with `timeframe`), `engine_fingerprint` (hash of the SCORING config: indicator params, thresholds, the regime weight table in force, TFs — `SignalAggregator.engine_config_fingerprint`, the gate fingerprint's twin), `code_version` (short git SHA), `penalty_tags_json`/`penalty_total`/`momentum_fading`/`exhaustion_count`, and the post-hoc `watch_level`/`watch_kinds`/`watch_line` written by `signal_log.update_watch` from `violent_watch.attach_watch` (level and kinds are stored even when the display line is off). Every `indicators_json` entry also carries `eff_weight` (base weight × regime multiplier = the weight the composite actually used; `engine.regime_weight_table` is the single source of truth) and `wscore` (its contribution) — null for skipped indicators (weight 0, `_`-prefixed); **divergence and candlestick entries exist only when they fire, so absence is "no vote", not a zero vote**. **Table `scan_outcomes`:** model-free forward moves per scan (`fwd_1h/4h/24h` in $ and ATRs, `mfe_24h_atr`/`mae_24h_atr`, `burst_1h` = a ≥3×ATR bar within the hour, `bars_used`, `complete`), measured from the row's own last closed bar on the stored candle stream; a horizon stays NULL until it has elapsed, and `complete` needs the full 24 h with no gap over 3 bars beyond the scheduled weekend/17:00-NY closures (`advisory_outcome._has_continuity`). `SignalLogger.label_scan_outcomes` runs opportunistically at the end of `log()` (at most once per 60 s per process, 150 rows and ~2 s per pass, idempotent — a row is rewritten only when its numbers change), so no daemon is needed; `scripts/label_scan_outcomes.py` is the backlog driver (`--dry-run` counts read-only). Readers must dedup by the FIRST row per (`timeframe`, `bar_ts`) — several surfaces log the same bar. Tools: `analyze_signal_log.py` (`--stored-candles` DB-only replay with event/opportunity/one-position books + persona scorecard, `--news` news-sentiment lead/lag + extremes + outcome buckets, `--parity` telegram-vs-dashboard divergence report, `--scan-outcomes` forward-move + hit-rate table by TF×signal/grade with a NEUTRAL baseline and burst rate by watch level), `scripts/backfill_signal_outcomes.py` (seed/dedup/advance legacy rows), `scripts/label_scan_outcomes.py` (label/relabel scan outcomes), `scripts/migrate_signal_log_news.py` (intern legacy blobs; `--vacuum` with both writers stopped). - **`news_engine.py`** — Forex Factory calendar (MyFXBook RSS fallback) + RSS keyword sentiment with gold-inverse keywords. Class-level caches (fresh engine per scan). Optional DeepSeek enrichment via `llm_client.py` (requests-only, no SDK): batched headline sentiment overriding keyword scores, gold-direction on release surprises, and a market-summary blurb (`SignalReport.news_summary`) — **display/warning-only, keyword fallback on any failure, inert without a key**; prompts are static module constants (DeepSeek prefix-cache pricing) and requests send `thinking: disabled` (V4-Flash reasoning otherwise blows the timeout). By XAU-tuned default, event risk and sentiment **warn but do not gate signals** (`event_risk_caps_signals=False`, `sentiment_affects_signals=False`) — the LLM does not change that. **Headline pool (2026-09-04):** only headlines ≤ 4 h old with a parsable publish time (`headline_max_age_seconds=14400`, undated items dropped), and `get_fundamental_context` drops headlines whose every relevant score is inside ±0.1 (`_is_neutral_headline`, `NEUTRAL_HEADLINE_THRESHOLD`) before the per-currency mean — so `news_sentiment` from that date is a mean over directional headlines only (bolder values; log comparisons must split at the deploy date). The market summary prompt writes for someone who has never traded (no acronyms), sends headlines newest-first with the platform's own gold direction, and caps at 400 chars. **Verdict 2026-09-04** (`backtest/research/news_sentiment_value_2026-09-04.md`): `news_sentiment` is a trailing-24h echo of price (r≈0.67 with the past day, ≈0.03 with the next; flips 44 h after tops / 16 h after bottoms) — both gating flags stay False; re-check with `analyze_signal_log.py --stored-candles --news`. - **`agent_panel.py`** — ai-hedge-fund-style advisory panel: 4 DeepSeek persona calls (macro/technical/risk/contrarian, concurrent on a panel-private pool) + 1 synthesizer producing an **ADVISORY-ONLY** consensus (`AGREE/NEUTRAL/OPPOSE` + confidence + rationale) written onto `SignalReport.panel_*` fields. Attached **post-hoc** by consumers (`telegram_bot` alert path, `main --panel`, `dashboard_publisher`) — never runs inside `analyze_from_dataframes` (backtest parity, test-locked) and no gate reads its fields. Persona payloads are role-filtered: macro gets high-impact events plus only code-scored directional news; technical gets strategy/pattern, primary indicators, MTF, levels, and ATR-normalized candles; risk gets setup strategy, strategy cost assumption, trailing semantics, volatility/events/warnings (never the median candle range as spread); contrarian gets directional news, setup, timestamped latest-closed price, indicators, reasons, and warnings. Spend is bounded by exact cache, full-evidence-identity TTL reuse, and `max_panels_per_day`; changed evidence never inherits a direction-only cached verdict (test-locked — which also means TTL reuse effectively never fires live because price/candles change every bar; the daily cap is the real spend bound). Verdict provenance logs creation time, `MISS`/`EXACT_HIT`/`TTL_REUSE`, and a SHA-256 payload hash. Prompts forbid outside knowledge and outputs remain short. **OPPOSE guard (2026-09):** the 3-week signal-log scorecard showed the contrarian/risk/technical personas oppose almost every trend entry (chair said OPPOSE 73% of the time; opposed trades made +30R, PF 2.6) while the macro persona was the only informative voice (macro=support → 88% win). `_apply_oppose_guard` therefore downgrades a chair OPPOSE to NEUTRAL unless macro opposes, or technical opposes while macro is not supportive; the synthesizer prompt states the same rule. Re-score with `analyze_signal_log.py --stored-candles` (persona scorecard) before changing it again. **Verdict source (2026-09-03, owner decision):** all four personas still run and are logged (`panel_takes_json`), but `agent_panel.verdict_source` defaults to `"macro"` — the alert verdict is the macro persona's stance (support/neutral/oppose → AGREE/NEUTRAL/OPPOSE) and the synthesizer call is skipped (4 LLM calls per panel instead of 5); `"chair"` restores the four-voice synthesizer + OPPOSE guard. Re-score personas monthly with `analyze_signal_log.py --stored-candles`; weights are not to be tuned before ~300 graded outcomes with a held-out half. - **`market_outlook.py`** — hourly DeepSeek market brief for XAU/USD, using ATR-normalized M15/H1/H4 candles, visible indicator states, levels, news, and recent signal history (read `mode=ro`). It returns a written **ADVISORY-ONLY** bias/setup-forming assessment and persists it independently in `outlooks` within `logs/signals.db` together with the news, indicator/regime/level, and candle evidence sent to the model. It may be sent post-scan by Telegram or requested with `main.py --outlook`; it never imports or runs inside the signal path. Hourly dedup is by UTC hour BUCKET (the old `now − last < 3600` rule dropped every other hour because the H1 loop fires at boundary+10 s+scan time) plus its own UTC-day cap (default 12); every silent early return logs `market outlook: skipped (...)` and an empty LLM reply logs a warning — a dark feature must be visible in `logs/telegram.log`. Provider/news failures leave a safe partial brief while LLM/database failures degrade to no outlook. diff --git a/analyze_signal_log.py b/analyze_signal_log.py index fe94128..5f18679 100644 --- a/analyze_signal_log.py +++ b/analyze_signal_log.py @@ -1152,6 +1152,101 @@ def export_csv(events: Sequence[SignalEvent], path: str) -> None: print(f"\nWrote {len(recs)} events to {path}") +# ── Scan outcomes (model-free forward returns; no bar data needed) ──── + + +def scan_outcome_rows(db_path: str) -> List[dict]: + """One labelled scan per (timeframe, bar_ts) — the LOWEST signal id. + + Several surfaces log the same bar, so the raw table is duplicated; + SQLite takes the bare columns from the row min() selected.""" + con = open_readonly(db_path) + try: + if not con.execute( + "SELECT 1 FROM sqlite_master WHERE type='table'" + " AND name='scan_outcomes'").fetchone(): + return [] + return [ + {"id": r[0], "timeframe": r[1], "signal": r[3] or "", + "quality": r[4] or "-", "watch_level": r[5], + "fwd_4h_atr": r[6], "burst_1h": r[7], "complete": r[8]} + for r in con.execute( + "SELECT MIN(o.signal_id), o.timeframe, o.bar_ts, s.signal," + " s.quality, s.watch_level, o.fwd_4h_atr, o.burst_1h," + " o.complete FROM scan_outcomes o" + " JOIN signals s ON s.id = o.signal_id" + " GROUP BY o.timeframe, o.bar_ts") + ] + finally: + con.close() + + +def _scan_stats(rows: Sequence[dict], directional: bool) -> Optional[dict]: + """n / mean / median forward move and the hit rate for one bucket. + + 'Hit' = the 4 h move agreed with the signal; for the NEUTRAL baseline + it degrades to the share of UP moves.""" + vals = sorted(r["fwd_4h_atr"] for r in rows if r["fwd_4h_atr"] is not None) + if not vals: + return None + hits = 0 + for row in rows: + move = row["fwd_4h_atr"] + if move is None: + continue + sign = -1 if (directional and "SELL" in row["signal"]) else 1 + hits += int(sign * move > 0) + mid = len(vals) // 2 + return {"n": len(vals), "mean": sum(vals) / len(vals), + "median": vals[mid] if len(vals) % 2 else + (vals[mid - 1] + vals[mid]) / 2, + "hit_rate": hits / len(vals)} + + +def print_scan_outcomes(rows: Sequence[dict]) -> None: + """Forward-move report: no replay, no model — just what happened.""" + if not rows: + print("No scan_outcomes rows yet " + "(python scripts/label_scan_outcomes.py)") + return + complete = [r for r in rows if r["complete"]] + print(f"\n{'═' * 66}\n SCAN OUTCOMES — 4 h forward move in ATRs " + f"({len(complete)} complete of {len(rows)} labelled bars)") + print(f"{'═' * 66}") + directional = [r for r in complete if r["signal"] not in ("", "NEUTRAL")] + neutral = [r for r in complete if r["signal"] == "NEUTRAL"] + for title, groups, keyfn, is_dir in ( + ("by timeframe x signal", directional, + lambda r: (r["timeframe"], r["signal"]), True), + ("by timeframe x grade", directional, + lambda r: (r["timeframe"], r["quality"]), True), + ("NEUTRAL baseline", neutral, + lambda r: (r["timeframe"], "NEUTRAL"), False)): + print(f"\n── {title} " + "─" * (52 - len(title))) + buckets: Dict[Tuple[str, str], List[dict]] = {} + for row in groups: + buckets.setdefault(keyfn(row), []).append(row) + for key in sorted(buckets): + stat = _scan_stats(buckets[key], is_dir) + if stat is None: + continue + print(f" {key[0]:6} {key[1]:12} n={stat['n']:5d} " + f"mean {stat['mean']:+.3f} med {stat['median']:+.3f} " + f"{'hit' if is_dir else 'up '} " + f"{100 * stat['hit_rate']:.0f}%") + print("\n── burst (>=3x ATR bar) within 1 h, by watch level " + "─" * 15) + levels: Dict[str, List[int]] = {} + for row in complete: + if row["burst_1h"] is None: + continue + levels.setdefault(row["watch_level"] or "NULL", []).append( + int(row["burst_1h"])) + for level in sorted(levels): + hits = levels[level] + print(f" {level:6} n={len(hits):5d} burst " + f"{100 * sum(hits) / len(hits):.1f}%") + + # ── main ────────────────────────────────────────────────────────────── @@ -1172,8 +1267,15 @@ def main(argv: Optional[Sequence[str]] = None) -> int: ap.add_argument("--news", action="store_true", help="news-sentiment lead/lag + outcome buckets " "(requires --stored-candles)") + ap.add_argument("--scan-outcomes", action="store_true", + help="labelled forward moves per scan (no bar data " + "needed); see scripts/label_scan_outcomes.py") args = ap.parse_args(argv) + if args.scan_outcomes: + print_scan_outcomes(scan_outcome_rows(args.db)) + return 0 + if args.news and not args.stored_candles: print("--news requires --stored-candles") return 2 diff --git a/engine.py b/engine.py index a78995d..2c3628f 100755 --- a/engine.py +++ b/engine.py @@ -13,7 +13,7 @@ Design philosophy: import logging from dataclasses import dataclass, field -from typing import Dict, List, Tuple +from typing import Dict, List, Optional, Tuple import numpy as np import pandas as pd @@ -32,6 +32,13 @@ class IndicatorSignal: confidence: float # 0-100 description: str weight: float = 1.0 + # RECORD-KEEPING ONLY (stamped by _compute_composite; nothing reads + # them back). effective_weight = weight x regime multiplier actually + # used, weighted_score = its contribution to the composite. Both stay + # None for indicators the composite skipped (weight 0, `_`-prefixed + # private series) and for analyses that never reached _compute_composite. + effective_weight: Optional[float] = None + weighted_score: Optional[float] = None @dataclass @@ -50,6 +57,46 @@ class TechnicalAnalysis: pattern_signals: List[str] = field(default_factory=list) +# NOTE: ROC is directionally redundant with MACD/Keltner/EMA +# (0.8-0.94 correlation), but zeroing its weight was backtested +# 2026-07 and LOST 13R — the redundant vote effectively raises +# the composite bar and the thresholds are calibrated with it. +# Don't remove without retuning score thresholds. +_TRENDING_WEIGHTS = { + "ema_trend": 2.0, "rsi": 1.0, "macd": 2.0, + "bollinger": 0.5, "keltner": 1.5, "roc": 1.5, + "candlestick": 0.8, "divergence": 1.0, "adx": 0.8, +} +_DEFAULT_REGIME_WEIGHTS = { + "trending_up": _TRENDING_WEIGHTS, + "trending_down": _TRENDING_WEIGHTS, + "trending": _TRENDING_WEIGHTS, + "ranging": { + "ema_trend": 0.3, "rsi": 1.5, "macd": 1.5, + "bollinger": 1.5, "keltner": 2.0, "roc": 0.5, + "candlestick": 1.0, "divergence": 2.5, "adx": 0.5, + }, + "transitioning": { + "ema_trend": 1.5, "rsi": 1.0, "macd": 2.5, + "bollinger": 1.0, "keltner": 2.5, "roc": 2.0, + "candlestick": 1.0, "divergence": 2.0, "adx": 1.0, + }, +} + + +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). + + 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) + table = override if override else _DEFAULT_REGIME_WEIGHTS + return {regime: dict(weights) for regime, weights in table.items()} + + class TechnicalEngine: """ Core technical analysis engine. @@ -1081,35 +1128,9 @@ class TechnicalEngine: total_weight = 0.0 total_confidence = 0.0 - # Use override weights if provided, else gold-optimized defaults - if self.config.regime_weights_override: - regime_weights = self.config.regime_weights_override - else: - # NOTE: ROC is directionally redundant with MACD/Keltner/EMA - # (0.8-0.94 correlation), but zeroing its weight was backtested - # 2026-07 and LOST 13R — the redundant vote effectively raises - # the composite bar and the thresholds are calibrated with it. - # Don't remove without retuning score thresholds. - _trending = { - "ema_trend": 2.0, "rsi": 1.0, "macd": 2.0, - "bollinger": 0.5, "keltner": 1.5, "roc": 1.5, - "candlestick": 0.8, "divergence": 1.0, "adx": 0.8, - } - regime_weights = { - "trending_up": _trending, - "trending_down": _trending, - "trending": _trending, - "ranging": { - "ema_trend": 0.3, "rsi": 1.5, "macd": 1.5, - "bollinger": 1.5, "keltner": 2.0, "roc": 0.5, - "candlestick": 1.0, "divergence": 2.5, "adx": 0.5, - }, - "transitioning": { - "ema_trend": 1.5, "rsi": 1.0, "macd": 2.5, - "bollinger": 1.0, "keltner": 2.5, "roc": 2.0, - "candlestick": 1.0, "divergence": 2.0, "adx": 1.0, - }, - } + # Override weights if provided, else gold-optimized defaults + # (regime_weight_table is the single source of truth). + regime_weights = regime_weight_table(self.config) active_regime_weights = regime_weights.get(result.market_regime, {}) @@ -1123,6 +1144,9 @@ class TechnicalEngine: signal_score = ind.signal.value weighted_score = signal_score * weight * (ind.confidence / 100.0) + # Record-keeping stamp (no maths depends on these two lines). + ind.effective_weight = float(weight) + ind.weighted_score = float(weighted_score) total_score += weighted_score total_weight += weight diff --git a/scripts/label_scan_outcomes.py b/scripts/label_scan_outcomes.py new file mode 100755 index 0000000..7ba76ad --- /dev/null +++ b/scripts/label_scan_outcomes.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +""" +label_scan_outcomes.py — batch driver for the scan_outcomes table. + +Labels model-free forward returns (1/4/24 h, $ and ATR-normalized, plus +MFE/MAE, a 3xATR burst flag and a completeness flag) for every logged scan +whose 24 h window has closed, from the closed-candle streams already +stored in the DB. No provider, no network. + +The live loggers do the same work opportunistically at the end of every +scan (time-boxed to ~2 s); this script is the catch-up pass for a backlog +— a freshly imported database, or rows whose bars only arrived later. + +Idempotent: a row is rewritten only when its numbers change, so a settled +database costs one read pass. Safe beside the live writers (short per-row +transactions with a busy timeout). Read-only preview with --dry-run. + + python scripts/label_scan_outcomes.py --dry-run + python scripts/label_scan_outcomes.py + python scripts/label_scan_outcomes.py --db logs/signals.db --budget 5000 +""" +import argparse +import os +import sqlite3 +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import app_config # noqa: E402 +from signal_log import SCAN_LABEL_MIN_AGE_H, SignalLogger # noqa: E402 + +CHUNK = 2000 + + +def _counts(con: sqlite3.Connection, symbol: str, + labelled: bool = True) -> dict: + """Candidate / labelled / complete / pending counts (read-only). + + The age filter uses logged_at (always an ISO string) as a proxy for + the bar close — the labeller itself parses the real bar label.""" + scans = ("SELECT COUNT(*) FROM signals s WHERE s.symbol = ?" + " AND s.indicators_json IS NOT NULL AND s.price > 0") + due = (scans + " AND s.logged_at <" + f" datetime('now', '-{int(SCAN_LABEL_MIN_AGE_H)} hours')") + out = { + "scan rows with indicators": con.execute(scans, (symbol,)).fetchone()[0], + "of those, window elapsed": con.execute(due, (symbol,)).fetchone()[0], + } + if labelled: + out["labelled"] = con.execute( + "SELECT COUNT(*) FROM scan_outcomes").fetchone()[0] + out["complete"] = con.execute( + "SELECT COUNT(*) FROM scan_outcomes WHERE complete = 1" + ).fetchone()[0] + out["pending (labelled, incomplete)"] = con.execute( + "SELECT COUNT(*) FROM scan_outcomes WHERE complete = 0" + ).fetchone()[0] + return out + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(description=__doc__.splitlines()[1]) + ap.add_argument("--db", default=None, help="defaults to signal_log.path") + ap.add_argument("--dry-run", action="store_true", + help="read-only candidate counts, writes nothing") + ap.add_argument("--budget", type=int, default=100000, + help="maximum rows to write in this run") + ap.add_argument("--symbol", default="XAU/USD") + args = ap.parse_args(argv) + path = args.db or app_config.signal_log_path() + if not os.path.exists(path): + print(f"no database at {path}") + return 1 + + if args.dry_run: + con = sqlite3.connect(f"file:{os.path.abspath(path)}?mode=ro", uri=True) + try: + have = {r[0] for r in con.execute( + "SELECT name FROM sqlite_master WHERE type='table'")} + if "scan_outcomes" not in have: + print("scan_outcomes table not present yet (created on the" + " first non-dry run / next logged scan)") + for key, value in _counts(con, args.symbol, + labelled="scan_outcomes" in have).items(): + print(f" {key}: {value}") + finally: + con.close() + return 0 + + lg = SignalLogger(path) + try: + print("before:", _counts(lg._conn, args.symbol)) + written = 0 + previous_complete = -1 + while written < args.budget: + n = lg.label_scan_outcomes( + budget=min(CHUNK, args.budget - written), time_budget=120.0) + written += n + complete = lg._conn.execute( + "SELECT COUNT(*) FROM scan_outcomes WHERE complete = 1" + ).fetchone()[0] + # Stop on a pass that wrote nothing, or one that could not + # complete a single further row (only re-labelling stragglers). + if n == 0 or complete == previous_complete: + break + previous_complete = complete + print(f"labelled (rows written): {written}") + print("after:", _counts(lg._conn, args.symbol)) + finally: + lg.close() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/signal_aggregator.py b/signal_aggregator.py index 7254b6c..2bfe558 100755 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -9,7 +9,7 @@ import hashlib import json import logging import os -from dataclasses import dataclass, field +from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from typing import Dict, List, Optional from zoneinfo import ZoneInfo @@ -21,7 +21,7 @@ _NY_TZ = ZoneInfo("America/New_York") from chart_patterns import detect_chart_pattern from config import PlatformConfig, SignalStrength, Timeframe from providers import BaseDataProvider, make_data_provider -from engine import TechnicalEngine, TechnicalAnalysis +from engine import TechnicalEngine, TechnicalAnalysis, regime_weight_table from news_engine import NewsEngine, FundamentalContext import signal_log @@ -322,11 +322,13 @@ class SignalAggregator: logger.debug("signal log: news context unavailable", exc_info=True) extra = {"scan_kind": "blocked" if blocked else "scan"} - try: - extra["config_fingerprint"] = self.gate_config_fingerprint() - except Exception: - logger.debug("signal log: config fingerprint unavailable", - exc_info=True) + for key, fingerprint in ( + ("config_fingerprint", self.gate_config_fingerprint), + ("engine_fingerprint", self.engine_config_fingerprint)): + try: + extra[key] = fingerprint() + except Exception: + logger.debug("signal log: %s unavailable", key, exc_info=True) signal_log.log_signal( report, context=context, @@ -365,6 +367,27 @@ class SignalAggregator: json.dumps(payload, sort_keys=True, default=str).encode("utf-8") ).hexdigest()[:12] + def engine_config_fingerprint(self) -> str: + """Short hash of the SCORING configuration this process runs with. + + gate_config_fingerprint covers the post-grade gates; this covers + what produced the composite in the first place — indicator + parameters, signal thresholds/geometry, the regime weight + multipliers actually in force (override included) and the + timeframe set. Logged on every row so indicator-weight research + can tell which rows share a scoring configuration.""" + payload = { + "technical": asdict(self.config.technical), + "signals": asdict(self.config.signals), + "regime_weights": regime_weight_table(self.config.technical), + "primary": self.config.primary_timeframe.value, + "confirm": [tf.value for tf in self.config.confirmation_timeframes], + "display_name": self.config.display_name, + } + return hashlib.sha1( + json.dumps(payload, sort_keys=True, default=str).encode("utf-8") + ).hexdigest()[:12] + def _apply_regime_filter_live( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> None: diff --git a/signal_log.py b/signal_log.py index 1f7f0ed..b717345 100644 --- a/signal_log.py +++ b/signal_log.py @@ -56,6 +56,29 @@ or news_engine) so this module stays at the bottom of the import graph. (``scan`` / ``blocked`` for weekend/day early returns, which carry no news or candles) and ``config_fingerprint`` (hash of the gate config the producing process ran with — surfaces config drift between processes). + +2026-09 weight-research additions: + +* Row columns ``bar_ts`` (the primary bar LABEL this scan closed on, + stored verbatim so it joins to indicators_json), ``engine_fingerprint`` + (hash of the SCORING config — indicator params, thresholds, regime + weight multipliers, timeframes), ``code_version`` (short git SHA), + ``penalty_tags_json`` / ``penalty_total`` / ``momentum_fading`` / + ``exhaustion_count`` (the decay/exhaustion record), and the post-hoc + ``watch_level`` / ``watch_kinds`` / ``watch_line`` written by + ``update_watch`` from violent_watch.attach_watch. +* Every per-indicator entry in ``indicators_json`` also carries + ``eff_weight`` (base weight x regime multiplier — the weight the + composite actually used) and ``wscore`` (its contribution). Both are + null for indicators the composite skipped. Divergence and candlestick + entries exist only when they fire: absence is "no vote", not zero. +* Table ``scan_outcomes`` — model-free forward returns per scan + (1/4/24 h, $ and ATR-normalized, plus MFE/MAE, a 3xATR burst flag and + a completeness flag), labelled from the stored candle streams by + ``label_scan_outcomes`` (called opportunistically at the end of log(), + time-boxed; batch driver: scripts/label_scan_outcomes.py). Readers + must dedup by the FIRST row per (timeframe, bar_ts) — several surfaces + log the same bar. """ import hashlib @@ -64,9 +87,10 @@ import logging import math import os import sqlite3 +import subprocess import threading import time -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import Any, Dict, Iterable, List, Optional, Tuple import app_config @@ -74,8 +98,13 @@ from advisory_outcome import ( advance_outcome, canonical_r, initial_outcome, + normalize_timeframe, timeframe_value, ) +# Private on purpose — the scan-outcome labeller must apply EXACTLY the +# advisory model's bar validation and scheduled-closure rule (NY-anchored +# weekend + the daily 17:00 NY maintenance hour), never a second copy. +from advisory_outcome import _TF_SECONDS, _has_continuity, _valid_bars logger = logging.getLogger(__name__) @@ -128,6 +157,27 @@ CREATE TABLE IF NOT EXISTS news_snapshots ( first_seen TEXT NOT NULL, last_seen TEXT ); +CREATE TABLE IF NOT EXISTS scan_outcomes ( + signal_id INTEGER PRIMARY KEY REFERENCES signals(id), + timeframe TEXT, + bar_ts TEXT, + atr REAL, + close0 REAL, + fwd_1h REAL, + fwd_4h REAL, + fwd_24h REAL, + fwd_1h_atr REAL, + fwd_4h_atr REAL, + fwd_24h_atr REAL, + mfe_24h_atr REAL, + mae_24h_atr REAL, + burst_1h INTEGER, + bars_used INTEGER, + complete INTEGER, + labeled_at TEXT +); +CREATE INDEX IF NOT EXISTS idx_scan_outcomes_tf_bar + ON scan_outcomes (timeframe, bar_ts); CREATE TABLE IF NOT EXISTS outlooks ( id INTEGER PRIMARY KEY AUTOINCREMENT, logged_at TEXT NOT NULL, @@ -170,6 +220,18 @@ _ADDITIVE_SIGNAL_COLS = ( ("scan_kind", "TEXT"), ("config_fingerprint", "TEXT"), ("panel_verdict_source", "TEXT"), + # 2026-09 weight-research columns (see module docstring). + ("bar_ts", "TEXT"), + ("engine_fingerprint", "TEXT"), + ("code_version", "TEXT"), + ("penalty_tags_json", "TEXT"), + ("penalty_total", "REAL"), + ("momentum_fading", "INTEGER"), + ("exhaustion_count", "INTEGER"), + # Violent-move watch, written post-hoc by update_watch. + ("watch_level", "TEXT"), + ("watch_kinds", "TEXT"), + ("watch_line", "TEXT"), ) # Read-side compatibility view `signals_with_news`: legacy rows keep @@ -203,6 +265,28 @@ _GATE_PREFIXES = ( ) _ACTIONABLE_GRADES = ("A+", "A", "B", "C") +# ── Scan outcomes (forward returns for indicator-weight research) ───── +# Horizons are CALENDAR hours from the signal bar's close; a bar label is +# its OPEN time, so the bar closing at label+interval is the one counted. +SCAN_OUTCOME_HORIZONS_H = (1, 4, 24) +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 +# Missing bars tolerated inside a window ON TOP of the scheduled closures +# advisory_outcome._has_continuity already allows. +SCAN_LABEL_MAX_GAP_BARS = 3 +# "Burst" bar: range >= this multiple of the scan's ATR (violent_watch). +SCAN_BURST_ATR_MULTIPLE = 3.0 +# Soft wall-clock cap for ONE opportunistic (in-scan) labelling pass. +SCAN_LABEL_TIME_BUDGET = 2.0 +# Minimum spacing between opportunistic passes in ONE process. Rows that +# can never complete (a real hole in the stored stream) are re-examined +# on every pass, and that read costs ~0.5 s on the live database — too +# much to pay on the alert path for every logged scan, and pointless more +# often than a candle close. Explicit calls (batch script, tests) ignore it. +SCAN_LABEL_MIN_INTERVAL = 60.0 + _ADDITIVE_OUTLOOK_COLS = ( ("news_json", "TEXT"), ("indicators_json", "TEXT"), @@ -220,6 +304,8 @@ _SIGNAL_INSERT_COLS = ( "setup_key", "headlines_hash", "events_hash", "gate_reason", "actionable", "cleaned_warnings_json", "bar_range_median", "scan_kind", "config_fingerprint", + "bar_ts", "engine_fingerprint", "code_version", "penalty_tags_json", + "penalty_total", "momentum_fading", "exhaustion_count", ) _INSERT = ( "INSERT INTO signals (" + ", ".join(_SIGNAL_INSERT_COLS) + ") VALUES (" @@ -263,6 +349,30 @@ def _blob_hash(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() +_code_version_cache: Optional[str] = None + + +def _code_version() -> str: + """Short git SHA of the tree this process runs from (computed once). + + Record-keeping only — "" whenever git is unavailable, slow (2 s cap) + or the tree is not a checkout. There is no VERSION/__version__ + convention in this repo, so the commit is the version.""" + global _code_version_cache + if _code_version_cache is None: + _code_version_cache = "" + try: + done = subprocess.run( + ["git", "rev-parse", "--short", "HEAD"], + cwd=os.path.dirname(os.path.abspath(__file__)), + capture_output=True, text=True, timeout=2.0) + if done.returncode == 0: + _code_version_cache = done.stdout.strip() + except Exception: + logger.debug("signal_log: code version unavailable", exc_info=True) + return _code_version_cache + + def setup_key(setup: Optional[Dict[str, Any]], timeframe: Any) -> Optional[str]: """Identity of a logged trade setup for event dedup. @@ -310,6 +420,118 @@ def gate_reason( return None +_SCAN_OUTCOME_COLS = ( + "timeframe", "bar_ts", "atr", "close0", + "fwd_1h", "fwd_4h", "fwd_24h", + "fwd_1h_atr", "fwd_4h_atr", "fwd_24h_atr", + "mfe_24h_atr", "mae_24h_atr", "burst_1h", "bars_used", "complete", +) +_SCAN_OUTCOME_INSERT = ( + "INSERT OR REPLACE INTO scan_outcomes (signal_id, " + + ", ".join(_SCAN_OUTCOME_COLS) + ", labeled_at) VALUES (" + + ", ".join("?" for _ in range(len(_SCAN_OUTCOME_COLS) + 2)) + ")" +) + + +def _utc(stamp: datetime) -> datetime: + """Tz-aware UTC (stored bar labels are UTC; a naive one is assumed).""" + if stamp.tzinfo is None: + return stamp.replace(tzinfo=timezone.utc) + return stamp.astimezone(timezone.utc) + + +def _window_continuous(bars: List[Dict[str, Any]], anchor: datetime, + timeframe: Any, + max_gap_bars: int = SCAN_LABEL_MAX_GAP_BARS) -> bool: + """True when ``bars`` walk forward from ``anchor`` without a real hole. + + The scheduled-closure judgement is advisory_outcome._has_continuity's + (NY-anchored weekend + the daily maintenance hour) applied one step at + a time; on top of it a gap of at most ``max_gap_bars`` missing bars is + tolerated, because a scan-outcome window is measured from closes, not + replayed bar by bar like an advisory trade.""" + seconds = _TF_SECONDS.get(normalize_timeframe(timeframe)) + if not seconds: + return False + previous = anchor + for bar in bars: + stamp = bar["_timestamp"] + if stamp <= previous: + continue + if not _has_continuity([bar], previous, timeframe): + missing = int((stamp - previous).total_seconds() // seconds) - 1 + if missing > max_gap_bars: + return False + previous = stamp + return True + + +def scan_outcome(bar_ts: Any, timeframe: Any, bars: Iterable[Dict[str, Any]], + atr: Optional[float], price: Optional[float] + ) -> Optional[Dict[str, Any]]: + """Forward-return label for ONE scan, from stored closed candles. + + ``bars`` is the union of stored bars for the row's primary timeframe. + ``close0`` is the close of the bar labelled ``bar_ts`` (the row's own + last closed bar), falling back to the logged scan price. A horizon is + reported only once it has ELAPSED inside the stream (an unfinished + window reports NULLs rather than a truncated move), and ``complete`` + additionally requires an unbroken window. Pure: no DB, no clock.""" + seconds = _TF_SECONDS.get(normalize_timeframe(timeframe)) + anchor = None + try: + anchor = _utc(datetime.fromisoformat(str(bar_ts).replace(" ", "T"))) + except (TypeError, ValueError): + anchor = None + if not seconds or anchor is None: + return None + + valid = _valid_bars(bars) + for bar in valid: + bar["_timestamp"] = _utc(bar["_timestamp"]) + close0 = next((b["close"] for b in valid if b["_timestamp"] == anchor), + None) + if close0 is None: + close0 = _finite(price) + if close0 is None: + return None + atr = atr if (atr is not None and atr > 0) else None + + forward = [b for b in valid if b["_timestamp"] > anchor] + reach = forward[-1]["_timestamp"] if forward else anchor + out: Dict[str, Any] = { + "timeframe": str(getattr(timeframe, "value", timeframe)), + "bar_ts": str(bar_ts), "atr": atr, "close0": close0, + "mfe_24h_atr": None, "mae_24h_atr": None, "burst_1h": None, + } + window: List[Dict[str, Any]] = [] + for hours in SCAN_OUTCOME_HORIZONS_H: + edge = anchor + timedelta(hours=hours) + upto = [b for b in forward if b["_timestamp"] <= edge] + elapsed = reach >= edge and bool(upto) + move = (upto[-1]["close"] - close0) if elapsed else None + out[f"fwd_{hours}h"] = move + out[f"fwd_{hours}h_atr"] = (move / atr) if (move is not None and atr) \ + else None + if hours == 1 and elapsed and atr: + out["burst_1h"] = int(any( + (b["high"] - b["low"]) >= SCAN_BURST_ATR_MULTIPLE * atr + for b in upto)) + if hours == SCAN_OUTCOME_WINDOW_H: + window = upto + if elapsed and atr: + out["mfe_24h_atr"] = (max(b["high"] for b in upto) + - close0) / atr + out["mae_24h_atr"] = (close0 + - min(b["low"] for b in upto)) / atr + complete = (reach >= anchor + timedelta(hours=SCAN_OUTCOME_WINDOW_H) + and bool(window) + and _window_continuous(window, anchor, timeframe)) + out["bars_used"] = len(window) + out["complete"] = int(complete) + return out + + # ── Serializers (duck-typed; deliberately NOT imported from ── # ── dashboard_publisher — the dashboard's hard independence rule) ── @@ -396,7 +618,12 @@ def _analysis_to_dict(analysis: Any) -> Optional[Dict[str, Any]]: "signal": getattr(signal, "name", str(signal)), "value": _finite(getattr(indicator, "value", None)), "confidence": _finite(getattr(indicator, "confidence", None)), + # `weight` is the BASE weight; `eff_weight` is that weight + # times the regime multiplier the composite actually used and + # `wscore` its contribution (null for skipped indicators). "weight": _finite(getattr(indicator, "weight", None)), + "eff_weight": _finite(getattr(indicator, "effective_weight", None)), + "wscore": _finite(getattr(indicator, "weighted_score", None)), "description": getattr(indicator, "description", "") or "", } composite = getattr(analysis, "composite_signal", None) @@ -510,6 +737,11 @@ def _report_row( cleaned = list(getattr(report, "cleaned_warnings", None) or []) setup_dict = json.loads(setup_json) if setup_json else None bar_range = _finite(getattr(report, "spread_estimate", None)) + # The primary bar LABEL this scan closed on, stored verbatim (the + # same string indicators_json carries, so readers can join on it). + primary = getattr(report, "primary_analysis", None) + bar_ts = str(getattr(primary, "timestamp", "") or "") or None + penalty_tags = getattr(report, "penalty_tags", None) or {} return { "logged_at": datetime.now(timezone.utc).isoformat(), "source": source, @@ -554,6 +786,17 @@ def _report_row( "bar_range_median": bar_range, "scan_kind": str(extra.get("scan_kind") or "scan"), "config_fingerprint": extra.get("config_fingerprint"), + # Weight-research record keeping (nothing live reads these back). + "bar_ts": bar_ts, + "engine_fingerprint": extra.get("engine_fingerprint"), + "code_version": _code_version() or None, + "penalty_tags_json": (_dumps({str(k): _finite(v) + for k, v in penalty_tags.items()}) + if penalty_tags else None), + "penalty_total": _finite(getattr(report, "penalty_total", None)), + "momentum_fading": int(bool(getattr(report, "momentum_fading", False))), + "exhaustion_count": int( + _finite(getattr(report, "exhaustion_count", 0)) or 0), } @@ -573,6 +816,7 @@ class SignalLogger: # switch plus the additive migration need brief exclusive locks. self._conn.execute("PRAGMA busy_timeout=5000") self._lock = threading.Lock() + self._next_scan_label = 0.0 # opportunistic labeller throttle with self._lock: self._with_lock_retry( lambda: self._conn.execute("PRAGMA journal_mode=WAL")) @@ -642,6 +886,11 @@ class SignalLogger: self._conn.execute( "CREATE INDEX IF NOT EXISTS idx_signals_event" " ON signals (advisory_event_id)") + # bar_ts is an additive column, so its index cannot live in + # _SCHEMA (which runs before the ALTERs on a legacy database). + self._conn.execute( + "CREATE INDEX IF NOT EXISTS idx_signals_tf_bar" + " ON signals (timeframe, bar_ts)") self._conn.commit() def _migrate_outlook_columns(self) -> None: @@ -703,6 +952,17 @@ class SignalLogger: except Exception: logger.warning("signal_log: advisory outcome resolver failed", exc_info=True) + # Opportunistic, time-boxed and throttled: the same "advance what + # the stored bars now allow" pass, for scan outcomes. Keeps the + # research table current without a daemon or a cron entry. + try: + if time.monotonic() >= self._next_scan_label: + self._next_scan_label = (time.monotonic() + + SCAN_LABEL_MIN_INTERVAL) + self.label_scan_outcomes() + except Exception: + logger.warning("signal_log: scan-outcome labeller failed", + exc_info=True) return rowid # ── news interning ──────────────────────────────────────────── @@ -822,6 +1082,26 @@ class SignalLogger: advisory_status, advisory_json, rowid)) self._conn.commit() + def update_watch(self, rowid: int, level: str, kinds: Any, + line: str) -> None: + """Fill the violent-move watch fields on an already-logged row. + + Post-hoc writer mirroring update_panel: violent_watch.attach_watch + runs on the surfaces AFTER analyze_pair inserted the row. Level and + kinds are written even when the display line is switched off — the + evaluation ("did a burst follow a 'hot' scan?") needs the + assessment, not the text.""" + if isinstance(kinds, (list, tuple, set)): + kinds_text = ",".join(str(k) for k in sorted(kinds)) + else: + kinds_text = str(kinds or "") + with self._lock: + self._conn.execute( + "UPDATE signals SET watch_level=?, watch_kinds=?," + " watch_line=? WHERE id=?", + (str(level or ""), kinds_text, str(line or ""), int(rowid))) + self._conn.commit() + def resolve_pending( self, symbol: str, candles_json: Optional[str], limit: int = 100 ) -> int: @@ -927,6 +1207,90 @@ class SignalLogger: continue return [bars[k] for k in sorted(bars)] + def label_scan_outcomes(self, budget: int = 150, now: Any = None, + time_budget: Optional[float] = None) -> int: + """Label forward returns for scans whose 24 h window has closed. + + 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. + + 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) + deadline = time.monotonic() + float( + SCAN_LABEL_TIME_BUDGET if time_budget is None else time_budget) + written = 0 + with self._lock: + 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" + " 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 bar IS NOT NULL" + " ORDER BY COALESCE(o.labeled_at, '') ASC, s.id ASC" + " LIMIT ?", (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: + try: + stamp = _utc(datetime.fromisoformat( + str(bar).replace(" ", "T"))) + except (TypeError, ValueError): + continue + if stamp > cutoff: + continue # window has not elapsed yet + key = (str(symbol), str(tf)) + if 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]]] = {} + for rowid, symbol, tf, price, bar, atr, key in due: + if time.monotonic() > deadline: + break + try: + if key not in streams: + streams[key] = self._stream_since( + symbol, tf, oldest[key][1]) + label = scan_outcome(bar, tf, streams[key], + _finite(atr), price) + if label is None: + continue + values = tuple(label[col] for col in _SCAN_OUTCOME_COLS) + prior = self._conn.execute( + "SELECT " + ", ".join(_SCAN_OUTCOME_COLS) + + " FROM scan_outcomes WHERE signal_id=?", + (rowid,)).fetchone() + if prior is not None and tuple(prior) == values: + continue + self._conn.execute( + _SCAN_OUTCOME_INSERT, + (int(rowid),) + values + + (datetime.now(timezone.utc).isoformat(),)) + self._conn.commit() + written += 1 + except Exception: + logger.warning( + "signal_log: scan-outcome labelling failed for row %s", + rowid, exc_info=True) + return written + def advance_with_bars(self, rowid: int, bars: Iterable[Dict[str, Any]], evidence: str = "external_bars") -> str: """Advance ONE pending outcome from caller-supplied bars (backfill @@ -1222,6 +1586,31 @@ def log_panel_update(report: Any) -> None: exc_info=True) +def log_watch_update(report: Any, watch: Any) -> None: + """Write the post-hoc violent-move watch fields onto this scan's row. + + Mirrors log_panel_update: matches on the private rowid hand-off from + log_signal, and a report that was never logged (backtest replay, + logging disabled, another process) is a silent no-op. NEVER raises.""" + try: + rowid = getattr(report, "_signal_log_rowid", None) + if rowid is None or not app_config.signal_log_enabled(): + return + path = app_config.signal_log_path() + with _loggers_lock: + lg = _loggers.get(path) + if lg is None: + return + lg.update_watch( + int(rowid), + str(getattr(watch, "level", "") or ""), + getattr(watch, "kinds", None) or [], + getattr(report, "watch_line", "") or "") + except Exception: + logger.warning("signal_log: failed to persist violent-move watch", + exc_info=True) + + def log_outlook(outlook: Any, surface: str = "live", payload: Any = None) -> None: """Persist one advisory outlook when signal logging is enabled. diff --git a/test_all.py b/test_all.py index 1a93aac..43fc3b4 100644 --- a/test_all.py +++ b/test_all.py @@ -10082,6 +10082,415 @@ def _(): assert rep.watch_line == "", "analyze_from_dataframes must not stamp it" +# ══════════════════════════════════════════════════════════════════════ +# MODULE 31: signal-log data points (indicator-weight research) +# ══════════════════════════════════════════════════════════════════════ + +print("\n═══ Module 31: signal-log data points ═══") + +from engine import _DEFAULT_REGIME_WEIGHTS, regime_weight_table + + +def _m31_report(ts, tf=Timeframe.M5, price=2450.0, quality="B", + signal=SignalStrength.BUY, atr=5.0, setup=False, + analysis=None): + """Logged scan with the record-keeping fields populated.""" + r = _make_log_report() + r.timestamp = ts + r.final_signal = signal + r.signal_quality = quality + r.current_price = price + r.trade_setup = r.trade_setup if setup else None + r.exhaustion_count = 2 + r.penalty_tags = {"rsi_overbought": 8.0, "extended_from_ema": 4.0} + r.penalty_total = 6.5 + r.momentum_fading = True + if analysis is not None: + r.primary_analysis = analysis + r.primary_analysis.timeframe = tf.value + r.primary_analysis.timestamp = ts + r.primary_analysis.key_levels = {"support": 2440.0, "atr": atr} + return r + + +def _m31_bars(start, n, step_min=5, up=0.1, base=2450.0, burst_at=None): + """Deterministic ramp: bar i opens base+i*up and closes one `up` + higher, with 0.3 wicks — every forward return is exact arithmetic.""" + out = [] + t = pd.Timestamp(start) + for i in range(n): + o = base + i * up + c = o + up + high = max(o, c) + (20.0 if i == burst_at else 0.3) + out.append({ + "timestamp": (t + pd.Timedelta(minutes=step_min * i)).isoformat(), + "open": o, "high": high, "low": min(o, c) - 0.3, + "close": c, "volume": 1.0}) + return out + + +def _m31_log_stream(lg, bars, chunk=14): + """Log the bar stream the way live scans store it (14 bars per row).""" + for i in range(0, len(bars), chunk): + window = bars[i:i + chunk] + lg.log(_m31_report(str(pd.Timestamp(window[-1]["timestamp"])), + quality="D", signal=SignalStrength.NEUTRAL), + source="dashboard", timeframe="5min", + dataframes={Timeframe.M5: _m27_df(window)}) + + +@test("engine: effective_weight = weight x regime multiplier, composite unchanged") +def _(): + config = TechnicalConfig() + result = TechnicalEngine(config).analyze( + make_xau_ohlcv(300, seed=11), "XAU/USD", "1h") + table = regime_weight_table(config).get(result.market_regime, {}) + total = weights = 0.0 + scored = 0 + for name, ind in result.indicators.items(): + if name.startswith("_") or ind.weight == 0: + assert ind.effective_weight is None, name + assert ind.weighted_score is None, name + continue + scored += 1 + assert abs(ind.effective_weight + - ind.weight * table.get(name, 1.0)) < 1e-12, name + assert abs(ind.weighted_score - ind.signal.value + * ind.effective_weight * ind.confidence / 100.0) < 1e-12 + total += ind.weighted_score + weights += ind.effective_weight + assert scored >= 5, scored + recomputed = max(-100.0, min(100.0, (total / weights) * 50)) + assert abs(recomputed - result.composite_score) < 1e-9, ( + recomputed, result.composite_score) + + +@test("engine: regime_weight_table honours regime_weights_override") +def _(): + df = make_xau_ohlcv(300, seed=12) + base = TechnicalEngine(TechnicalConfig()).analyze(df, "XAU/USD", "1h") + assert regime_weight_table(TechnicalConfig()) == _DEFAULT_REGIME_WEIGHTS + # The table is a COPY: mutating it cannot reach the shipped weights. + regime_weight_table(TechnicalConfig())["ranging"]["rsi"] = 99.0 + again = TechnicalEngine(TechnicalConfig()).analyze(df, "XAU/USD", "1h") + assert again.composite_score == base.composite_score + + regime = base.market_regime + override = {r: dict(w) for r, w in _DEFAULT_REGIME_WEIGHTS.items()} + override[regime]["rsi"] = override[regime].get("rsi", 1.0) * 4 + 1 + config = TechnicalConfig(regime_weights_override=override) + assert regime_weight_table(config) == override + tuned = TechnicalEngine(config).analyze(df, "XAU/USD", "1h") + assert tuned.market_regime == regime + assert tuned.composite_score != base.composite_score, "override ignored" + total = weights = 0.0 + for name, ind in tuned.indicators.items(): + if ind.effective_weight is None: + continue + assert abs(ind.effective_weight + - ind.weight * override[regime].get(name, 1.0)) < 1e-12 + total += ind.weighted_score + weights += ind.effective_weight + assert abs(max(-100.0, min(100.0, (total / weights) * 50)) + - tuned.composite_score) < 1e-9 + + +@test("aggregator: engine_config_fingerprint tracks scoring config only") +def _(): + from pair_signals.xauusd import make_config + m5 = SignalAggregator(make_config(Timeframe.M5)) + m15 = SignalAggregator(make_config(Timeframe.M15)) + fingerprint = m5.engine_config_fingerprint() + assert len(fingerprint) == 12 and fingerprint.isalnum() + assert fingerprint == m5.engine_config_fingerprint(), "not stable" + assert fingerprint == SignalAggregator( + make_config(Timeframe.M5)).engine_config_fingerprint() + assert fingerprint != m15.engine_config_fingerprint(), "timeframe blind" + + flipped = make_config(Timeframe.M5) + flipped.technical.ema_partial_neutral = ( + not flipped.technical.ema_partial_neutral) + flipped_agg = SignalAggregator(flipped) + assert flipped_agg.engine_config_fingerprint() != fingerprint + assert flipped_agg.gate_config_fingerprint() == m5.gate_config_fingerprint() + + weighted = make_config(Timeframe.M5) + weighted.technical.regime_weights_override = {"ranging": {"rsi": 9.0}} + weighted_agg = SignalAggregator(weighted) + assert weighted_agg.engine_config_fingerprint() != fingerprint + assert weighted_agg.gate_config_fingerprint() == m5.gate_config_fingerprint() + + +@test("signal_log: migration adds the research columns, scan_outcomes and indexes") +def _(): + import sqlite3 + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + try: + con = sqlite3.connect(path) + con.execute( + "CREATE TABLE signals (id INTEGER PRIMARY KEY, logged_at TEXT," + " symbol TEXT, timeframe TEXT)") + con.commit() + con.close() + lg = slog.SignalLogger(path) + columns = {r[1] for r in lg._conn.execute("PRAGMA table_info(signals)")} + assert {"bar_ts", "engine_fingerprint", "code_version", + "penalty_tags_json", "penalty_total", "momentum_fading", + "exhaustion_count", "watch_level", "watch_kinds", + "watch_line"} <= columns, sorted(columns) + outcome_cols = [r[1] for r in + lg._conn.execute("PRAGMA table_info(scan_outcomes)")] + assert outcome_cols[0] == "signal_id" + assert {"signal_id", "timeframe", "bar_ts", "atr", "close0", + "fwd_1h", "fwd_4h", "fwd_24h", "fwd_1h_atr", "fwd_4h_atr", + "fwd_24h_atr", "mfe_24h_atr", "mae_24h_atr", "burst_1h", + "bars_used", "complete", "labeled_at"} == set(outcome_cols), \ + outcome_cols + names = {r[0] for r in lg._conn.execute( + "SELECT name FROM sqlite_master WHERE type='index'")} + assert "idx_signals_tf_bar" in names + assert "idx_scan_outcomes_tf_bar" in names + slog.SignalLogger(path).close() # idempotent re-migration + lg.close() + finally: + slog.reset_loggers_for_tests() + os.remove(path) + + +@test("signal_log: a row records bar_ts, fingerprints, penalties and eff weights") +def _(): + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + try: + analysis = TechnicalEngine(TechnicalConfig()).analyze( + make_xau_ohlcv(300, seed=13), "XAU/USD", Timeframe.M5.value) + bar = "2026-08-19 12:00:00+00:00" + report = _m31_report(bar, analysis=analysis) + lg = slog.SignalLogger(path) + rowid = lg.log(report, source="telegram", timeframe="5min", + extra={"scan_kind": "scan", + "config_fingerprint": "gate12345678", + "engine_fingerprint": "engn12345678"}) + row = lg._conn.execute( + "SELECT bar_ts, engine_fingerprint, config_fingerprint," + " code_version, penalty_tags_json, penalty_total," + " momentum_fading, exhaustion_count, indicators_json" + " FROM signals WHERE id=?", (rowid,)).fetchone() + assert row[0] == bar, row[0] + assert row[1] == "engn12345678" and row[2] == "gate12345678" + assert row[3] == slog._code_version() + assert json.loads(row[4]) == {"rsi_overbought": 8.0, + "extended_from_ema": 4.0} + assert row[5] == 6.5 and row[6] == 1 and row[7] == 2 + stored = json.loads(row[8])["5min"]["indicators"] + scored = [n for n, e in stored.items() if e["eff_weight"] is not None] + assert scored, stored + for name in scored: + entry = stored[name] + live = analysis.indicators[name] + assert entry["eff_weight"] == live.effective_weight + assert entry["wscore"] == live.weighted_score + assert entry["weight"] == live.weight # base weight untouched + assert stored["atr"]["eff_weight"] is None, "context indicator scored" + lg.close() + # The aggregator is what hands the engine fingerprint over. + captured = {} + with patch("signal_log.log_signal", + lambda *a, **k: captured.update(k.get("extra") or {})): + agg = SignalAggregator(PlatformConfig()) + agg._log_signal(_make_log_report(), "test", blocked=True) + assert captured["engine_fingerprint"] == agg.engine_config_fingerprint() + assert captured["config_fingerprint"] == agg.gate_config_fingerprint() + finally: + slog.reset_loggers_for_tests() + os.remove(path) + + +@test("signal_log: attach_watch fills watch_level/kinds/line post-hoc") +def _(): + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + os.remove(path) + try: + os.environ["SIGNAL_LOG_PATH"] = path + os.environ["SIGNAL_LOG_ENABLED"] = "1" + analysis = _m30_analysis(indicators={ + "bollinger": _m30_ind("BB", -0.2), "rsi": _m30_ind("RSI", 22.0)}) + report = _m31_report("2026-08-19 12:00:00+00:00", analysis=analysis) + slog.log_signal(report, source="telegram", timeframe="5min") + rowid = report._signal_log_rowid + with patch("news_engine.NewsEngine.get_week_events", return_value=[]): + # alert_line off: the assessment is still recorded. + watch = attach_watch(report, Timeframe.M5, + {"enabled": True, "alert_line": False}, + _M30_QUIET) + assert watch.level == "watch" and report.watch_line == "" + lg = slog._loggers[path] + assert lg._conn.execute( + "SELECT watch_level, watch_kinds, watch_line FROM signals" + " WHERE id=?", (rowid,)).fetchone() == ( + "watch", ",".join(sorted(watch.kinds)), "") + # With the line on, the displayed text is stored too. + watch = attach_watch(report, Timeframe.M5, {"enabled": True}, + _M30_QUIET) + assert lg._conn.execute( + "SELECT watch_level, watch_line FROM signals WHERE id=?", + (rowid,)).fetchone() == (watch.level, report.watch_line) + # A report that was never logged is a silent no-op. + orphan = _m31_report("2026-08-19 12:05:00+00:00", + analysis=analysis) + attach_watch(orphan, Timeframe.M5, {"enabled": True}, _M30_QUIET) + assert lg._conn.execute( + "SELECT COUNT(*) FROM signals").fetchone()[0] == 1 + 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) + + +@test("signal_log: label_scan_outcomes computes exact forward moves") +def _(): + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + try: + bars = _m31_bars("2026-08-19T12:00:00+00:00", 320, up=0.1, + burst_at=6) # bar 6 spans 20.3 = 4x the ATR + anchor = str(pd.Timestamp(bars[0]["timestamp"])) + lg = slog.SignalLogger(path) + rowid = lg.log(_m31_report(anchor), source="telegram", + timeframe="5min", + dataframes={Timeframe.M5: _m27_df(bars[0:1])}) + # log() labels opportunistically: with no forward bars stored yet + # the row exists but nothing is known. + assert lg._conn.execute( + "SELECT complete, fwd_1h FROM scan_outcomes WHERE signal_id=?", + (rowid,)).fetchone() == (0, None) + # Only 6 h of stream: the 24 h window is not judgeable yet. + _m31_log_stream(lg, bars[:73]) + lg.label_scan_outcomes(budget=500, time_budget=30.0) + cols = ("fwd_1h", "fwd_4h", "fwd_24h", "complete", "bars_used") + partial = lg._conn.execute( + "SELECT " + ", ".join(cols) + " FROM scan_outcomes" + " WHERE signal_id=?", (rowid,)).fetchone() + assert partial[3] == 0, partial + assert abs(partial[0] - 1.2) < 1e-9 and abs(partial[1] - 4.8) < 1e-9 + assert partial[2] is None, "unfinished horizon must stay NULL" + + _m31_log_stream(lg, bars[73:]) # the rest arrives later + lg.label_scan_outcomes(budget=500, time_budget=30.0) + row = lg._conn.execute( + "SELECT timeframe, bar_ts, atr, close0, fwd_1h, fwd_4h, fwd_24h," + " fwd_1h_atr, fwd_4h_atr, fwd_24h_atr, mfe_24h_atr, mae_24h_atr," + " burst_1h, bars_used, complete FROM scan_outcomes" + " WHERE signal_id=?", (rowid,)).fetchone() + assert row[0] == "5min" and row[1] == anchor + assert row[2] == 5.0 and abs(row[3] - 2450.1) < 1e-9 + for value, expected in zip(row[4:10], + (1.2, 4.8, 28.8, 0.24, 0.96, 5.76)): + assert abs(value - expected) < 1e-9, (row, expected) + assert abs(row[10] - 29.1 / 5.0) < 1e-9 # MFE = last high + assert abs(row[11] - 0.3 / 5.0) < 1e-9 # MAE = first wick + assert row[12] == 1, "the 4x ATR bar in the first hour is a burst" + assert row[13] == 288 and row[14] == 1 # 24 h of 5-min bars + + # Idempotent: a settled database writes nothing. + assert lg.label_scan_outcomes(budget=500, time_budget=30.0) == 0 + # Budget caps the rows examined per pass. + lg._conn.execute("DELETE FROM scan_outcomes") + lg._conn.commit() + assert lg.label_scan_outcomes(budget=1, time_budget=30.0) == 1 + assert lg._conn.execute( + "SELECT COUNT(*) FROM scan_outcomes").fetchone()[0] == 1 + lg.close() + finally: + slog.reset_loggers_for_tests() + os.remove(path) + + +@test("signal_log: a failing scan-outcome labeller never breaks the insert") +def _(): + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + try: + lg = slog.SignalLogger(path) + logging.disable(logging.CRITICAL) + try: + with patch.object(slog.SignalLogger, "label_scan_outcomes", + side_effect=RuntimeError("label boom")): + rowid = lg.log(_m31_report("2026-08-19 12:00:00+00:00"), + source="telegram", timeframe="5min") + finally: + logging.disable(logging.NOTSET) + assert lg._conn.execute( + "SELECT id FROM signals").fetchone()[0] == rowid + lg.close() + finally: + slog.reset_loggers_for_tests() + os.remove(path) + + +@test("backtest parity: the replay path knows nothing about the new writers") +def _(): + root = os.path.dirname(os.path.abspath(__file__)) + with open(os.path.join(root, "backtest", "gpu_backtest.py"), + encoding="utf-8") as fh: + src = fh.read() + for banned in ("scan_outcomes", "label_scan_outcomes", "update_watch", + "log_watch_update", "import signal_log"): + assert banned not in src, f"gpu_backtest must not reference {banned}" + # The stamped record-keeping fields are engine-side, so the replay + # engine gets them for free without any signal change. + from backtest.gpu_backtest import GPUTechnicalEngine + assert "_compute_composite" not in GPUTechnicalEngine.__dict__, \ + "the backtester must keep inheriting the composite maths" + + +@test("analyze_signal_log: --scan-outcomes reports on the stored labels") +def _(): + import contextlib + import io as _io + import analyze_signal_log as asl + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + try: + bars = _m31_bars("2026-08-19T12:00:00+00:00", 300, up=0.1) + lg = slog.SignalLogger(path) + anchor = str(pd.Timestamp(bars[0]["timestamp"])) + lg.log(_m31_report(anchor, quality="A"), source="telegram", + timeframe="5min", + dataframes={Timeframe.M5: _m27_df(bars[0:1])}) + lg.log(_m31_report(anchor, quality="A"), source="dashboard", + timeframe="5min") # same bar, second surface + _m31_log_stream(lg, bars) + lg.label_scan_outcomes(budget=500, time_budget=30.0) + lg.update_watch(1, "hot", ["clock", "extreme"], "line") + lg.close() + rows = asl.scan_outcome_rows(path) + assert rows, "no labelled rows" + assert len({(r["timeframe"], r["id"]) for r in rows}) == len(rows) + assert min(r["id"] for r in rows) == 1, "dedup keeps the lowest id" + buf = _io.StringIO() + with contextlib.redirect_stdout(buf): + asl.print_scan_outcomes(rows) + asl.print_scan_outcomes([]) + out = buf.getvalue() + assert "SCAN OUTCOMES" in out and "by timeframe x signal" in out + assert "5min BUY" in out, out + assert "hot" in out and "No scan_outcomes rows yet" in out + finally: + slog.reset_loggers_for_tests() + os.remove(path) + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════ diff --git a/violent_watch.py b/violent_watch.py index 73d8021..c167879 100644 --- a/violent_watch.py +++ b/violent_watch.py @@ -350,6 +350,21 @@ def assess(report: Any, readings=readings, likely_direction=direction) +def _log_watch(report: Any, watch: "ViolentWatch") -> None: + """Persist the assessment onto the scan's signals.db row (fail-safe). + + The row is inserted inside analyze_pair — BEFORE a surface attaches + the watch — so these columns arrive by post-hoc update, exactly like + the panel verdict. Level and kinds are written even when the display + line is off: evaluating the watch needs the assessment, not the text. + """ + try: + from signal_log import log_watch_update + log_watch_update(report, watch) + except Exception: + logger.debug("violent watch: logging failed", exc_info=True) + + def attach_watch(report: Any, primary_tf: Optional[Timeframe], settings: Dict[str, Any], @@ -377,6 +392,7 @@ def attach_watch(report: Any, watch = assess(report, moment, events, primary_tf) if settings.get("alert_line", True): report.watch_line = watch.line() + _log_watch(report, watch) return watch except Exception: logger.exception("violent watch: assessment failed")