Audit timing contracts in IBKR notifications and reports

Author: Pigbibi

application/rebalance_service.py

@@ -314,16 +314,45 @@ def _format_dashboard_text(text) -> str:
     return "\n".join(lines)
 
 
-def _strategy_dashboard_text(signal_metadata) -> str:
+def _build_timing_audit_lines(signal_metadata, *, translator) -> list[str]:
     metadata = signal_metadata if isinstance(signal_metadata, Mapping) else {}
     raw_annotations = metadata.get("execution_annotations")
     annotations = raw_annotations if isinstance(raw_annotations, Mapping) else {}
-    return _format_dashboard_text(
+    signal_date = str(annotations.get("signal_date") or metadata.get("signal_date") or "").strip()
+    effective_date = str(annotations.get("effective_date") or metadata.get("effective_date") or "").strip()
+    contract = str(
+        annotations.get("execution_timing_contract")
+        or metadata.get("execution_timing_contract")
+        or ""
+    ).strip()
+    if not signal_date and not effective_date and not contract:
+        return []
+    label = "⏱ 执行时点" if _translator_uses_zh(translator) else "⏱ Timing"
+    if signal_date and effective_date:
+        value = f"{signal_date} -> {effective_date}"
+    else:
+        value = signal_date or effective_date or contract
+    if contract and contract not in value:
+        value = f"{value} ({contract})" if value else contract
+    return [f"{label}: {value}"]
+
+
+def _strategy_dashboard_text(signal_metadata, *, translator) -> str:
+    metadata = signal_metadata if isinstance(signal_metadata, Mapping) else {}
+    raw_annotations = metadata.get("execution_annotations")
+    annotations = raw_annotations if isinstance(raw_annotations, Mapping) else {}
+    dashboard_text = _format_dashboard_text(
         annotations.get("dashboard_text")
         or metadata.get("dashboard_text")
         or metadata.get("dashboard")
         or ""
     )
+    timing_lines = _build_timing_audit_lines(metadata, translator=translator)
+    if not timing_lines:
+        return dashboard_text
+    if not dashboard_text:
+        return "\n".join(timing_lines)
+    return f"{dashboard_text}\n" + "\n".join(timing_lines)
 
 
 def build_dashboard(
@@ -341,7 +370,7 @@ def build_dashboard(
     status_icon="🐤",
 ):
     signal_metadata = signal_metadata or {}
-    strategy_dashboard = _strategy_dashboard_text(signal_metadata)
+    strategy_dashboard = _strategy_dashboard_text(signal_metadata, translator=translator)
     if strategy_dashboard:
         return strategy_dashboard
     equity = account_values.get("equity", 0)
@@ -550,7 +579,7 @@ def run_strategy_core(
             separator=config.separator,
             status_icon=signal_metadata.get("status_icon", "🐤"),
         )
-        strategy_dashboard = _strategy_dashboard_text(signal_metadata)
+        strategy_dashboard = _strategy_dashboard_text(signal_metadata, translator=config.translator)
 
         if target_weights is None:
             decision = signal_metadata.get("snapshot_guard_decision")

application/runtime_composer.py

@@ -26,6 +26,7 @@ class IBKRRuntimeComposer:
     strategy_display_name: str
     strategy_display_name_localized: str
     managed_symbols: tuple[str, ...]
+    signal_effective_after_trading_days: int | None
     signal_source: str
     status_icon: str
     safe_haven: str
@@ -89,6 +90,7 @@ def build_reporting_adapters(self):
             strategy_display_name=self.strategy_display_name,
             strategy_display_name_localized=self.strategy_display_name_localized,
             dry_run=self.dry_run_only,
+            signal_effective_after_trading_days=self.signal_effective_after_trading_days,
             strategy_config_source=self.strategy_config_source,
             ib_gateway_host_resolver=self.ib_gateway_host_resolver,
             ib_gateway_port=self.ib_gateway_port,
@@ -145,6 +147,7 @@ def build_runtime_composer(
     strategy_display_name: str,
     strategy_display_name_localized: str,
     managed_symbols: tuple[str, ...],
+    signal_effective_after_trading_days: int | None,
     signal_source: str,
     status_icon: str,
     safe_haven: str,
@@ -188,6 +191,7 @@ def build_runtime_composer(
         strategy_display_name=str(strategy_display_name or ""),
         strategy_display_name_localized=str(strategy_display_name_localized or ""),
         managed_symbols=tuple(managed_symbols),
+        signal_effective_after_trading_days=signal_effective_after_trading_days,
         signal_source=str(signal_source or ""),
         status_icon=str(status_icon or ""),
         safe_haven=str(safe_haven or ""),

application/runtime_reporting_adapters.py

@@ -7,6 +7,7 @@
 from datetime import datetime, timezone
 from typing import Any
 
+from quant_platform_kit.strategy_contracts import build_execution_timing_metadata
 from runtime_logging import RuntimeLogContext
 
 
@@ -33,6 +34,7 @@ class IBKRRuntimeReportingAdapters:
     strategy_display_name: str = ""
     strategy_display_name_localized: str = ""
     dry_run: bool = False
+    signal_effective_after_trading_days: int | None = None
     strategy_config_source: str | None = None
     ib_gateway_host_resolver: Callable[[], str] | None = None
     ib_gateway_port: int = 0
@@ -84,6 +86,11 @@ def build_log_context(self, *, trace_header: str | None = None) -> RuntimeLogCon
         )
 
     def build_report(self, log_context: RuntimeLogContext) -> dict[str, Any]:
+        started_at = self.clock()
+        timing_summary = build_execution_timing_metadata(
+            signal_date=started_at,
+            signal_effective_after_trading_days=self.signal_effective_after_trading_days,
+        )
         return self.report_builder(
             platform=log_context.platform,
             deploy_target=log_context.deploy_target,
@@ -95,7 +102,7 @@ def build_report(self, log_context: RuntimeLogContext) -> dict[str, Any]:
             run_id=log_context.run_id,
             run_source="cloud_run",
             dry_run=self.dry_run,
-            started_at=self.clock(),
+            started_at=started_at,
             summary={
                 "account_ids": list(self.extra_context_fields.get("account_ids") or ()),
                 "managed_symbols": list(self.managed_symbols),
@@ -104,6 +111,7 @@ def build_report(self, log_context: RuntimeLogContext) -> dict[str, Any]:
                 "safe_haven": self.safe_haven,
                 "strategy_display_name": self.strategy_display_name,
                 "strategy_display_name_localized": self.strategy_display_name_localized,
+                **timing_summary,
             },
             diagnostics={
                 "strategy_config_source": self.strategy_config_source,
@@ -165,6 +173,7 @@ def build_runtime_reporting_adapters(
     strategy_display_name: str,
     strategy_display_name_localized: str,
     dry_run: bool,
+    signal_effective_after_trading_days: int | None,
     strategy_config_source: str | None,
     ib_gateway_host_resolver: Callable[[], str],
     ib_gateway_port: int,
@@ -204,6 +213,7 @@ def build_runtime_reporting_adapters(
         strategy_display_name=str(strategy_display_name or ""),
         strategy_display_name_localized=str(strategy_display_name_localized or ""),
         dry_run=bool(dry_run),
+        signal_effective_after_trading_days=signal_effective_after_trading_days,
         strategy_config_source=strategy_config_source,
         ib_gateway_host_resolver=ib_gateway_host_resolver,
         ib_gateway_port=int(ib_gateway_port),

main.py

@@ -196,6 +196,11 @@ def _env_flag(name: str) -> bool:
     else "market_data"
 )
 STRATEGY_STATUS_ICON = STRATEGY_RUNTIME.status_icon
+SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS = getattr(
+    getattr(STRATEGY_RUNTIME.runtime_adapter, "runtime_policy", None),
+    "signal_effective_after_trading_days",
+    None,
+)
 FEATURE_RUNTIME_PARAMETERS = dict(STRATEGY_RUNTIME.runtime_config)
 STRATEGY_RUNTIME_CONFIG = dict(STRATEGY_RUNTIME.merged_runtime_config)
 SAFE_HAVEN = str(STRATEGY_RUNTIME_CONFIG.get("safe_haven") or "BIL")
@@ -332,6 +337,7 @@ def build_composer():
         strategy_display_name=STRATEGY_DISPLAY_NAME,
         strategy_display_name_localized=strategy_display_name,
         managed_symbols=resolve_reporting_managed_symbols(),
+        signal_effective_after_trading_days=SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS,
         signal_source=STRATEGY_SIGNAL_SOURCE,
         status_icon=STRATEGY_STATUS_ICON,
         safe_haven=SAFE_HAVEN,

notifications/renderers.py

@@ -303,16 +303,45 @@ def _format_dashboard_text(text) -> str:
     return "\n".join(lines)
 
 
-def _strategy_dashboard_text(signal_metadata) -> str:
+def _build_timing_audit_lines(signal_metadata, *, translator) -> list[str]:
     metadata = signal_metadata if isinstance(signal_metadata, Mapping) else {}
     raw_annotations = metadata.get("execution_annotations")
     annotations = raw_annotations if isinstance(raw_annotations, Mapping) else {}
-    return _format_dashboard_text(
+    signal_date = str(annotations.get("signal_date") or metadata.get("signal_date") or "").strip()
+    effective_date = str(annotations.get("effective_date") or metadata.get("effective_date") or "").strip()
+    contract = str(
+        annotations.get("execution_timing_contract")
+        or metadata.get("execution_timing_contract")
+        or ""
+    ).strip()
+    if not signal_date and not effective_date and not contract:
+        return []
+    label = "⏱ 执行时点" if _translator_uses_zh(translator) else "⏱ Timing"
+    if signal_date and effective_date:
+        value = f"{signal_date} -> {effective_date}"
+    else:
+        value = signal_date or effective_date or contract
+    if contract and contract not in value:
+        value = f"{value} ({contract})" if value else contract
+    return [f"{label}: {value}"]
+
+
+def _strategy_dashboard_text(signal_metadata, *, translator) -> str:
+    metadata = signal_metadata if isinstance(signal_metadata, Mapping) else {}
+    raw_annotations = metadata.get("execution_annotations")
+    annotations = raw_annotations if isinstance(raw_annotations, Mapping) else {}
+    dashboard_text = _format_dashboard_text(
         annotations.get("dashboard_text")
         or metadata.get("dashboard_text")
         or metadata.get("dashboard")
         or ""
     )
+    timing_lines = _build_timing_audit_lines(metadata, translator=translator)
+    if not timing_lines:
+        return dashboard_text
+    if not dashboard_text:
+        return "\n".join(timing_lines)
+    return f"{dashboard_text}\n" + "\n".join(timing_lines)
 
 
 def build_dashboard(
@@ -330,7 +359,7 @@ def build_dashboard(
     status_icon="🐤",
 ):
     signal_metadata = signal_metadata or {}
-    strategy_dashboard = _strategy_dashboard_text(signal_metadata)
+    strategy_dashboard = _strategy_dashboard_text(signal_metadata, translator=translator)
     if strategy_dashboard:
         return strategy_dashboard
     equity = account_values.get("equity", 0)
@@ -375,6 +404,7 @@ def build_dashboard(
         translator("snapshot_as_of_detail", value=_format_text(snapshot_as_of, fallback="<none>")) if snapshot_as_of else None,
     ]
     diagnostics_lines = [f"  - {part}" for part in diagnostics if part]
+    diagnostics_lines.extend(_build_timing_audit_lines(signal_metadata, translator=translator))
     diagnostics_text = "\n".join(diagnostics_lines)
     localized_status_desc = _localize_notification_text(status_desc, translator=translator)
     localized_signal_desc = _localize_notification_text(signal_desc, translator=translator)

strategy_runtime.py

@@ -23,6 +23,8 @@
     StrategyDecision,
     StrategyEntrypoint,
     StrategyRuntimeAdapter,
+    apply_runtime_policy_to_runtime_config,
+    build_execution_timing_metadata,
     build_strategy_context_from_available_inputs,
     build_strategy_evaluation_inputs,
 )
@@ -193,6 +195,7 @@ def _evaluate_market_data_strategy(
         runtime_config = dict(self.runtime_config)
         runtime_config.setdefault("translator", translator)
         runtime_config.setdefault("pacing_sec", float(pacing_sec))
+        apply_runtime_policy_to_runtime_config(runtime_config, self.runtime_adapter)
         portfolio_snapshot = self._fetch_portfolio_snapshot_for_context(ib, required=False)
         ctx = self._build_strategy_context(
             runtime_adapter=self.runtime_adapter,
@@ -215,6 +218,12 @@ def _evaluate_market_data_strategy(
             "managed_symbols": managed_symbols,
             "status_icon": self.status_icon,
             "dry_run_only": self.runtime_settings.dry_run_only,
+            **build_execution_timing_metadata(
+                signal_date=run_as_of,
+                signal_effective_after_trading_days=(
+                    self.runtime_adapter.runtime_policy.signal_effective_after_trading_days
+                ),
+            ),
         }
         if portfolio_snapshot is not None:
             metadata["portfolio_total_equity"] = float(getattr(portfolio_snapshot, "total_equity", 0.0) or 0.0)
@@ -235,6 +244,7 @@ def _evaluate_value_target_strategy(
     ) -> StrategyEvaluationResult:
         runtime_config = dict(self.runtime_config)
         runtime_config.setdefault("translator", translator)
+        apply_runtime_policy_to_runtime_config(runtime_config, self.runtime_adapter)
         portfolio_snapshot = fetch_portfolio_snapshot(ib)
         market_inputs = self._build_value_target_market_inputs(
             ib=ib,
@@ -265,6 +275,12 @@ def _evaluate_value_target_strategy(
             "status_icon": self.status_icon,
             "dry_run_only": self.runtime_settings.dry_run_only,
             "portfolio_total_equity": float(portfolio_snapshot.total_equity),
+            **build_execution_timing_metadata(
+                signal_date=run_as_of,
+                signal_effective_after_trading_days=(
+                    self.runtime_adapter.runtime_policy.signal_effective_after_trading_days
+                ),
+            ),
         }
         if safe_haven_symbol:
             metadata["safe_haven_symbol"] = str(safe_haven_symbol)

tests/test_rebalance_service.py

@@ -202,6 +202,9 @@ def fake_execute_rebalance(
             {
                 "managed_symbols": ("AAA", "BOXX"),
                 "status_icon": "📏",
+                "signal_date": "2026-04-01",
+                "effective_date": "2026-04-02",
+                "execution_timing_contract": "next_trading_day",
                 "execution_annotations": {
                     "dashboard_text": (
                         "📌 Strategy portfolio\n"
@@ -237,6 +240,7 @@ def fake_execute_rebalance(
     assert "Total assets (strategy symbols + cash): $1,000.00" in observed["messages"][0]
     assert "💼 Strategy holdings" in observed["messages"][0]
     assert "📏 breadth=60.0%" in observed["messages"][0]
+    assert "⏱ Timing: 2026-04-01 -> 2026-04-02 (next_trading_day)" in observed["messages"][0]
     assert "Target Weights" not in observed["messages"][0]
 
 

tests/test_request_handling.py

@@ -112,6 +112,9 @@ def test_handle_request_persists_machine_readable_report(strategy_module, monkey
     assert observed["report"]["run_source"] == "cloud_run"
     assert observed["report"]["account_scope"] == strategy_module.ACCOUNT_GROUP
     assert observed["report"]["summary"]["signal_source"] == strategy_module.STRATEGY_SIGNAL_SOURCE
+    assert observed["report"]["summary"]["execution_timing_contract"] == "next_trading_day"
+    assert observed["report"]["summary"]["signal_date"]
+    assert observed["report"]["summary"]["effective_date"]
 
 
 def test_execution_report_prefers_configured_managed_symbols_without_ranking_pool(strategy_module_factory):
@@ -120,6 +123,7 @@ def test_execution_report_prefers_configured_managed_symbols_without_ranking_poo
 
     assert report["summary"]["managed_symbols"] == ["SOXL", "SOXX", "BOXX", "QQQI", "SPYI"]
     assert report["summary"]["safe_haven"] == "BIL"
+    assert report["summary"]["execution_timing_contract"] == "next_trading_day"
 
 
 def test_handle_request_enriches_runtime_report_with_cycle_details(strategy_module, monkeypatch):

tests/test_runtime_composer.py

@@ -34,6 +34,7 @@ def fake_reporting_builder(**kwargs):
         strategy_display_name="Global ETF Rotation",
         strategy_display_name_localized="全球 ETF 轮动",
         managed_symbols=("AAA", "BIL"),
+        signal_effective_after_trading_days=1,
         signal_source="market_data",
         status_icon="🐤",
         safe_haven="BIL",
@@ -79,6 +80,7 @@ def fake_reporting_builder(**kwargs):
     assert observed["notification_builder"]["send_message"]
     assert observed["reporting_builder"]["account_scope"] == "default"
     assert observed["reporting_builder"]["managed_symbols"] == ("AAA", "BIL")
+    assert observed["reporting_builder"]["signal_effective_after_trading_days"] == 1
     assert runtime.connect_ib() == "ib-connection"
     assert runtime.portfolio_port_factory("ib").get_portfolio_snapshot() == ("portfolio-snapshot", "ib")
     assert runtime.compute_signals == "compute-signals"