feat(backtest): 接真实 CTA 策略跑通端到端回测（DoubleMaStrategy on 600000）

修复 cta_engine 在真数据上的多个 bug（Phase 2 未在真数据验证）：
- interval "1d" -> "d"（vnpy Interval.DAILY.value）
- capital 0 -> 1_000_000（0 致首笔交易即爆仓，统计全 0）
- statistics 改用 calculate_statistics(df)（旧代码误用 calculate_result 拿 DataFrame）
- statistics JSON-safe（vnpy 可能含 Timestamp）
- test_cta_engine mock 匹配新流程（calculate_statistics 返回统计字典）

验证：diag_cta.py 真实回测 DoubleMaStrategy on 600000 (2024H1, 111 天)
→ 真实统计 total_return -0.017% / sharpe -1.03 / max_drawdown -2.17 / 1 trade
容器 79 tests passed。

sanguo_backtest/cta_engine.py

@@ -79,14 +79,14 @@ def run_cta_backtest(strategy_class, symbol: str, params: dict, start: str, end:
         # Set parameters with A-share specific values
         engine.set_parameters(
             vt_symbol=vt_symbol,
-            interval="1d",  # Daily interval for A-shares
+            interval="d",  # Interval.DAILY.value — vnpy enum uses "d" not "1d"
             start=start_dt,
             end=end_dt,
             rate=0.001,  # Commission rate (0.1% for A-shares)
             slippage=0,  # No slippage for simplicity
             size=1,  # Contract size (1 for stocks)
             pricetick=0.01,  # Minimum price tick (0.01 yuan for A-shares)
-            capital=0  # No initial capital limit
+            capital=1_000_000  # Starting capital — 0 causes instant liquidation on first trade
         )
 
         # Add strategy
@@ -98,8 +98,15 @@ def run_cta_backtest(strategy_class, symbol: str, params: dict, start: str, end:
         # Run backtesting
         engine.run_backtesting()
 
-        # Calculate statistics
-        statistics = engine.calculate_result()
+        # Calculate statistics — calculate_result() returns a daily DataFrame,
+        # calculate_statistics(df) returns the stats dict (sharpe/drawdown/etc.)
+        daily_df = engine.calculate_result()
+        raw_stats = engine.calculate_statistics(daily_df, output=False) or {}
+        # Ensure JSON-serializable (vnpy may include Timestamp / non-numeric values)
+        statistics = {
+            k: (v if isinstance(v, (int, float, str, bool)) or v is None else str(v))
+            for k, v in raw_stats.items()
+        }
 
         # Get daily results for equity curve
         daily_results = engine.get_all_daily_results()

scripts/diag_cta.py

@@ -0,0 +1,51 @@
+"""Diagnostic: real CTA backtest of DoubleMaStrategy on 600000 (A-share daily).
+Guarded entry for spawn. Throwaway."""
+import sys
+import os
+import traceback
+
+_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "vnpy_v4.4.0"))
+_REPO = os.path.dirname(_VNPY_SRC)
+for _p in (_REPO, _VNPY_SRC):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+
+def main():
+    # Configure vnpy DB to the real quant_trading.db BEFORE engine.load_data()
+    from vnpy.trader.setting import SETTINGS
+    from sanguo_data.config import load_config
+    cfg = load_config("/app/config/data_platform.yaml")
+    SETTINGS["database.name"] = "sqlite"
+    SETTINGS["database.database"] = cfg.data_paths["vnpy_db"]
+    print("DB:", SETTINGS["database.database"])
+
+    from vnpy_ctastrategy.strategies.double_ma_strategy import DoubleMaStrategy
+    print("DoubleMaStrategy.parameters:", getattr(DoubleMaStrategy, "parameters", "?"))
+
+    from sanguo_backtest.cta_engine import run_cta_backtest
+    # Classic double-MA params (vnpy example defaults); fixed_size=1
+    params = {"fast_window": 10, "slow_window": 20, "fixed_size": 1}
+
+    print(f"run_cta_backtest DoubleMaStrategy on 600000, 2024-01-01..2024-06-30, params={params}")
+    try:
+        result = run_cta_backtest(
+            DoubleMaStrategy, "600000", params,
+            "2024-01-01", "2024-06-30", cfg, "/tmp/cta_results.db",
+        )
+        print("=== status:", result.status)
+        print("=== statistics ===")
+        if result.statistics:
+            for k, v in result.statistics.items():
+                print(f"  {k}: {v}")
+        else:
+            print("  (empty)")
+        if result.status == "failed":
+            print("=== error ===")
+            print(result.error_msg)
+    except Exception:
+        traceback.print_exc()
+
+
+if __name__ == "__main__":
+    main()

tests/backtest/test_cta_engine.py

@@ -15,9 +15,11 @@ class TestRunCtaBacktest:
         mock_strategy_class = Mock()
         mock_strategy_class.__name__ = "TestStrategy"
 
-        # Mock BacktestingEngine
+        # Mock BacktestingEngine — calculate_result() returns daily_df (DataFrame),
+        # calculate_statistics(df) returns the stats dict (vnpy API, matches cta_engine)
         mock_engine = MagicMock()
-        mock_engine.calculate_result.return_value = {
+        mock_engine.calculate_result.return_value = MagicMock(name="daily_df")
+        mock_engine.calculate_statistics.return_value = {
             "total_return": 0.15,
             "sharpe_ratio": 1.2,
             "max_drawdown": -0.08,
@@ -68,6 +70,7 @@ class TestRunCtaBacktest:
         mock_engine.load_data.assert_called_once()
         mock_engine.run_backtesting.assert_called_once()
         mock_engine.calculate_result.assert_called_once()
+        mock_engine.calculate_statistics.assert_called_once()
 
     def test_run_cta_backtest_handles_exception(self, temp_db_path):
         """Test that exceptions during backtest are handled properly."""