feat(backtest): 回测流程集成基准对比—产出相对指标+时序json

sanguo_backtest/cta_engine.py

@@ -15,6 +15,8 @@ if _VNPY_SRC not in sys.path:
     sys.path.insert(0, _VNPY_SRC)
 
 from sanguo_backtest.result_store import BacktestResult, save_result
+from sanguo_data.datareader import read_index_daily
+from sanguo_backtest.metrics import compute_metrics, BENCHMARK_SYMBOL
 
 
 # Mock Exchange enum for local use (replaces vnpy.trader.constant.Exchange)
@@ -121,6 +123,66 @@ def run_cta_backtest(strategy_class, symbol: str, params: dict, start: str, end:
             for k, v in raw_stats.items()
         }
 
+        # Calculate relative metrics against benchmark (Task 3)
+        # Ensure daily_df index is datetime for compute_metrics
+        if daily_df is not None and not daily_df.empty:
+            if not isinstance(daily_df.index, pd.DatetimeIndex):
+                daily_df.index = pd.to_datetime(daily_df.index)
+
+            # Get benchmark config (default hs300)
+            benchmark_key = getattr(cfg, "benchmark", "hs300") if hasattr(cfg, "benchmark") else "hs300"
+            benchmark_code = BENCHMARK_SYMBOL.get(benchmark_key, "sh000300")
+
+            # Load benchmark data
+            start_date = start_dt if isinstance(start_dt, datetime) else datetime.strptime(start, "%Y-%m-%d")
+            end_date = end_dt if isinstance(end_dt, datetime) else datetime.strptime(end, "%Y-%m-%d")
+
+            try:
+                bench_df = read_index_daily(benchmark_code, start_date, end_date, cfg)
+                if bench_df is not None and not bench_df.empty and "close" in bench_df.columns:
+                    # Calculate benchmark daily returns
+                    bench_df["date"] = pd.to_datetime(bench_df["date"])
+                    bench_df = bench_df.sort_values("date")
+                    benchmark_returns = bench_df["close"].pct_change().dropna()
+                    benchmark_returns.index = pd.to_datetime(bench_df["date"].iloc[1:])
+
+                    # vnpy daily_df must have "return" column for compute_metrics
+                    # If not present, calculate from balance
+                    if "return" not in daily_df.columns:
+                        if "balance" in daily_df.columns:
+                            daily_df["return"] = daily_df["balance"].pct_change().fillna(0)
+                        elif "net_pnl" in daily_df.columns:
+                            daily_df["return"] = (daily_df["net_pnl"] / 1_000_000).fillna(0)
+                        else:
+                            daily_df["return"] = 0.0
+
+                    # Compute relative metrics
+                    metrics_result = compute_metrics(daily_df, benchmark_returns)
+
+                    # Merge scalars into statistics (for API response)
+                    statistics.update(metrics_result.scalars)
+
+                    # Serialize series to JSON (separate file, same as equity_curve/trades)
+                    import json
+                    series_data = {}
+                    for key, series in metrics_result.series.items():
+                        if isinstance(series, pd.Series):
+                            series_data[key] = {
+                                "dates": series.index.astype(str).tolist(),
+                                "values": series.tolist()
+                            }
+
+                    # Write metrics series to JSON file
+                    file_dir = os.path.dirname(os.path.abspath(db_path))
+                    metrics_file = os.path.join(file_dir, f"{task_id}_metrics.json")
+                    with open(metrics_file, "w") as f:
+                        json.dump({"series": series_data}, f, indent=2)
+
+            except Exception as metrics_error:
+                # Log but don't fail backtest if metrics calculation fails
+                import logging
+                logging.warning(f"Failed to compute relative metrics: {metrics_error}")
+
         # Build equity curve DataFrame (S1.2): use the daily_df returned by
         # calculate_result (index=date, has a 'balance' column). get_all_daily_results
         # returns DailyResult objects (not dicts), so prefer daily_df.