Annotate snapshot-priced IB dry runs

Author: Pigbibi

application/execution_service.py

@@ -18,21 +18,10 @@ def get_market_prices(
     symbols,
     *,
     fetch_quote_snapshots,
-    dry_run_only: bool = False,
-    snapshot_price_fallbacks: dict[str, float] | None = None,
 ):
     """Fetch market prices for multiple symbols in one pass."""
     quotes = fetch_quote_snapshots(ib, symbols)
-    prices = {symbol: quote.last_price for symbol, quote in quotes.items()}
-    if dry_run_only and snapshot_price_fallbacks:
-        for symbol in symbols:
-            normalized = str(symbol).strip().upper()
-            if normalized in prices:
-                continue
-            fallback_price = snapshot_price_fallbacks.get(normalized)
-            if fallback_price and float(fallback_price) > 0:
-                prices[normalized] = float(fallback_price)
-    return prices
+    return {symbol: quote.last_price for symbol, quote in quotes.items()}
 
 
 def check_order_submitted(report, *, translator):
@@ -192,6 +181,38 @@ def _display_text(value: Any, *, fallback: str) -> str:
     return text or fallback
 
 
+def _apply_snapshot_price_fallbacks(
+    prices: dict[str, float],
+    symbols,
+    *,
+    dry_run_only: bool,
+    snapshot_price_fallbacks: dict[str, float] | None,
+) -> tuple[dict[str, float], tuple[str, ...]]:
+    if not dry_run_only or not snapshot_price_fallbacks:
+        return dict(prices), ()
+    resolved = dict(prices)
+    fallback_symbols: list[str] = []
+    for symbol in symbols:
+        normalized = str(symbol).strip().upper()
+        if normalized in resolved:
+            continue
+        fallback_price = snapshot_price_fallbacks.get(normalized)
+        if fallback_price and float(fallback_price) > 0:
+            resolved[normalized] = float(fallback_price)
+            fallback_symbols.append(normalized)
+    return resolved, tuple(fallback_symbols)
+
+
+def _format_symbol_preview(symbols: tuple[str, ...], *, limit: int = 3) -> str:
+    if not symbols:
+        return ""
+    shown = [str(symbol).strip().upper() for symbol in symbols[:limit]]
+    remaining = len(symbols) - len(shown)
+    if remaining > 0:
+        shown.append(f"+{remaining}")
+    return ",".join(shown)
+
+
 def _resolve_execution_lock_path(
     *,
     strategy_profile: str | None,
@@ -364,6 +385,10 @@ def execute_rebalance(
         "residual_cash_estimate": float(account_values.get("buying_power", 0.0) or 0.0),
         "current_stock_weight": 0.0,
         "current_safe_haven_weight": 0.0,
+        "price_source_mode": "market_quote",
+        "snapshot_price_fallback_used": False,
+        "snapshot_price_fallback_symbols": [],
+        "snapshot_price_fallback_count": 0,
         "lock_path": None,
     }
     if equity <= 0:
@@ -389,9 +414,18 @@ def execute_rebalance(
         ib,
         all_symbols,
         fetch_quote_snapshots=fetch_quote_snapshots,
+    )
+    prices, snapshot_price_fallback_symbols = _apply_snapshot_price_fallbacks(
+        prices,
+        all_symbols,
         dry_run_only=dry_run_only,
         snapshot_price_fallbacks=snapshot_price_fallbacks,
     )
+    execution_summary["snapshot_price_fallback_used"] = bool(snapshot_price_fallback_symbols)
+    execution_summary["snapshot_price_fallback_symbols"] = list(snapshot_price_fallback_symbols)
+    execution_summary["snapshot_price_fallback_count"] = len(snapshot_price_fallback_symbols)
+    if snapshot_price_fallback_symbols:
+        execution_summary["price_source_mode"] = "mixed_market_quote_snapshot_close"
 
     current_mv = {}
     for symbol in all_symbols:
@@ -459,6 +493,14 @@ def execute_rebalance(
             ]
         )
     )
+    if snapshot_price_fallback_symbols:
+        trade_logs.append(
+            translator(
+                "dry_run_snapshot_prices",
+                count=len(snapshot_price_fallback_symbols),
+                symbols=_format_symbol_preview(snapshot_price_fallback_symbols),
+            )
+        )
     trade_logs.extend(_format_target_lines(target_weights, current_mv, equity, translator=translator))
 
     missing_price_symbols: list[str] = []

application/rebalance_service.py

@@ -105,6 +105,7 @@ def run_strategy_core(
     translator,
     separator,
     reconciliation_output_path=None,
+    result_hook=None,
 ):
     ib = None
     try:
