initial import: etf strategy project

scripts/report_last_run.py

@@ -0,0 +1,93 @@
+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()