fix(backtest): result_id 用 DB 行 id + equity/trades 落 JSON(S1.1+S1.2)

- BacktestResult 加 id;save_result 设 result.id=lastrowid(修 get_result bug)
- runner._on_done 用 result.id(getattr 兜底 FactorReport)
- cta_engine 构建 equity_curve/trades DataFrame;save 传 file_dir
- result_store parquet→JSON(去 pyarrow 依赖,本机/容器都稳)
- 16 tests passed
This commit is contained in:
2026-07-07 06:06:10 +08:00
parent 198321c4a9
commit 510f77e6ea
13 changed files with 904 additions and 13 deletions
+9 -7
View File
@@ -22,6 +22,7 @@ class BacktestResult:
equity_curve: Optional[pd.DataFrame] = None
trades: Optional[pd.DataFrame] = None
error_msg: Optional[str] = None
id: Optional[int] = None
# SQLite schema for backtest stats
@@ -73,12 +74,12 @@ def save_result(result: BacktestResult, db_path: str, file_dir: Optional[str] =
fdir.mkdir(parents=True, exist_ok=True)
if result.equity_curve is not None and not result.equity_curve.empty:
equity_path = str(fdir / f"{result.task_id}_equity.parquet")
result.equity_curve.to_parquet(equity_path)
equity_path = str(fdir / f"{result.task_id}_equity.json")
result.equity_curve.to_json(equity_path, orient="records", date_format="iso", force_ascii=False)
if result.trades is not None and not result.trades.empty:
trades_path = str(fdir / f"{result.task_id}_trades.parquet")
result.trades.to_parquet(trades_path)
trades_path = str(fdir / f"{result.task_id}_trades.json")
result.trades.to_json(trades_path, orient="records", date_format="iso", force_ascii=False)
# Insert record into database
cur = conn.execute(
@@ -102,6 +103,7 @@ def save_result(result: BacktestResult, db_path: str, file_dir: Optional[str] =
)
)
conn.commit()
result.id = cur.lastrowid
return cur.lastrowid
finally:
conn.close()
@@ -131,9 +133,9 @@ def load_result(rid: int, db_path: str) -> BacktestResult:
cols = [d[0] for d in conn.execute("SELECT * FROM backtest_stats LIMIT 0").description]
d = dict(zip(cols, row))
# Load parquet files if paths exist
equity = pd.read_parquet(d["equity_path"]) if d.get("equity_path") else None
trades = pd.read_parquet(d["trades_path"]) if d.get("trades_path") else None
# Load JSON files if paths exist (equity_curve/trades persisted as JSON)
equity = pd.read_json(d["equity_path"], orient="records") if d.get("equity_path") else None
trades = pd.read_json(d["trades_path"], orient="records") if d.get("trades_path") else None
return BacktestResult(
task_id=d["task_id"],