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
@@ -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.
|
||||
|
||||
+39
-1
@@ -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(
|
||||
|
||||
@@ -48,5 +48,9 @@
|
||||
"max_panels_per_day": 20,
|
||||
"reuse_ttl_seconds": 3600
|
||||
},
|
||||
"signal_log": {
|
||||
"enabled": true,
|
||||
"path": "logs/signals.db"
|
||||
},
|
||||
"log_level": "INFO"
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
+2
-1
@@ -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 "
|
||||
|
||||
@@ -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
|
||||
|
||||
+34
-1
@@ -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:
|
||||
|
||||
+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()
|
||||
+1
-1
@@ -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(
|
||||
|
||||
+287
@@ -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
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
|
||||
Reference in New Issue
Block a user