signal_log: persist every live scan (signal, grade, price, news) to SQLite
New signal_log.py logs each SignalReport to logs/signals.db via a fail-safe hook in SignalAggregator.analyze_pair — the live-only entry point used by the bot, daemon, CLI and dashboard. Backtester calls analyze_from_dataframes directly, so live==backtest parity is preserved (test-locked). Rows carry signal/grade/price/spread, MTF + trade-setup + reasons/warnings JSON, and news-at-scan-time (calendar events + scored headlines JSON via NewsEngine class-level caches). Config: signal_log block (enabled, path; env SIGNAL_LOG_ENABLED/PATH), source tagged per consumer (telegram/oanda_daemon/cli/dashboard). 381 tests pass (24 modules).
This commit is contained in:
Harwinder Sekhon
+313
@@ -0,0 +1,313 @@
|
||||
"""
|
||||
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).
|
||||
|
||||
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
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_signals_sym_tf_ts
|
||||
ON signals (symbol, timeframe, logged_at);
|
||||
"""
|
||||
|
||||
_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
|
||||
) 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)
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
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)
|
||||
|
||||
@property
|
||||
def path(self) -> str:
|
||||
return self._path
|
||||
|
||||
def log(
|
||||
self,
|
||||
report: Any,
|
||||
context: Any = None,
|
||||
source: str = "live",
|
||||
timeframe: str = "",
|
||||
) -> None:
|
||||
row = _report_row(report, context, source, timeframe)
|
||||
with self._lock:
|
||||
self._conn.execute(_INSERT, row)
|
||||
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.
|
||||
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
|
||||
lg.log(report, context=context, source=source, timeframe=timeframe)
|
||||
except Exception:
|
||||
logger.warning("signal_log: failed to persist signal",
|
||||
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()
|
||||
Reference in New Issue
Block a user