-
{{ isFactor ? '因子分析' : '回测' }}进行中
+
{{ isFactor ? '因子分析' : isOptimize ? '参数优化' : '回测' }}进行中
任务 ID：{{ taskId }}
diff --git a/sanguo_api/routes.py b/sanguo_api/routes.py
index 9bba9c4..e90b0d7 100644
--- a/sanguo_api/routes.py
+++ b/sanguo_api/routes.py
@@ -316,8 +316,21 @@ def list_tasks(type: str | None = None, status: str | None = None):
@router.get("/task/{task_id}/optimization-results", dependencies=[Depends(verify_token)])
def optimization_results(task_id: str):
- """Optimization results: list of {params, statistics} per parameter combo."""
- raw = get_orchestrator().get_raw_result(task_id)
+ """Optimization results: list of {params, statistics} per parameter combo.
+
+ Bug2: Read from DB aggregate record (persistent across restarts).
+ Falls back to in-memory raw_result for backward compat (in-flight tasks).
+ """
+ orch = get_orchestrator()
+
+ # Primary: DB aggregate record (survives API restart)
+ from sanguo_backtest.result_store import load_result_by_task_id
+ agg = load_result_by_task_id(task_id, orch.db_path)
+ if agg and agg.statistics and "combos" in agg.statistics:
+ return {"task_id": task_id, "results": agg.statistics["combos"]}
+
+ # Fallback: in-memory raw_result (backward compat for in-flight tasks)
+ raw = orch.get_raw_result(task_id)
if raw is None:
raise HTTPException(status_code=404, detail="optimization results not ready")
rows = []
diff --git a/sanguo_backtest/cta_optimizer.py b/sanguo_backtest/cta_optimizer.py
index a35192b..057558d 100644
--- a/sanguo_backtest/cta_optimizer.py
+++ b/sanguo_backtest/cta_optimizer.py
@@ -1,6 +1,8 @@
"""CTA strategy parameter optimization wrapper using vnpy_ctastrategy.backtesting."""
import sys
import os
+import json
+import logging
import traceback
import uuid
from datetime import datetime
@@ -16,6 +18,46 @@ if _VNPY_SRC not in sys.path:
from sanguo_backtest.result_store import BacktestResult, save_result
from sanguo_backtest.cta_engine import guess_exchange, Exchange
+logger = logging.getLogger(__name__)
+
+
+def _ensure_vnpy_settings_for_spawn(cfg) -> None:
+ """Write vt_setting.json so vnpy multiprocessing spawn grandchildren inherit
+ the correct database path.
+
+ Background: vnpy's run_optimization uses multiprocessing spawn, which
+ re-imports vnpy.trader.setting in each grandchild process. On re-import,
+ SETTINGS resets to defaults (database.database="database.db"), losing the
+ DB path configured in the parent. However, setting.py also runs
+ SETTINGS.update(load_json("vt_setting.json")) on import, reading from
+ Path.home()/.vntrader/vt_setting.json. Since sanguo-api runs as
+ Administrator (home=C:\\Users\\Administrator) and spawn children inherit
+ the same home, writing this file ensures grandchildren also load the
+ correct DB → load_data succeeds → calculate_result produces real
+ statistics instead of empty {}.
+
+ Adapter only — vnpy_v4.4.0/ source is untouched.
+ """
+ try:
+ # Resolve vnpy data DB path: prefer cfg param, fall back to config file
+ # (same independent load as the SETTINGS setup below — cfg may be None)
+ if cfg and hasattr(cfg, "data_paths") and cfg.data_paths.get("vnpy_db"):
+ vnpy_db = cfg.data_paths["vnpy_db"]
+ else:
+ from sanguo_data.config import load_config, find_config_path
+ vnpy_db = load_config(find_config_path()).data_paths["vnpy_db"]
+
+ vntrader_dir = Path.home() / ".vntrader"
+ vntrader_dir.mkdir(parents=True, exist_ok=True)
+ setting = {
+ "database.name": "sqlite",
+ "database.database": vnpy_db,
+ }
+ with open(vntrader_dir / "vt_setting.json", "w", encoding="utf-8") as f:
+ json.dump(setting, f, ensure_ascii=False, indent=2)
+ except Exception as e:
+ logger.warning("vt_setting.json adapter failed (spawn may use default DB): %s", e)
+
def run_cta_optimization(
strategy_class,
@@ -25,7 +67,8 @@ def run_cta_optimization(
end: str,
cfg,
db_path: str,
- max_workers: int = 2
+ max_workers: int = 2,
+ task_id: str = None
) -> List[BacktestResult]:
"""
Run CTA strategy parameter optimization using vnpy_ctastrategy BacktestingEngine.
@@ -43,8 +86,14 @@ def run_cta_optimization(
Returns:
List[BacktestResult]: List of result objects with optimization statistics
"""
- # Generate unique task ID for this optimization run
- task_id = f"opt_{uuid.uuid4().hex[:8]}"
+ # Generate unique task ID for this optimization run (use caller-provided task_id if any)
+ if not task_id:
+ task_id = f"opt_{uuid.uuid4().hex[:8]}"
+
+ # Bug1 fix: vnpy multiprocessing spawn grandchildren re-import setting.py,
+ # resetting SETTINGS to defaults (database.db). Write vt_setting.json so they
+ # load the correct DB path. Adapter only — vnpy source untouched.
+ _ensure_vnpy_settings_for_spawn(cfg)
try:
# Lazy import of BacktestingEngine and OptimizationSetting
@@ -107,6 +156,7 @@ def run_cta_optimization(
# Parse optimization results and convert to BacktestResult objects
results = []
+ combos = [] # Bug2: collect valid combos for aggregate record
for item in optimization_results:
try:
# Handle both tuple format (params, target_value, statistics)
@@ -121,6 +171,11 @@ def run_cta_optimization(
# Unknown format, skip this result
continue
+ # Bug2 fix: skip items with empty statistics (vnpy returns these
+ # for combos that loaded 0 bars or failed). Don't persist empty records.
+ if not statistics or not isinstance(statistics, dict):
+ continue
+
# Create individual result for each optimization run
result = BacktestResult(
task_id=f"opt_{uuid.uuid4().hex[:8]}", # Unique ID per result
@@ -139,25 +194,36 @@ def run_cta_optimization(
# Save each result to database
save_result(result, db_path=db_path)
+ combos.append({"params": params, "statistics": statistics})
except Exception as e:
- # Handle individual result parsing error
- error_result = BacktestResult(
- task_id=f"opt_{uuid.uuid4().hex[:8]}",
- type="optimize",
- status="failed",
- strategy=strategy_class.__name__,
- symbol=symbol,
- params={},
- start=start,
- end=end,
- statistics={},
- equity_curve=None,
- trades=None,
- error_msg=f"Result parsing error: {type(e).__name__}: {e}"
- )
- results.append(error_result)
- save_result(error_result, db_path=db_path)
+ # Bug2: log parse errors but don't save empty error records
+ logger.warning("optimize: skipped result parse error: %s: %s",
+ type(e).__name__, e)
+
+ # Bug2: save aggregate record under the main task_id for persistence.
+ # Stores all valid combos in statistics["combos"] so results survive
+ # API restart and can be retrieved by the optimization-results endpoint.
+ if combos:
+ grid_summary = {name: list(rng) for name, rng in grid.items()}
+ aggregate = BacktestResult(
+ task_id=task_id,
+ type="optimize",
+ status="done",
+ strategy=strategy_class.__name__,
+ symbol=symbol,
+ params={
+ "strategy": strategy_class.__name__,
+ "symbol": symbol,
+ "grid": grid_summary,
+ },
+ start=start,
+ end=end,
+ statistics={"combos": combos},
+ equity_curve=None,
+ trades=None,
+ )
+ save_result(aggregate, db_path=db_path)
return results
diff --git a/sanguo_backtest/result_store.py b/sanguo_backtest/result_store.py
index ba8d5f4..6716f72 100644
--- a/sanguo_backtest/result_store.py
+++ b/sanguo_backtest/result_store.py
@@ -10,6 +10,26 @@ import pandas as pd
logger = logging.getLogger(__name__)
+def _json_default(obj):
+ """json.dumps default: vnpy calculate_result 的 statistics 含 numpy.int64/float64,
+ 标准 json 不能序列化（optimize 的统计直接来自 vnpy，未像 cta 那样被 compute_metrics
+ 覆盖成 python float）。用 duck typing 把 numpy 标量/数组转原生，不引入 numpy 硬依赖。"""
+ import datetime as _dt
+ if isinstance(obj, (_dt.date, _dt.datetime)):
+ return obj.isoformat()
+ if hasattr(obj, "item") and callable(getattr(obj, "item")):
+ try:
+ return obj.item()
+ except Exception:
+ pass
+ if hasattr(obj, "tolist") and callable(getattr(obj, "tolist")):
+ try:
+ return obj.tolist()
+ except Exception:
+ pass
+ raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
+
+
@dataclass
class BacktestResult:
"""Backtest result data structure."""
@@ -96,10 +116,10 @@ def save_result(result: BacktestResult, db_path: str, file_dir: Optional[str] =
result.status,
result.strategy,
result.symbol,
- json.dumps(result.params),
+ json.dumps(result.params, default=_json_default),
result.start,
result.end,
- json.dumps(result.statistics),
+ json.dumps(result.statistics, default=_json_default),
equity_path,
trades_path,
result.error_msg
diff --git a/sanguo_orchestrator/runner.py b/sanguo_orchestrator/runner.py
index a90d308..973faf2 100644
--- a/sanguo_orchestrator/runner.py
+++ b/sanguo_orchestrator/runner.py
@@ -82,7 +82,7 @@ class Orchestrator:
spec = self._pending[task_id]
fut: Future = self.pool.submit_work(
task_id, _opt_worker, spec["strategy_class"], spec["symbol"],
- spec["grid"], spec["start"], spec["end"], spec["cfg"], self.db_path
+ spec["grid"], spec["start"], spec["end"], spec["cfg"], self.db_path, task_id
)
task = self.pool.get_task(task_id)
@@ -178,10 +178,10 @@ def _cta_worker(strategy_class, symbol: str, params: dict, start: str, end: str,
return run_cta_backtest(strategy_class, symbol, params, start, end, cfg, db_path, benchmark=benchmark, task_id=task_id, capital=capital, position_pct=position_pct)
-def _opt_worker(strategy_class, symbol: str, grid: dict, start: str, end: str, cfg, db_path: str) -> any:
+def _opt_worker(strategy_class, symbol: str, grid: dict, start: str, end: str, cfg, db_path: str, task_id: str) -> any:
"""Worker for CTA optimization (lazy import, spawn-friendly)"""
from sanguo_backtest.cta_optimizer import run_cta_optimization
- return run_cta_optimization(strategy_class, symbol, grid, start, end, cfg, db_path)
+ return run_cta_optimization(strategy_class, symbol, grid, start, end, cfg, db_path, task_id=task_id)
def _factor_worker(symbols: list, factor_names: list, start: str, end: str, cfg, output_dir: str) -> any: