ForexBot2/signal_log.py

"""
signal_log.py — SQLite signal-history logger
============================================

Persists every live scan's SignalReport (signal, grade, price at scan
time, news at scan time) to a SQLite database so the full decision
history can be queried later.

Hooked ONLY into SignalAggregator.analyze_pair (the live entry point);
the backtester calls analyze_from_dataframes directly and therefore
never writes here (live == backtest parity — CLAUDE.md rule 4).

Panel fields arrive POST-HOC: the row is inserted during analyze_pair
(before consumers attach the advisory panel), so log_signal stashes the
rowid on the report and agent_panel.attach_panel calls
log_panel_update() to fill the verdict, takes, creation/cache provenance,
payload hash, and advisory-outcome seed on that same row. Later closed-candle
snapshots advance the hypothetical next-open/trailing outcome. A report
without a rowid (backtest replay, logging disabled) is a silent no-op.

Deployment settings come from the config.json "signal_log" block via
app_config (env vars SIGNAL_LOG_ENABLED / SIGNAL_LOG_PATH win):

    "signal_log": {"enabled": true, "path": "logs/signals.db"}

Fail-safe contract: logging must NEVER break a scan. Every public entry
point catches all exceptions and downgrades them to a warning log line.
Report/context objects are duck-typed (no imports from signal_aggregator
or news_engine) so this module stays at the bottom of the import graph.
"""

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
from advisory_outcome import (
    advance_outcome,
    canonical_r,
    initial_outcome,
    timeframe_value,
)

logger = logging.getLogger(__name__)

_SCHEMA = """
CREATE TABLE IF NOT EXISTS signals (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    logged_at TEXT NOT NULL,
    source TEXT NOT NULL,
    symbol TEXT NOT NULL,
    timeframe TEXT NOT NULL,
    signal TEXT NOT NULL,
    score REAL,
    confidence REAL,
    quality TEXT,
    action TEXT,
    price REAL,
    spread REAL,
    market_regime TEXT,
    volatility_regime TEXT,
    mtf_json TEXT,
    setup_json TEXT,
    reasons_json TEXT,
    warnings_json TEXT,
    event_risk TEXT,
    event_description TEXT,
    news_sentiment REAL,
    news_description TEXT,
    news_summary TEXT,
    events_json TEXT,
    headlines_json TEXT,
    indicators_json TEXT,
    candles_json TEXT,
    panel_verdict TEXT,
    panel_confidence REAL,
    panel_rationale TEXT,
    panel_takes_json TEXT,
    panel_created_at TEXT,
    panel_cache_status TEXT,
    panel_payload_hash TEXT,
    advisory_outcome_status TEXT,
    advisory_r_multiple REAL,
    advisory_outcome_json TEXT
);
CREATE INDEX IF NOT EXISTS idx_signals_sym_tf_ts
    ON signals (symbol, timeframe, logged_at);
CREATE TABLE IF NOT EXISTS outlooks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    logged_at TEXT NOT NULL,
    surface TEXT NOT NULL,
    bias TEXT,
    setup_forming TEXT,
    confidence REAL,
    watch TEXT,
    note TEXT,
    news_json TEXT,
    indicators_json TEXT,
    candles_json TEXT
);
"""

# Additive migration for columns introduced after the original schema.
# Idempotent — skips columns already present.
_ADDITIVE_SIGNAL_COLS = (
    ("panel_verdict", "TEXT"),
    ("panel_confidence", "REAL"),
    ("panel_rationale", "TEXT"),
    ("panel_takes_json", "TEXT"),
    ("indicators_json", "TEXT"),
    ("candles_json", "TEXT"),
    ("panel_created_at", "TEXT"),
    ("panel_cache_status", "TEXT"),
    ("panel_payload_hash", "TEXT"),
    ("advisory_outcome_status", "TEXT"),
    ("advisory_r_multiple", "REAL"),
    ("advisory_outcome_json", "TEXT"),
)

