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). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RWV2yLt3YXHHGVYHSQU2EE
This commit is contained in:
+390
-1
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user