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

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

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
+        # Override weights if provided, else gold-optimized defaults
-        if self.config.regime_weights_override:
+        # (regime_weight_table is the single source of truth).
-            regime_weights = self.config.regime_weights_override
+        regime_weights = regime_weight_table(self.config)
-        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,
-                },
-            }
 
         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

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())

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"}
+        for key, fingerprint in (
+                ("config_fingerprint", self.gate_config_fingerprint),
+                ("engine_fingerprint", self.engine_config_fingerprint)):
             try:
-            extra["config_fingerprint"] = self.gate_config_fingerprint()
+                extra[key] = fingerprint()
             except Exception:
-            logger.debug("signal log: config fingerprint unavailable",
+                logger.debug("signal log: %s unavailable", key, exc_info=True)
-                         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:

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.

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
 # ══════════════════════════════════════════════════════════════════════

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")