_ADDITIVE_OUTLOOK_COLS = (
    ("news_json", "TEXT"),
    ("indicators_json", "TEXT"),
    ("candles_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,
    indicators_json, candles_json,
    panel_verdict, panel_confidence, panel_rationale, panel_takes_json
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""

_OUTLOOK_INSERT = """
INSERT INTO outlooks (
    logged_at, surface, bias, setup_forming, confidence, watch, note,
    news_json, indicators_json, candles_json
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""


def _finite(x: Any) -> Optional[float]:
    """float(x) or None for non-numeric / non-finite values (NaN/inf —
    SQLite has no NaN and strict JSON rejects it)."""
    try:
        f = float(x)
    except (TypeError, ValueError):
        return None
    return f if math.isfinite(f) else None


def _iso(ts: Any) -> Optional[str]:
    if ts is None:
        return None
    try:
        if ts.tzinfo is None:
            ts = ts.replace(tzinfo=timezone.utc)
        return ts.astimezone(timezone.utc).isoformat()
    except (AttributeError, ValueError):
        return str(ts)


def _dumps(obj: Any) -> str:
    return json.dumps(obj, default=str, allow_nan=False)


# ── Serializers (duck-typed; deliberately NOT imported from            ──
# ── dashboard_publisher — the dashboard's hard independence rule)      ──

def _event_to_dict(ev: Any) -> Dict[str, Any]:
    try:
        surprise = ev.surprise
    except Exception:
        surprise = None
    return {
        "time": _iso(getattr(ev, "time", None)),
        "currency": getattr(ev, "currency", None),
        "impact": getattr(ev, "impact", None),
        "event": getattr(ev, "event", None),
        "forecast": getattr(ev, "forecast", None),
        "previous": getattr(ev, "previous", None),
        "actual": getattr(ev, "actual", None),
        "surprise": surprise,
    }


def _headline_to_dict(h: Any) -> Dict[str, Any]:
    return {
        "headline": getattr(h, "headline", None),
        "source": getattr(h, "source", None),
        "timestamp": _iso(getattr(h, "timestamp", None)),
        "score": _finite(getattr(h, "sentiment_score", None)),
        "score_by_currency": {
            str(currency): _finite(score)
            for currency, score in (
                getattr(h, "sentiment_by_currency", None) or {}).items()
        },
        "currencies": list(getattr(h, "currencies_mentioned", None) or []),
        "keywords": list(getattr(h, "keywords", None) or []),
        "link": getattr(h, "link", None),
        "description": getattr(h, "description", None),
        "publisher": getattr(h, "publisher", None),
    }


def _context_news_json(context: Any) -> "tuple[Optional[str], Optional[str]]":
    """(events_json, headlines_json) from a FundamentalContext, or
    (None, None) when no context was captured."""
    if context is None:
        return None, None
    events = {
        "high_impact_warning": bool(
            getattr(context, "high_impact_warning", False)),
        "warning_message": getattr(context, "warning_message", "") or "",
        "calendar_available": bool(
            getattr(context, "calendar_available", False)),
        "sentiment_by_currency": dict(
            getattr(context, "news_sentiment", None) or {}),
        "events_today": [
            _event_to_dict(e)
            for e in getattr(context, "events_today", None) or []
        ],
        "events_upcoming_1h": [
            _event_to_dict(e)
            for e in getattr(context, "events_upcoming_1h", None) or []
        ],
    }
    headlines = [
        _headline_to_dict(h)
        for h in getattr(context, "news_headlines", None) or []
    ]
    return _dumps(events), _dumps(headlines)


def _key_str(k: Any) -> str:
    return getattr(k, "value", str(k))


def _analysis_to_dict(analysis: Any) -> Optional[Dict[str, Any]]:
    """Compact, JSON-safe snapshot of one TechnicalAnalysis."""
    if analysis is None:
        return None
    indicators = {}
    for name, indicator in (
            getattr(analysis, "indicators", None) or {}).items():
        if str(name).startswith("_"):
            continue
        signal = getattr(indicator, "signal", None)
        indicators[str(name)] = {
            "signal": getattr(signal, "name", str(signal)),
            "value": _finite(getattr(indicator, "value", None)),
            "confidence": _finite(getattr(indicator, "confidence", None)),
            "weight": _finite(getattr(indicator, "weight", None)),
            "description": getattr(indicator, "description", "") or "",
        }
    composite = getattr(analysis, "composite_signal", None)
    return {
        "timestamp": str(getattr(analysis, "timestamp", "")),
        "composite_signal": getattr(composite, "name", str(composite)),
        "composite_score": _finite(
            getattr(analysis, "composite_score", None)),
        "confidence": _finite(getattr(analysis, "confidence", None)),
        "market_regime": getattr(analysis, "market_regime", None),
        "volatility_regime": getattr(analysis, "volatility_regime", None),
        "key_levels": {
            str(name): _finite(value)
            for name, value in (
                getattr(analysis, "key_levels", None) or {}).items()
        },
        "pattern_signals": list(
            getattr(analysis, "pattern_signals", None) or []),
        "indicators": indicators,
    }


def _report_indicators_json(report: Any) -> Optional[str]:
    analyses = [getattr(report, "primary_analysis", None)]
    analyses.extend(getattr(report, "confirmation_analyses", None) or [])
    snapshot = {}
    for analysis in analyses:
        if analysis is None:
            continue
        timeframe = str(getattr(analysis, "timeframe", ""))
        if timeframe:
            snapshot[timeframe] = _analysis_to_dict(analysis)
    return _dumps(snapshot) if snapshot else None


def _candles_json(dataframes: Any, limit: int = 14) -> Optional[str]:
    """Last closed OHLCV bars per fetched timeframe, oldest first."""
    snapshot = {}
    for timeframe, df in (dataframes or {}).items():
        if df is None or getattr(df, "empty", True):
            continue
        bars = []
        for timestamp, row in df.tail(limit).iterrows():
            bars.append({
                "timestamp": _iso(timestamp),
                "open": _finite(row.get("Open")),
                "high": _finite(row.get("High")),
                "low": _finite(row.get("Low")),
                "close": _finite(row.get("Close")),
                "volume": _finite(row.get("Volume")),
            })
        if bars:
            snapshot[_key_str(timeframe)] = bars
    return _dumps(snapshot) if snapshot else None


def _report_row(
    report: Any,
    context: Any,
    source: str,
    timeframe: str,
    dataframes: Any = None,
) -> 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),
            "setup_pattern": getattr(setup, "setup_pattern", None),
            "round_trip_cost": _finite(
                getattr(setup, "round_trip_cost", None)),
            "trail_arm_r": _finite(getattr(setup, "trail_arm_r", None)),
            "trail_dist_r": _finite(getattr(setup, "trail_dist_r", None)),
        })

    reasons_json = _dumps({
        "bullish": list(getattr(report, "bullish_reasons", None) or []),
        "bearish": list(getattr(report, "bearish_reasons", None) or []),
    })
    warnings_json = _dumps(list(getattr(report, "warnings", None) or []))
    events_json, headlines_json = _context_news_json(context)

    signal = getattr(report, "final_signal", None)
    panel_takes = getattr(report, "panel_takes", None) or []
    panel_takes_json = _dumps(panel_takes) if panel_takes else None
    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,
        _report_indicators_json(report),
        _candles_json(dataframes),
        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_signal_columns()
            self._migrate_outlook_columns()

    @property
    def path(self) -> str:
        return self._path

    def _migrate_signal_columns(self) -> None:
        """Add newer signal columns to pre-existing tables (idempotent)."""
        existing = {r[1] for r in self._conn.execute("PRAGMA table_info(signals)")}
        for name, col_type in _ADDITIVE_SIGNAL_COLS:
            if name not in existing:
                try:
                    self._conn.execute(
                        f"ALTER TABLE signals ADD COLUMN {name} {col_type}")
                except sqlite3.OperationalError as exc:
                    # Another dashboard/bot process may have migrated the
                    # shared database after our PRAGMA snapshot.
                    if "duplicate column name" not in str(exc).lower():
                        raise
        self._conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_signals_advisory_pending"
            " ON signals (symbol, advisory_outcome_status, id)")
        self._conn.commit()

    def _migrate_outlook_columns(self) -> None:
        """Add snapshot columns to pre-existing outlook tables."""
        existing = {
            r[1] for r in self._conn.execute("PRAGMA table_info(outlooks)")
        }
        for name, col_type in _ADDITIVE_OUTLOOK_COLS:
            if name not in existing:
                try:
                    self._conn.execute(
                        f"ALTER TABLE outlooks ADD COLUMN {name} {col_type}")
                except sqlite3.OperationalError as exc:
                    if "duplicate column name" not in str(exc).lower():
                        raise
        self._conn.commit()

    def log(
        self,
        report: Any,
        context: Any = None,
        source: str = "live",
        timeframe: str = "",
        dataframes: Any = None,
    ) -> int:
        """Insert one scan row; returns the rowid (panel-update key)."""
        row = _report_row(report, context, source, timeframe, dataframes)
        with self._lock:
            cur = self._conn.execute(_INSERT, row)
            self._conn.commit()
        rowid = int(cur.lastrowid)
        try:
            self.resolve_pending(str(row[2]), row[25])
        except Exception:
            logger.warning("signal_log: advisory outcome resolver failed",
                           exc_info=True)
        return rowid

    def update_panel(
        self,
        rowid: int,
        verdict: str,
        confidence: Optional[float],
        rationale: str,
        takes_json: str,
        *,
        created_at: Optional[str] = None,
        cache_status: Optional[str] = None,
        payload_hash: Optional[str] = None,
    ) -> None:
        """Fill panel provenance and initialize its advisory outcome."""
        with self._lock:
            row = self._conn.execute(
                "SELECT setup_json, indicators_json, timeframe,"
                " advisory_outcome_status FROM signals WHERE id=?",
                (rowid,),
            ).fetchone()
            if row is None:
                return
            advisory_status = row[3]
            advisory_json = None
            if advisory_status is None:
                setup = json.loads(row[0]) if row[0] else None
                indicators = json.loads(row[1]) if row[1] else None
                advisory_status, state = initial_outcome(
                    setup, indicators, row[2])
                advisory_json = _dumps(state)
            self._conn.execute(
                "UPDATE signals SET panel_verdict=?, panel_confidence=?,"
                " panel_rationale=?, panel_takes_json=?,"
                " panel_created_at=COALESCE(?, panel_created_at),"
                " panel_cache_status=COALESCE(?, panel_cache_status),"
                " panel_payload_hash=COALESCE(?, panel_payload_hash),"
                " advisory_outcome_status=COALESCE(advisory_outcome_status, ?),"
                " advisory_outcome_json=COALESCE(advisory_outcome_json, ?)"
                " WHERE id=?",
                (verdict, confidence, rationale, takes_json,
                 created_at, cache_status, payload_hash,
                 advisory_status, advisory_json, rowid))
            self._conn.commit()

    def resolve_pending(
        self, symbol: str, candles_json: Optional[str], limit: int = 100
    ) -> int:
        """Advance pending advisory outcomes from a new candle snapshot."""
        if not candles_json:
            return 0
        snapshots = json.loads(candles_json)
        if not isinstance(snapshots, dict):
            return 0
        changed = 0
        with self._lock:
            self._conn.execute("BEGIN IMMEDIATE")
            try:
                rows = self._conn.execute(
                    "SELECT id, timeframe, advisory_outcome_json FROM signals"
                    " WHERE symbol=? AND advisory_outcome_status='PENDING'"
                    " ORDER BY id",
                    (symbol,),
                ).fetchall()
                eligible = 0
                for rowid, timeframe, state_json in rows:
                    bars = timeframe_value(snapshots, timeframe)
                    if not bars or not state_json:
                        continue
                    if eligible >= max(1, int(limit)):
                        break
                    eligible += 1
                    state = json.loads(state_json)
                    before = _dumps(state)
                    state = advance_outcome(state, bars)
                    after = _dumps(state)
                    if after == before:
                        continue
                    status = state.get("status", "UNRESOLVABLE")
                    self._conn.execute(
                        "UPDATE signals SET advisory_outcome_status=?,"
                        " advisory_r_multiple=?, advisory_outcome_json=?"
                        " WHERE id=? AND advisory_outcome_status='PENDING'",
                        (status, canonical_r(state), after, rowid),
                    )
                    changed += 1
                self._conn.commit()
            except Exception:
                self._conn.rollback()
                raise
        return changed

    def log_outlook(self, outlook: Any, surface: str,
                    payload: Any = None) -> int:
        """Insert one advisory market outlook; returns its rowid."""
        payload = payload if isinstance(payload, dict) else {}
        technical = {
            key: payload.get(key)
            for key in ("indicators", "market_regime",
                        "volatility_regime", "key_levels")
            if key in payload
        }
        row = (
            datetime.now(timezone.utc).isoformat(),
            surface,
            getattr(outlook, "bias", None),
            getattr(outlook, "setup_forming", None),
            _finite(getattr(outlook, "confidence", None)),
            getattr(outlook, "watch", None),
            getattr(outlook, "note", None),
            _dumps(payload.get("news")) if "news" in payload else None,
            _dumps(technical) if technical else None,
            _dumps(payload.get("candles")) if "candles" in payload else None,
        )
        with self._lock:
            cur = self._conn.execute(_OUTLOOK_INSERT, row)
            self._conn.commit()
        return int(cur.lastrowid)

    def close(self) -> None:
        with self._lock:
            self._conn.close()


# ── Module-level entry point (config-driven, fail-safe) ───────────────

_loggers: Dict[str, SignalLogger] = {}
_loggers_lock = threading.Lock()


def log_signal(
    report: Any,
    context: Any = None,
    source: str = "live",
    timeframe: str = "",
    dataframes: Any = None,
) -> 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, dataframes=dataframes)
        try:
            report._signal_log_rowid = rowid
        except Exception:
            pass  # exotic report object — panel update just won't happen
    except Exception:
        logger.warning("signal_log: failed to persist signal",
                       exc_info=True)


def log_panel_update(report: Any) -> None:
    """Write post-hoc panel fields onto the row logged for this scan.

    Matches on the private rowid hand-off from log_signal; a report that
    was never logged (backtest replay, logging disabled, logger in
    another process) is a silent no-op. NEVER raises."""
    try:
        rowid = getattr(report, "_signal_log_rowid", None)
        if rowid is None or not app_config.signal_log_enabled():
            return
        verdict = getattr(report, "panel_verdict", "") or ""
        if not verdict:
            return
        path = app_config.signal_log_path()
        with _loggers_lock:
            lg = _loggers.get(path)
        if lg is None:
            return
        lg.update_panel(
            int(rowid), verdict,
            _finite(getattr(report, "panel_confidence", None)),
            getattr(report, "panel_rationale", "") or "",
            _dumps(getattr(report, "panel_takes", None) or []),
            created_at=getattr(report, "panel_created_at", "") or None,
            cache_status=getattr(report, "panel_cache_status", "") or None,
            payload_hash=getattr(report, "panel_payload_hash", "") or None)
    except Exception:
        logger.warning("signal_log: failed to persist panel verdict",
                       exc_info=True)


def log_outlook(outlook: Any, surface: str = "live",
                payload: Any = None) -> None:
    """Persist one advisory outlook when signal logging is enabled.

    Outlook persistence is deliberately independent of a signal row: the
    feature runs on its own hourly cadence and must never affect a scan.
    """
    try:
        if not app_config.signal_log_enabled():
            return
        path = app_config.signal_log_path()
        with _loggers_lock:
            lg = _loggers.get(path)
            if lg is None:
                lg = SignalLogger(path)
                _loggers[path] = lg
        lg.log_outlook(outlook, surface, payload=payload)
    except Exception:
        logger.warning("signal_log: failed to persist market outlook",
                       exc_info=True)


def reset_loggers_for_tests() -> None:
    """Close and drop all cached loggers (test isolation)."""
    with _loggers_lock:
        for lg in _loggers.values():
            try:
                lg.close()
            except Exception:
                pass
        _loggers.clear()