feat(s1): 回测核心端到端跑通(vnpy client 对齐)

- 修 submit_cta/optimize 策略字符串→类解析(get_strategy_class)
- cta_engine: worker 进程设 vnpy DB→quant_trading.db(修 0 根数据)
- equity_curve 取自 calculate_result 的 daily_df(修 get_all_daily_results 对象问题)
- kline 补 cfg(find_config_path 共享)
- 端到端冒烟通过:DoubleMaStrategy 600000 → equity111/pnl111/trades1/kline117
This commit is contained in:
2026-07-07 06:21:17 +08:00
parent 80d7f58589
commit 28aea67232
5 changed files with 132 additions and 10 deletions
+3
View File
@@ -16,7 +16,10 @@ def load_kline(symbol: str, start: str, end: str, cfg=None) -> list[dict]:
cfg: Optional data config; None uses default data_platform.yaml.
"""
from sanguo_data.datareader import read_db_daily
from sanguo_data.config import load_config, find_config_path
if cfg is None:
cfg = load_config(find_config_path())
bars = read_db_daily(symbol, start, end, cfg)
return [
{
+9 -3
View File
@@ -6,7 +6,7 @@ from pydantic import BaseModel
from .schemas import CtaBacktestRequest, OptimizeRequest, FactorAnalysisRequest
from .auth import verify_token as verify_token_impl, verify_password, create_token
from .ws import manager
from .strategy_registry import list_strategies, strategy_params
from .strategy_registry import list_strategies, strategy_params, get_strategy_class
from .kline import load_kline
@@ -60,8 +60,11 @@ def login(req: LoginRequest):
@router.post("/backtest/cta", dependencies=[Depends(verify_token)])
async def submit_cta(req: CtaBacktestRequest):
"""Submit CTA backtest task"""
cls = get_strategy_class(req.strategy)
if cls is None:
raise HTTPException(status_code=400, detail=f"未知策略: {req.strategy}")
tid = await get_orchestrator().submit_cta(
strategy_class=req.strategy,
strategy_class=cls,
symbol=req.symbol,
params=req.params,
start=req.start,
@@ -74,8 +77,11 @@ async def submit_cta(req: CtaBacktestRequest):
@router.post("/backtest/optimize", dependencies=[Depends(verify_token)])
async def submit_optimize(req: OptimizeRequest):
"""Submit optimization task"""
cls = get_strategy_class(req.strategy)
if cls is None:
raise HTTPException(status_code=400, detail=f"未知策略: {req.strategy}")
tid = await get_orchestrator().submit_optimize(
strategy_class=req.strategy,
strategy_class=cls,
symbol=req.symbol,
grid=req.grid,
start=req.start,
+25 -7
View File
@@ -94,6 +94,17 @@ def run_cta_backtest(strategy_class, symbol: str, params: dict, start: str, end:
# Add strategy
engine.add_strategy(strategy_class, params)
# Configure vnpy DB → A-share quant_trading.db. Worker process (spawn)
# doesn't inherit main-process SETTINGS, so set before engine.load_data.
try:
from vnpy.trader.setting import SETTINGS
from sanguo_data.config import load_config, find_config_path
_dcfg = load_config(find_config_path())
SETTINGS["database.name"] = "sqlite"
SETTINGS["database.database"] = _dcfg.data_paths["vnpy_db"]
except Exception:
pass
# Load historical data
engine.load_data()
@@ -110,13 +121,20 @@ def run_cta_backtest(strategy_class, symbol: str, params: dict, start: str, end:
for k, v in raw_stats.items()
}
# Build equity curve DataFrame (S1.2): engine.get_all_daily_results()
# returns a list of dicts; keep date + balance for the chart + parquet.
daily_results = engine.get_all_daily_results()
if isinstance(daily_results, list) and daily_results:
equity_df = pd.DataFrame(daily_results)
cols = [c for c in ("date", "balance") if c in equity_df.columns]
equity_df = equity_df[cols] if cols else pd.DataFrame()
# 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.
if daily_df is not None and hasattr(daily_df, "empty") and not daily_df.empty:
if "balance" in daily_df.columns:
_bal = daily_df["balance"].astype(float)
elif "net_pnl" in daily_df.columns:
_bal = daily_df["net_pnl"].astype(float).cumsum() + 1_000_000
else:
_bal = None
equity_df = pd.DataFrame({
"date": daily_df.index.astype(str),
"balance": _bal.tolist(),
}) if _bal is not None else pd.DataFrame()
else:
equity_df = pd.DataFrame()
+14
View File
@@ -1,5 +1,6 @@
# sanguo_data/config.py
from dataclasses import dataclass
import os
import yaml
@dataclass(frozen=True)
@@ -27,3 +28,16 @@ def load_config(path: str) -> DataConfig:
validation=raw.get("validation", {}),
performance=raw.get("performance", {}),
)
def find_config_path() -> str:
"""Locate data_platform.yaml: container /app/config first, then repo config/."""
candidates = [
"/app/config/data_platform.yaml",
os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config", "data_platform.yaml"),
"config/data_platform.yaml",
]
for p in candidates:
if os.path.exists(p):
return p
return candidates[0]
+81
View File
@@ -0,0 +1,81 @@
#!/usr/bin/env python3
"""Phase 3b S1 end-to-end smoke.
Login -> submit CTA backtest (DoubleMaStrategy on 600000) -> poll status ->
verify the result-page endpoints (equity-curve / daily-pnl / trades / kline)
return non-empty data.
Runs from the Mac against the NAS container (http://192.168.2.154:8000).
No third-party deps (urllib only).
"""
import json
import sys
import time
import urllib.request
BASE = "http://192.168.2.154:8000"
def _request(method: str, path: str, token: str | None = None, body: dict | None = None) -> dict:
headers = {"Content-Type": "application/json"}
if token:
headers["Authorization"] = f"Bearer {token}"
data = json.dumps(body).encode() if body is not None else None
req = urllib.request.Request(BASE + path, data=data, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=30) as resp:
return json.loads(resp.read())
def main() -> int:
tok = _request("POST", "/api/v1/auth/login", body={"username": "admin", "password": "admin"})["token"]
print("[1] login OK")
sub = _request("POST", "/api/v1/backtest/cta", token=tok, body={
"symbol": "600000",
"strategy": "DoubleMaStrategy",
"params": {"fast_window": 10, "slow_window": 20, "fixed_size": 1},
"start": "2024-01-01",
"end": "2024-06-30",
})
tid = sub["task_id"]
print(f"[2] submitted: {tid}")
status = "pending"
for i in range(60):
s = _request("GET", f"/api/v1/task/{tid}", token=tok)
status = s["status"]
print(f" [{i:02d}] status={status} stage={s.get('stage', '')}")
if status in ("done", "failed"):
break
time.sleep(3)
if status != "done":
print(f"[!] backtest did not complete: {status}")
return 1
eq = _request("GET", f"/api/v1/task/{tid}/equity-curve", token=tok)
pnl = _request("GET", f"/api/v1/task/{tid}/daily-pnl", token=tok)
tr = _request("GET", f"/api/v1/task/{tid}/trades", token=tok)
kl = _request("GET", "/api/v1/kline?symbol=600000&start=2024-01-01&end=2024-06-30", token=tok)
n_eq = len(eq.get("equity_curve", []))
n_pnl = len(pnl.get("daily_pnl", []))
n_tr = len(tr.get("trades", []))
n_kl = len(kl.get("kline", []))
print(f"[3] equity={n_eq} pnl={n_pnl} trades={n_tr} kline={n_kl}")
assert n_eq > 0, "equity_curve empty"
assert n_kl > 0, "kline empty"
print("[4] SMOKE PASSED")
return 0
if __name__ == "__main__":
try:
sys.exit(main())
except AssertionError as e:
print(f"[SMOKE FAILED] {e}")
sys.exit(2)
except Exception as e:
print(f"[SMOKE ERROR] {type(e).__name__}: {e}")
sys.exit(3)