diff --git a/application/signal_snapshot.py b/application/signal_snapshot.py index 4e0463d..19f5c68 100644 --- a/application/signal_snapshot.py +++ b/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")), diff --git a/tests/test_signal_snapshot.py b/tests/test_signal_snapshot.py new file mode 100644 index 0000000..7a35af1 --- /dev/null +++ b/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()