claude_dev
28aea67232
- 修 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
82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
#!/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)
|