From 4ffdd6bcb2c3a46dcc582e84ab1bf54119e9c34c Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Thu, 13 Aug 2026 10:24:49 -0400 Subject: [PATCH] signal_log: persist every live scan (signal, grade, price, news) to SQLite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- CLAUDE.md | 3 +- app_config.py | 40 ++++- config.json.example | 4 + main.py | 2 +- oanda_daemon.py | 3 +- pair_signals/__init__.py | 3 +- signal_aggregator.py | 35 ++++- signal_log.py | 313 +++++++++++++++++++++++++++++++++++++++ telegram_bot.py | 2 +- test_all.py | 287 +++++++++++++++++++++++++++++++++++ 10 files changed, 685 insertions(+), 7 deletions(-) create mode 100644 signal_log.py diff --git a/CLAUDE.md b/CLAUDE.md index 31325e4..eeb74bd 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -11,7 +11,7 @@ Every numeric threshold in `config.py` and `pair_signals/xauusd.py` was backtest ## Commands ```bash -python test_all.py # QA suite: 374 tests, 23 modules, no network (mocked) +python test_all.py # QA suite: 381 tests, 24 modules, no network (mocked) python main.py --pair XAUUSD --timeframe H1 --detail # CLI signal report (--json for full dump, # --panel to run the advisory LLM agent panel on-demand) python telegram_bot.py # the bot (refuses to boot without credentials) @@ -59,6 +59,7 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz - **`engine.py`** — `TechnicalEngine`: 9 scored indicators (ema_trend, rsi, macd, bollinger, adx, roc, keltner, divergence, candlestick) + ATR as context. Regime classification (trending_up/down, `trending` = mixed/direction-UNKNOWN, transitioning, ranging) drives per-indicator weight multipliers and regime-aware RSI reinterpretation. Compute order matters (ATR first; MACD always computed even when not a pillar — divergence needs the histogram). Private series ride on the result as `_underscore` attrs; scorers/formatters skip anything `_`-prefixed or weight-0. - **`signal_aggregator.py`** — MTF agreement, diminishing-decay penalty system, exhaustion hard block, quality grading, then an ordered chain of post-grade gates (session, XAU SELL quality/vol/hour, regime blocks, M5 noise gates), then three setup generators: XAU ATR (A+/A), generic swing-based, plus the M5 pullback (`XAU_PB`) and chart-pattern (`XAU_CP`, `chart_patterns.py`) entry paths. The PB/CP generators run AFTER the gate chain and force quality to "B"; the CP path deliberately accepts grade-F composites and re-checks only session/day/big-candle (via warning prefixes) plus its own SELL/vol/RSI gates — it does NOT re-check exhaustion or the M5 min-score/MTF gates. **This is validated behavior, not a bug** (2026-08-04 audit: 226/586 trades in the shipped M5 12mo book are XAU_CP, +38.9R, 104 of them exhaustion-flagged and net-positive) — changing it is a rule-1 revalidation event. `TradeSetup.strategy` selects trailing params; trail params are stamped on the setup from the *producing* config so per-TF overrides reach formatters. - **`pair_signals/`** — per-pair config factories. **Always use the factory (`produce_signal` / `make_config`), never bare `PlatformConfig()`** — the factory applies the config.json overlay first, then the validated per-TF deltas (M5 hour-9 block, M15 trail arm 0.70, M15 SELL-hour block, etc.; evidence in the docstring). A test enforces factory ≡ inline-config equivalence. Confirmation map: M5→[M15,H1], M15→[M5,H1], H1→[M15,H4], H4→[H1,D1]. +- **`signal_log.py`** — SQLite signal-history logger (stdlib only). Hooked into `SignalAggregator.analyze_pair` only (the live entry point — the backtester never writes here, preserving parity), so every live scan from every surface (Telegram `telegram`, CLI `cli`, Oanda daemon `oanda_daemon`, dashboard `dashboard`) lands one row in `logs/signals.db`: signal/grade/price, MTF + trade-setup + reasons JSON, and the news at scan time (event risk + calendar events + scored headlines JSON, re-fetched via the NewsEngine class-level caches). Config.json `signal_log` block (`enabled`, `path`; env `SIGNAL_LOG_ENABLED`/`SIGNAL_LOG_PATH`). **Fail-safe by contract** — a logging failure can never break a scan. Panel fields are NOT logged (attach_panel runs post-hoc, after analyze_pair returns). - **`news_engine.py`** — Forex Factory calendar (MyFXBook RSS fallback) + RSS keyword sentiment with gold-inverse keywords. Class-level caches (fresh engine per scan). Optional DeepSeek enrichment via `llm_client.py` (requests-only, no SDK): batched headline sentiment overriding keyword scores, gold-direction on release surprises, and a market-summary blurb (`SignalReport.news_summary`) — **display/warning-only, keyword fallback on any failure, inert without a key**; prompts are static module constants (DeepSeek prefix-cache pricing) and requests send `thinking: disabled` (V4-Flash reasoning otherwise blows the timeout). By XAU-tuned default, event risk and sentiment **warn but do not gate signals** (`event_risk_caps_signals=False`, `sentiment_affects_signals=False`) — the LLM does not change that. - **`agent_panel.py`** — ai-hedge-fund-style advisory panel: 4 DeepSeek persona calls (macro/technical/risk/contrarian, concurrent on a panel-private pool) + 1 synthesizer producing an **ADVISORY-ONLY** consensus (`AGREE/NEUTRAL/OPPOSE` + confidence + rationale) written onto `SignalReport.panel_*` fields. Attached **post-hoc** by consumers (`telegram_bot` alert path, `main --panel`, `dashboard_publisher`) — never runs inside `analyze_from_dataframes` (backtest parity, test-locked) and no gate reads its fields. Static prompts (prefix caching); spend bounded by an exact per-bar cache, a reuse TTL keyed (symbol, TF, direction) so persisting signals don't re-spend every bar, and a `max_panels_per_day` UTC budget (default 20 ≈ ≤100 requests) on top of the shared 500/day client cap. `<2` valid persona takes skips the synthesizer; any failure degrades to no panel, never a changed signal. - **`telegram_bot.py`** — owner-locked bot + subscriber channel; candle-boundary auto-scan loops (M5/M15/H1/H4, boundary+10 s), alert quality threshold + fingerprint dedup (committed only after a successful send), 📰 LLM market-context line on alerts when available, WebSocket level-breach monitor fed by the streaming provider (`data_providers.streaming`, falling back to the active provider; auto-disabled when the resolved provider can't stream or has no API key — Oanda does not stream, Tiingo does). Rotating log: `logs/telegram.log`, 3-h rotation × 240 backups. diff --git a/app_config.py b/app_config.py index 813b700..a983170 100644 --- a/app_config.py +++ b/app_config.py @@ -86,7 +86,7 @@ _VALID_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") _SCANNABLE_TFS = ("M5", "M15", "H1", "H4") _KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level", - "oanda", "deepseek", "agent_panel"} + "oanda", "deepseek", "agent_panel", "signal_log"} _KNOWN_OANDA = { "account_id", "api_token", "base_url", "lot_size", "quality_gate", "max_hold_bars", @@ -98,6 +98,7 @@ _KNOWN_AGENT_PANEL = { "enabled", "telegram", "dashboard", "min_quality", "max_panels_per_day", "reuse_ttl_seconds", } +_KNOWN_SIGNAL_LOG = {"enabled", "path"} _KNOWN_TELEGRAM = {"token", "chat_id", "channel_id", "alert_min_quality"} _KNOWN_SCANNING = {"pairs", "timeframes"} _KNOWN_PAIR_KEYS = {"blocked_hours_utc"} @@ -123,6 +124,7 @@ class AppConfig: oanda_settings: Dict[str, Any] = field(default_factory=dict) deepseek_settings: Dict[str, Any] = field(default_factory=dict) agent_panel_settings: Dict[str, Any] = field(default_factory=dict) + signal_log_settings: Dict[str, Any] = field(default_factory=dict) log_level: str = "INFO" @@ -306,6 +308,17 @@ def load_app_config(path: str = CONFIG_PATH) -> AppConfig: "non-negative number") app.agent_panel_settings = dict(panel) + slog = raw.get("signal_log", {}) + _check_keys(slog, _KNOWN_SIGNAL_LOG, '"signal_log"') + if "enabled" in slog and not isinstance(slog["enabled"], bool): + raise AppConfigError( + "config.json: signal_log.enabled must be true or false") + if "path" in slog and (not isinstance(slog["path"], str) + or not slog["path"]): + raise AppConfigError( + "config.json: signal_log.path must be a non-empty string") + app.signal_log_settings = dict(slog) + level = raw.get("log_level", app.log_level) if level not in _VALID_LOG_LEVELS: raise AppConfigError( @@ -405,6 +418,31 @@ def deepseek_api_key(app: Optional[AppConfig] = None) -> str: app.deepseek_settings.get("api_key", "")) +# ── Signal-log resolution helpers (env → config.json → default) ────── + +def signal_log_enabled(app: Optional[AppConfig] = None) -> bool: + """Default ON — signal history is the platform's audit trail.""" + env = os.environ.get("SIGNAL_LOG_ENABLED") + if env: + return env.strip().lower() in ("1", "true", "yes", "on") + app = app or get_app_config() + return bool(app.signal_log_settings.get("enabled", True)) + + +def signal_log_path(app: Optional[AppConfig] = None) -> str: + """Absolute path to the SQLite signal-history DB. A relative config + value resolves against the repo root so CWD never matters.""" + env = os.environ.get("SIGNAL_LOG_PATH") + if env: + path = env + else: + app = app or get_app_config() + path = str(app.signal_log_settings.get("path", "logs/signals.db")) + if not os.path.isabs(path): + path = os.path.join(os.path.dirname(os.path.abspath(__file__)), path) + return path + + # ── PlatformConfig overlay ──────────────────────────────────────────── def apply_pair_overrides( diff --git a/config.json.example b/config.json.example index 8cffe08..5d4cd11 100644 --- a/config.json.example +++ b/config.json.example @@ -48,5 +48,9 @@ "max_panels_per_day": 20, "reuse_ttl_seconds": 3600 }, + "signal_log": { + "enabled": true, + "path": "logs/signals.db" + }, "log_level": "INFO" } diff --git a/main.py b/main.py index 27be14c..e53f862 100755 --- a/main.py +++ b/main.py @@ -336,7 +336,7 @@ def build_config(args) -> PlatformConfig: def run_live(args, config: PlatformConfig): print(f"{Colors.GRAY} Fetching market data and running analysis...{Colors.RESET}\n") aggregator = SignalAggregator(config) - report = aggregator.analyze_pair(symbol=config.symbol) + report = aggregator.analyze_pair(symbol=config.symbol, log_source="cli") # Explicit user intent: force bypasses the enabled/surface/quality # toggles, but the daily panel budget and LLM cap still apply. diff --git a/oanda_daemon.py b/oanda_daemon.py index 47a7614..af76151 100644 --- a/oanda_daemon.py +++ b/oanda_daemon.py @@ -466,7 +466,8 @@ class TradingDaemon: aggregator = SignalAggregator(config) aggregator.data_provider = self._provider - report = aggregator.analyze_pair(symbol=SYMBOL) + report = aggregator.analyze_pair(symbol=SYMBOL, + log_source="oanda_daemon") logger.info( "%s scan: signal=%s score=%.1f conf=%.0f quality=%s " diff --git a/pair_signals/__init__.py b/pair_signals/__init__.py index 5dd873a..13a4e48 100644 --- a/pair_signals/__init__.py +++ b/pair_signals/__init__.py @@ -30,6 +30,7 @@ def produce_signal( primary_tf: Timeframe, data_provider: BaseDataProvider, confirmation_tfs: Optional[List[Timeframe]] = None, + log_source: str = "dashboard", ) -> SignalReport: config_factory = PAIR_PRODUCERS.get(symbol) if config_factory is None: @@ -41,7 +42,7 @@ def produce_signal( config = config_factory(primary_tf, confirmation_tfs) aggregator = SignalAggregator(config) aggregator.data_provider = data_provider - return aggregator.analyze_pair(symbol=symbol) + return aggregator.analyze_pair(symbol=symbol, log_source=log_source) from pair_signals import xauusd as _xau_mod diff --git a/signal_aggregator.py b/signal_aggregator.py index 50a00c1..66774c5 100755 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -21,6 +21,7 @@ from config import PlatformConfig, SignalStrength, Timeframe from providers import BaseDataProvider, make_data_provider from engine import TechnicalEngine, TechnicalAnalysis from news_engine import NewsEngine, FundamentalContext +import signal_log logger = logging.getLogger(__name__) @@ -142,8 +143,14 @@ class SignalAggregator: self, symbol: Optional[str] = None, context: Optional[FundamentalContext] = None, + log_source: Optional[str] = None, ) -> SignalReport: - """Live-mode entry: fetch all needed timeframes then delegate.""" + """Live-mode entry: fetch all needed timeframes then delegate. + + Every report produced here (including weekend/day-filter early + returns) is persisted to the signal-history DB via _log_signal — + fail-safe, and never reached by the backtester, which calls + analyze_from_dataframes directly (live == backtest parity).""" if symbol is None: symbol = self.config.symbol @@ -164,6 +171,7 @@ class SignalAggregator: report.trade_setup = None report.warnings.append("WEEKEND FILTER: XAU/USD market closed — no liquidity") report.action_recommendation = "NO TRADE — Weekend (market closed)" + self._log_signal(report, log_source) return report if xau_cfg.allowed_weekdays and now_utc.weekday() not in xau_cfg.allowed_weekdays: @@ -181,6 +189,7 @@ class SignalAggregator: f"DAY FILTER: {day_names.get(now_utc.weekday(), '?')} not in allowed days ({allowed})" ) report.action_recommendation = f"NO TRADE — Only trading {allowed}" + self._log_signal(report, log_source) return report timeframes = [self.config.primary_timeframe] + list( @@ -258,8 +267,32 @@ class SignalAggregator: f"{symbol}: live session filter errored — filter SKIPPED this scan" ) + self._log_signal(report, log_source) return report + def _log_signal( + self, report: SignalReport, source: Optional[str] + ) -> None: + """Persist the scan to the signal-history DB (signal_log.py). + + Fail-safe by contract — a logging failure must never break a + live scan. The news context re-fetch rides the NewsEngine's + class-level caches, so it costs nothing within a scan and + matches the context the signal just used; if it fails, the row + is still written, just without the news JSON blobs.""" + context = None + try: + context = self._get_news_engine().get_fundamental_context() + except Exception: + logger.debug("signal log: news context unavailable", + exc_info=True) + signal_log.log_signal( + report, + context=context, + source=source or "live", + timeframe=self.config.primary_timeframe.value, + ) + def _apply_regime_filter_live( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> None: diff --git a/signal_log.py b/signal_log.py new file mode 100644 index 0000000..f1fff9d --- /dev/null +++ b/signal_log.py @@ -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() diff --git a/telegram_bot.py b/telegram_bot.py index a2bbb5c..4c3d77f 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -536,7 +536,7 @@ def _run_signal_sync( config.confirmation_timeframes = confirmation_tfs aggregator = SignalAggregator(config) aggregator.data_provider = _get_shared_dp() - return aggregator.analyze_pair(symbol=symbol) + return aggregator.analyze_pair(symbol=symbol, log_source="telegram") async def run_signal( diff --git a/test_all.py b/test_all.py index 2e825fb..8d909f8 100644 --- a/test_all.py +++ b/test_all.py @@ -9,6 +9,7 @@ import sys import os import time import threading +import tempfile import traceback from datetime import datetime, timedelta, timezone from unittest.mock import patch, MagicMock @@ -7325,6 +7326,292 @@ def _(): ap.reset_panel_state() +print("\n═══ Module 24: signal_log.py (SQLite signal history) ═══") + + +def _make_log_report(): + """SignalReport with a full trade setup, MTF data, reasons + warnings.""" + report = SignalReport( + symbol="XAU/USD", + display_name="XAU/USD", + timestamp="2026-08-01T12:00:00+00:00", + final_signal=SignalStrength.BUY, + final_score=0.72, + final_confidence=0.85, + mtf_agreement=True, + mtf_signals={Timeframe.M5: SignalStrength.STRONG_BUY, + Timeframe.M15: SignalStrength.BUY}, + mtf_scores={Timeframe.M5: 0.6, Timeframe.M15: 0.7}, + event_risk="caution", + event_description="NFP in 20m", + news_sentiment=-0.4, + news_description="Dovish Fed headlines", + news_summary="Gold supported by soft dollar.", + action_recommendation="BUY", + signal_quality="A", + market_regime="trending_up", + volatility_regime="normal", + exhaustion_count=0, + ) + report.current_price = 2450.50 + report.spread_estimate = 0.30 + report.bullish_reasons = ["EMA trend up", "MTF agree"] + report.bearish_reasons = [] + report.warnings = ["STALE DATA on H1 — signal may be unreliable"] + report.trade_setup = TradeSetup( + pair="XAU/USD", display_name="XAU/USD", direction="BUY", + entry_price=2450.50, stop_loss=2445.00, take_profit_1=2458.00, + take_profit_2=2468.00, risk_reward=1.6, risk_pips=5.5, + reward_pips=8.8, pip_value=1.0, strategy="XAU_PB", + trail_arm_r=0.70, trail_dist_r=0.85, + ) + return report + + +def _make_log_context(): + now = datetime.now(timezone.utc) + return FundamentalContext( + events_today=[EconomicEvent( + time=now + timedelta(minutes=20), currency="USD", + impact="high", event="Non-Farm Employment", + forecast="228K", previous="200K", actual=None)], + events_upcoming_1h=[], + high_impact_warning=True, + warning_message="High-impact event within 1h", + news_sentiment={"USD": -0.4}, + news_headlines=[NewsSentiment( + headline="Fed signals rate hike", source="test", + timestamp=now, currencies_mentioned=["USD", "XAU"], + sentiment_score=1.0, keywords=["rate hike"])], + calendar_available=True, + llm_summary="Gold supported by soft dollar.", + ) + + +@test("signal_log: schema creation + full roundtrip") +def _(): + import sqlite3 + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + try: + lg = slog.SignalLogger(path) + report = _make_log_report() + context = _make_log_context() + lg.log(report, context=context, source="test", timeframe="M5") + lg.close() + conn = sqlite3.connect(path) + rows = conn.execute("SELECT * FROM signals").fetchall() + cur = conn.execute("SELECT * FROM signals") + cols = [c[0] for c in cur.description] + conn.close() + assert len(rows) == 1, "exactly one row written" + row = dict(zip(cols, rows[0])) + assert row["symbol"] == "XAU/USD" + assert row["timeframe"] == "M5" + assert row["source"] == "test" + assert row["signal"] == "BUY" + assert row["quality"] == "A" + assert row["action"] == "BUY" + assert row["price"] == 2450.50 + assert row["spread"] == 0.30 + assert row["market_regime"] == "trending_up" + assert row["event_risk"] == "caution" + assert row["news_sentiment"] == -0.4 + mtf = json.loads(row["mtf_json"]) + assert mtf["agreement"] is True + assert mtf["signals"][Timeframe.M15.value] == "BUY" + assert mtf["scores"][Timeframe.M5.value] == 0.6 + setup = json.loads(row["setup_json"]) + assert setup["strategy"] == "XAU_PB" + assert setup["entry"] == 2450.50 and setup["trail_arm_r"] == 0.70 + reasons = json.loads(row["reasons_json"]) + assert "MTF agree" in reasons["bullish"] + assert "STALE" in json.loads(row["warnings_json"])[0] + events = json.loads(row["events_json"]) + assert events["high_impact_warning"] is True + assert events["events_today"][0]["event"] == "Non-Farm Employment" + assert events["events_today"][0]["surprise"] is None + headlines = json.loads(row["headlines_json"]) + assert headlines[0]["headline"] == "Fed signals rate hike" + assert headlines[0]["score"] == 1.0 + assert headlines[0]["currencies"] == ["USD", "XAU"] + finally: + slog.reset_loggers_for_tests() + os.remove(path) + + +@test("signal_log: null context → news JSON NULL, row still written") +def _(): + import sqlite3 + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + try: + lg = slog.SignalLogger(path) + lg.log(_make_log_report(), context=None, source="test", timeframe="H1") + lg.close() + conn = sqlite3.connect(path) + row = conn.execute( + "SELECT events_json, headlines_json, news_summary, signal " + "FROM signals").fetchone() + conn.close() + assert row[0] is None and row[1] is None + assert row[2] == "Gold supported by soft dollar." + assert row[3] == "BUY" + finally: + slog.reset_loggers_for_tests() + os.remove(path) + + +@test("signal_log: module entry honors SIGNAL_LOG_ENABLED/PATH env") +def _(): + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + os.remove(path) # want a fresh file from SignalLogger + try: + os.environ["SIGNAL_LOG_PATH"] = path + os.environ.pop("SIGNAL_LOG_ENABLED", None) + slog.log_signal(_make_log_report(), source="cli", timeframe="H4") + assert os.path.exists(path), "enabled default must write the DB" + slog.reset_loggers_for_tests() + os.remove(path) + os.environ["SIGNAL_LOG_ENABLED"] = "false" + slog.log_signal(_make_log_report(), source="cli", timeframe="H4") + assert not os.path.exists(path), "disabled must not create the DB" + finally: + os.environ.pop("SIGNAL_LOG_PATH", None) + os.environ.pop("SIGNAL_LOG_ENABLED", None) + slog.reset_loggers_for_tests() + if os.path.exists(path): + os.remove(path) + + +@test("signal_log: unwritable path degrades to warning, never raises") +def _(): + import signal_log as slog + # Route the DB through a plain file so makedirs fails + # (NotADirectoryError). log_signal must swallow it. + fd, blocker = tempfile.mkstemp() + os.close(fd) + try: + os.environ["SIGNAL_LOG_PATH"] = os.path.join(blocker, "sub", "x.db") + os.environ.pop("SIGNAL_LOG_ENABLED", None) + logging.disable(logging.CRITICAL) # silence the expected traceback + try: + slog.log_signal(_make_log_report(), source="test", timeframe="H1") + finally: + logging.disable(logging.NOTSET) + finally: + os.environ.pop("SIGNAL_LOG_PATH", None) + os.environ.pop("SIGNAL_LOG_ENABLED", None) + slog.reset_loggers_for_tests() + os.remove(blocker) + + +@test("signal_log: analyze_pair weekend early-return is logged") +def _(): + import signal_log as slog + from signal_aggregator import _is_forex_weekend + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + os.remove(path) + try: + os.environ["SIGNAL_LOG_PATH"] = path + os.environ.pop("SIGNAL_LOG_ENABLED", None) + agg = SignalAggregator(PlatformConfig()) + agg._get_news_engine = lambda: _NoNews() + with patch("signal_aggregator._is_forex_weekend", return_value=True): + report = agg.analyze_pair(symbol="XAU/USD", log_source="test") + assert report.final_signal == SignalStrength.NEUTRAL + assert report.signal_quality == "F" + import sqlite3 + conn = sqlite3.connect(path) + row = conn.execute( + "SELECT source, signal, quality, timeframe FROM signals").fetchone() + conn.close() + assert row is not None, "weekend report must be persisted" + assert row[0] == "test" + assert row[1] == "NEUTRAL" + assert row[2] == "F" + assert row[3] == Timeframe.H1.value + finally: + os.environ.pop("SIGNAL_LOG_PATH", None) + os.environ.pop("SIGNAL_LOG_ENABLED", None) + slog.reset_loggers_for_tests() + if os.path.exists(path): + os.remove(path) + + +@test("signal_log parity: analyze_from_dataframes never logs") +def _(): + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + os.remove(path) + try: + os.environ["SIGNAL_LOG_PATH"] = path + os.environ.pop("SIGNAL_LOG_ENABLED", None) + agg = SignalAggregator(PlatformConfig()) + agg.primary_timeframe = Timeframe.H1 # unused; keeps intent explicit + dfs = {Timeframe.H1: make_ohlcv(300, seed=42), + Timeframe.M15: make_ohlcv(300, seed=43), + Timeframe.H4: make_ohlcv(300, seed=44)} + report = agg.analyze_from_dataframes( + "XAU/USD", dfs, context=FundamentalContext(), use_news=False) + assert report.final_signal in (SignalStrength.NEUTRAL, + SignalStrength.BUY, + SignalStrength.SELL) + assert not os.path.exists(path), \ + "the replayable core must never write the signal-history DB" + finally: + os.environ.pop("SIGNAL_LOG_PATH", None) + os.environ.pop("SIGNAL_LOG_ENABLED", None) + slog.reset_loggers_for_tests() + if os.path.exists(path): + os.remove(path) + + +@test("signal_log: _log_signal tolerates a failing news engine") +def _(): + import signal_log as slog + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + os.remove(path) + try: + os.environ["SIGNAL_LOG_PATH"] = path + os.environ.pop("SIGNAL_LOG_ENABLED", None) + agg = SignalAggregator(PlatformConfig()) + agg._get_news_engine = lambda: _BoomNews() + with patch("signal_aggregator._is_forex_weekend", return_value=True): + report = agg.analyze_pair(symbol="XAU/USD", log_source="test") + assert report.final_signal == SignalStrength.NEUTRAL + import sqlite3 + conn = sqlite3.connect(path) + row = conn.execute( + "SELECT events_json, headlines_json FROM signals").fetchone() + conn.close() + assert row == (None, None), \ + "row still written with NULL news when context fetch fails" + finally: + os.environ.pop("SIGNAL_LOG_PATH", None) + os.environ.pop("SIGNAL_LOG_ENABLED", None) + slog.reset_loggers_for_tests() + if os.path.exists(path): + os.remove(path) + + +class _NoNews: + def get_fundamental_context(self): + return FundamentalContext() + + +class _BoomNews: + def get_fundamental_context(self): + raise RuntimeError("news engine down") + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════