Fill signal snapshot date from status text

application/signal_snapshot.py

@@ -4,10 +4,16 @@
 
 from collections.abc import Mapping
 from datetime import date, datetime, timezone
+import re
 from typing import Any
 
 SIGNAL_SNAPSHOT_SCHEMA_VERSION = "signal_snapshot.v1"
 
+_SNAPSHOT_DATE_RE = re.compile(
+    r"(?:snapshot_as_of|snapshot_date|snapshot|快照日期)\s*[:=]\s*(\d{4}-\d{2}-\d{2})",
+    re.IGNORECASE,
+)
+
 _INDICATOR_FIELDS = (
     "benchmark_symbol",
     "benchmark_price",
@@ -52,6 +58,16 @@ def _first_value(*values: Any) -> Any:
     return None
 
 
+def _extract_snapshot_date_from_text(*values: Any) -> str | None:
+    for value in values:
+        if not isinstance(value, str):
+            continue
+        match = _SNAPSHOT_DATE_RE.search(value)
+        if match is not None:
+            return match.group(1)
+    return None
+
+
 def _merge_signal_sources(*sources: Mapping[str, Any] | None) -> dict[str, Any]:
     merged: dict[str, Any] = {}
     for source in sources:
@@ -108,6 +124,13 @@ def build_signal_snapshot(
         allocation=allocation,
         explicit_target_weights=target_weights,
     )
+    parsed_snapshot_date = _extract_snapshot_date_from_text(
+        source.get("status_display"),
+        source.get("status_description"),
+        source.get("signal_display"),
+        source.get("signal_description"),
+        source.get("signal_message"),
+    )
     indicators = {
         field: _json_safe(source[field])
         for field in _INDICATOR_FIELDS
@@ -125,6 +148,7 @@ def build_signal_snapshot(
                 source.get("signal_date"),
                 source.get("snapshot_as_of"),
                 source.get("trade_date"),
+                parsed_snapshot_date,
             )
         ),
         "market_date": _json_safe(
@@ -133,6 +157,7 @@ def build_signal_snapshot(
                 source.get("signal_date"),
                 source.get("snapshot_as_of"),
                 source.get("trade_date"),
+                parsed_snapshot_date,
             )
         ),
         "effective_date": _json_safe(source.get("effective_date")),

tests/test_signal_snapshot.py

@@ -0,0 +1,54 @@
+import sys
+import unittest
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from application.signal_snapshot import build_signal_snapshot
+
+
+class SignalSnapshotTests(unittest.TestCase):
+    def test_derives_snapshot_date_from_zh_status_display(self):
+        snapshot = build_signal_snapshot(
+            platform="longbridge",
+            strategy_profile="mega_cap_leader_rotation_top50_balanced",
+            execution={
+                "status_display": "不执行 | 原因=当前不在月度执行窗口 | 快照日期=2026-06-01 | 允许日期=2026-06-02",
+                "signal_display": "月度快照节奏 | 等待进入执行窗口",
+                "latest_price_source": "longbridge_candlesticks",
+            },
+        )
+
+        self.assertEqual(snapshot["market_date"], "2026-06-01")
+        self.assertEqual(snapshot["signal_as_of"], "2026-06-01")
+
+    def test_derives_snapshot_date_from_runtime_diagnostic_status_display(self):
+        snapshot = build_signal_snapshot(
+            platform="longbridge",
+            execution={
+                "status_display": "no-op | reason=outside_monthly_execution_window snapshot=2026-04-10 allowed=2026-04-13",
+                "latest_price_source": "longbridge_candlesticks",
+            },
+        )
+
+        self.assertEqual(snapshot["market_date"], "2026-04-10")
+        self.assertEqual(snapshot["signal_as_of"], "2026-04-10")
+
+    def test_prefers_structured_market_date_over_status_text(self):
+        snapshot = build_signal_snapshot(
+            platform="longbridge",
+            execution={
+                "market_date": "2026-06-02",
+                "status_display": "no-op | snapshot=2026-06-01 allowed=2026-06-02",
+            },
+        )
+
+        self.assertEqual(snapshot["market_date"], "2026-06-02")
+        self.assertEqual(snapshot["signal_as_of"], "2026-06-01")
+
+
+if __name__ == "__main__":
+    unittest.main()