94 lines
2.6 KiB
Python
94 lines
2.6 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import json
|
||
|
|
import sqlite3
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
|
||
|
|
def fetch_top(con: sqlite3.Connection, run_id: str, limit: int) -> list[dict[str, Any]]:
|
||
|
|
cols = [r[1] for r in con.execute("PRAGMA table_info(trials)")]
|
||
|
|
sql = "SELECT * FROM trials WHERE run_id = ? ORDER BY ann_return DESC LIMIT ?"
|
||
|
|
rows = []
|
||
|
|
for r in con.execute(sql, [run_id, int(limit)]):
|
||
|
|
rows.append(dict(zip(cols, r)))
|
||
|
|
return rows
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
ap = argparse.ArgumentParser()
|
||
|
|
ap.add_argument("--state", default="data/opt_state.json")
|
||
|
|
ap.add_argument("--db", default="data/experiments.sqlite")
|
||
|
|
ap.add_argument("--top", type=int, default=5)
|
||
|
|
args = ap.parse_args()
|
||
|
|
|
||
|
|
state_path = Path(args.state)
|
||
|
|
state = json.loads(state_path.read_text(encoding="utf-8"))
|
||
|
|
|
||
|
|
hist = state.get("history") or []
|
||
|
|
if not hist:
|
||
|
|
raise SystemExit("no history in opt_state.json")
|
||
|
|
|
||
|
|
last = hist[-1]
|
||
|
|
run_id = str(last.get("run_id"))
|
||
|
|
|
||
|
|
best = state.get("best")
|
||
|
|
|
||
|
|
print("last_run_id", run_id)
|
||
|
|
print("last_run", {k: last.get(k) for k in ["timestamp", "seed", "trials", "jobs", "best_ann_return", "code_version"] if k in last})
|
||
|
|
if best:
|
||
|
|
print(
|
||
|
|
"global_best",
|
||
|
|
{
|
||
|
|
"ann_return": best.get("ann_return"),
|
||
|
|
"ann_vol": best.get("ann_vol"),
|
||
|
|
"max_drawdown": best.get("max_drawdown"),
|
||
|
|
"sharpe": best.get("sharpe"),
|
||
|
|
"trades_per_year": best.get("trades_per_year"),
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
db_path = Path(args.db)
|
||
|
|
with sqlite3.connect(str(db_path)) as con:
|
||
|
|
rows = fetch_top(con, run_id=run_id, limit=int(args.top))
|
||
|
|
|
||
|
|
if not rows:
|
||
|
|
print("no rows for run_id")
|
||
|
|
return
|
||
|
|
|
||
|
|
def slim(r: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
keys = [
|
||
|
|
"id",
|
||
|
|
"trial",
|
||
|
|
"ann_return",
|
||
|
|
"ann_vol",
|
||
|
|
"max_drawdown",
|
||
|
|
"sharpe",
|
||
|
|
"trades_per_year",
|
||
|
|
"sma_fast",
|
||
|
|
"sma_slow",
|
||
|
|
"lazy_days",
|
||
|
|
"min_hold_days",
|
||
|
|
"replace_score_gap",
|
||
|
|
"min_score",
|
||
|
|
"macro_min_breadth",
|
||
|
|
"macro_down_frac",
|
||
|
|
"desired_positions_min",
|
||
|
|
"atr_mult",
|
||
|
|
"stop_loss_atr",
|
||
|
|
"profit_tighten_atr",
|
||
|
|
"atr_mult_profit",
|
||
|
|
"bias_exit",
|
||
|
|
"vol_ratio_exit",
|
||
|
|
]
|
||
|
|
return {k: r.get(k) for k in keys if k in r}
|
||
|
|
|
||
|
|
print("top_trials")
|
||
|
|
for r in rows:
|
||
|
|
print(json.dumps(slim(r), ensure_ascii=False))
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|