Add standard signal snapshots

Pigbibi · Pigbibi · commit 697c5d9d51b1 · 2026-05-28T04:50:36.000+08:00
diff --git a/application/rebalance_service.py b/application/rebalance_service.py
@@ -13,6 +13,7 @@
     build_reconciliation_record,
     write_reconciliation_record,
 )
+from application.signal_snapshot import build_signal_snapshot
 from notifications.events import NotificationPublisher
 from notifications import renderers as notification_renderers
 from quant_platform_kit.common.models import PortfolioSnapshot, Position
@@ -581,6 +582,18 @@ def run_strategy_core(
             signal_metadata = {}
         allocation = _resolve_weight_allocation(signal_metadata, required=target_weights is not None)
         resolved_target_weights = dict(allocation.get("targets") or {}) if target_weights is not None else None
+        signal_metadata = dict(signal_metadata or {})
+        signal_metadata["signal_snapshot"] = build_signal_snapshot(
+            platform="ibkr",
+            strategy_profile=signal_metadata.get("strategy_profile"),
+            metadata={
+                **signal_metadata,
+                "latest_price_source": signal_metadata.get("price_source_mode")
+                or "ibkr_strategy_market_data",
+            },
+            allocation={**allocation, "target_mode": "weight"},
+            target_weights=resolved_target_weights,
+        )
 
         dashboard = notification_renderers.build_dashboard(
             positions,
diff --git a/application/signal_snapshot.py b/application/signal_snapshot.py
@@ -0,0 +1,166 @@
+"""Shared signal snapshot payload helpers for platform reports."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from datetime import date, datetime, timezone
+from typing import Any
+
+SIGNAL_SNAPSHOT_SCHEMA_VERSION = "signal_snapshot.v1"
+
+_INDICATOR_FIELDS = (
+    "benchmark_symbol",
+    "benchmark_price",
+    "long_trend_value",
+    "exit_line",
+    "active_risk_asset",
+    "allocation_mode",
+    "trend_symbol",
+    "trend_price",
+    "trend_ma",
+    "trend_ma20",
+    "trend_ma20_slope",
+    "trend_rsi14",
+    "trend_rsi14_dynamic_threshold",
+    "trend_rsi14_effective_threshold",
+    "trend_bb_upper",
+    "blend_gate_volatility_delever_metric",
+    "blend_gate_volatility_delever_triggered",
+)
+
+
+def _utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _json_safe(value: Any) -> Any:
+    if isinstance(value, datetime):
+        return value.isoformat()
+    if isinstance(value, date):
+        return value.isoformat()
+    if isinstance(value, Mapping):
+        return {str(key): _json_safe(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple, set)):
+        return [_json_safe(item) for item in value]
+    return value
+
+
+def _first_value(*values: Any) -> Any:
+    for value in values:
+        if value is not None and value != "":
+            return value
+    return None
+
+
+def _merge_signal_sources(*sources: Mapping[str, Any] | None) -> dict[str, Any]:
+    merged: dict[str, Any] = {}
+    for source in sources:
+        if not isinstance(source, Mapping):
+            continue
+        annotations = source.get("execution_annotations")
+        if isinstance(annotations, Mapping):
+            merged.update(annotations)
+        merged.update(source)
+    return merged
+
+
+def _normalized_numeric_mapping(value: Any) -> dict[str, float]:
+    if not isinstance(value, Mapping):
+        return {}
+    normalized: dict[str, float] = {}
+    for key, raw_value in value.items():
+        symbol = str(key or "").strip().upper()
+        if not symbol:
+            continue
+        try:
+            normalized[symbol] = float(raw_value)
+        except (TypeError, ValueError):
+            continue
+    return normalized
+
+
+def _target_payload(
+    *,
+    allocation: Mapping[str, Any] | None,
+    explicit_target_weights: Mapping[str, Any] | None,
+) -> tuple[str | None, dict[str, float], dict[str, float]]:
+    allocation = allocation if isinstance(allocation, Mapping) else {}
+    target_mode = str(allocation.get("target_mode") or "").strip() or None
+    targets = _normalized_numeric_mapping(explicit_target_weights or allocation.get("targets"))
+    if target_mode == "value":
+        return target_mode, {}, targets
+    return target_mode, targets, {}
+
+
+def build_signal_snapshot(
+    *,
+    platform: str,
+    strategy_profile: str | None = None,
+    generated_at: datetime | None = None,
+    diagnostics: Mapping[str, Any] | None = None,
+    execution: Mapping[str, Any] | None = None,
+    allocation: Mapping[str, Any] | None = None,
+    metadata: Mapping[str, Any] | None = None,
+    target_weights: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    source = _merge_signal_sources(metadata, diagnostics, execution)
+    target_mode, normalized_weights, normalized_values = _target_payload(
+        allocation=allocation,
+        explicit_target_weights=target_weights,
+    )
+    indicators = {
+        field: _json_safe(source[field])
+        for field in _INDICATOR_FIELDS
+        if source.get(field) not in (None, "")
+    }
+    snapshot = {
+        "schema_version": SIGNAL_SNAPSHOT_SCHEMA_VERSION,
+        "platform": str(platform or "").strip(),
+        "strategy_profile": _first_value(strategy_profile, source.get("strategy_profile")),
+        "strategy_version": source.get("strategy_version"),
+        "generated_at": _json_safe(generated_at or _utcnow()),
+        "signal_as_of": _json_safe(
+            _first_value(
+                source.get("signal_as_of"),
+                source.get("signal_date"),
+                source.get("snapshot_as_of"),
+                source.get("trade_date"),
+            )
+        ),
+        "market_date": _json_safe(
+            _first_value(
+                source.get("market_date"),
+                source.get("signal_date"),
+                source.get("snapshot_as_of"),
+                source.get("trade_date"),
+            )
+        ),
+        "effective_date": _json_safe(source.get("effective_date")),
+        "latest_price_source": _first_value(
+            source.get("latest_price_source"),
+            source.get("price_source_mode"),
+            source.get("market_data_source"),
+            source.get("signal_source"),
+        ),
+        "quote_overlay_used": source.get("quote_overlay_used"),
+        "data_freshness_warning": _first_value(
+            source.get("data_freshness_warning"),
+            source.get("snapshot_price_fallback_used"),
+        ),
+        "signal": _first_value(
+            source.get("signal_display"),
+            source.get("signal_description"),
+            source.get("signal_message"),
+        ),
+        "status": _first_value(
+            source.get("status_display"),
+            source.get("status_description"),
+            source.get("market_status"),
+            source.get("canary_status"),
+        ),
+        "target_mode": target_mode,
+        "target_weights": normalized_weights,
+        "target_values": normalized_values,
+        "indicators": indicators,
+    }
+    return {key: _json_safe(value) for key, value in snapshot.items()}
diff --git a/main.py b/main.py
@@ -22,6 +22,7 @@
 from application.runtime_composer import build_runtime_composer
 from application.runtime_strategy_adapters import build_runtime_strategy_adapters
 from application.rebalance_service import run_strategy_core as run_rebalance_cycle
+from application.signal_snapshot import build_signal_snapshot
 from decision_mapper import map_strategy_decision
 from entrypoints.cloud_run import is_market_open_today
 from notifications.telegram import build_strategy_display_name, build_translator, send_telegram_message
@@ -530,6 +531,22 @@ def build_strategy_plugin_alert_context_label() -> str:
     )
 
 
+def _has_signal_snapshot_details(snapshot: dict[str, object]) -> bool:
+    return any(
+        snapshot.get(field_name)
+        for field_name in (
+            "signal_as_of",
+            "market_date",
+            "latest_price_source",
+            "target_weights",
+            "target_values",
+            "indicators",
+            "signal",
+            "status",
+        )
+    )
+
+
 def publish_strategy_plugin_alerts(signals, *, report=None):
     result = dispatch_strategy_plugin_alerts(
         signals,
@@ -688,6 +705,35 @@ def _handle_request(*, dry_run_only_override: bool | None = None, response_body:
         )
         execution_summary = dict(cycle_result.execution_summary or {})
         reconciliation_record = dict(cycle_result.reconciliation_record or {})
+        signal_metadata = dict(cycle_result.signal_metadata or {})
+        signal_snapshot = dict(signal_metadata.get("signal_snapshot") or {})
+        if not signal_snapshot:
+            signal_snapshot = build_signal_snapshot(
+                platform="ibkr",
+                strategy_profile=signal_metadata.get("strategy_profile") or STRATEGY_PROFILE,
+                metadata=signal_metadata,
+                target_weights=cycle_result.target_weights,
+            )
+        if execution_summary.get("price_source_mode") or reconciliation_record.get("price_source_mode"):
+            signal_snapshot["latest_price_source"] = (
+                execution_summary.get("price_source_mode")
+                or reconciliation_record.get("price_source_mode")
+            )
+        fallback_used = bool(
+            execution_summary.get("snapshot_price_fallback_used")
+            or reconciliation_record.get("snapshot_price_fallback_used")
+        )
+        if fallback_used:
+            signal_snapshot["data_freshness_warning"] = "snapshot_price_fallback_used"
+        has_signal_snapshot = _has_signal_snapshot_details(signal_snapshot)
+        if has_signal_snapshot:
+            log_runtime_event(
+                log_context,
+                "strategy_signal_snapshot",
+                message="Strategy signal snapshot",
+                execution_window="precheck" if dry_run_only_override else "execution",
+                **signal_snapshot,
+            )
         finalize_runtime_report(
             report,
             status="ok",
@@ -721,6 +767,7 @@ def _handle_request(*, dry_run_only_override: bool | None = None, response_body:
                 "snapshot_price_fallback_symbols": execution_summary.get("snapshot_price_fallback_symbols")
                 or reconciliation_record.get("snapshot_price_fallback_symbols")
                 or [],
+                **({"signal_snapshot": signal_snapshot} if has_signal_snapshot else {}),
             },
             artifacts={
                 "reconciliation_record_path": cycle_result.reconciliation_record_path,
diff --git a/notifications/renderers.py b/notifications/renderers.py
@@ -300,6 +300,36 @@ def _build_timing_audit_lines(signal_metadata, *, translator) -> list[str]:
     return [f"{label}: {value}"]
 
 
+def _format_signal_snapshot_line(snapshot, *, translator) -> str:
+    if not isinstance(snapshot, Mapping):
+        return ""
+    market_date = str(snapshot.get("market_date") or snapshot.get("signal_as_of") or "").strip()
+    source = str(snapshot.get("latest_price_source") or "").strip()
+    overlay = snapshot.get("quote_overlay_used")
+    warning = snapshot.get("data_freshness_warning")
+    if not market_date and not source and overlay is None and warning in (None, "", False):
+        return ""
+    if _translator_uses_zh(translator):
+        overlay_text = "是" if overlay is True else "否" if overlay is False else "未知"
+        parts = [
+            f"日期 {market_date or '未知'}",
+            f"数据源 {source or '未知'}",
+            f"报价覆盖 {overlay_text}",
+        ]
+        if warning not in (None, "", False):
+            parts.append(f"提示 {warning}")
+        return "🧾 信号快照: " + " | ".join(parts)
+    overlay_text = "yes" if overlay is True else "no" if overlay is False else "unknown"
+    parts = [
+        f"date {market_date or 'unknown'}",
+        f"source {source or 'unknown'}",
+        f"quote overlay {overlay_text}",
+    ]
+    if warning not in (None, "", False):
+        parts.append(f"warning {warning}")
+    return "🧾 Signal snapshot: " + " | ".join(parts)
+
+
 def _strategy_dashboard_text(signal_metadata, *, translator) -> str:
     metadata = signal_metadata if isinstance(signal_metadata, Mapping) else {}
     raw_annotations = metadata.get("execution_annotations")
@@ -311,11 +341,13 @@ def _strategy_dashboard_text(signal_metadata, *, translator) -> str:
         or ""
     )
     timing_lines = _build_timing_audit_lines(metadata, translator=translator)
-    if not timing_lines:
+    snapshot_line = _format_signal_snapshot_line(metadata.get("signal_snapshot"), translator=translator)
+    audit_lines = [*timing_lines, *([snapshot_line] if snapshot_line else [])]
+    if not audit_lines:
         return dashboard_text
     if not dashboard_text:
-        return "\n".join(timing_lines)
-    return f"{dashboard_text}\n" + "\n".join(timing_lines)
+        return "\n".join(audit_lines)
+    return f"{dashboard_text}\n" + "\n".join(audit_lines)
 
 
 def build_dashboard(
