claude_dev
54fc1b656f
- result_store.load_result_by_task_id + orchestrator.get_result DB 兜底（历史回看）
- GET /task 列表、GET /task/{id}/optimization-results
- Task.raw_result 存优化结果 list（内存）
- cta_optimizer 修同款 bug（interval d / capital 1M / vnpy DB SETTINGS）
- get_status 返回 error_msg（str 守卫）
- 前端 优化页（网格输入+轮询+结果表）、历史页（任务列表+回看）、侧栏子菜单
- 修 5 个旧 test_routes 回归；73 tests passed
- 冒烟：历史 3 任务 + 优化 9 组合
68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Phase 3b S3 smoke: history list + parameter optimization end-to-end."""
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
|
|
BASE = "http://192.168.2.154:8000"
|
|
|
|
|
|
def _req(method, path, token=None, body=None):
|
|
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=60) as resp:
|
|
return json.loads(resp.read())
|
|
|
|
|
|
def main() -> int:
|
|
tok = _req("POST", "/api/v1/auth/login", body={"username": "admin", "password": "admin"})["token"]
|
|
|
|
tasks = _req("GET", "/api/v1/task", token=tok)["tasks"]
|
|
print(f"[history] {len(tasks)} tasks; latest: {tasks[0]['task_id'] if tasks else 'none'}")
|
|
assert isinstance(tasks, list)
|
|
|
|
sub = _req("POST", "/api/v1/backtest/optimize", token=tok, body={
|
|
"symbol": "600000",
|
|
"strategy": "DoubleMaStrategy",
|
|
"grid": {"fast_window": [5, 15, 5], "slow_window": [15, 25, 5]},
|
|
"start": "2024-01-01",
|
|
"end": "2024-06-30",
|
|
"max_workers": 2,
|
|
})
|
|
tid = sub["task_id"]
|
|
print(f"[optimize] submitted: {tid}")
|
|
|
|
status = "pending"
|
|
for i in range(60):
|
|
s = _req("GET", f"/api/v1/task/{tid}", token=tok)
|
|
status = s["status"]
|
|
print(f" [{i:02d}] {status} {s.get('stage', '')}")
|
|
if status in ("done", "failed"):
|
|
break
|
|
time.sleep(3)
|
|
|
|
if status != "done":
|
|
print(f"[!] optimize failed: {status}")
|
|
return 1
|
|
|
|
res = _req("GET", f"/api/v1/task/{tid}/optimization-results", token=tok)["results"]
|
|
print(f"[results] {len(res)} param combos")
|
|
assert len(res) > 0, "no optimization results"
|
|
for r in res[:3]:
|
|
print(f" params={r['params']} sharpe={r['statistics'].get('sharpe_ratio')}")
|
|
print("SMOKE PASSED")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except AssertionError as e:
|
|
print(f"[FAILED] {e}"); sys.exit(2)
|
|
except Exception as e:
|
|
print(f"[ERROR] {type(e).__name__}: {e}"); sys.exit(3)
|