""" run_marathon.py — Autonomous multi-cycle optimizer for unattended runs. Alternates random → optuna search cycles across all 20 combos, applying sign-stability locks between cycles and sending ntfy progress notifications at each milestone. Cycle pattern: odd cycles → random search (broad exploration) even cycles → optuna search (exploit DB history) After every random cycle → lock_params --apply --cross-asset After every cycle → mine_sweep_db + write results/marathon_progress.md Notifications: - ntfy per-combo: brief verdict ping after each combo completes - ntfy per-cycle: full summary (tiers, locks, OOS hits, ETA) - results/marathon_progress.md updated after each cycle Usage: python3 run_marathon.py # 96h, 1h/combo, all 20 combos python3 run_marathon.py --hours 48 # 48h total python3 run_marathon.py --hours-per-combo 2 # 2h per combo per cycle python3 run_marathon.py --assets COINBASE_ETHUSD BINANCE_SOLUSD # alts only python3 run_marathon.py --start-mode optuna # start with optuna (if random already done) python3 run_marathon.py --dry-run # print plan only, don't run """ import subprocess import sys import os import argparse import time import sqlite3 import math import re from datetime import datetime, timedelta NTFY_TOPIC = "jlo_alerts" DB_FILE = "results/sweep_database.db" OOS_FILE = "results/oos_dashboard.md" PROGRESS_FILE = "results/marathon_progress.md" DB_SCHEMA_VER = 3 ALL_ASSETS = [ "COINBASE_BTCUSD", "COINBASE_ETHUSD", "BINANCE_SOLUSD", "BINANCE_LINKUSD", ] # Short-name aliases so users can type e.g. --assets BTCUSD instead of COINBASE_BTCUSD _ASSET_ALIASES = {a.split("_")[-1]: a for a in ALL_ASSETS} # e.g. {"BTCUSD": "COINBASE_BTCUSD", ...} TIMEFRAMES = ["4H", "6H", "8H", "12H", "1D"] _TF_TO_MIN = {"4H": "240", "6H": "360", "8H": "480", "12H": "720", "1D": "1D"} _ntfy_topic = NTFY_TOPIC # ── Notifications ────────────────────────────────────────────────────────── def ntfy(message, title=None): cmd = ["curl", "-s", "-H", f"Title: {title or 'Marathon'}", "-d", message, f"ntfy.sh/{_ntfy_topic}"] try: subprocess.run(cmd, timeout=10, capture_output=True) except Exception: pass # ── DB helpers ───────────────────────────────────────────────────────────── def tier_label(n): if n < 50: return "🔴 SPARSE" if n < 200: return "🟡 EMERGING" if n < 500: return "🟢 USABLE" if n < 2000: return "🟢🟢 RICH" return "🔵 DEEP" def get_db_counts(): """Return {(asset, tf): n_rows} for current schema version.""" if not os.path.exists(DB_FILE): return {} try: conn = sqlite3.connect(DB_FILE) rows = conn.execute( f"SELECT asset, timeframe, COUNT(*) FROM sweep_results " f"WHERE schema_ver={DB_SCHEMA_VER} GROUP BY asset, timeframe" ).fetchall() conn.close() return {(a, t): n for a, t, n in rows} except Exception: return {} def tier_summary_str(counts, assets, tfs): buckets = {"🔵 DEEP": 0, "🟢🟢 RICH": 0, "🟢 USABLE": 0, "🟡 EMERGING": 0, "🔴 SPARSE": 0} for a in assets: for t in tfs: n = counts.get((a, t), 0) buckets[tier_label(n)] += 1 parts = [f"{v}×{k.split()[-1]}" for k, v in buckets.items() if v] return " | ".join(parts) # ── OOS dashboard reader ─────────────────────────────────────────────────── def read_oos_table(): """Return list of OOS row dicts from oos_dashboard.md.""" rows = [] if not os.path.exists(OOS_FILE): return rows try: with open(OOS_FILE) as f: for line in f: if "|" not in line or "---" in line or "Asset" in line: continue parts = [p.strip() for p in line.strip().strip("|").split("|")] if len(parts) >= 10: rows.append({ "asset": parts[0], "tf": parts[1], "is_sort": parts[5], "oos_pnl": parts[8], "verdict": parts[-1], }) except Exception: pass return rows def get_oos_verdict(asset, tf): """Look up the OOS verdict for a specific combo from oos_dashboard.md.""" short = asset.split("_")[-1] # e.g. BTCUSD for row in read_oos_table(): if (row["asset"] == asset or row["asset"] == short) and row["tf"] == tf: return row["verdict"], row["oos_pnl"], row["is_sort"] return "—", "—", "—" def get_oos_highlights(): """Return list of EXCELLENT/ACCEPT lines.""" hits = [] for row in read_oos_table(): if "✅" in row["verdict"]: hits.append(f"{row['asset']} {row['tf']}: {row['oos_pnl']} {row['verdict']}") return hits # ── Lock application ─────────────────────────────────────────────────────── def apply_locks(): cmd = [sys.executable, "tools/lock_params.py", "--apply", "--cross-asset"] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) output = result.stdout + result.stderr applied = len(re.findall(r"APPLY\s+Lock", output)) skipped = len(re.findall(r"SKIP\s+Lock", output)) return applied, skipped, output except Exception as e: return 0, 0, str(e) # ── Dashboard update ─────────────────────────────────────────────────────── def update_dashboard(): try: subprocess.run([sys.executable, "mine_sweep_db.py"], capture_output=True, timeout=120) except Exception: pass # ── Progress markdown ────────────────────────────────────────────────────── def write_progress_md(run_state): """Write results/marathon_progress.md from run_state dict.""" now = datetime.now() s = run_state counts = get_db_counts() oos_rows = read_oos_table() oos_lookup = {(r["asset"], r["tf"]): r for r in oos_rows} lines = [] lines.append("# Marathon Progress") lines.append("") lines.append(f"**Started:** {s['start_time']:%Y-%m-%d %H:%M} ") lines.append(f"**Updated:** {now:%Y-%m-%d %H:%M} ") lines.append(f"**Plan:** {s['max_cycles']} cycles × {s['n_combos']} combos × {s['hours_per_combo']:.1f}h = ~{s['total_hours']:.0f}h ") lines.append(f"**ETA:** {s['eta']:%Y-%m-%d %H:%M} ") lines.append(f"**ntfy topic:** `{_ntfy_topic}` ") lines.append("") # Status elapsed_h = (time.time() - s["wall_start"]) / 3600 lines.append("## Status") lines.append("") lines.append(f"- Cycle **{s['current_cycle']}/{s['max_cycles']}** `[{s['current_mode']}]` — " f"{'IN PROGRESS' if s['current_cycle'] <= s['max_cycles'] else 'COMPLETE'}") lines.append(f"- Elapsed: {elapsed_h:.1f}h / {s['hours']:.0f}h budget") lines.append(f"- Runs: {s['total_ok']} OK / {s['total_fail']} failed") lines.append(f"- Locks applied (total): {s['cumulative_locks']}") lines.append("") # DB tier table lines.append("## DB Tier Progress") lines.append("") lines.append("| Asset | TF | Rows | Tier |") lines.append("|---|---|---|---|") for a in s["assets"]: for t in s["tfs"]: n = counts.get((a, t), 0) lines.append(f"| {a} | {t} | {n:,} | {tier_label(n)} |") lines.append("") # OOS dashboard snapshot lines.append("## OOS Dashboard (latest)") lines.append("") lines.append("| Asset | TF | IS Sort | OOS P&L | Verdict |") lines.append("|---|---|---|---|---|") for a in s["assets"]: for t in s["tfs"]: key = (a, t) if key in oos_lookup: r = oos_lookup[key] lines.append(f"| {a} | {t} | {r['is_sort']} | {r['oos_pnl']} | {r['verdict']} |") else: lines.append(f"| {a} | {t} | — | — | not run yet |") lines.append("") # Cycle history if s["cycle_history"]: lines.append("## Cycle History") lines.append("") for ch in s["cycle_history"]: lines.append(f"### Cycle {ch['num']}/{s['max_cycles']} [{ch['mode']}] " f"— {ch['elapsed_h']:.1f}h — {ch['ok']} OK / {ch['fail']} failed") lines.append("") if ch["locks"]: lines.append(f"- **Locks applied:** {ch['locks']}") lines.append("") if ch["combos"]: lines.append("| Combo | IS Sortino | OOS P&L | Verdict | DB rows |") lines.append("|---|---|---|---|---|") for cr in ch["combos"]: lines.append(f"| {cr['asset']}-{cr['tf']} | {cr['is_sort']} | " f"{cr['oos_pnl']} | {cr['verdict']} | +{cr['db_rows']:,} |") lines.append("") with open(PROGRESS_FILE, "w", encoding="utf-8") as f: f.write("\n".join(lines) + "\n") # ── Single combo run ─────────────────────────────────────────────────────── def run_combo(asset, tf, hours, search_mode, log_fh, regime="all", regime_threshold=15.0, tighten=False): data_path = f"data/mlp/{asset}, {_TF_TO_MIN.get(tf, tf)}.csv" label = f"{asset}-{tf} [{search_mode}]" cmd = [sys.executable, "tools/auto_optimize_loop.py", "--data", data_path, "--search", search_mode, "--hours", str(hours)] if regime != "all": cmd += ["--regime", regime, "--regime-threshold", str(regime_threshold)] if tighten: cmd += ["--tighten"] print(f" → {label} ({hours:.2f}h)", flush=True) log_fh.write(f"\n[{datetime.now():%H:%M:%S}] START {label}\n") log_fh.flush() db_before = get_db_counts().get((asset, tf), 0) start = time.time() try: result = subprocess.run(cmd, timeout=hours * 3600 * 1.5) elapsed = time.time() - start ok = result.returncode == 0 except subprocess.TimeoutExpired: elapsed = time.time() - start ok = False log_fh.write(f" TIMEOUT after {elapsed/60:.1f}m\n") db_after = get_db_counts().get((asset, tf), 0) db_added = db_after - db_before status = "OK" if ok else "FAIL" log_fh.write(f"[{datetime.now():%H:%M:%S}] END {label} {elapsed/60:.1f}m [{status}] +{db_added} rows\n") log_fh.flush() return ok, elapsed, db_added # ── Main ─────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Autonomous multi-cycle optimizer.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument("--hours", type=float, default=96.0, help="Total wall-clock budget in hours (default: 96)") parser.add_argument("--hours-per-combo", type=float, default=1.0, help="Hours per combo per cycle (default: 1.0)") parser.add_argument("--cycles", type=int, default=None, help="Max cycles to run (default: auto from budget)") parser.add_argument("--assets", nargs="*", default=None, metavar="ASSET") parser.add_argument("--tfs", nargs="*", default=None, metavar="TF") parser.add_argument("--start-mode", default="random", choices=["random", "optuna"], help="Search mode for cycle 1 (default: random)") parser.add_argument("--ntfy-topic", default=NTFY_TOPIC, help=f"ntfy topic (default: {NTFY_TOPIC})") parser.add_argument("--dry-run", action="store_true", help="Print plan and exit without running") parser.add_argument("--regime", type=str, default="all", choices=["all", "bull", "bear"], help="Regime filter for all combos: 'bull'/'bear' trains on regime-filtered IS data only. " "Use to build separate bull/bear param sets (Phase 4a). Default: 'all' (no filter).") parser.add_argument("--regime-threshold", type=float, default=15.0, help="MVRV zscore threshold for bull/bear split (default: 15.0).") parser.add_argument("--tighten", action="store_true", help="Auto-apply q5-q95 range tightening after each iteration (tighten_params.py --apply)") args = parser.parse_args() global _ntfy_topic _ntfy_topic = args.ntfy_topic assets = [_ASSET_ALIASES.get(a.upper(), a.upper()) for a in args.assets] if args.assets else ALL_ASSETS tfs = [tf.upper() for tf in args.tfs] if args.tfs else TIMEFRAMES combos = [(a, t) for a in assets for t in tfs] n = len(combos) hours_per_cycle = n * args.hours_per_combo max_cycles = args.cycles or max(1, math.floor(args.hours / hours_per_cycle)) # If one full cycle exceeds the budget, scale hours_per_combo down to fit if hours_per_cycle > args.hours: args.hours_per_combo = args.hours / n hours_per_cycle = n * args.hours_per_combo modes = [] mode = args.start_mode for _ in range(max_cycles): modes.append(mode) mode = "optuna" if mode == "random" else "random" total_hours = max_cycles * hours_per_cycle eta = datetime.now() + timedelta(hours=total_hours) print(f"\n{'='*65}") print(f" MARATHON OPTIMIZER") print(f" Assets: {', '.join(assets)}") print(f" Timeframes: {', '.join(tfs)}") print(f" Combos: {n}") print(f" Cycles: {max_cycles} ({' → '.join(modes)})") print(f" Per combo/cycle: {args.hours_per_combo:.1f}h") print(f" Per cycle: {hours_per_cycle:.1f}h") print(f" Total budget: {args.hours:.0f}h (using ~{total_hours:.0f}h)") print(f" ETA: {eta:%Y-%m-%d %H:%M}") print(f" ntfy topic: {_ntfy_topic}") print(f" Progress file: {PROGRESS_FILE} (written after each cycle)") if args.regime != "all": print(f" Regime filter: {args.regime} (zscore threshold={args.regime_threshold})") print(f"{'='*65}\n") if args.dry_run: for i, m in enumerate(modes, 1): print(f" Cycle {i}/{max_cycles}: {m} ({n} combos × {args.hours_per_combo:.1f}h = {hours_per_cycle:.0f}h)") if m == "random": print(f" → lock_params --apply --cross-asset") print(f" → mine_sweep_db + write {PROGRESS_FILE}") return # ── Shared run state (for progress markdown) ─────────────────────────── wall_start = time.time() run_state = { "start_time": datetime.now(), "max_cycles": max_cycles, "n_combos": n, "hours_per_combo": args.hours_per_combo, "total_hours": total_hours, "hours": args.hours, "eta": eta, "assets": assets, "tfs": tfs, "wall_start": wall_start, "current_cycle": 1, "current_mode": modes[0], "total_ok": 0, "total_fail": 0, "cumulative_locks": 0, "cycle_history": [], } os.makedirs("output", exist_ok=True) log_path = f"output/marathon_{datetime.now():%Y%m%d_%H%M%S}.log" log_fh = open(log_path, "w", encoding="utf-8") log_fh.write(f"Marathon started: {datetime.now()}\n") log_fh.write(f"Cycles: {max_cycles} Modes: {' → '.join(modes)}\n\n") ntfy( f"Marathon started: {max_cycles} cycles × {n} combos × {args.hours_per_combo:.1f}h = ~{total_hours:.0f}h.\n" f"Plan: {' → '.join(modes)}. ETA {eta:%b %d %H:%M}.\n" f"Progress: github.com/jamesjlopez/TradingBot25/blob/main/{PROGRESS_FILE}", title="🚀 Marathon started" ) total_ok = 0 total_fail = 0 cumulative_locks = 0 for cycle_num, search_mode in enumerate(modes, 1): cycle_start = time.time() run_state["current_cycle"] = cycle_num run_state["current_mode"] = search_mode print(f"\n{'='*65}") print(f" CYCLE {cycle_num}/{max_cycles} [{search_mode.upper()}] started {datetime.now():%Y-%m-%d %H:%M}") print(f"{'='*65}") log_fh.write(f"\n{'='*65}\nCYCLE {cycle_num}/{max_cycles} [{search_mode}] {datetime.now()}\n") cycle_ok = 0 cycle_fail = 0 cycle_combos = [] # records for progress MD for i, (asset, tf) in enumerate(combos, 1): elapsed_total = (time.time() - wall_start) / 3600 if elapsed_total + args.hours_per_combo > args.hours * 1.02: msg = (f"Budget reached after {elapsed_total:.1f}h. " f"Completed {cycle_num-1} full cycles + {i-1}/{n} in cycle {cycle_num}.") print(f"\n [BUDGET] {msg}") ntfy(msg, title="⏱ Marathon budget reached") log_fh.close() return remaining = n - i + n * (max_cycles - cycle_num) eta_final = datetime.now() + timedelta(hours=remaining * args.hours_per_combo) print(f"\n [{i}/{n}] {asset}-{tf} | cycle {cycle_num}/{max_cycles} | ETA {eta_final:%b %d %H:%M}") ok, elapsed, db_added = run_combo(asset, tf, args.hours_per_combo, search_mode, log_fh, regime=args.regime, regime_threshold=args.regime_threshold, tighten=args.tighten) verdict, oos_pnl, is_sort = get_oos_verdict(asset, tf) cycle_combos.append({ "asset": asset, "tf": tf, "is_sort": is_sort, "oos_pnl": oos_pnl, "verdict": verdict, "db_rows": db_added, "ok": ok, }) if ok: cycle_ok += 1; total_ok += 1 else: cycle_fail += 1; total_fail += 1 run_state["total_ok"] = total_ok run_state["total_fail"] = total_fail # Per-combo ntfy — brief verdict ping icon = "✅" if "✅" in verdict else ("⚠" if "⚠" in verdict else "❌") ntfy( f"[{cycle_num}/{max_cycles}] {asset.split('_')[-1]}-{tf} [{search_mode}]\n" f"IS Sortino: {is_sort} OOS: {oos_pnl} {verdict}\n" f"+{db_added:,} DB rows ({i}/{n} combos done)", title=f"{icon} {asset.split('_')[-1]}-{tf} done" ) cycle_elapsed_h = (time.time() - cycle_start) / 3600 # ── Post-cycle: apply locks ──────────────────────────────────────── locks_applied = 0 if search_mode == "random": print(f"\n [LOCKS] Applying sign-stability locks...") n_applied, n_skipped, lock_output = apply_locks() locks_applied = n_applied cumulative_locks += n_applied run_state["cumulative_locks"] = cumulative_locks log_fh.write(f"\n[LOCKS] Applied={n_applied} Skipped={n_skipped}\n{lock_output}\n") print(f" Applied {n_applied} lock(s), {n_skipped} already OK.") # ── Post-cycle: dashboard update ─────────────────────────────────── print(f" [DASHBOARD] Updating sweep dashboard...") update_dashboard() # ── Record cycle history ─────────────────────────────────────────── run_state["cycle_history"].append({ "num": cycle_num, "mode": search_mode, "elapsed_h": cycle_elapsed_h, "ok": cycle_ok, "fail": cycle_fail, "locks": locks_applied, "combos": cycle_combos, }) # ── Write progress markdown ──────────────────────────────────────── print(f" [PROGRESS] Writing {PROGRESS_FILE}...") write_progress_md(run_state) # ── Per-cycle ntfy ───────────────────────────────────────────────── counts = get_db_counts() t_str = tier_summary_str(counts, assets, tfs) oos_hits = get_oos_highlights() msg_lines = [ f"Cycle {cycle_num}/{max_cycles} [{search_mode}] done in {cycle_elapsed_h:.1f}h.", f"Combos: {cycle_ok} OK / {cycle_fail} failed.", f"Tiers: {t_str}.", ] if locks_applied: msg_lines.append(f"Locks: {locks_applied} applied (total: {cumulative_locks}).") if oos_hits: msg_lines.append("✅ OOS: " + " | ".join(oos_hits[:3])) if cycle_num < max_cycles: next_eta = datetime.now() + timedelta(hours=hours_per_cycle) msg_lines.append(f"Next: cycle {cycle_num+1} [{modes[cycle_num]}], ETA {next_eta:%b %d %H:%M}.") ntfy("\n".join(msg_lines), title=f"📊 Cycle {cycle_num}/{max_cycles} [{search_mode}] done") log_fh.write(f"\nCYCLE {cycle_num} SUMMARY: ok={cycle_ok} fail={cycle_fail} " f"locks={locks_applied} elapsed={cycle_elapsed_h:.1f}h\n") # ── Final ────────────────────────────────────────────────────────────── total_elapsed = (time.time() - wall_start) / 3600 run_state["current_cycle"] = max_cycles + 1 # signals completion write_progress_md(run_state) counts = get_db_counts() t_str = tier_summary_str(counts, assets, tfs) oos_hits = get_oos_highlights() final_msg = [ f"🏁 Marathon complete: {max_cycles} cycles in {total_elapsed:.1f}h.", f"Runs: {total_ok} OK / {total_fail} failed.", f"Final tiers: {t_str}.", f"Locks applied: {cumulative_locks}.", ] if oos_hits: final_msg.append("✅ EXCELLENT: " + " | ".join(oos_hits)) ntfy("\n".join(final_msg), title="🏁 Marathon complete") print(f"\n{'='*65}") print(f" MARATHON COMPLETE — {total_elapsed:.1f}h — {total_ok} OK / {total_fail} failed") print(f" Locks: {cumulative_locks} | Tiers: {t_str}") if oos_hits: print(f" OOS: {' | '.join(oos_hits)}") print(f"{'='*65}\n") log_fh.write(f"\nMARATHON COMPLETE: {max_cycles} cycles, {total_elapsed:.1f}h, " f"{total_ok} OK / {total_fail} failed, {cumulative_locks} locks\n") log_fh.close() if __name__ == "__main__": main()