Harwinder Sekhon
b1e880be4f
- seed an advisory outcome on every setup row; same-setup re-logs within 60 min are DUPLICATE rows (advisory_event_id); resolver rebuilds bar streams across gaps - headlines/events interned into news_snapshots (DB 667 MB -> 262 MB); readers use signals_with_news / read_news and mode=ro (immutable=1 missed WAL rows) - post-gate truth columns: gate_reason, actionable, cleaned_warnings, scan_kind, config_fingerprint; SELL-hour gates fail closed - telegram: breach rescan reuses this bar's report; grade dropped from alert fingerprint (upgrade-only re-alert) - market_outlook: hour-bucket dedup (old rule skipped alternate hours), skip heartbeats - agent_panel: macro-anchored OPPOSE guard (opposed trades were +30R over 3 weeks) - analyze_signal_log --stored-candles/--parity; backfill + news migration scripts - tests: Module 25 (analyze_signal_log) + 27 (observability); 429 total Claude-Session: https://claude.ai/code/session_01FC4s6VCURFte4dub4xKxYL
1245 lines
52 KiB
Python
1245 lines
52 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 the
|
|
XAU_CP acceptance removed from ``warnings_json``), ``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) and ``config_fingerprint`` (hash of the gate config the
|
|
producing process ran with — surfaces config drift between processes).
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, Iterable, List, Optional, Tuple
|
|
|
|
import app_config
|
|
from advisory_outcome import (
|
|
advance_outcome,
|
|
canonical_r,
|
|
initial_outcome,
|
|
timeframe_value,
|
|
)
|
|
|
|
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 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"),
|
|
)
|
|
|
|
# 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")
|
|
|
|
_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",
|
|
)
|
|
_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()
|
|
|
|
|
|
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
|
|
|
|
|
|
# ── 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": _finite(getattr(indicator, "weight", 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))
|
|
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"),
|
|
}
|
|
|
|
|
|
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):
|
|
self._path = path
|
|
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()
|
|
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))
|
|
|
|
@property
|
|
def path(self) -> str:
|
|
return self._path
|
|
|
|
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)")
|
|
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)
|
|
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
|
|
|
|
def news_for(self, rowid: int) -> Tuple[Optional[str], Optional[str]]:
|
|
"""(events_json, headlines_json) for a row, legacy or interned."""
|
|
with self._lock:
|
|
return read_news(self._conn, rowid)
|
|
|
|
# ── 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
|
|
if gap is not None and 0 <= 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,
|
|
) -> 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),"
|
|
" 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,
|
|
advisory_status, advisory_json, 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 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
|
|
from provider history). Returns the resulting status."""
|
|
with self._lock:
|
|
row = self._conn.execute(
|
|
"SELECT advisory_outcome_status, advisory_outcome_json"
|
|
" FROM signals WHERE id=?", (rowid,)).fetchone()
|
|
if row is None or row[0] != "PENDING" or not row[1]:
|
|
return str(row[0]) if row else "MISSING"
|
|
state = advance_outcome(json.loads(row[1]), list(bars))
|
|
state["evidence"] = evidence
|
|
status = state.get("status", "UNRESOLVABLE")
|
|
self._conn.execute(
|
|
"UPDATE signals SET advisory_outcome_status=?,"
|
|
" advisory_r_multiple=?, advisory_outcome_json=?"
|
|
" WHERE id=? AND advisory_outcome_status='PENDING'",
|
|
(status, canonical_r(state), _dumps(state), rowid))
|
|
self._conn.commit()
|
|
return status
|
|
|
|
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 0 <= (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)
|
|
except Exception:
|
|
logger.warning("signal_log: failed to persist panel verdict",
|
|
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()
|