@@ -161,6 +162,17 @@ def run_strategy_core(
             message = f"{translator('heartbeat_title')}\n{dashboard}\n{separator}\n{no_op_text}"
             send_tg_message(message)
             print(message, flush=True)
+            if callable(result_hook):
+                result_hook(
+                    {
+                        "result": "OK - heartbeat",
+                        "signal_metadata": dict(signal_metadata or {}),
+                        "target_weights": None,
+                        "execution_summary": None,
+                        "reconciliation_record": dict(record),
+                        "reconciliation_record_path": str(record_path),
+                    }
+                )
             return "OK - heartbeat"
 
         execution_result = execute_rebalance(
@@ -213,6 +225,17 @@ def run_strategy_core(
 
         send_tg_message(message)
         print(message, flush=True)
+        if callable(result_hook):
+            result_hook(
+                {
+                    "result": "OK - executed",
+                    "signal_metadata": dict(signal_metadata or {}),
+                    "target_weights": dict(target_weights or {}),
+                    "execution_summary": dict(execution_summary or {}),
+                    "reconciliation_record": dict(record),
+                    "reconciliation_record_path": str(record_path),
+                }
+            )
         return "OK - executed"
     finally:
         if ib is not None and ib.isConnected():

application/reconciliation_service.py

@@ -70,6 +70,10 @@ def build_reconciliation_record(
         "cash_reserve_dollars": execution_summary.get("cash_reserve_dollars"),
         "current_stock_weight": execution_summary.get("current_stock_weight"),
         "current_safe_haven_weight": execution_summary.get("current_safe_haven_weight"),
+        "price_source_mode": execution_summary.get("price_source_mode"),
+        "snapshot_price_fallback_used": execution_summary.get("snapshot_price_fallback_used"),
+        "snapshot_price_fallback_count": execution_summary.get("snapshot_price_fallback_count"),
+        "snapshot_price_fallback_symbols": execution_summary.get("snapshot_price_fallback_symbols") or [],
         "execution_status": execution_summary.get("execution_status") or ("no_op" if no_op_reason else "executed"),
         "lock_path": execution_summary.get("lock_path"),
         "no_op_reason": no_op_reason or execution_summary.get("no_op_reason"),

main.py

@@ -204,6 +204,7 @@ def get_ib_port():
     instance_name=RUNTIME_SETTINGS.ib_gateway_instance_name,
     extra_fields={"account_ids": list(ACCOUNT_IDS)},
 )
+LAST_CYCLE_DETAILS: dict[str, object] = {}
 
 def t(key, **kwargs):
     return build_translator(NOTIFY_LANG)(key, **kwargs)
@@ -412,7 +413,9 @@ def execute_rebalance(
 # Main strategy runner
 # ---------------------------------------------------------------------------
 def run_strategy_core():
-    return run_rebalance_cycle(
+    global LAST_CYCLE_DETAILS
+    cycle_details: dict[str, object] = {}
+    result = run_rebalance_cycle(
         connect_ib=connect_ib,
         get_current_portfolio=get_current_portfolio,
         compute_signals=compute_signals,
@@ -421,7 +424,10 @@ def run_strategy_core():
         translator=t,
         separator=SEPARATOR,
         reconciliation_output_path=RECONCILIATION_OUTPUT_PATH,
+        result_hook=lambda payload: cycle_details.update(payload or {}),
     )
+    LAST_CYCLE_DETAILS = cycle_details
+    return result
 
 
 # ---------------------------------------------------------------------------
@@ -432,6 +438,8 @@ def handle_request():
     if request.method == "GET":
         return "OK - use POST to execute strategy", 200
 
+    global LAST_CYCLE_DETAILS
+    LAST_CYCLE_DETAILS = {}
     log_context = build_request_log_context()
     report = build_execution_report(log_context)
     try:
@@ -459,10 +467,38 @@ def handle_request():
             message="Starting strategy execution",
         )
         result = run_strategy_core()
+        cycle_details = dict(LAST_CYCLE_DETAILS or {})
+        execution_summary = dict(cycle_details.get("execution_summary") or {})
+        reconciliation_record = dict(cycle_details.get("reconciliation_record") or {})
         finalize_runtime_report(
             report,
             status="ok",
-            diagnostics={"result": result},
+            summary={
+                "result": result,
+                "execution_status": execution_summary.get("execution_status") or reconciliation_record.get("execution_status"),
+                "no_op_reason": execution_summary.get("no_op_reason") or reconciliation_record.get("no_op_reason"),
+                "orders_submitted_count": len(execution_summary.get("orders_submitted") or reconciliation_record.get("orders_submitted") or ()),
+                "orders_skipped_count": len(execution_summary.get("orders_skipped") or reconciliation_record.get("orders_skipped") or ()),
+                "snapshot_price_fallback_used": bool(
+                    execution_summary.get("snapshot_price_fallback_used")
+                    or reconciliation_record.get("snapshot_price_fallback_used")
+                ),
+                "snapshot_price_fallback_count": int(
+                    execution_summary.get("snapshot_price_fallback_count")
+                    or reconciliation_record.get("snapshot_price_fallback_count")
+                    or 0
+                ),
+            },
+            diagnostics={
+                "result": result,
+                "price_source_mode": execution_summary.get("price_source_mode") or reconciliation_record.get("price_source_mode"),
+                "snapshot_price_fallback_symbols": execution_summary.get("snapshot_price_fallback_symbols")
+                or reconciliation_record.get("snapshot_price_fallback_symbols")
+                or [],
+            },
+            artifacts={
+                "reconciliation_record_path": cycle_details.get("reconciliation_record_path"),
+            },
         )
         log_runtime_event(
             log_context,

notifications/telegram.py

@@ -27,6 +27,7 @@
         "snapshot_file_ts_detail": "快照文件时间={value}",
         "snapshot_path_detail": "快照路径={value}",
         "config_source_detail": "配置来源={value}",
+        "dry_run_snapshot_prices": "🧪 dry-run估价: 使用快照收盘价 {count}个标的 ({symbols})",
         "trade_date_detail": "交易日={value}",
         "target_diff": "目标差异 {symbol}: 当前={current} 目标={target} 变化={delta}",
         "pending_orders_detected": "检测到未完成订单: profile={profile} symbols={symbols}",
@@ -71,6 +72,7 @@
         "snapshot_file_ts_detail": "snapshot_file_ts={value}",
         "snapshot_path_detail": "snapshot_path={value}",
         "config_source_detail": "config_source={value}",
+        "dry_run_snapshot_prices": "🧪 dry-run pricing: snapshot close for {count} symbols ({symbols})",
         "trade_date_detail": "trade_date={value}",
         "target_diff": "target_diff {symbol}: current={current} target={target} delta={delta}",
         "pending_orders_detected": "pending_orders_detected profile={profile} symbols={symbols}",

tests/test_execution_service.py

@@ -22,6 +22,7 @@ def translate(key, **kwargs):
         "same_day_fills_detected": "same_day_fills_detected profile={profile} mode={mode} symbols={symbols} trade_date={trade_date}",
         "same_day_execution_locked": "same_day_execution_locked profile={profile} mode={mode} trade_date={trade_date} snapshot_date={snapshot_date} target_hash={target_hash} lock_path={lock_path}",
         "execution_lock_acquired": "execution_lock_acquired mode={mode} trade_date={trade_date} snapshot_date={snapshot_date} lock_path={lock_path}",
+        "dry_run_snapshot_prices": "dry_run_snapshot_prices count={count} symbols={symbols}",
         "no_equity": "❌ No equity",
     }
     template = templates[key]
@@ -404,5 +405,10 @@ def accountValues(self):
 
     assert summary["execution_status"] == "executed"
     assert len(summary["orders_submitted"]) == 2
+    assert summary["snapshot_price_fallback_used"] is True
+    assert summary["snapshot_price_fallback_count"] == 2
+    assert set(summary["snapshot_price_fallback_symbols"]) == {"VOO", "BOXX"}
+    assert summary["price_source_mode"] == "mixed_market_quote_snapshot_close"
+    assert any(log.startswith("dry_run_snapshot_prices count=2") for log in trade_logs)
     assert any(log.startswith("DRY_RUN buy VOO") for log in trade_logs)
     assert any(log.startswith("DRY_RUN buy BOXX") for log in trade_logs)

tests/test_request_handling.py

@@ -83,6 +83,47 @@ def test_handle_request_persists_machine_readable_report(strategy_module, monkey
     assert observed["report"]["summary"]["signal_source"] == strategy_module.STRATEGY_SIGNAL_SOURCE
 
 
+def test_handle_request_enriches_runtime_report_with_cycle_details(strategy_module, monkeypatch):
+    observed = {}
+
+    monkeypatch.setattr(strategy_module, "build_run_id", lambda: "run-001")
+    monkeypatch.setattr(strategy_module, "is_market_open_today", lambda: True)
+
+    def fake_run_strategy_core():
+        strategy_module.LAST_CYCLE_DETAILS = {
+            "execution_summary": {
+                "execution_status": "executed",
+                "orders_submitted": [{"symbol": "AAA"}],
+                "orders_skipped": [],
+                "price_source_mode": "mixed_market_quote_snapshot_close",
+                "snapshot_price_fallback_used": True,
+                "snapshot_price_fallback_count": 1,
+                "snapshot_price_fallback_symbols": ["AAA"],
+            },
+            "reconciliation_record_path": "/tmp/reconciliation.json",
+        }
+        return "OK - executed"
+
+    monkeypatch.setattr(strategy_module, "run_strategy_core", fake_run_strategy_core)
+    monkeypatch.setattr(
+        strategy_module,
+        "persist_execution_report",
+        lambda report: observed.setdefault("report", dict(report)) or "/tmp/runtime-report.json",
+    )
+
+    with strategy_module.app.test_request_context("/", method="POST"):
+        body, status = strategy_module.handle_request()
+
+    assert status == 200
+    assert body == "OK - executed"
+    assert observed["report"]["summary"]["execution_status"] == "executed"
+    assert observed["report"]["summary"]["snapshot_price_fallback_used"] is True
+    assert observed["report"]["summary"]["snapshot_price_fallback_count"] == 1
+    assert observed["report"]["diagnostics"]["price_source_mode"] == "mixed_market_quote_snapshot_close"
+    assert observed["report"]["diagnostics"]["snapshot_price_fallback_symbols"] == ["AAA"]
+    assert observed["report"]["artifacts"]["reconciliation_record_path"] == "/tmp/reconciliation.json"
+
+
 def test_handle_request_post_returns_market_closed_when_schedule_empty(strategy_module, monkeypatch):
     observed = {}