From 697c5d9d51b101d23d21809404ac83c83e168d96 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 28 May 2026 04:50:36 +0800 Subject: [PATCH] Add standard signal snapshots --- application/rebalance_service.py | 13 +++ application/signal_snapshot.py | 166 +++++++++++++++++++++++++++++++ main.py | 47 +++++++++ notifications/renderers.py | 38 ++++++- 4 files changed, 261 insertions(+), 3 deletions(-) create mode 100644 application/signal_snapshot.py diff --git a/application/rebalance_service.py b/application/rebalance_service.py index af27a5f..b4e7a69 100644 --- 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 new file mode 100644 index 0000000..4e0463d --- /dev/null +++ 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 index b90644e..e284c36 100644 --- 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 index 05bc616..bfc2e46 100644 --- 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(