signal_log: persist market snapshots

CLAUDE.md

@@ -59,10 +59,10 @@ 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 arrive post-hoc: `log_signal` stashes the rowid on the report and `agent_panel.attach_panel` writes the verdict back onto the same row via `log_panel_update` (no rowid — backtest replay or logging disabled — is a silent no-op).
+- **`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, full per-TF indicator snapshots, the last 14 closed OHLCV candles per fetched TF, 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 arrive post-hoc: `log_signal` stashes the rowid on the report and `agent_panel.attach_panel` writes the verdict back onto the same row via `log_panel_update` (no rowid — backtest replay or logging disabled — is a silent no-op).
 - **`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. The technical persona's payload is enriched (2026-08-15): raw indicator values plus ATR-normalized recent candles per TF (`candles_enabled`/`candle_bars_primary` 20/`candle_bars_confirm` 10 config keys; code does the arithmetic — raw prices are the model's weakest input, per `backtest/research/llm_news_eval_2026-08-06.md`). Prompts forbid training-data/outside knowledge and cap notes at 15 words / rationales at 30 to bound output tokens.
-- **`market_outlook.py`** — hourly DeepSeek market brief for XAU/USD, using ATR-normalized M15/H1/H4 candles, visible indicator states, levels, news, and immutable recent signal history. It returns a written **ADVISORY-ONLY** bias/setup-forming assessment, persists it independently in `outlooks` within `logs/signals.db`, and may be sent post-scan by Telegram or requested with `main.py --outlook`; it never imports or runs inside the signal path. A lock enforces hourly dedup plus its own UTC-day cap (default 12), and provider/news failures leave a safe partial brief while LLM/database failures degrade to no outlook.
+- **`market_outlook.py`** — hourly DeepSeek market brief for XAU/USD, using ATR-normalized M15/H1/H4 candles, visible indicator states, levels, news, and immutable recent signal history. It returns a written **ADVISORY-ONLY** bias/setup-forming assessment and persists it independently in `outlooks` within `logs/signals.db` together with the news, indicator/regime/level, and candle evidence sent to the model. It may be sent post-scan by Telegram or requested with `main.py --outlook`; it never imports or runs inside the signal path. A lock enforces hourly dedup plus its own UTC-day cap (default 12), and provider/news failures leave a safe partial brief while LLM/database failures degrade to no outlook.
 - **`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.
 - **`oanda_daemon.py` / `oanda_executor.py`** — practice-account auto-execution. Book-parity rules (see daemon docstring): no broker-side TP (trailing model — TP1/TP2 advisory), 5-bar per-TF cooldown (wall-clock approximation, so it expires over weekends where the backtester would still count bars), trail ratchets at bar close once per *observed* bar (a missed/failed cycle skips those bars entirely — the intra-gap high/low never enters the peak). Accepted divergence: one position total; orders send `positionFill=OPEN_ONLY` so a state/broker divergence cancels the order instead of silently netting. State/broker re-sync runs at boot and after any suspicious fill (`_reconcile_positions`).
 - **`dashboard_publisher.py` / `static/dashboard.html`** — web dashboard as a **static publisher** (live LAMP server): standalone daemon scans at candle closes via `produce_signal` and atomically writes `snapshot.json`/`news.json`/`dashboard.html` to `$DASHBOARD_OUT_DIR` (Apache serves them; page is public by owner decision). **Hard independence rule: never imports, and is never imported by, `telegram_bot`/`oanda_daemon`** — confirmation map and boundary math are deliberate local copies, so a web-side failure can only mean a stale page. Due-TF classification uses the RAW wake time (flooring wake−buffer skips every H1/H4 scan — 2026-08-02 bug). News panel reads `NewsEngine.get_fundamental_context()` plus the additive `get_week_events()` accessor (display-only; no signal path calls it); as a separate process it has its own news caches and DeepSeek daily cap. News/calendar rows are clickable (detail dialog); headlines carry `id` = sha1(headline) in news.json, and exactly one on-demand endpoint exists: `static/news_summary.php` (installed by `install_static`), which serves a DeepSeek gold-impact summary per headline — whitelist-gated against news.json ids, cache-forever in `ai/`, own fail-closed daily cap (150/day, file-based, separate from the publisher's), key at `/etc/gold-dashboard/deepseek.key`, never reads repo config. Deployment checklist: `deploy/README-dashboard.md`.

market_outlook.py

@@ -377,7 +377,7 @@ def run_outlook(settings: Dict[str, Any],
             return None
         try:
             from signal_log import log_outlook
-            log_outlook(result, surface=surface)
+            log_outlook(result, surface=surface, payload=payload)
         except Exception:
             logger.debug("market outlook: persistence unavailable", exc_info=True)
         return result

signal_aggregator.py

@@ -267,11 +267,12 @@ class SignalAggregator:
                     f"{symbol}: live session filter errored — filter SKIPPED this scan"
                 )
 
-        self._log_signal(report, log_source)
+        self._log_signal(report, log_source, dfs)
         return report
 
     def _log_signal(
-        self, report: SignalReport, source: Optional[str]
+        self, report: SignalReport, source: Optional[str],
+        dataframes: Optional[Dict[Timeframe, pd.DataFrame]] = None,
     ) -> None:
         """Persist the scan to the signal-history DB (signal_log.py).
 
@@ -291,6 +292,7 @@ class SignalAggregator:
             context=context,
             source=source or "live",
             timeframe=self.config.primary_timeframe.value,
+            dataframes=dataframes,
         )
 
     def _apply_regime_filter_live(

signal_log.py

@@ -68,6 +68,8 @@ CREATE TABLE IF NOT EXISTS signals (
     news_summary TEXT,
     events_json TEXT,
     headlines_json TEXT,
+    indicators_json TEXT,
+    candles_json TEXT,
     panel_verdict TEXT,
     panel_confidence REAL,
     panel_rationale TEXT,
@@ -83,17 +85,28 @@ CREATE TABLE IF NOT EXISTS outlooks (
     setup_forming TEXT,
     confidence REAL,
     watch TEXT,
-    note TEXT
+    note TEXT,
+    news_json TEXT,
+    indicators_json TEXT,
+    candles_json TEXT
 );
 """
 
-# Additive migration: older DBs created before the panel columns existed
-# get them added here. Idempotent — skips columns already present.
-_PANEL_COLS = (
+# 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"),
+)
+
+_ADDITIVE_OUTLOOK_COLS = (
+    ("news_json", "TEXT"),
+    ("indicators_json", "TEXT"),
+    ("candles_json", "TEXT"),
 )
 
 _INSERT = """
@@ -103,14 +116,16 @@ INSERT INTO signals (
     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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
 """
 
 _OUTLOOK_INSERT = """
 INSERT INTO outlooks (
-    logged_at, surface, bias, setup_forming, confidence, watch, note
-) VALUES (?, ?, ?, ?, ?, ?, ?)
+    logged_at, surface, bias, setup_forming, confidence, watch, note,
+    news_json, indicators_json, candles_json
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
 """
 
 
@@ -165,8 +180,16 @@ def _headline_to_dict(h: Any) -> Dict[str, Any]:
         "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),
     }
 
 
@@ -203,11 +226,83 @@ 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({
@@ -271,6 +366,8 @@ def _report_row(
         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),
@@ -294,30 +391,43 @@ class SignalLogger:
         self._lock = threading.Lock()
         with self._lock:
             self._conn.executescript(_SCHEMA)
-            self._migrate_panel_columns()
+            self._migrate_signal_columns()
+            self._migrate_outlook_columns()
 
     @property
     def path(self) -> str:
         return self._path
 
-    def _migrate_panel_columns(self) -> None:
-        """Add panel_* columns to pre-existing tables (idempotent)."""
+    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 _PANEL_COLS:
+        for name, col_type in _ADDITIVE_SIGNAL_COLS:
             if name not in existing:
                 self._conn.execute(
                     f"ALTER TABLE signals ADD COLUMN {name} {col_type}")
         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:
+                self._conn.execute(
+                    f"ALTER TABLE outlooks ADD COLUMN {name} {col_type}")
+        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)
+        row = _report_row(report, context, source, timeframe, dataframes)
         with self._lock:
             cur = self._conn.execute(_INSERT, row)
             self._conn.commit()
@@ -339,8 +449,16 @@ class SignalLogger:
                 (verdict, confidence, rationale, takes_json, rowid))
             self._conn.commit()
 
-    def log_outlook(self, outlook: Any, surface: str) -> int:
+    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,
@@ -349,6 +467,9 @@ class SignalLogger:
             _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)
@@ -371,6 +492,7 @@ def log_signal(
     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
@@ -386,7 +508,7 @@ def log_signal(
                 lg = SignalLogger(path)
                 _loggers[path] = lg
         rowid = lg.log(report, context=context, source=source,
-                       timeframe=timeframe)
+                       timeframe=timeframe, dataframes=dataframes)
         try:
             report._signal_log_rowid = rowid
         except Exception:
@@ -424,7 +546,8 @@ def log_panel_update(report: Any) -> None:
                        exc_info=True)
 
 
-def log_outlook(outlook: Any, surface: str = "live") -> None:
+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
@@ -439,7 +562,7 @@ def log_outlook(outlook: Any, surface: str = "live") -> None:
             if lg is None:
                 lg = SignalLogger(path)
                 _loggers[path] = lg
-        lg.log_outlook(outlook, surface)
+        lg.log_outlook(outlook, surface, payload=payload)
     except Exception:
         logger.warning("signal_log: failed to persist market outlook",
                        exc_info=True)

test_all.py

@@ -7530,6 +7530,25 @@ def _make_log_report():
         reward_pips=8.8, pip_value=1.0, strategy="XAU_PB",
         trail_arm_r=0.70, trail_dist_r=0.85,
     )
+    report.primary_analysis = TechnicalAnalysis(
+        symbol="XAU/USD", timeframe=Timeframe.M5.value,
+        timestamp=report.timestamp,
+        indicators={
+            "rsi": IndicatorSignal(
+                "RSI", 62.5, SignalStrength.BUY, 70.0,
+                "Momentum is bullish", 1.2),
+            "atr": IndicatorSignal(
+                "ATR", 5.25, SignalStrength.NEUTRAL, 0.0,
+                "Volatility context", 0.0),
+        },
+        composite_signal=SignalStrength.BUY,
+        composite_score=42.5,
+        confidence=76.0,
+        market_regime="trending_up",
+        volatility_regime="normal",
+        key_levels={"support": 2440.0, "atr": 5.25},
+        pattern_signals=["bullish_engulfing"],
+    )
     return report
 
 
@@ -7547,7 +7566,10 @@ def _make_log_context():
         news_headlines=[NewsSentiment(
             headline="Fed signals rate hike", source="test",
             timestamp=now, currencies_mentioned=["USD", "XAU"],
-            sentiment_score=1.0, keywords=["rate hike"])],
+            sentiment_score=1.0, keywords=["rate hike"],
+            sentiment_by_currency={"USD": 1.0, "XAU": -1.0},
+            link="https://example.test/fed", description="Policy update",
+            publisher="Example News")],
         calendar_available=True,
         llm_summary="Gold supported by soft dollar.",
     )
@@ -7563,7 +7585,12 @@ def _():
         lg = slog.SignalLogger(path)
         report = _make_log_report()
         context = _make_log_context()
-        lg.log(report, context=context, source="test", timeframe="M5")
+        candles = {
+            Timeframe.M5: make_ohlcv(20, seed=91),
+            Timeframe.H1: make_ohlcv(10, seed=92),
+        }
+        lg.log(report, context=context, source="test", timeframe="M5",
+               dataframes=candles)
         lg.close()
         conn = sqlite3.connect(path)
         rows = conn.execute("SELECT * FROM signals").fetchall()
@@ -7600,7 +7627,21 @@ def _():
         headlines = json.loads(row["headlines_json"])
         assert headlines[0]["headline"] == "Fed signals rate hike"
         assert headlines[0]["score"] == 1.0
+        assert headlines[0]["score_by_currency"]["XAU"] == -1.0
         assert headlines[0]["currencies"] == ["USD", "XAU"]
+        assert headlines[0]["description"] == "Policy update"
+        indicators = json.loads(row["indicators_json"])
+        primary = indicators[Timeframe.M5.value]
+        assert primary["composite_score"] == 42.5
+        assert primary["indicators"]["rsi"]["signal"] == "BUY"
+        assert primary["indicators"]["atr"]["value"] == 5.25
+        candle_snapshot = json.loads(row["candles_json"])
+        assert len(candle_snapshot[Timeframe.M5.value]) == 14
+        assert len(candle_snapshot[Timeframe.H1.value]) == 10
+        bar = candle_snapshot[Timeframe.M5.value][0]
+        assert set(bar) == {"timestamp", "open", "high", "low", "close",
+                            "volume"}
+        assert bar["timestamp"].endswith("+00:00")
     finally:
         slog.reset_loggers_for_tests()
         os.remove(path)
@@ -7969,23 +8010,48 @@ def _():
         con.execute(
             "CREATE TABLE signals (id INTEGER PRIMARY KEY, logged_at TEXT,"
             " symbol TEXT, timeframe TEXT)")
+        con.execute(
+            "CREATE TABLE outlooks (id INTEGER PRIMARY KEY, logged_at TEXT,"
+            " surface TEXT, bias TEXT, setup_forming TEXT, confidence REAL,"
+            " watch TEXT, note TEXT)")
         con.commit()
         con.close()
         logger = slog.SignalLogger(path)
+        columns = {
+            row[1] for row in logger._conn.execute("PRAGMA table_info(signals)")
+        }
+        assert {"indicators_json", "candles_json"} <= columns
+        outlook_columns = {
+            row[1] for row in logger._conn.execute("PRAGMA table_info(outlooks)")
+        }
+        assert {"news_json", "indicators_json", "candles_json"} <= \
+            outlook_columns
         result = mo.OutlookResult(
             bias="bearish", setup_forming="sell", confidence=71.5,
             watch="Wait for price to fail at resistance.",
             note="Momentum is fading after a sharp rise.", created_ts=time.time())
-        rowid = logger.log_outlook(result, surface="test")
+        payload = {
+            "news": {"headlines": ["Gold firms"]},
+            "indicators": {"1h": {"rsi": {"value": 55.0}}},
+            "market_regime": {"1h": "trending_up"},
+            "volatility_regime": {"1h": "normal"},
+            "key_levels": {"1h": {"support": 4400.0}},
+            "candles": {"1h": {"bars": [[-0.2, 0.1, -0.3, 0.0]]}},
+        }
+        rowid = logger.log_outlook(result, surface="test", payload=payload)
         logger.close()
         con = sqlite3.connect(path)
         row = con.execute(
-            "SELECT surface, bias, setup_forming, confidence, watch, note "
+            "SELECT surface, bias, setup_forming, confidence, watch, note, "
+            "news_json, indicators_json, candles_json "
             "FROM outlooks WHERE id=?", (rowid,)).fetchone()
         con.close()
-        assert row == ("test", "bearish", "sell", 71.5,
-                       "Wait for price to fail at resistance.",
-                       "Momentum is fading after a sharp rise.")
+        assert row[:6] == ("test", "bearish", "sell", 71.5,
+                           "Wait for price to fail at resistance.",
+                           "Momentum is fading after a sharp rise.")
+        assert json.loads(row[6])["headlines"] == ["Gold firms"]
+        assert json.loads(row[7])["market_regime"]["1h"] == "trending_up"
+        assert len(json.loads(row[8])["1h"]["bars"]) == 1
     finally:
         slog.reset_loggers_for_tests()
         os.remove(path)