Harwinder Sekhon
2e332545a9
Delivery-side fixes from the 2026-09-08 three-day signal-log + telegram-log review. Nothing in the signal core changes: the 3-year H1 and M15 books at $0.65 were regenerated before/after and are identical on every trade column (H1 253 / +70.02R / PF 1.98, M15 1092 / +151.90R / PF 1.38); only the exported `warnings` text differs, on 50/253 H1 and 358/1092 M15 trades. - signal_aggregator: step 7 moves into `_attach_core_setup`; a final pass after the PB/CP archetypes (`_drop_stale_spread_gate_vetoes`) moves a core generator's "SPREAD GATE: ... costs would eat the edge" veto to `cleaned_warnings` whenever a later generator shipped a setup. The 2026-09-08 07:45 UTC A-grade alert carried that line beside a $26.49 TP1; a fifth of H1 and a third of M15 A-grade alerts had been doing the same. The prefix is now the `_SPREAD_GATE_PREFIX` constant shared by both emitters and both cleanups. - telegram_bot: alerts require a trade setup (`_is_alert_candidate`, shared by the scan and breach paths; no-setup candidates log `alert skipped: ... (gate reason)`). The 07:50 UTC B-grade SELL with no setup had gone out under the same warning. - telegram_bot: every delivered alert / breach alert / violent-move heads-up logs an `... sent:` INFO line with the signal-log row id (only failures were logged before, so the review had to reconstruct sends from panel rows). Send fan-out is one `_broadcast` helper for both paths. - telegram_bot: `telegram.log` rotates daily x 30 instead of 3-hourly x 240 (old `_HH` files on the live box need one manual delete). - signal_log / CLAUDE.md: `cleaned_warnings_json` is no longer a CP-only marker; docs say so. - test_all: 468 tests (+2): stale-veto cleanup incl. the PB-shaped case and the both-vetoed case, and the alert-candidate rule with its skip line; the rotation test asserts MIDNIGHT / interval 86400 / 30 backups. Claude-Session: https://claude.ai/code/session_018xY4NPPrx6NJoG6QWHzwpd
1619 lines
69 KiB
Python
1619 lines
69 KiB
Python
"""
|
|
signal_log.py — SQLite signal-history logger
|
|
============================================
|
|
|
|
Persists every live scan's SignalReport (signal, grade, price at scan
|
|
time, news at scan time) to a SQLite database so the full decision
|
|
history can be queried later.
|
|
|
|
Hooked ONLY into SignalAggregator.analyze_pair (the live entry point);
|
|
the backtester calls analyze_from_dataframes directly and therefore
|
|
never writes here (live == backtest parity — CLAUDE.md rule 4).
|
|
|
|
Panel fields arrive POST-HOC: the row is inserted during analyze_pair
|
|
(before consumers attach the advisory panel), so log_signal stashes the
|
|
rowid on the report and agent_panel.attach_panel calls
|
|
log_panel_update() to fill the verdict, takes, creation/cache provenance,
|
|
payload hash, and advisory-outcome seed on that same row. Later closed-candle
|
|
snapshots advance the hypothetical next-open/trailing outcome. A report
|
|
without a rowid (backtest replay, logging disabled) is a silent no-op.
|
|
|
|
Deployment settings come from the config.json "signal_log" block via
|
|
app_config (env vars SIGNAL_LOG_ENABLED / SIGNAL_LOG_PATH win):
|
|
|
|
"signal_log": {"enabled": true, "path": "logs/signals.db"}
|
|
|
|
Fail-safe contract: logging must NEVER break a scan. Every public entry
|
|
point catches all exceptions and downgrades them to a warning log line.
|
|
Report/context objects are duck-typed (no imports from signal_aggregator
|
|
or news_engine) so this module stays at the bottom of the import graph.
|
|
|
|
2026-09 observability additions (signal-log audit):
|
|
|
|
* Every row with a trade setup is an advisory-outcome candidate, not only
|
|
panel rows. Re-logs of the same setup (same ``setup_key`` — tf,
|
|
direction, entry, stop, strategy — within OUTCOME_DEDUP_GAP_SECONDS,
|
|
from ANY surface) are marked ``DUPLICATE`` and point at the canonical
|
|
row through ``advisory_event_id``; only the canonical row is seeded and
|
|
advanced, so summing ``advisory_r_multiple`` over
|
|
``advisory_outcome_status='RESOLVED'`` counts each trade once.
|
|
* Pending outcomes advance from the inserting scan's own snapshot first;
|
|
when that snapshot no longer overlaps the outcome's last processed bar
|
|
(downtime, weekend straddle) the resolver rebuilds the bar stream from
|
|
every stored snapshot of that timeframe, so a >14-bar gap in one
|
|
process's scans is not a terminal ``UNRESOLVABLE``.
|
|
* News blobs are content-addressed: ``headlines_json``/``events_json`` are
|
|
interned once into ``news_snapshots`` and rows carry ``headlines_hash``
|
|
/ ``events_hash`` (the legacy columns stay NULL on new rows; the
|
|
``signals_with_news`` view and ``read_news`` resolve both eras).
|
|
* Post-gate truth columns: ``gate_reason`` (the warning that suppressed
|
|
the setup), ``actionable`` (grade >= C AND setup present — what an
|
|
alert surface would act on; the ``action`` text is written BEFORE the
|
|
gate chain and can disagree), ``cleaned_warnings_json`` (warnings a
|
|
shipped setup superseded, removed from ``warnings_json``: the vetoes
|
|
the XAU_CP acceptance stripped and, since 2026-09-08, a core
|
|
generator's stale spread-gate veto when a later generator produced
|
|
the setup — so non-empty does NOT mean "XAU_CP row"), ``bar_range_median``
|
|
(the value historically stored as ``spread`` — it is the median
|
|
High-Low of the primary frame, NOT a broker spread), ``scan_kind``
|
|
(``scan``; ``blocked`` for weekend/day early returns, which carry no
|
|
news or candles; ``failed`` when the primary frame could not be
|
|
fetched — price 0, no signal — readers skip it like ``blocked``) 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
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import sqlite3
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, Dict, Iterable, List, Optional, Tuple
|
|
|
|
import app_config
|
|
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__)
|
|
|
|
_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS signals (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
logged_at TEXT NOT NULL,
|
|
source TEXT NOT NULL,
|
|
symbol TEXT NOT NULL,
|
|
timeframe TEXT NOT NULL,
|
|
signal TEXT NOT NULL,
|
|
score REAL,
|
|
confidence REAL,
|
|
quality TEXT,
|
|
action TEXT,
|
|
price REAL,
|
|
spread REAL,
|
|
market_regime TEXT,
|
|
volatility_regime TEXT,
|
|
mtf_json TEXT,
|
|
setup_json TEXT,
|
|
reasons_json TEXT,
|
|
warnings_json TEXT,
|
|
event_risk TEXT,
|
|
event_description TEXT,
|
|
news_sentiment REAL,
|
|
news_description TEXT,
|
|
news_summary TEXT,
|
|
events_json TEXT,
|
|
headlines_json TEXT,
|
|
indicators_json TEXT,
|
|
candles_json TEXT,
|
|
panel_verdict TEXT,
|
|
panel_confidence REAL,
|
|
panel_rationale TEXT,
|
|
panel_takes_json TEXT,
|
|
panel_created_at TEXT,
|
|
panel_cache_status TEXT,
|
|
panel_payload_hash TEXT,
|
|
advisory_outcome_status TEXT,
|
|
advisory_r_multiple REAL,
|
|
advisory_outcome_json TEXT
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_signals_sym_tf_ts
|
|
ON signals (symbol, timeframe, logged_at);
|
|
CREATE TABLE IF NOT EXISTS news_snapshots (
|
|
hash TEXT PRIMARY KEY,
|
|
kind TEXT NOT NULL,
|
|
body TEXT NOT NULL,
|
|
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,
|
|
surface TEXT NOT NULL,
|
|
bias TEXT,
|
|
setup_forming TEXT,
|
|
confidence REAL,
|
|
watch TEXT,
|
|
note TEXT,
|
|
news_json TEXT,
|
|
indicators_json TEXT,
|
|
candles_json TEXT
|
|
);
|
|
"""
|
|
|
|
# Additive migration for columns introduced after the original schema.
|
|
# Idempotent — skips columns already present.
|
|
_ADDITIVE_SIGNAL_COLS = (
|
|
("panel_verdict", "TEXT"),
|
|
("panel_confidence", "REAL"),
|
|
("panel_rationale", "TEXT"),
|
|
("panel_takes_json", "TEXT"),
|
|
("indicators_json", "TEXT"),
|
|
("candles_json", "TEXT"),
|
|
("panel_created_at", "TEXT"),
|
|
("panel_cache_status", "TEXT"),
|
|
("panel_payload_hash", "TEXT"),
|
|
("advisory_outcome_status", "TEXT"),
|
|
("advisory_r_multiple", "REAL"),
|
|
("advisory_outcome_json", "TEXT"),
|
|
# 2026-09 observability columns (see module docstring).
|
|
("setup_key", "TEXT"),
|
|
("advisory_event_id", "INTEGER"),
|
|
("headlines_hash", "TEXT"),
|
|
("events_hash", "TEXT"),
|
|
("gate_reason", "TEXT"),
|
|
("actionable", "INTEGER"),
|
|
("cleaned_warnings_json", "TEXT"),
|
|
("bar_range_median", "REAL"),
|
|
("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
|
|
# their inline news blobs, new rows resolve them through news_snapshots.
|
|
# Recreated on every logger init (SignalLogger._ensure_news_view) so it
|
|
# always exposes the current column set.
|
|
|
|
# Same-setup re-logs (cross-surface double logs, persisting-signal
|
|
# re-fires) within this gap are ONE advisory event. Mirrors
|
|
# analyze_signal_log's 60-minute clustering.
|
|
OUTCOME_DEDUP_GAP_SECONDS = 3600
|
|
|
|
# Warning prefixes that mean "a gate removed the setup", checked in gate
|
|
# order (live wrapper first, then the post-grade chain, then the setup
|
|
# generators). Kept here — signal_log must not import signal_aggregator.
|
|
_GATE_PREFIXES = (
|
|
"WEEKEND FILTER",
|
|
"DAY FILTER",
|
|
"SESSION FILTER",
|
|
"Primary timeframe stale",
|
|
"Primary timeframe data unavailable",
|
|
"REGIME FILTER",
|
|
"EXHAUSTION BLOCK",
|
|
"EXHAUSTED TREND FILTER",
|
|
"SELL FILTER",
|
|
"MTF OPPOSITION",
|
|
"BIG CANDLE FILTER",
|
|
"M5 FILTER",
|
|
"LOW ATR FILTER",
|
|
"SPREAD GATE",
|
|
)
|
|
_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"),
|
|
("candles_json", "TEXT"),
|
|
)
|
|
|
|
_SIGNAL_INSERT_COLS = (
|
|
"logged_at", "source", "symbol", "timeframe", "signal", "score",
|
|
"confidence", "quality", "action", "price", "spread", "market_regime",
|
|
"volatility_regime", "mtf_json", "setup_json", "reasons_json",
|
|
"warnings_json", "event_risk", "event_description", "news_sentiment",
|
|
"news_description", "news_summary", "events_json", "headlines_json",
|
|
"indicators_json", "candles_json", "panel_verdict", "panel_confidence",
|
|
"panel_rationale", "panel_takes_json",
|
|
"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 ("
|
|
+ ", ".join("?" for _ in _SIGNAL_INSERT_COLS) + ")"
|
|
)
|
|
|
|
_OUTLOOK_INSERT = """
|
|
INSERT INTO outlooks (
|
|
logged_at, surface, bias, setup_forming, confidence, watch, note,
|
|
news_json, indicators_json, candles_json
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
"""
|
|
|
|
|
|
def _finite(x: Any) -> Optional[float]:
|
|
"""float(x) or None for non-numeric / non-finite values (NaN/inf —
|
|
SQLite has no NaN and strict JSON rejects it)."""
|
|
try:
|
|
f = float(x)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return f if math.isfinite(f) else None
|
|
|
|
|
|
def _iso(ts: Any) -> Optional[str]:
|
|
if ts is None:
|
|
return None
|
|
try:
|
|
if ts.tzinfo is None:
|
|
ts = ts.replace(tzinfo=timezone.utc)
|
|
return ts.astimezone(timezone.utc).isoformat()
|
|
except (AttributeError, ValueError):
|
|
return str(ts)
|
|
|
|
|
|
def _dumps(obj: Any) -> str:
|
|
return json.dumps(obj, default=str, allow_nan=False)
|
|
|
|
|
|
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.
|
|
|
|
``"<tf>|<direction>|<entry>|<stop>|<strategy>"`` — the same tuple
|
|
analyze_signal_log clusters on (tf, direction, round(entry, 2),
|
|
round(SL, 2), strategy). None when the setup lacks a usable price."""
|
|
if not isinstance(setup, dict):
|
|
return None
|
|
entry = _finite(setup.get("entry"))
|
|
stop = _finite(setup.get("stop_loss"))
|
|
if entry is None or stop is None:
|
|
return None
|
|
return "|".join((
|
|
str(getattr(timeframe, "value", timeframe) or ""),
|
|
str(setup.get("direction") or ""),
|
|
f"{entry:.2f}",
|
|
f"{stop:.2f}",
|
|
str(setup.get("strategy") or ""),
|
|
))
|
|
|
|
|
|
def gate_reason(
|
|
warnings: Iterable[str],
|
|
cleaned_warnings: Iterable[str],
|
|
quality: str,
|
|
has_setup: bool,
|
|
) -> Optional[str]:
|
|
"""The warning that explains why a row carries no setup, or None.
|
|
|
|
Rows WITH a setup have no gate reason. Rows without one report the
|
|
first gate-prefixed warning (in gate order); an actionable grade with
|
|
no setup and no gate warning is the setup generator returning None
|
|
(RR floor / geometry), which the aggregator does not warn about."""
|
|
if has_setup:
|
|
return None
|
|
pool = list(warnings or []) + list(cleaned_warnings or [])
|
|
for prefix in _GATE_PREFIXES:
|
|
for warning in pool:
|
|
if str(warning).startswith(prefix):
|
|
return str(warning)
|
|
if quality in _ACTIONABLE_GRADES:
|
|
return "SETUP GENERATOR: no setup (RR floor / geometry)"
|
|
if quality:
|
|
return f"GRADE {quality}: below setup threshold"
|
|
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) ──
|
|
|
|
def _event_to_dict(ev: Any) -> Dict[str, Any]:
|
|
try:
|
|
surprise = ev.surprise
|
|
except Exception:
|
|
surprise = None
|
|
return {
|
|
"time": _iso(getattr(ev, "time", None)),
|
|
"currency": getattr(ev, "currency", None),
|
|
"impact": getattr(ev, "impact", None),
|
|
"event": getattr(ev, "event", None),
|
|
"forecast": getattr(ev, "forecast", None),
|
|
"previous": getattr(ev, "previous", None),
|
|
"actual": getattr(ev, "actual", None),
|
|
"surprise": surprise,
|
|
}
|
|
|
|
|
|
def _headline_to_dict(h: Any) -> Dict[str, Any]:
|
|
return {
|
|
"headline": getattr(h, "headline", None),
|
|
"source": getattr(h, "source", None),
|
|
"timestamp": _iso(getattr(h, "timestamp", None)),
|
|
"score": _finite(getattr(h, "sentiment_score", None)),
|
|
"score_by_currency": {
|
|
str(currency): _finite(score)
|
|
for currency, score in (
|
|
getattr(h, "sentiment_by_currency", None) or {}).items()
|
|
},
|
|
"currencies": list(getattr(h, "currencies_mentioned", None) or []),
|
|
"keywords": list(getattr(h, "keywords", None) or []),
|
|
"link": getattr(h, "link", None),
|
|
"description": getattr(h, "description", None),
|
|
"publisher": getattr(h, "publisher", None),
|
|
}
|
|
|
|
|
|
def _context_news_json(context: Any) -> "tuple[Optional[str], Optional[str]]":
|
|
"""(events_json, headlines_json) from a FundamentalContext, or
|
|
(None, None) when no context was captured."""
|
|
if context is None:
|
|
return None, None
|
|
events = {
|
|
"high_impact_warning": bool(
|
|
getattr(context, "high_impact_warning", False)),
|
|
"warning_message": getattr(context, "warning_message", "") or "",
|
|
"calendar_available": bool(
|
|
getattr(context, "calendar_available", False)),
|
|
"sentiment_by_currency": dict(
|
|
getattr(context, "news_sentiment", None) or {}),
|
|
"events_today": [
|
|
_event_to_dict(e)
|
|
for e in getattr(context, "events_today", None) or []
|
|
],
|
|
"events_upcoming_1h": [
|
|
_event_to_dict(e)
|
|
for e in getattr(context, "events_upcoming_1h", None) or []
|
|
],
|
|
}
|
|
headlines = [
|
|
_headline_to_dict(h)
|
|
for h in getattr(context, "news_headlines", None) or []
|
|
]
|
|
return _dumps(events), _dumps(headlines)
|
|
|
|
|
|
def _key_str(k: Any) -> str:
|
|
return getattr(k, "value", str(k))
|
|
|
|
|
|
def _analysis_to_dict(analysis: Any) -> Optional[Dict[str, Any]]:
|
|
"""Compact, JSON-safe snapshot of one TechnicalAnalysis."""
|
|
if analysis is None:
|
|
return None
|
|
indicators = {}
|
|
for name, indicator in (
|
|
getattr(analysis, "indicators", None) or {}).items():
|
|
if str(name).startswith("_"):
|
|
continue
|
|
signal = getattr(indicator, "signal", None)
|
|
indicators[str(name)] = {
|
|
"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)
|
|
return {
|
|
"timestamp": str(getattr(analysis, "timestamp", "")),
|
|
"composite_signal": getattr(composite, "name", str(composite)),
|
|
"composite_score": _finite(
|
|
getattr(analysis, "composite_score", None)),
|
|
"confidence": _finite(getattr(analysis, "confidence", None)),
|
|
"market_regime": getattr(analysis, "market_regime", None),
|
|
"volatility_regime": getattr(analysis, "volatility_regime", None),
|
|
"key_levels": {
|
|
str(name): _finite(value)
|
|
for name, value in (
|
|
getattr(analysis, "key_levels", None) or {}).items()
|
|
},
|
|
"pattern_signals": list(
|
|
getattr(analysis, "pattern_signals", None) or []),
|
|
"indicators": indicators,
|
|
}
|
|
|
|
|
|
def _report_indicators_json(report: Any) -> Optional[str]:
|
|
analyses = [getattr(report, "primary_analysis", None)]
|
|
analyses.extend(getattr(report, "confirmation_analyses", None) or [])
|
|
snapshot = {}
|
|
for analysis in analyses:
|
|
if analysis is None:
|
|
continue
|
|
timeframe = str(getattr(analysis, "timeframe", ""))
|
|
if timeframe:
|
|
snapshot[timeframe] = _analysis_to_dict(analysis)
|
|
return _dumps(snapshot) if snapshot else None
|
|
|
|
|
|
def _candles_json(dataframes: Any, limit: int = 14) -> Optional[str]:
|
|
"""Last closed OHLCV bars per fetched timeframe, oldest first."""
|
|
snapshot = {}
|
|
for timeframe, df in (dataframes or {}).items():
|
|
if df is None or getattr(df, "empty", True):
|
|
continue
|
|
bars = []
|
|
for timestamp, row in df.tail(limit).iterrows():
|
|
bars.append({
|
|
"timestamp": _iso(timestamp),
|
|
"open": _finite(row.get("Open")),
|
|
"high": _finite(row.get("High")),
|
|
"low": _finite(row.get("Low")),
|
|
"close": _finite(row.get("Close")),
|
|
"volume": _finite(row.get("Volume")),
|
|
})
|
|
if bars:
|
|
snapshot[_key_str(timeframe)] = bars
|
|
return _dumps(snapshot) if snapshot else None
|
|
|
|
|
|
def _report_row(
|
|
report: Any,
|
|
context: Any,
|
|
source: str,
|
|
timeframe: str,
|
|
dataframes: Any = None,
|
|
extra: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Column-name → value mapping for one scan row (see
|
|
_SIGNAL_INSERT_COLS). ``extra`` carries producer-side metadata
|
|
(``scan_kind``, ``config_fingerprint``)."""
|
|
extra = extra or {}
|
|
mtf_signals = getattr(report, "mtf_signals", None) or {}
|
|
mtf_json = _dumps({
|
|
"agreement": bool(getattr(report, "mtf_agreement", False)),
|
|
"signals": {_key_str(k): getattr(v, "name", str(v))
|
|
for k, v in mtf_signals.items()},
|
|
"scores": {_key_str(k): _finite(v)
|
|
for k, v in (getattr(report, "mtf_scores", None)
|
|
or {}).items()},
|
|
})
|
|
|
|
setup = getattr(report, "trade_setup", None)
|
|
setup_json = None
|
|
if setup is not None:
|
|
setup_json = _dumps({
|
|
"direction": getattr(setup, "direction", None),
|
|
"entry": _finite(getattr(setup, "entry_price", None)),
|
|
"stop_loss": _finite(getattr(setup, "stop_loss", None)),
|
|
"take_profit_1": _finite(getattr(setup, "take_profit_1", None)),
|
|
"take_profit_2": _finite(getattr(setup, "take_profit_2", None)),
|
|
"risk_reward": _finite(getattr(setup, "risk_reward", None)),
|
|
"risk_pips": _finite(getattr(setup, "risk_pips", None)),
|
|
"reward_pips": _finite(getattr(setup, "reward_pips", None)),
|
|
"strategy": getattr(setup, "strategy", None),
|
|
"setup_pattern": getattr(setup, "setup_pattern", None),
|
|
"round_trip_cost": _finite(
|
|
getattr(setup, "round_trip_cost", None)),
|
|
"trail_arm_r": _finite(getattr(setup, "trail_arm_r", None)),
|
|
"trail_dist_r": _finite(getattr(setup, "trail_dist_r", None)),
|
|
})
|
|
|
|
reasons_json = _dumps({
|
|
"bullish": list(getattr(report, "bullish_reasons", None) or []),
|
|
"bearish": list(getattr(report, "bearish_reasons", None) or []),
|
|
})
|
|
warnings_json = _dumps(list(getattr(report, "warnings", None) or []))
|
|
events_json, headlines_json = _context_news_json(context)
|
|
|
|
signal = getattr(report, "final_signal", None)
|
|
panel_takes = getattr(report, "panel_takes", None) or []
|
|
panel_takes_json = _dumps(panel_takes) if panel_takes else None
|
|
quality = getattr(report, "signal_quality", "") or ""
|
|
warnings = list(getattr(report, "warnings", None) or [])
|
|
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,
|
|
"symbol": str(getattr(report, "symbol", "")),
|
|
"timeframe": timeframe,
|
|
"signal": getattr(signal, "name", str(signal)),
|
|
"score": _finite(getattr(report, "final_score", None)),
|
|
"confidence": _finite(getattr(report, "final_confidence", None)),
|
|
"quality": quality,
|
|
"action": getattr(report, "action_recommendation", "") or "",
|
|
"price": _finite(getattr(report, "current_price", None)),
|
|
"spread": bar_range,
|
|
"market_regime": getattr(report, "market_regime", "") or "",
|
|
"volatility_regime": getattr(report, "volatility_regime", "") or "",
|
|
"mtf_json": mtf_json,
|
|
"setup_json": setup_json,
|
|
"reasons_json": reasons_json,
|
|
"warnings_json": warnings_json,
|
|
"event_risk": getattr(report, "event_risk", "") or "",
|
|
"event_description": getattr(report, "event_description", "") or "",
|
|
"news_sentiment": _finite(getattr(report, "news_sentiment", None)),
|
|
"news_description": getattr(report, "news_description", "") or "",
|
|
"news_summary": getattr(report, "news_summary", "") or "",
|
|
# Interned by SignalLogger.log into news_snapshots; the inline
|
|
# columns stay NULL on new rows (legacy rows keep theirs).
|
|
"events_json": events_json,
|
|
"headlines_json": headlines_json,
|
|
"indicators_json": _report_indicators_json(report),
|
|
"candles_json": _candles_json(dataframes),
|
|
"panel_verdict": getattr(report, "panel_verdict", None),
|
|
"panel_confidence": _finite(getattr(report, "panel_confidence", None)),
|
|
"panel_rationale": getattr(report, "panel_rationale", None),
|
|
"panel_takes_json": panel_takes_json,
|
|
"setup_key": setup_key(setup_dict, timeframe),
|
|
"headlines_hash": None,
|
|
"events_hash": None,
|
|
"gate_reason": gate_reason(warnings, cleaned, quality,
|
|
setup_dict is not None),
|
|
"actionable": int(setup_dict is not None
|
|
and quality in _ACTIONABLE_GRADES),
|
|
"cleaned_warnings_json": _dumps(cleaned) if cleaned else None,
|
|
"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),
|
|
}
|
|
|
|
|
|
class SignalLogger:
|
|
"""Thread-safe SQLite writer for signal history. One instance per
|
|
database path; safe to share across the bot's executor threads and
|
|
the daemon's thread pool."""
|
|
|
|
def __init__(self, path: str):
|
|
parent = os.path.dirname(path)
|
|
if parent:
|
|
os.makedirs(parent, exist_ok=True)
|
|
self._conn = sqlite3.connect(path, check_same_thread=False)
|
|
# busy_timeout BEFORE the journal-mode switch: several processes
|
|
# (bot, dashboard, tests) open the same file at once and the WAL
|
|
# 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"))
|
|
self._with_lock_retry(
|
|
lambda: self._conn.executescript(_SCHEMA))
|
|
self._with_lock_retry(self._migrate_signal_columns)
|
|
self._with_lock_retry(self._migrate_outlook_columns)
|
|
self._with_lock_retry(self._ensure_news_view)
|
|
|
|
def _ensure_news_view(self) -> None:
|
|
"""(Re)create the read-side view atomically. Several processes
|
|
open the DB at once; DROP+CREATE inside one IMMEDIATE transaction
|
|
serialises them, and a view another opener created in between is
|
|
not an error."""
|
|
try:
|
|
self._conn.execute("BEGIN IMMEDIATE")
|
|
self._conn.execute("DROP VIEW IF EXISTS signals_with_news")
|
|
self._conn.execute(
|
|
"CREATE VIEW signals_with_news AS"
|
|
" SELECT s.*,"
|
|
" COALESCE(s.headlines_json, nh.body) AS headlines_resolved,"
|
|
" COALESCE(s.events_json, ne.body) AS events_resolved"
|
|
" FROM signals s"
|
|
" LEFT JOIN news_snapshots nh ON nh.hash = s.headlines_hash"
|
|
" LEFT JOIN news_snapshots ne ON ne.hash = s.events_hash")
|
|
self._conn.commit()
|
|
except sqlite3.OperationalError as exc:
|
|
self._conn.rollback()
|
|
if "already exists" not in str(exc).lower():
|
|
raise
|
|
|
|
@staticmethod
|
|
def _with_lock_retry(fn, attempts: int = 8, delay: float = 0.05):
|
|
"""Run ``fn`` again on 'database is locked' (concurrent openers
|
|
racing the same schema step); other errors propagate at once."""
|
|
for attempt in range(attempts):
|
|
try:
|
|
return fn()
|
|
except sqlite3.OperationalError as exc:
|
|
if "locked" not in str(exc).lower() or attempt == attempts - 1:
|
|
raise
|
|
time.sleep(delay * (attempt + 1))
|
|
|
|
def _migrate_signal_columns(self) -> None:
|
|
"""Add newer signal columns to pre-existing tables (idempotent)."""
|
|
existing = {r[1] for r in self._conn.execute("PRAGMA table_info(signals)")}
|
|
for name, col_type in _ADDITIVE_SIGNAL_COLS:
|
|
if name not in existing:
|
|
try:
|
|
self._conn.execute(
|
|
f"ALTER TABLE signals ADD COLUMN {name} {col_type}")
|
|
except sqlite3.OperationalError as exc:
|
|
# Another dashboard/bot process may have migrated the
|
|
# shared database after our PRAGMA snapshot.
|
|
if "duplicate column name" not in str(exc).lower():
|
|
raise
|
|
self._conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_signals_advisory_pending"
|
|
" ON signals (symbol, advisory_outcome_status, id)")
|
|
self._conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_signals_setup_key"
|
|
" ON signals (symbol, setup_key, id)")
|
|
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:
|
|
"""Add snapshot columns to pre-existing outlook tables."""
|
|
existing = {
|
|
r[1] for r in self._conn.execute("PRAGMA table_info(outlooks)")
|
|
}
|
|
for name, col_type in _ADDITIVE_OUTLOOK_COLS:
|
|
if name not in existing:
|
|
try:
|
|
self._conn.execute(
|
|
f"ALTER TABLE outlooks ADD COLUMN {name} {col_type}")
|
|
except sqlite3.OperationalError as exc:
|
|
if "duplicate column name" not in str(exc).lower():
|
|
raise
|
|
self._conn.commit()
|
|
|
|
def log(
|
|
self,
|
|
report: Any,
|
|
context: Any = None,
|
|
source: str = "live",
|
|
timeframe: str = "",
|
|
dataframes: Any = None,
|
|
extra: Optional[Dict[str, Any]] = None,
|
|
) -> int:
|
|
"""Insert one scan row; returns the rowid (panel-update key).
|
|
|
|
One transaction interns the news blobs, inserts the row and seeds
|
|
(or dedups) its advisory outcome; a seeding failure is logged and
|
|
the row still commits. Pending outcomes are then advanced from
|
|
this scan's candle snapshot."""
|
|
row = _report_row(report, context, source, timeframe, dataframes,
|
|
extra=extra)
|
|
now_iso = row["logged_at"]
|
|
with self._lock:
|
|
self._conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
row["headlines_hash"] = self._intern_news(
|
|
"headlines", row.pop("headlines_json"), now_iso)
|
|
row["events_hash"] = self._intern_news(
|
|
"events", row.pop("events_json"), now_iso)
|
|
row["headlines_json"] = None
|
|
row["events_json"] = None
|
|
cur = self._conn.execute(
|
|
_INSERT, tuple(row[c] for c in _SIGNAL_INSERT_COLS))
|
|
rowid = int(cur.lastrowid)
|
|
try:
|
|
self._seed_outcome(rowid, row)
|
|
except Exception:
|
|
logger.warning("signal_log: outcome seeding failed",
|
|
exc_info=True)
|
|
self._conn.commit()
|
|
except Exception:
|
|
self._conn.rollback()
|
|
raise
|
|
try:
|
|
self.resolve_pending(str(row["symbol"]), row["candles_json"])
|
|
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 ────────────────────────────────────────────
|
|
|
|
def _intern_news(self, kind: str, body: Optional[str],
|
|
now_iso: str) -> Optional[str]:
|
|
"""Store one news blob once; returns its content hash."""
|
|
if not body:
|
|
return None
|
|
digest = _blob_hash(body)
|
|
self._conn.execute(
|
|
"INSERT OR IGNORE INTO news_snapshots"
|
|
" (hash, kind, body, first_seen, last_seen)"
|
|
" VALUES (?, ?, ?, ?, ?)",
|
|
(digest, kind, body, now_iso, now_iso))
|
|
self._conn.execute(
|
|
"UPDATE news_snapshots SET last_seen=? WHERE hash=?",
|
|
(now_iso, digest))
|
|
return digest
|
|
|
|
# ── advisory-event seeding / dedup ────────────────────────────
|
|
|
|
def _seed_outcome(self, rowid: int, row: Dict[str, Any]) -> None:
|
|
"""Seed a PENDING outcome for a new setup row, or mark it a
|
|
DUPLICATE of the most recent same-setup row within the gap.
|
|
Runs inside log()'s transaction."""
|
|
key = row.get("setup_key")
|
|
if not key or not row.get("setup_json"):
|
|
return
|
|
prior = self._conn.execute(
|
|
"SELECT id, advisory_event_id, logged_at FROM signals"
|
|
" WHERE symbol=? AND setup_key=? AND id<?"
|
|
" ORDER BY id DESC LIMIT 1",
|
|
(row["symbol"], key, rowid)).fetchone()
|
|
if prior is not None:
|
|
try:
|
|
gap = (datetime.fromisoformat(row["logged_at"])
|
|
- datetime.fromisoformat(prior[2])).total_seconds()
|
|
except (TypeError, ValueError):
|
|
gap = None
|
|
# abs(), not 0 <= gap: logged_at is stamped in _report_row
|
|
# BEFORE this row wins the write lock (busy_timeout 5 s), so
|
|
# two processes logging the same setup at one boundary can
|
|
# land with their timestamp order inverted against their id
|
|
# order. A negative gap is that inversion, not a different
|
|
# trade — treating it as "not a duplicate" seeded a second
|
|
# event and double-counted its R.
|
|
if gap is not None and abs(gap) <= OUTCOME_DEDUP_GAP_SECONDS:
|
|
event_id = int(prior[1] or prior[0])
|
|
self._conn.execute(
|
|
"UPDATE signals SET advisory_event_id=?,"
|
|
" advisory_outcome_status='DUPLICATE',"
|
|
" advisory_outcome_json=? WHERE id=?",
|
|
(event_id,
|
|
_dumps({"model": "advisory-event-dedup-v1",
|
|
"status": "DUPLICATE", "event_id": event_id}),
|
|
rowid))
|
|
return
|
|
setup = json.loads(row["setup_json"])
|
|
indicators = (json.loads(row["indicators_json"])
|
|
if row.get("indicators_json") else None)
|
|
status, state = initial_outcome(setup, indicators, row["timeframe"])
|
|
self._conn.execute(
|
|
"UPDATE signals SET advisory_event_id=?,"
|
|
" advisory_outcome_status=?, advisory_outcome_json=?"
|
|
" WHERE id=?",
|
|
(rowid, status, _dumps(state), rowid))
|
|
|
|
def update_panel(
|
|
self,
|
|
rowid: int,
|
|
verdict: str,
|
|
confidence: Optional[float],
|
|
rationale: str,
|
|
takes_json: str,
|
|
*,
|
|
created_at: Optional[str] = None,
|
|
cache_status: Optional[str] = None,
|
|
payload_hash: Optional[str] = None,
|
|
verdict_source: Optional[str] = None,
|
|
) -> None:
|
|
"""Fill panel provenance; seeds an outcome only for rows that
|
|
log() did not already seed (rows written by an older process)."""
|
|
with self._lock:
|
|
row = self._conn.execute(
|
|
"SELECT setup_json, indicators_json, timeframe,"
|
|
" advisory_outcome_status FROM signals WHERE id=?",
|
|
(rowid,),
|
|
).fetchone()
|
|
if row is None:
|
|
return
|
|
advisory_status = row[3]
|
|
advisory_json = None
|
|
if advisory_status is None:
|
|
setup = json.loads(row[0]) if row[0] else None
|
|
indicators = json.loads(row[1]) if row[1] else None
|
|
advisory_status, state = initial_outcome(
|
|
setup, indicators, row[2])
|
|
advisory_json = _dumps(state)
|
|
self._conn.execute(
|
|
"UPDATE signals SET panel_verdict=?, panel_confidence=?,"
|
|
" panel_rationale=?, panel_takes_json=?,"
|
|
" panel_created_at=COALESCE(?, panel_created_at),"
|
|
" panel_cache_status=COALESCE(?, panel_cache_status),"
|
|
" panel_payload_hash=COALESCE(?, panel_payload_hash),"
|
|
" panel_verdict_source=COALESCE(?, panel_verdict_source),"
|
|
" advisory_outcome_status=COALESCE(advisory_outcome_status, ?),"
|
|
" advisory_outcome_json=COALESCE(advisory_outcome_json, ?)"
|
|
" WHERE id=?",
|
|
(verdict, confidence, rationale, takes_json,
|
|
created_at, cache_status, payload_hash, verdict_source,
|
|
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:
|
|
"""Advance pending advisory outcomes from a new candle snapshot.
|
|
|
|
The inserting scan's own snapshot is tried first (cheap, covers the
|
|
steady state). When it does not overlap an outcome's last processed
|
|
bar — that process missed >14 bars of this timeframe, a weekend
|
|
straddle, a restart — the resolver rebuilds the bar stream for the
|
|
timeframe from every stored snapshot after that bar (any surface)
|
|
and retries, so the outcome is not marked UNRESOLVABLE for a gap
|
|
the database can fill. Each row commits on its own."""
|
|
if not candles_json:
|
|
return 0
|
|
snapshots = json.loads(candles_json)
|
|
if not isinstance(snapshots, dict):
|
|
return 0
|
|
changed = 0
|
|
with self._lock:
|
|
rows = self._conn.execute(
|
|
"SELECT id, timeframe FROM signals"
|
|
" WHERE symbol=? AND advisory_outcome_status='PENDING'"
|
|
" ORDER BY id",
|
|
(symbol,),
|
|
).fetchall()
|
|
eligible = 0
|
|
for rowid, timeframe in rows:
|
|
bars = timeframe_value(snapshots, timeframe)
|
|
if not bars:
|
|
continue
|
|
if eligible >= max(1, int(limit)):
|
|
break
|
|
eligible += 1
|
|
# One IMMEDIATE transaction per row: the state is re-read
|
|
# under the write lock, so the bot and the dashboard (each
|
|
# advancing from its own snapshot) serialise instead of
|
|
# overwriting each other's progress.
|
|
self._conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
hit = self._conn.execute(
|
|
"SELECT advisory_outcome_json FROM signals"
|
|
" WHERE id=? AND advisory_outcome_status='PENDING'",
|
|
(rowid,)).fetchone()
|
|
if hit is None or not hit[0]:
|
|
self._conn.rollback()
|
|
continue
|
|
before = _dumps(json.loads(hit[0]))
|
|
advanced = advance_outcome(json.loads(before), bars)
|
|
if (advanced.get("status") == "UNRESOLVABLE"
|
|
and "continuity" in str(advanced.get("reason", ""))):
|
|
union = self._stream_since(
|
|
symbol, timeframe,
|
|
json.loads(before)["progress"]["last_processed_bar"])
|
|
if union:
|
|
advanced = advance_outcome(json.loads(before), union)
|
|
after = _dumps(advanced)
|
|
if after == before:
|
|
self._conn.rollback()
|
|
continue
|
|
status = advanced.get("status", "UNRESOLVABLE")
|
|
self._conn.execute(
|
|
"UPDATE signals SET advisory_outcome_status=?,"
|
|
" advisory_r_multiple=?, advisory_outcome_json=?"
|
|
" WHERE id=?",
|
|
(status, canonical_r(advanced), after, rowid),
|
|
)
|
|
self._conn.commit()
|
|
changed += 1
|
|
except Exception:
|
|
self._conn.rollback()
|
|
raise
|
|
return changed
|
|
|
|
def _stream_since(self, symbol: str, timeframe: str,
|
|
last_bar: str, max_rows: int = 4000) -> List[Dict[str, Any]]:
|
|
"""Union of every stored closed bar of ``timeframe`` from
|
|
``last_bar`` onward, across all rows/surfaces (oldest first).
|
|
Only the requested timeframe's array is pulled out of each
|
|
candles_json (json_extract), so the cost is bounded by the number
|
|
of rows logged since that bar, not by blob size."""
|
|
tf = str(getattr(timeframe, "value", timeframe))
|
|
try:
|
|
since = datetime.fromisoformat(
|
|
str(last_bar).replace(" ", "T")).isoformat()
|
|
except (TypeError, ValueError):
|
|
return []
|
|
bars: Dict[str, Dict[str, Any]] = {}
|
|
for key in _tf_storage_keys(tf):
|
|
rows = self._conn.execute(
|
|
"SELECT json_extract(candles_json, ?) FROM signals"
|
|
" WHERE symbol=? AND candles_json IS NOT NULL"
|
|
" AND logged_at >= ? ORDER BY id LIMIT ?",
|
|
(f'$."{key}"', symbol, since, int(max_rows))).fetchall()
|
|
for (blob,) in rows:
|
|
if not blob:
|
|
continue
|
|
try:
|
|
for bar in json.loads(blob):
|
|
stamp = bar.get("timestamp")
|
|
if stamp and stamp >= since:
|
|
bars.setdefault(str(stamp), bar)
|
|
except (TypeError, ValueError, AttributeError):
|
|
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 backfill_seeds(self, limit: Optional[int] = None) -> Dict[str, int]:
|
|
"""Assign setup_key / advisory_event_id to legacy setup rows and
|
|
seed the canonical ones that carry no outcome yet. Existing
|
|
outcomes (panel-seeded rows) are never overwritten. Safe to run
|
|
beside live writers (short per-row transactions)."""
|
|
stats = {"keyed": 0, "duplicates": 0, "seeded": 0}
|
|
with self._lock:
|
|
rows = self._conn.execute(
|
|
"SELECT id, symbol, timeframe, logged_at, setup_json,"
|
|
" indicators_json, advisory_outcome_status, advisory_event_id"
|
|
" FROM signals WHERE setup_json IS NOT NULL"
|
|
" AND setup_json != 'null' ORDER BY id"
|
|
+ (f" LIMIT {int(limit)}" if limit else "")).fetchall()
|
|
last_by_key: Dict[Tuple[str, str], Tuple[int, int, datetime]] = {}
|
|
for (rowid, symbol, tf, logged_at, setup_json, ind_json,
|
|
status, event_id) in rows:
|
|
setup = json.loads(setup_json)
|
|
key = setup_key(setup, tf)
|
|
if not key:
|
|
continue
|
|
try:
|
|
ts = datetime.fromisoformat(logged_at)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
prior = last_by_key.get((symbol, key))
|
|
if (prior is not None
|
|
and abs((ts - prior[2]).total_seconds())
|
|
<= OUTCOME_DEDUP_GAP_SECONDS):
|
|
# Same setup re-logged inside the gap: a duplicate of
|
|
# the canonical event regardless of its legacy status
|
|
# (a panel-seeded RESOLVED re-log measured the very
|
|
# same trade — keeping it would double count R).
|
|
canonical = prior[1]
|
|
if status != "DUPLICATE" or event_id != canonical:
|
|
self._conn.execute(
|
|
"UPDATE signals SET setup_key=?,"
|
|
" advisory_event_id=?,"
|
|
" advisory_outcome_status='DUPLICATE',"
|
|
" advisory_r_multiple=NULL,"
|
|
" advisory_outcome_json=? WHERE id=?",
|
|
(key, canonical,
|
|
_dumps({"model": "advisory-event-dedup-v1",
|
|
"status": "DUPLICATE",
|
|
"event_id": canonical,
|
|
"superseded_status": status}), rowid))
|
|
stats["duplicates"] += 1
|
|
last_by_key[(symbol, key)] = (rowid, canonical, ts)
|
|
continue
|
|
canonical = int(event_id or rowid)
|
|
if status is None:
|
|
indicators = json.loads(ind_json) if ind_json else None
|
|
new_status, state = initial_outcome(setup, indicators, tf)
|
|
self._conn.execute(
|
|
"UPDATE signals SET setup_key=?, advisory_event_id=?,"
|
|
" advisory_outcome_status=?, advisory_outcome_json=?"
|
|
" WHERE id=?",
|
|
(key, canonical, new_status, _dumps(state), rowid))
|
|
stats["seeded"] += 1
|
|
else:
|
|
self._conn.execute(
|
|
"UPDATE signals SET setup_key=?, advisory_event_id=?"
|
|
" WHERE id=?", (key, canonical, rowid))
|
|
stats["keyed"] += 1
|
|
last_by_key[(symbol, key)] = (rowid, canonical, ts)
|
|
self._conn.commit()
|
|
return stats
|
|
|
|
def resolve_from_store(self, symbol: str, limit: int = 500) -> int:
|
|
"""Advance every PENDING outcome from the stored bar streams alone
|
|
(no provider). Used by the backfill script after seeding."""
|
|
changed = 0
|
|
with self._lock:
|
|
rows = self._conn.execute(
|
|
"SELECT id, timeframe, advisory_outcome_json FROM signals"
|
|
" WHERE symbol=? AND advisory_outcome_status='PENDING'"
|
|
" ORDER BY id LIMIT ?", (symbol, int(limit))).fetchall()
|
|
for rowid, timeframe, state_json in rows:
|
|
if not state_json:
|
|
continue
|
|
state = json.loads(state_json)
|
|
union = self._stream_since(
|
|
symbol, timeframe, state["progress"]["last_processed_bar"])
|
|
if not union:
|
|
continue
|
|
before = _dumps(state)
|
|
advanced = advance_outcome(json.loads(before), union)
|
|
after = _dumps(advanced)
|
|
if after == before:
|
|
continue
|
|
self._conn.execute(
|
|
"UPDATE signals SET advisory_outcome_status=?,"
|
|
" advisory_r_multiple=?, advisory_outcome_json=?"
|
|
" WHERE id=? AND advisory_outcome_status='PENDING'",
|
|
(advanced.get("status", "UNRESOLVABLE"),
|
|
canonical_r(advanced), after, rowid))
|
|
self._conn.commit()
|
|
changed += 1
|
|
return changed
|
|
|
|
def reseed_unresolvable(self, max_reseeds: int = 1) -> List[int]:
|
|
"""Bounded retry for continuity-broken outcomes: re-freeze the
|
|
seed from the row's own setup/indicators (at most ``max_reseeds``
|
|
times per row) so the next resolver pass can walk the now-filled
|
|
stream. Invalid-seed rows stay terminal."""
|
|
reseeded: List[int] = []
|
|
with self._lock:
|
|
rows = self._conn.execute(
|
|
"SELECT id, timeframe, setup_json, indicators_json,"
|
|
" advisory_outcome_json FROM signals"
|
|
" WHERE advisory_outcome_status='UNRESOLVABLE'"
|
|
" AND advisory_event_id = id").fetchall()
|
|
for rowid, tf, setup_json, ind_json, state_json in rows:
|
|
old = json.loads(state_json) if state_json else {}
|
|
if "continuity" not in str(old.get("reason", "")) and \
|
|
"gap" not in str(old.get("reason", "")):
|
|
continue
|
|
count = int((old.get("reseed") or {}).get("count", 0))
|
|
if count >= max_reseeds or not setup_json:
|
|
continue
|
|
indicators = json.loads(ind_json) if ind_json else None
|
|
status, state = initial_outcome(
|
|
json.loads(setup_json), indicators, tf)
|
|
if status != "PENDING":
|
|
continue
|
|
state["reseed"] = {
|
|
"count": count + 1,
|
|
"at": datetime.now(timezone.utc).isoformat()}
|
|
self._conn.execute(
|
|
"UPDATE signals SET advisory_outcome_status=?,"
|
|
" advisory_r_multiple=NULL, advisory_outcome_json=?"
|
|
" WHERE id=?", (status, _dumps(state), rowid))
|
|
reseeded.append(int(rowid))
|
|
self._conn.commit()
|
|
return reseeded
|
|
|
|
def log_outlook(self, outlook: Any, surface: str,
|
|
payload: Any = None) -> int:
|
|
"""Insert one advisory market outlook; returns its rowid."""
|
|
payload = payload if isinstance(payload, dict) else {}
|
|
technical = {
|
|
key: payload.get(key)
|
|
for key in ("indicators", "market_regime",
|
|
"volatility_regime", "key_levels")
|
|
if key in payload
|
|
}
|
|
row = (
|
|
datetime.now(timezone.utc).isoformat(),
|
|
surface,
|
|
getattr(outlook, "bias", None),
|
|
getattr(outlook, "setup_forming", None),
|
|
_finite(getattr(outlook, "confidence", None)),
|
|
getattr(outlook, "watch", None),
|
|
getattr(outlook, "note", None),
|
|
_dumps(payload.get("news")) if "news" in payload else None,
|
|
_dumps(technical) if technical else None,
|
|
_dumps(payload.get("candles")) if "candles" in payload else None,
|
|
)
|
|
with self._lock:
|
|
cur = self._conn.execute(_OUTLOOK_INSERT, row)
|
|
self._conn.commit()
|
|
return int(cur.lastrowid)
|
|
|
|
def close(self) -> None:
|
|
with self._lock:
|
|
self._conn.close()
|
|
|
|
|
|
def _tf_storage_keys(tf: str) -> List[str]:
|
|
"""Keys under which a timeframe's bars may be stored in candles_json
|
|
(enum value plus the D1 alias used by older rows)."""
|
|
keys = [tf]
|
|
if tf in ("1d", "1day"):
|
|
keys = ["1d", "1day"]
|
|
return keys
|
|
|
|
|
|
def read_news(con: sqlite3.Connection, rowid: int
|
|
) -> Tuple[Optional[str], Optional[str]]:
|
|
"""(events_json, headlines_json) for ``rowid`` on ANY connection —
|
|
resolves interned snapshots and falls back to the legacy inline
|
|
columns. Read-only."""
|
|
row = con.execute(
|
|
"SELECT events_json, headlines_json, events_hash, headlines_hash"
|
|
" FROM signals WHERE id=?", (rowid,)).fetchone()
|
|
if row is None:
|
|
return None, None
|
|
events, headlines, ev_hash, hl_hash = row
|
|
if events is None and ev_hash:
|
|
hit = con.execute("SELECT body FROM news_snapshots WHERE hash=?",
|
|
(ev_hash,)).fetchone()
|
|
events = hit[0] if hit else None
|
|
if headlines is None and hl_hash:
|
|
hit = con.execute("SELECT body FROM news_snapshots WHERE hash=?",
|
|
(hl_hash,)).fetchone()
|
|
headlines = hit[0] if hit else None
|
|
return events, headlines
|
|
|
|
|
|
# ── Module-level entry point (config-driven, fail-safe) ───────────────
|
|
|
|
_loggers: Dict[str, SignalLogger] = {}
|
|
_loggers_lock = threading.Lock()
|
|
|
|
|
|
def log_signal(
|
|
report: Any,
|
|
context: Any = None,
|
|
source: str = "live",
|
|
timeframe: str = "",
|
|
dataframes: Any = None,
|
|
extra: Optional[Dict[str, Any]] = None,
|
|
) -> None:
|
|
"""Persist one scan if the signal_log config block enables it.
|
|
Stashes the rowid on the report (``_signal_log_rowid``) so a post-hoc
|
|
panel attach can update this same row. NEVER raises — a logging
|
|
failure must not break a live scan. ``extra`` = producer metadata
|
|
(``scan_kind``, ``config_fingerprint``); blocked scans are skipped
|
|
when ``signal_log.log_blocked_scans`` is false."""
|
|
try:
|
|
if not app_config.signal_log_enabled():
|
|
return
|
|
if ((extra or {}).get("scan_kind") == "blocked"
|
|
and not app_config.signal_log_blocked_scans()):
|
|
return
|
|
path = app_config.signal_log_path()
|
|
with _loggers_lock:
|
|
lg = _loggers.get(path)
|
|
if lg is None:
|
|
lg = SignalLogger(path)
|
|
_loggers[path] = lg
|
|
rowid = lg.log(report, context=context, source=source,
|
|
timeframe=timeframe, dataframes=dataframes,
|
|
extra=extra)
|
|
try:
|
|
report._signal_log_rowid = rowid
|
|
except Exception:
|
|
pass # exotic report object — panel update just won't happen
|
|
except Exception:
|
|
logger.warning("signal_log: failed to persist signal",
|
|
exc_info=True)
|
|
|
|
|
|
def log_panel_update(report: Any) -> None:
|
|
"""Write post-hoc panel fields onto the row logged for this scan.
|
|
|
|
Matches on the private rowid hand-off from log_signal; a report that
|
|
was never logged (backtest replay, logging disabled, logger in
|
|
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
|
|
verdict = getattr(report, "panel_verdict", "") or ""
|
|
if not verdict:
|
|
return
|
|
path = app_config.signal_log_path()
|
|
with _loggers_lock:
|
|
lg = _loggers.get(path)
|
|
if lg is None:
|
|
return
|
|
lg.update_panel(
|
|
int(rowid), verdict,
|
|
_finite(getattr(report, "panel_confidence", None)),
|
|
getattr(report, "panel_rationale", "") or "",
|
|
_dumps(getattr(report, "panel_takes", None) or []),
|
|
created_at=getattr(report, "panel_created_at", "") or None,
|
|
cache_status=getattr(report, "panel_cache_status", "") or None,
|
|
payload_hash=getattr(report, "panel_payload_hash", "") or None,
|
|
verdict_source=getattr(report, "panel_verdict_source", "") or None)
|
|
except Exception:
|
|
logger.warning("signal_log: failed to persist panel verdict",
|
|
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.
|
|
|
|
Outlook persistence is deliberately independent of a signal row: the
|
|
feature runs on its own hourly cadence and must never affect a scan.
|
|
"""
|
|
try:
|
|
if not app_config.signal_log_enabled():
|
|
return
|
|
path = app_config.signal_log_path()
|
|
with _loggers_lock:
|
|
lg = _loggers.get(path)
|
|
if lg is None:
|
|
lg = SignalLogger(path)
|
|
_loggers[path] = lg
|
|
lg.log_outlook(outlook, surface, payload=payload)
|
|
except Exception:
|
|
logger.warning("signal_log: failed to persist market outlook",
|
|
exc_info=True)
|
|
|
|
|
|
def reset_loggers_for_tests() -> None:
|
|
"""Close and drop all cached loggers (test isolation)."""
|
|
with _loggers_lock:
|
|
for lg in _loggers.values():
|
|
try:
|
|
lg.close()
|
|
except Exception:
|
|
pass
|
|
_loggers.clear()
|