claude_dev
7fe3fb0844
根因: empyrical 0.5.5 引用 numpy2.0 已移除的 np.NINF → compute_metrics 静默崩 → _metrics.json 不生成 → 结果页回退 vnpy 原始字段(单位混乱: total_return当百分数、max_drawdown当元 → 前端×100显 3305%/-50M%)。 - metrics.py: 导入 empyrical 前补回 np 别名(NINF/Inf/PINF/NaN/NAN/infty) - routes.py: benchmark-curve/risk-series 缺 metrics 文件时返空200(不再404拖垮整页); get_result 从 statistics 抽 relative_metrics - cta_engine.py: bench_df 日期 strip tz 防 pct_change 崩; metrics 块加 traceback 日志 - Result.vue: onMounted 用 Promise.allSettled 隔离7端点, 单接口失败不拖垮整页 - result_store.py: _safe_read_json 容错迁移后残留 NAS 绝对路径, stale path 不崩 list_results - datareader.py: read_index_daily 改从 vnpy DB 读 + 前缀解析交易所(sh→SSE, 避免 000300 被 guess_exchange 误判 SZSE)
445 lines
15 KiB
Python
445 lines
15 KiB
Python
"""
|
|
FastAPI routes for Sanguo Quant API
|
|
"""
|
|
import os
|
|
from fastapi import APIRouter, HTTPException, Depends, WebSocket, Query, Header
|
|
from fastapi.responses import FileResponse
|
|
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, get_strategy_class
|
|
from .kline import load_kline
|
|
|
|
|
|
router = APIRouter()
|
|
_orchestrator = None
|
|
_auth_config = {"username": "admin", "password_hash": "", "jwt_secret": "x", "expire_minutes": 60}
|
|
|
|
|
|
def set_orchestrator(orch):
|
|
"""Set the global orchestrator instance"""
|
|
global _orchestrator
|
|
_orchestrator = orch
|
|
|
|
|
|
def get_orchestrator():
|
|
"""Get the global orchestrator instance"""
|
|
return _orchestrator
|
|
|
|
|
|
def set_auth_config(cfg):
|
|
"""Set authentication configuration"""
|
|
_auth_config.update(cfg)
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
"""Login request schema"""
|
|
username: str
|
|
password: str
|
|
|
|
|
|
async def verify_token(authorization: str | None = Header(None)):
|
|
"""Dependency to verify JWT token from Authorization header"""
|
|
if authorization is None:
|
|
raise HTTPException(status_code=401, detail="Missing authorization header")
|
|
|
|
if not authorization.startswith("Bearer "):
|
|
raise HTTPException(status_code=401, detail="Invalid authorization header format")
|
|
|
|
token = authorization.split(" ")[1]
|
|
return verify_token_impl(token)
|
|
|
|
|
|
@router.post("/auth/login")
|
|
def login(req: LoginRequest):
|
|
"""Authenticate user and return JWT token"""
|
|
if req.username != _auth_config["username"] or not verify_password(req.password, _auth_config["password_hash"]):
|
|
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
|
return {"token": create_token(req.username)}
|
|
|
|
|
|
@router.post("/backtest/cta", dependencies=[Depends(verify_token)])
|
|
async def submit_cta(req: CtaBacktestRequest):
|
|
"""Submit CTA backtest task"""
|
|
# Validate benchmark parameter
|
|
if req.benchmark not in ("hs300", "zz500"):
|
|
raise HTTPException(status_code=422, detail=f"Invalid benchmark: {req.benchmark}. Must be 'hs300' or 'zz500'")
|
|
|
|
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=cls,
|
|
symbol=req.symbol,
|
|
params=req.params,
|
|
start=req.start,
|
|
end=req.end,
|
|
cfg=None,
|
|
benchmark=req.benchmark,
|
|
capital=req.capital,
|
|
position_pct=req.position_pct
|
|
)
|
|
return {"task_id": tid}
|
|
|
|
|
|
@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=cls,
|
|
symbol=req.symbol,
|
|
grid=req.grid,
|
|
start=req.start,
|
|
end=req.end,
|
|
cfg=None
|
|
)
|
|
return {"task_id": tid}
|
|
|
|
|
|
@router.post("/factor/analyze", dependencies=[Depends(verify_token)])
|
|
async def submit_factor(req: FactorAnalysisRequest):
|
|
"""Submit factor analysis task"""
|
|
tid = await get_orchestrator().submit_factor(
|
|
symbols=req.symbols,
|
|
factor_names=req.factor_names,
|
|
start=req.start,
|
|
end=req.end,
|
|
cfg=None,
|
|
output_dir="/tmp/factor"
|
|
)
|
|
return {"task_id": tid}
|
|
|
|
|
|
@router.get("/task/{task_id}", dependencies=[Depends(verify_token)])
|
|
def get_status(task_id: str):
|
|
"""Get task status"""
|
|
s = get_orchestrator().get_status(task_id)
|
|
if s is None:
|
|
raise HTTPException(status_code=404, detail="task not found")
|
|
pool = get_orchestrator().pool
|
|
stage = pool.get_stage(task_id)
|
|
task = pool.get_task(task_id)
|
|
return {
|
|
"task_id": task_id,
|
|
"status": s.value if hasattr(s, "value") else str(s),
|
|
"stage": stage or "",
|
|
"error_msg": task.error_msg if (task and isinstance(task.error_msg, str)) else None,
|
|
}
|
|
|
|
|
|
@router.get("/task/{task_id}/result", dependencies=[Depends(verify_token)])
|
|
def get_result(task_id: str):
|
|
"""Get task result"""
|
|
r = get_orchestrator().get_result(task_id)
|
|
if r is None:
|
|
raise HTTPException(status_code=404, detail="result not ready")
|
|
|
|
# Extract relative metrics from statistics
|
|
relative_metrics = {}
|
|
relative_fields = [
|
|
"total_return", "annual_return", "sharpe_ratio", "max_drawdown",
|
|
"alpha", "beta", "sortino_ratio", "information_ratio",
|
|
"annual_volatility", "benchmark_return", "benchmark_volatility",
|
|
]
|
|
for field in relative_fields:
|
|
if field in r.statistics:
|
|
relative_metrics[field] = r.statistics[field]
|
|
|
|
return {
|
|
"task_id": task_id,
|
|
"statistics": r.statistics,
|
|
"relative_metrics": relative_metrics,
|
|
"symbol": r.symbol,
|
|
"start": r.start,
|
|
"end": r.end,
|
|
"strategy": r.strategy,
|
|
"params": r.params,
|
|
"status": r.status,
|
|
}
|
|
|
|
|
|
@router.websocket("/ws/task/{task_id}")
|
|
async def task_ws(websocket: WebSocket, task_id: str, token: str = Query(...)):
|
|
"""WebSocket endpoint for task status updates"""
|
|
try:
|
|
verify_token(token)
|
|
except Exception:
|
|
await websocket.close(code=4401)
|
|
return
|
|
|
|
await websocket.accept()
|
|
manager.connect(task_id, websocket)
|
|
|
|
try:
|
|
while True:
|
|
await websocket.receive_text() # Keep connection alive
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
manager.disconnect(task_id, websocket)
|
|
|
|
|
|
# ===== Backtest UI support endpoints (S1.4 / S1.5) =====
|
|
|
|
def _df_to_records(df) -> list[dict]:
|
|
"""DataFrame → list[dict] (empty-safe)."""
|
|
if df is None:
|
|
return []
|
|
if hasattr(df, "empty") and df.empty:
|
|
return []
|
|
if hasattr(df, "to_dict"):
|
|
return df.to_dict(orient="records")
|
|
return list(df)
|
|
|
|
|
|
@router.get("/strategy/list", dependencies=[Depends(verify_token)])
|
|
def strategy_list():
|
|
"""List available CTA strategies for the UI dropdown."""
|
|
return {"strategies": list_strategies()}
|
|
|
|
|
|
@router.get("/strategy/{name}/params", dependencies=[Depends(verify_token)])
|
|
def strategy_params_route(name: str):
|
|
"""Strategy parameters + defaults for the dynamic form."""
|
|
return strategy_params(name)
|
|
|
|
|
|
@router.get("/task/{task_id}/equity-curve", dependencies=[Depends(verify_token)])
|
|
def equity_curve(task_id: str):
|
|
r = get_orchestrator().get_result(task_id)
|
|
if r is None:
|
|
raise HTTPException(status_code=404, detail="result not ready")
|
|
return {"task_id": task_id, "equity_curve": _df_to_records(r.equity_curve)}
|
|
|
|
|
|
@router.get("/task/{task_id}/daily-pnl", dependencies=[Depends(verify_token)])
|
|
def daily_pnl(task_id: str):
|
|
r = get_orchestrator().get_result(task_id)
|
|
if r is None:
|
|
raise HTTPException(status_code=404, detail="result not ready")
|
|
ec = r.equity_curve
|
|
if ec is None or (hasattr(ec, "empty") and ec.empty) or "balance" not in ec.columns:
|
|
return {"task_id": task_id, "daily_pnl": []}
|
|
import pandas as pd
|
|
bal = pd.to_numeric(ec["balance"], errors="coerce")
|
|
pnl = bal.diff().fillna(0.0)
|
|
return {
|
|
"task_id": task_id,
|
|
"daily_pnl": [{"date": str(d), "pnl": float(p)} for d, p in zip(ec["date"], pnl)],
|
|
}
|
|
|
|
|
|
@router.get("/task/{task_id}/trades", dependencies=[Depends(verify_token)])
|
|
def trades_route(task_id: str):
|
|
r = get_orchestrator().get_result(task_id)
|
|
if r is None:
|
|
raise HTTPException(status_code=404, detail="result not ready")
|
|
return {"task_id": task_id, "trades": _df_to_records(r.trades)}
|
|
|
|
|
|
@router.get("/kline", dependencies=[Depends(verify_token)])
|
|
def kline(symbol: str, start: str, end: str):
|
|
"""Historical daily K-line for the backtest chart."""
|
|
try:
|
|
return {"symbol": symbol, "kline": load_kline(symbol, start, end)}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"kline load failed: {type(e).__name__}: {e}")
|
|
|
|
|
|
# ===== Factor (投研) endpoints (S2) =====
|
|
|
|
@router.get("/factor/list", dependencies=[Depends(verify_token)])
|
|
def factor_list():
|
|
"""List registered factors for the UI dropdown."""
|
|
from sanguo_factor.registry import list_factors
|
|
items = [{"name": f["name"], "category": f.get("category", "")} for f in list_factors()]
|
|
return {"factors": items}
|
|
|
|
|
|
@router.get("/task/{task_id}/ic-summary", dependencies=[Depends(verify_token)])
|
|
def ic_summary(task_id: str):
|
|
"""Factor IC summary (mean/std/icir/t_stat per period)."""
|
|
r = get_orchestrator().get_raw_result(task_id)
|
|
if r is None:
|
|
raise HTTPException(status_code=404, detail="result not ready")
|
|
ic = getattr(r, "ic_summary", None)
|
|
if ic is None:
|
|
raise HTTPException(status_code=404, detail="no ic_summary (not a factor result?)")
|
|
return {"task_id": task_id, "ic_summary": ic}
|
|
|
|
|
|
@router.get("/task/{task_id}/report/{factor}")
|
|
def factor_report(task_id: str, factor: str, token: str = Query(...)):
|
|
"""Serve the alphalens tears HTML report (token via query for iframe use)."""
|
|
try:
|
|
verify_token_impl(token)
|
|
except Exception:
|
|
raise HTTPException(status_code=401, detail="invalid token")
|
|
r = get_orchestrator().get_raw_result(task_id)
|
|
if r is None:
|
|
raise HTTPException(status_code=404, detail="result not ready")
|
|
paths = getattr(r, "report_paths", {}) or {}
|
|
path = paths.get(factor)
|
|
if not path or not os.path.exists(path):
|
|
raise HTTPException(status_code=404, detail=f"report for {factor} not found")
|
|
return FileResponse(path)
|
|
|
|
|
|
# ===== History + Optimization endpoints (S3) =====
|
|
|
|
@router.get("/task", dependencies=[Depends(verify_token)])
|
|
def list_tasks(type: str | None = None, status: str | None = None):
|
|
"""List historical tasks (from the results DB)."""
|
|
from sanguo_backtest.result_store import list_results
|
|
orch = get_orchestrator()
|
|
items = []
|
|
for r in list_results(type_filter=type, db_path=orch.db_path):
|
|
if status and r.status != status:
|
|
continue
|
|
items.append({
|
|
"id": r.id,
|
|
"task_id": r.task_id,
|
|
"type": r.type,
|
|
"status": r.status,
|
|
"strategy": r.strategy,
|
|
"symbol": r.symbol,
|
|
"start": r.start,
|
|
"end": r.end,
|
|
})
|
|
items.reverse() # newest first
|
|
return {"tasks": items}
|
|
|
|
|
|
@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)
|
|
if raw is None:
|
|
raise HTTPException(status_code=404, detail="optimization results not ready")
|
|
rows = []
|
|
for r in (raw if isinstance(raw, list) else [raw]):
|
|
rows.append({
|
|
"params": getattr(r, "params", {}),
|
|
"statistics": getattr(r, "statistics", {}),
|
|
})
|
|
return {"task_id": task_id, "results": rows}
|
|
|
|
|
|
# ===== Task 4: Backtest result API extensions =====
|
|
|
|
def _get_metrics_file_path(task_id: str) -> str | None:
|
|
"""Get the metrics file path. Tries the request task_id directly first, then
|
|
falls back to the result's internal task_id — run_cta_backtest writes the file
|
|
under ITS OWN uuid task_id (result.task_id), which differs from the runner's
|
|
task_id when submitted via the API."""
|
|
orch = get_orchestrator()
|
|
if not (hasattr(orch, 'db_path') and orch.db_path):
|
|
return None
|
|
file_dir = os.path.dirname(os.path.abspath(orch.db_path))
|
|
# 1. direct lookup by request task_id
|
|
direct = os.path.join(file_dir, f"{task_id}_metrics.json")
|
|
if os.path.exists(direct):
|
|
return direct
|
|
# 2. resolve via result.task_id (run_cta_backtest's own uuid)
|
|
r = orch.get_result(task_id)
|
|
if r is not None and getattr(r, "task_id", None):
|
|
resolved = os.path.join(file_dir, f"{r.task_id}_metrics.json")
|
|
if os.path.exists(resolved):
|
|
return resolved
|
|
return None
|
|
|
|
|
|
@router.get("/task/{task_id}/benchmark-curve", dependencies=[Depends(verify_token)])
|
|
def benchmark_curve(task_id: str):
|
|
"""Get benchmark curve data (strategy vs benchmark)."""
|
|
metrics_file = _get_metrics_file_path(task_id)
|
|
if not metrics_file:
|
|
# metrics 文件缺失时返回空 200（图表优雅降级），不再 404 触发前端整页空白
|
|
return {"dates": [], "strategy": [], "benchmark": []}
|
|
|
|
import json
|
|
with open(metrics_file, 'r') as f:
|
|
data = json.load(f)
|
|
|
|
series = data.get("series", {})
|
|
equity_curve = series.get("equity_curve", {})
|
|
benchmark_curve = series.get("benchmark_curve", {})
|
|
|
|
return {
|
|
"dates": equity_curve.get("dates", []),
|
|
"strategy": equity_curve.get("values", []),
|
|
"benchmark": benchmark_curve.get("values", [])
|
|
}
|
|
|
|
|
|
@router.get("/task/{task_id}/risk-series", dependencies=[Depends(verify_token)])
|
|
def risk_series(task_id: str):
|
|
"""Get risk series data (alpha, beta, drawdown)."""
|
|
metrics_file = _get_metrics_file_path(task_id)
|
|
if not metrics_file:
|
|
# metrics 文件缺失时返回空 200（图表优雅降级），不再 404 触发前端整页空白
|
|
return {"dates": [], "alpha": [], "beta": [], "drawdown": [], "strategy_vol": [], "benchmark_vol": []}
|
|
|
|
import json
|
|
with open(metrics_file, 'r') as f:
|
|
data = json.load(f)
|
|
|
|
series = data.get("series", {})
|
|
alpha = series.get("alpha", {})
|
|
beta = series.get("beta", {})
|
|
drawdown = series.get("drawdown", {})
|
|
vol_s = series.get("volatility_strategy", {})
|
|
vol_b = series.get("volatility_benchmark", {})
|
|
|
|
return {
|
|
"dates": alpha.get("dates", []),
|
|
"alpha": alpha.get("values", []),
|
|
"beta": beta.get("values", []),
|
|
"drawdown": drawdown.get("values", []),
|
|
"strategy_vol": vol_s.get("values", []),
|
|
"benchmark_vol": vol_b.get("values", [])
|
|
}
|
|
|
|
|
|
@router.get("/task/{task_id}/daily-holdings", dependencies=[Depends(verify_token)])
|
|
def daily_holdings(task_id: str):
|
|
"""Get daily holdings data from equity curve."""
|
|
r = get_orchestrator().get_result(task_id)
|
|
if r is None:
|
|
raise HTTPException(status_code=404, detail="result not ready")
|
|
|
|
ec = r.equity_curve
|
|
if ec is None or (hasattr(ec, "empty") and ec.empty):
|
|
return {"task_id": task_id, "daily_holdings": []}
|
|
|
|
# Return balance and return data as holdings
|
|
holdings = []
|
|
for _, row in ec.iterrows():
|
|
holding = {"date": str(row.get("date", ""))}
|
|
if "balance" in row:
|
|
holding["balance"] = float(row["balance"])
|
|
if "return" in row:
|
|
holding["return"] = float(row["return"])
|
|
holdings.append(holding)
|
|
|
|
return {"task_id": task_id, "daily_holdings": holdings}
|
|
|
|
|
|
@router.get("/task/{task_id}/log", dependencies=[Depends(verify_token)])
|
|
def log_endpoint(task_id: str):
|
|
"""Get backtest log text."""
|
|
orch = get_orchestrator()
|
|
if hasattr(orch, 'db_path') and orch.db_path:
|
|
file_dir = os.path.dirname(os.path.abspath(orch.db_path))
|
|
log_file = os.path.join(file_dir, f"{task_id}.log")
|
|
if os.path.exists(log_file):
|
|
with open(log_file, 'r') as f:
|
|
log_content = f.read()
|
|
return {"task_id": task_id, "log": log_content}
|
|
|
|
# Return empty log if file doesn't exist
|
|
return {"task_id": task_id, "log": ""} |