""" run_phased.py — Two-phase parameter optimisation. Phase 1 — Weight search (thresholds fixed to midpoints): Free params: all i_w_* weights Fixed params: i_long_entry_activation_threshold, i_long_exit_activation_threshold, i_trailing_stop_threshold (locked to midpoints of current ranges) Goal: Find signal weights that generalise, with threshold gambling removed. Phase 2 — Threshold tuning (weights fixed from Phase 1 winner): Free params: i_long_entry_activation_threshold, i_long_exit_activation_threshold, i_trailing_stop_threshold (narrowed ranges ±2 steps around Phase 1 winner) Fixed params: all i_w_* weights from Phase 1 winner (values: [winner_value]) Goal: Tune entry/exit timing given known-good weights. Result: two winner CSVs — phase1 and phase2 — both written to results/winners/. The phase2 winner is typically the better IS params; its weights are from the more generalisable phase1 search. Usage: python3 run_phased.py --data data/COINBASE_BTCUSD-1D.csv --hours 4 python3 run_phased.py --data data/COINBASE_ETHUSD-6H.csv --hours 2 --search optuna python3 run_phased.py --data data/COINBASE_BTCUSD-1D.csv --hours 6 \\ --regime bull --regime-threshold 15 # Run all 20 combos (2h each phase = 4h/combo, 80h total) python3 run_all_crypto.py --hours 4 --phased The --phased flag on run_all_crypto.py calls run_phased.py instead of auto_optimize_loop.py. """ import os import sys import json import copy import argparse import subprocess import shutil import time import pandas as pd sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from config import WINNERS_DIR, OUTPUT_DIR THRESHOLD_PARAMS = { "i_long_entry_activation_threshold", "i_long_exit_activation_threshold", "i_trailing_stop_threshold", # exit_confirmation threshold is unused (exit_conf=0 locked), but keep fixed anyway "i_long_exit_activation_confirmation_threshold", } WEIGHT_PARAMS_PREFIX = "i_w_" def _midpoint(cfg): """Return the midpoint of a range config, rounded to the nearest step.""" if "values" in cfg: vals = cfg["values"] return vals[len(vals) // 2] start, stop, step = cfg["start"], cfg["stop"], cfg["step"] mid = start + (stop - start) / 2.0 # Round to nearest step steps_from_start = round((mid - start) / step) return start + steps_from_start * step def make_phase1_params(base_params: dict) -> dict: """ Return a copy of base_params with all threshold params locked to their range midpoints. Weight params stay free (unchanged). """ p = copy.deepcopy(base_params) for key in THRESHOLD_PARAMS: if key in p: mid = _midpoint(p[key]) p[key] = {"values": [mid]} return p def make_phase2_params(base_params: dict, phase1_winner: dict, n_steps: int = 2) -> dict: """ Return a copy of base_params with: - all i_w_* weight params locked to phase1 winner values - threshold params narrowed to ±n_steps around phase1 winner values """ p = copy.deepcopy(base_params) # Lock weight params from winner for key in list(p.keys()): if key.startswith(WEIGHT_PARAMS_PREFIX): winner_val = phase1_winner.get(key) if winner_val is not None: try: p[key] = {"values": [float(winner_val)]} except (ValueError, TypeError): pass # keep original range if value is missing/non-numeric # Narrow threshold params: ±n_steps around winner value for key in THRESHOLD_PARAMS: if key in p and key in phase1_winner: try: winner_val = float(phase1_winner[key]) cfg = base_params[key] if "values" in cfg and len(cfg["values"]) == 1: # Already locked — keep as-is p[key] = cfg elif "start" in cfg: step = cfg["step"] start = winner_val - n_steps * step stop = winner_val + n_steps * step # Clamp to original range start = max(cfg["start"], start) stop = min(cfg["stop"], stop) p[key] = {"start": start, "stop": stop, "step": step} except (ValueError, TypeError, KeyError): pass return p def run_phase(phase_num, data_file, params_file, hours, search_mode, extra_args=None): """Run auto_optimize_loop.py for one phase. Returns exit code.""" cmd = [ sys.executable, "tools/auto_optimize_loop.py", "--data", data_file, "--params", params_file, "--hours", str(hours), "--search", search_mode, "--no-dashboard", # suppress per-iter dashboard; we run it at end ] if extra_args: cmd.extend(extra_args) print(f"\n{'='*60}") print(f" PHASE {phase_num} — {'Weight search (thresholds fixed)' if phase_num == 1 else 'Threshold tuning (weights fixed)'}") print(f" Data: {data_file}") print(f" Params: {params_file}") print(f" Hours: {hours:.2f}h Search: {search_mode}") print(f"{'='*60}\n") result = subprocess.run(cmd) return result.returncode def main(): parser = argparse.ArgumentParser( description="Two-phase optimizer: weights first, thresholds second.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument("--data", required=True, help="Path to data CSV (e.g. data/COINBASE_BTCUSD-1D.csv)") parser.add_argument("--hours", type=float, default=4.0, help="Total hours budget split evenly between phases (default: 4 = 2h each)") parser.add_argument("--phase1-hours", type=float, default=None, help="Override Phase 1 duration (default: hours/2)") parser.add_argument("--phase2-hours", type=float, default=None, help="Override Phase 2 duration (default: hours/2)") parser.add_argument("--search", type=str, default="random", choices=["random", "optuna"], help="Search mode for Phase 1 (default: random). Phase 2 always uses optuna.") parser.add_argument("--regime", type=str, default="all", choices=["all", "bull", "bear"], help="Regime filter for Phase 1 data (see auto_optimize_loop.py --regime)") parser.add_argument("--regime-threshold", type=float, default=15.0, help="MVRV zscore threshold for regime filter (default: 15.0)") parser.add_argument("--phase2-steps", type=int, default=2, help="Threshold narrowing: ±N steps around phase1 winner (default: 2)") parser.add_argument("--phase1-only", action="store_true", help="Only run Phase 1 (weight search), skip Phase 2") parser.add_argument("--phase2-only", action="store_true", help="Only run Phase 2 (threshold tuning) — requires an existing winner CSV") args = parser.parse_args() # Derive asset/timeframe from data filename stem = os.path.splitext(os.path.basename(args.data))[0] parts = stem.rsplit("-", 1) asset = parts[0] if len(parts) == 2 else stem tf = parts[1].upper() if len(parts) == 2 else "1D" base_params_file = f"strategies/params/params_strategy_activation_scores_{asset}_{tf}.json" if not os.path.exists(base_params_file): print(f"[ERROR] Params file not found: {base_params_file}") sys.exit(1) with open(base_params_file) as f: base_params = json.load(f) hours_p1 = args.phase1_hours or args.hours / 2.0 hours_p2 = args.phase2_hours or args.hours / 2.0 os.makedirs(OUTPUT_DIR, exist_ok=True) phase1_params_file = os.path.join(OUTPUT_DIR, f"phased_p1_params_{asset}_{tf}.json") phase2_params_file = os.path.join(OUTPUT_DIR, f"phased_p2_params_{asset}_{tf}.json") winner_file = os.path.join(WINNERS_DIR, f"optimization_winner_activation_scores_{asset}_{tf}.csv") regime_extra = [] if args.regime != "all": regime_extra = ["--regime", args.regime, "--regime-threshold", str(args.regime_threshold)] # ── Phase 1 ────────────────────────────────────────────────────────────── if not args.phase2_only: p1_params = make_phase1_params(base_params) # Show what was locked locked = [k for k in THRESHOLD_PARAMS if k in p1_params] print(f"\n[PHASE 1 CONFIG] Locking threshold params to range midpoints:") for k in locked: print(f" {k} → {p1_params[k]}") print(f" Free weights: {sum(1 for k in p1_params if k.startswith(WEIGHT_PARAMS_PREFIX) and 'start' in p1_params[k])}") with open(phase1_params_file, "w") as f: json.dump(p1_params, f, indent=2) rc = run_phase(1, args.data, phase1_params_file, hours_p1, args.search, extra_args=regime_extra) if rc != 0: print(f"[ERROR] Phase 1 exited with code {rc}") sys.exit(rc) # Copy phase1 winner to a labelled file so it isn't overwritten by phase 2 p1_winner_file = os.path.join(WINNERS_DIR, f"optimization_winner_activation_scores_{asset}_{tf}_phase1.csv") if os.path.exists(winner_file): shutil.copy(winner_file, p1_winner_file) print(f"\n[PHASE 1 DONE] Winner saved → {p1_winner_file}") else: print("[ERROR] Phase 1 produced no winner file.") sys.exit(1) if args.phase1_only: print("[run_phased] --phase1-only set; stopping after Phase 1.") return # ── Phase 2 ────────────────────────────────────────────────────────────── p1_winner_file = os.path.join(WINNERS_DIR, f"optimization_winner_activation_scores_{asset}_{tf}_phase1.csv") if not os.path.exists(p1_winner_file): # Fall back to regular winner file if phase1 copy doesn't exist p1_winner_file = winner_file if not os.path.exists(p1_winner_file): print(f"[ERROR] Phase 1 winner not found at {p1_winner_file}. Run Phase 1 first.") sys.exit(1) phase1_winner = pd.read_csv(p1_winner_file).iloc[0].to_dict() p2_params = make_phase2_params(base_params, phase1_winner, n_steps=args.phase2_steps) locked_weights = [k for k in p2_params if k.startswith(WEIGHT_PARAMS_PREFIX) and "values" in p2_params[k]] free_thresholds = [k for k in THRESHOLD_PARAMS if k in p2_params and "start" in p2_params[k]] print(f"\n[PHASE 2 CONFIG] Weights locked from phase1 winner: {len(locked_weights)}") print(f" Free threshold params: {free_thresholds}") for k in free_thresholds: print(f" {k}: {p2_params[k]}") with open(phase2_params_file, "w") as f: json.dump(p2_params, f, indent=2) # Phase 2 always uses optuna — low-dimensional threshold search is ideal for it rc = run_phase(2, args.data, phase2_params_file, hours_p2, "optuna") if rc != 0: print(f"[ERROR] Phase 2 exited with code {rc}") sys.exit(rc) # Copy phase2 winner p2_winner_file = os.path.join(WINNERS_DIR, f"optimization_winner_activation_scores_{asset}_{tf}_phase2.csv") if os.path.exists(winner_file): shutil.copy(winner_file, p2_winner_file) print(f"\n[PHASE 2 DONE] Winner saved → {p2_winner_file}") # ── Final comparison ────────────────────────────────────────────────────── print(f"\n{'='*60}") print(" PHASED OPTIMISATION COMPLETE") print(f"{'='*60}") def _report(label, fpath): if os.path.exists(fpath): row = pd.read_csv(fpath).iloc[0] sort = row.get("Sortino Ratio", "?") pnl = row.get("Total P&L %", "?") dd = row.get("Max Drawdown %", "?") tr = row.get("Total Trades", "?") try: print(f" {label}: Sortino={float(sort):.4f} P&L={float(pnl):,.0f}% DD={float(dd):.2f}% Trades={int(float(tr))}") except Exception: print(f" {label}: (could not parse winner)") else: print(f" {label}: no winner file found") _report("Phase 1 winner (weights only)", p1_winner_file) _report("Phase 2 winner (thresholds tuned)", p2_winner_file if os.path.exists(p2_winner_file) else winner_file) print(f"\n Note: Phase 2 winner uses Phase 1 weights — it is the recommended IS params.") print(f" Compare Phase 2 OOS vs standard marathon winner to assess generalisability gain.") if __name__ == "__main__": main()