""" 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 panel_verdict / panel_confidence / panel_rationale / panel_takes_json on that same row. 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. """ import json import logging import math import os import sqlite3 import threading from datetime import datetime, timezone from typing import Any, Dict, List, Optional import app_config 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, panel_verdict TEXT, panel_confidence REAL, panel_rationale TEXT, panel_takes_json TEXT ); CREATE INDEX IF NOT EXISTS idx_signals_sym_tf_ts ON signals (symbol, timeframe, logged_at); """ # Additive migration: older DBs created before the panel columns existed # get them added here. Idempotent — skips columns already present. _PANEL_COLS = ( ("panel_verdict", "TEXT"), ("panel_confidence", "REAL"), ("panel_rationale", "TEXT"), ("panel_takes_json", "TEXT"), ) _INSERT = """ INSERT INTO signals ( 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, panel_verdict, panel_confidence, panel_rationale, panel_takes_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) # ── 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)), "currencies": list(getattr(h, "currencies_mentioned", None) or []), "keywords": list(getattr(h, "keywords", None) or []), } 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 _report_row( report: Any, context: Any, source: str, timeframe: str, ) -> tuple: 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), "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 return ( datetime.now(timezone.utc).isoformat(), source, str(getattr(report, "symbol", "")), timeframe, getattr(signal, "name", str(signal)), _finite(getattr(report, "final_score", None)), _finite(getattr(report, "final_confidence", None)), getattr(report, "signal_quality", "") or "", getattr(report, "action_recommendation", "") or "", _finite(getattr(report, "current_price", None)), _finite(getattr(report, "spread_estimate", None)), getattr(report, "market_regime", "") or "", getattr(report, "volatility_regime", "") or "", mtf_json, setup_json, reasons_json, warnings_json, getattr(report, "event_risk", "") or "", getattr(report, "event_description", "") or "", _finite(getattr(report, "news_sentiment", None)), getattr(report, "news_description", "") or "", getattr(report, "news_summary", "") or "", events_json, headlines_json, getattr(report, "panel_verdict", None), _finite(getattr(report, "panel_confidence", None)), getattr(report, "panel_rationale", None), panel_takes_json, ) 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) self._conn.execute("PRAGMA journal_mode=WAL") self._conn.execute("PRAGMA busy_timeout=5000") self._lock = threading.Lock() with self._lock: self._conn.executescript(_SCHEMA) self._migrate_panel_columns() @property def path(self) -> str: return self._path def _migrate_panel_columns(self) -> None: """Add panel_* columns to pre-existing tables (idempotent).""" existing = {r[1] for r in self._conn.execute("PRAGMA table_info(signals)")} for name, col_type in _PANEL_COLS: if name not in existing: self._conn.execute( f"ALTER TABLE signals ADD COLUMN {name} {col_type}") self._conn.commit() def log( self, report: Any, context: Any = None, source: str = "live", timeframe: str = "", ) -> int: """Insert one scan row; returns the rowid (panel-update key).""" row = _report_row(report, context, source, timeframe) with self._lock: cur = self._conn.execute(_INSERT, row) self._conn.commit() return int(cur.lastrowid) def update_panel( self, rowid: int, verdict: str, confidence: Optional[float], rationale: str, takes_json: str, ) -> None: """Fill the panel columns on an already-inserted scan row.""" with self._lock: self._conn.execute( "UPDATE signals SET panel_verdict=?, panel_confidence=?," " panel_rationale=?, panel_takes_json=? WHERE id=?", (verdict, confidence, rationale, takes_json, rowid)) self._conn.commit() def close(self) -> None: with self._lock: self._conn.close() # ── 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 = "", ) -> 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.""" 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 rowid = lg.log(report, context=context, source=source, timeframe=timeframe) 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 [])) except Exception: logger.warning("signal_log: failed to persist panel verdict", 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()