""" Deeper threshold sweep for BTC MLP presets. This intentionally keeps MLP weights fixed and explores threshold/config parameters around the current winners and artifact recommendations. It reuses strategy_mlp_scores' score-to-signal block and canonical calculate_metrics so candidate metrics stay comparable to optimizer winners. Examples: .venv/bin/python tools/run_mlp_deep_sweep.py --timeframes 6H 8H 12H 1D --samples 50000 .venv/bin/python tools/run_mlp_deep_sweep.py --timeframes 6H --samples 10000 --promote """ from __future__ import annotations import argparse import concurrent.futures import csv import os import json import multiprocessing as mp import pickle from pathlib import Path import subprocess import tempfile from typing import Any import numpy as np import pandas as pd import sys REPO = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO)) from config import ( # noqa: E402 OOS_START, SCORE_START, SUBPERIOD_SPLIT, TRAIN_END, TRAIN_START, WFO_FOLDS, WFO_MIN_OOS_TRADES, WFO_MIN_VALID_FOLDS, composite_score, get_min_trades, ) from strategies.strategy_mlp_scores import ( # noqa: E402 FEATURE_COLS as _FEATURE_COLS, _prepare_features, _score_to_signals, calculate_metrics, load_mlp_artifact, mlp_forward, ) from tools.worker_utils import DEFAULT_WORKER_FRACTION, resolve_worker_count # noqa: E402 _DEFAULT_ASSET = "COINBASE_BTCUSD" ASSET = _DEFAULT_ASSET # may be overridden by --asset arg in main() TFS = ["4H", "6H", "8H", "12H", "1D"] WINNER_DIR = REPO / "results" / "winners" SWEEP_DIR = REPO / "results" / "sweeps" REPORT_DIR = REPO / "results" / "reports" _WORKER_TF: str | None = None _WORKER_CACHE: dict[str, tuple[pd.DataFrame, np.ndarray, np.ndarray, np.ndarray]] | None = None PARAM_COLS = [ "mlp_weights_file", "i_long_entry_activation_threshold", "i_long_exit_activation_threshold", "i_long_exit_activation_confirmation_threshold", "i_trailing_stop_threshold", "i_use_long_exit_confirmation", "i_use_long_entry_confirmation", "i_regime_window", "i_regime_entry_min_score", "i_mvrv_suppress_bear", "i_exit_score_window", "i_entry_score_window", ] METRIC_COLS = [ "Total Trades", "Total P&L %", "Sharpe Ratio", "Sortino Ratio", "Calmar Ratio", "P&L/DD Ratio", "Max Drawdown %", "% In Market", # OOS metrics persisted so promotable() can use stored baseline on next run. "OOS_Total Trades", "OOS_Total P&L %", "OOS_Sortino Ratio", "OOS_Bear_Pct", ] # OOS Sortino protection gate (see promotable()). OOS_SORTINO_DELTA = 1.0 # max tolerated Sortino degradation vs current winner OOS_MIN_TRADES = 1 # gate skipped when either side has fewer OOS trades OOS_BEAR_PCT_SKIP = 85.0 # gate skipped when OOS is this bear-dominated (%) def winner_path(tf: str) -> Path: return WINNER_DIR / f"optimization_winner_strategy_mlp_scores_{ASSET}_{tf}.csv" def artifact_path(tf: str) -> str: return f"strategies/params/mlp/mlp_weights_{ASSET}_{tf}.json" def data_path(tf: str) -> Path: # MLP data lives in data/mlp/ with TradingView export naming (minutes or "1D") _TF_MINUTES = {"4H": "240", "6H": "360", "8H": "480", "12H": "720", "1D": "1D"} period = _TF_MINUTES.get(tf, tf) mlp_path = REPO / "data" / "mlp" / f"{ASSET}, {period}.csv" if mlp_path.exists(): return mlp_path return REPO / "data" / f"{ASSET}-{tf}.csv" def as_bool(v: Any) -> bool: if isinstance(v, str): return v.strip().lower() in {"true", "1", "yes"} return bool(v) def as_float(v: Any, default: float = 0.0) -> float: try: if pd.isna(v): return default return float(v) except Exception: return default def load_winner(tf: str) -> dict[str, Any] | None: path = winner_path(tf) if not path.exists(): return None rows = list(csv.DictReader(path.open())) if not rows: return None row = dict(rows[0]) row["i_mvrv_suppress_bear"] = as_bool(row.get("i_mvrv_suppress_bear", False)) row["i_use_long_entry_confirmation"] = as_bool(row.get("i_use_long_entry_confirmation", False)) row["i_use_long_exit_confirmation"] = 1 return row def recommended_from_artifact(weights_path: str) -> dict[str, Any]: art = load_mlp_artifact(weights_path) rec = art["meta"].get("training", {}).get("recommended_thresholds", {}) params = { "mlp_weights_file": weights_path, "i_long_entry_activation_threshold": rec.get("i_long_entry_activation_threshold", 0.0), "i_long_exit_activation_threshold": rec.get("i_long_exit_activation_threshold", 0.0), "i_long_exit_activation_confirmation_threshold": rec.get( "i_long_exit_activation_confirmation_threshold", 0.0 ), "i_trailing_stop_threshold": rec.get("i_trailing_stop_threshold", 0.0), "i_use_long_exit_confirmation": 1, "i_use_long_entry_confirmation": False, "i_regime_window": 0, "i_regime_entry_min_score": -1000.0, "i_mvrv_suppress_bear": False, "i_exit_score_window": 1, "i_entry_score_window": 1, } return params def snap(value: float, start: float, stop: float, step: float) -> float: value = min(max(value, start), stop - step) n = round((value - start) / step) return round(start + n * step, 8) def param_key(p: dict[str, Any]) -> tuple[Any, ...]: return ( p["mlp_weights_file"], as_float(p["i_long_entry_activation_threshold"]), as_float(p["i_long_exit_activation_threshold"]), as_float(p["i_long_exit_activation_confirmation_threshold"]), as_float(p["i_trailing_stop_threshold"]), bool(as_bool(p.get("i_mvrv_suppress_bear", False))), ) def normalise_param(p: dict[str, Any]) -> dict[str, Any]: return { "mlp_weights_file": str(p["mlp_weights_file"]), "i_long_entry_activation_threshold": as_float(p["i_long_entry_activation_threshold"]), "i_long_exit_activation_threshold": as_float(p["i_long_exit_activation_threshold"]), "i_long_exit_activation_confirmation_threshold": as_float( p["i_long_exit_activation_confirmation_threshold"] ), "i_trailing_stop_threshold": max(0.0, as_float(p.get("i_trailing_stop_threshold", 0.0))), "i_use_long_exit_confirmation": 1, "i_use_long_entry_confirmation": False, "i_regime_window": 0, "i_regime_entry_min_score": -1000.0, "i_mvrv_suppress_bear": as_bool(p.get("i_mvrv_suppress_bear", False)), "i_exit_score_window": max(1, int(as_float(p.get("i_exit_score_window", 1)))), "i_entry_score_window": max(1, int(as_float(p.get("i_entry_score_window", 1)))), } def generate_candidates(tf: str, samples: int, seed: int, extra_weights: list[str] | None = None, wide: bool = False, entry_lo: float | None = None, entry_hi: float | None = None, exit_lo: float | None = None, exit_hi: float | None = None) -> list[dict[str, Any]]: rng = np.random.default_rng(seed) current = load_winner(tf) own_artifact = artifact_path(tf) extra_weights = extra_weights or [] seeds: list[dict[str, Any]] = [] if current: seeds.append(normalise_param(current)) if os.path.exists(own_artifact): seeds.append(normalise_param(recommended_from_artifact(own_artifact))) for weights_path in extra_weights: seeds.append(normalise_param(recommended_from_artifact(weights_path))) if tf == "1D" and ASSET == "COINBASE_BTCUSD": # Pine aliases 1D weights to 12H, but keeps 1D thresholds. Include both # 1D and 12H weights so the report can prove whether the alias still wins. alt_12h = artifact_path("12H") if os.path.exists(alt_12h): seeds.append(normalise_param(recommended_from_artifact(alt_12h))) seen: set[tuple[Any, ...]] = set() out: list[dict[str, Any]] = [] def add(p: dict[str, Any], source: str) -> None: pp = normalise_param(p) key = param_key(pp) if key in seen: return seen.add(key) pp["source"] = source out.append(pp) _entry_lo_def, _entry_hi_def = (-600.0, 600.0) if wide else (-300.0, 300.0) _exit_lo_def, _exit_hi_def = (-750.0, 750.0) if wide else (-300.0, 400.0) entry_lo = entry_lo if entry_lo is not None else _entry_lo_def entry_hi = entry_hi if entry_hi is not None else _entry_hi_def exit_lo = exit_lo if exit_lo is not None else _exit_lo_def exit_hi = exit_hi if exit_hi is not None else _exit_hi_def trail_hi = 100.0 if wide else 50.0 allowed_trail = np.arange(0.0, trail_hi, 5.0) for s in seeds: s = dict(s) raw_trail = as_float(s.get("i_trailing_stop_threshold", 0.0)) s["i_trailing_stop_threshold"] = float(allowed_trail[np.argmin(np.abs(allowed_trail - raw_trail))]) add(s, "seed") local_sigma_entry = max(5.0, (entry_hi - entry_lo) / 6.0) local_sigma_exit = max(10.0, (exit_hi - exit_lo) / 6.0) local_n = int(samples * 0.65) broad_n = max(0, samples - local_n) if seeds: per_seed = max(1, local_n // len(seeds)) for i, s in enumerate(seeds): for _ in range(per_seed): p = dict(s) p["i_long_entry_activation_threshold"] = snap( rng.normal(as_float(s["i_long_entry_activation_threshold"]), local_sigma_entry), entry_lo, entry_hi, 5.0, ) p["i_long_exit_activation_threshold"] = snap( rng.normal(as_float(s["i_long_exit_activation_threshold"]), local_sigma_exit), exit_lo, exit_hi, 10.0, ) p["i_long_exit_activation_confirmation_threshold"] = snap( rng.normal(as_float(s["i_long_exit_activation_confirmation_threshold"]), local_sigma_exit), exit_lo, exit_hi, 10.0, ) p["i_trailing_stop_threshold"] = float(rng.choice(np.arange(0.0, trail_hi, 5.0))) p["i_mvrv_suppress_bear"] = bool(rng.choice([False, True])) p["i_exit_score_window"] = int(rng.choice([1, 1, 1, 2, 3, 4, 5])) p["i_entry_score_window"] = int(rng.choice([1, 1, 1, 2, 3, 4, 5])) add(p, f"local_{i}") weight_choices = sorted({s["mlp_weights_file"] for s in seeds}) for _ in range(broad_n): p = { "mlp_weights_file": str(rng.choice(weight_choices)), "i_long_entry_activation_threshold": float(rng.choice(np.arange(entry_lo, entry_hi, 5.0))), "i_long_exit_activation_threshold": float(rng.choice(np.arange(exit_lo, exit_hi, 10.0))), "i_long_exit_activation_confirmation_threshold": float(rng.choice(np.arange(exit_lo, exit_hi, 10.0))), "i_trailing_stop_threshold": float(rng.choice(np.arange(0.0, trail_hi, 5.0))), "i_use_long_exit_confirmation": 1, "i_use_long_entry_confirmation": False, "i_regime_window": 0, "i_regime_entry_min_score": -1000.0, "i_mvrv_suppress_bear": bool(rng.choice([False, True])), "i_exit_score_window": int(rng.choice([1, 1, 1, 2, 3, 4, 5])), "i_entry_score_window": int(rng.choice([1, 1, 1, 2, 3, 4, 5])), } add(p, "broad") return out def load_scored_frame(tf: str, weights_path: str) -> tuple[pd.DataFrame, np.ndarray, np.ndarray, np.ndarray]: full = pd.read_csv(data_path(tf)) full.columns = full.columns.str.lower().str.strip() full["time"] = pd.to_datetime(full["time"], utc=True).dt.tz_localize(None) full = full.loc[full["time"] >= TRAIN_START].copy().reset_index(drop=True) art = load_mlp_artifact(weights_path) art_cols = art.get("feature_cols") or [] missing_from_current = [c for c in art_cols if c not in _FEATURE_COLS and c not in full.columns] if missing_from_current: raise ValueError( f"Stale artifact {os.path.basename(weights_path)}: expects features not in " f"current FEATURE_COLS or data: {missing_from_current}. Retrain to skip." ) X = _prepare_features(full, art_cols or None) scored = full[["time", "close", "high"]].copy() for col in ("mvrv_regime",): if col in full.columns: scored[col] = full[col] scored["activation_score"] = mlp_forward(X, art["layers"]) return scored, X[:, 9], full["close"].to_numpy(np.float64), full["high"].to_numpy(np.float64) def fragile_count(scores: pd.Series, params: dict[str, Any], margin: float) -> int: thresholds = np.array( [ as_float(params["i_long_entry_activation_threshold"]), as_float(params["i_long_exit_activation_threshold"]), as_float(params["i_long_exit_activation_confirmation_threshold"]), ], dtype=np.float64, ) dist = np.min(np.abs(scores.to_numpy(np.float64)[:, None] - thresholds[None, :]), axis=1) return int(np.sum(dist < margin)) def is_years(df: pd.DataFrame, score_start: str, end: str) -> float: sl = df[(df["time"] >= pd.Timestamp(score_start)) & (df["time"] <= pd.Timestamp(end))] if len(sl) < 2: return 0.5 return max(0.5, (sl["time"].iloc[-1] - sl["time"].iloc[0]).days / 365.25) _STALE_SENTINEL = object() # sentinel stored in cache when an artifact is stale def evaluate(tf: str, params: dict[str, Any], cache: dict[str, tuple[pd.DataFrame, np.ndarray, np.ndarray, np.ndarray]]) -> dict[str, Any]: weights = params["mlp_weights_file"] if weights not in cache: try: cache[weights] = load_scored_frame(tf, weights) except ValueError as exc: print(f" [SKIP] {os.path.basename(weights)}: {exc}", flush=True) cache[weights] = _STALE_SENTINEL if cache[weights] is _STALE_SENTINEL: # Stale artifact — return zero-score result so it can never win result = dict(params) result.update({"Total Trades": 0, "Total P&L %": 0.0, "Sharpe Ratio": -999.0, "Sortino Ratio": -999.0, "Calmar Ratio": -999.0, "P&L/DD Ratio": 0.0, "Max Drawdown %": 0.0, "% In Market": 0.0, "Composite": 0.0, "Min_Trades_Floor": 0, "Sub1_Calmar": -999.0, "Sub2_Calmar": -999.0, "WFO_Score": -999.0, "WFO_Min_Fold": -999.0, "WFO_Neg_Folds": 99, "WFO_Fold_Calmars": [], "OOS_Total Trades": 0, "OOS_Total P&L %": 0.0, "OOS_Sortino Ratio": -999.0, "OOS_Calmar Ratio": -999.0, "OOS_Max Drawdown %": 0.0, "OOS_Bear_Pct": 0.0, "OOS_Neutral_Pct": 0.0, "OOS_Bull_Pct": 0.0, "Fragile_0.05": 99, "Fragile_1.0": 99, "source": params.get("source", "stale")}) return result scored, stoch_peak, close_arr, high_arr = cache[weights] signal_params = dict(params) signal_params["_pine_time_start"] = SCORE_START sigs = _score_to_signals(scored.copy(), signal_params, stoch_peak, close_arr, high_arr) train_sigs = sigs[sigs["time"] <= pd.Timestamp(TRAIN_END)].copy() metrics = calculate_metrics(train_sigs, score_start=SCORE_START) years = is_years(train_sigs, SCORE_START, TRAIN_END) comp = composite_score( metrics.get("Calmar Ratio", 0.0), metrics.get("Sortino Ratio", 0.0), int(metrics.get("Total Trades", 0)), years, metrics.get("Total P&L %", 0.0), ) split = pd.Timestamp(SUBPERIOD_SPLIT) sub1 = calculate_metrics(train_sigs[train_sigs["time"] <= split].copy(), score_start=SCORE_START) sub2 = calculate_metrics(train_sigs[train_sigs["time"] > split].copy(), score_start=SUBPERIOD_SPLIT) fold_calmars = [] for _, oos_start, oos_end in WFO_FOLDS: fold_window = train_sigs[train_sigs["time"] <= pd.Timestamp(oos_end)].copy() fold_metrics = calculate_metrics(fold_window, score_start=oos_start, min_trades=WFO_MIN_OOS_TRADES) if int(fold_metrics.get("Total Trades", 0)) >= WFO_MIN_OOS_TRADES: fold_calmars.append(float(fold_metrics.get("Calmar Ratio", -99.0))) if len(fold_calmars) >= WFO_MIN_VALID_FOLDS: wfo_score = float(np.mean(fold_calmars)) wfo_min = float(np.min(fold_calmars)) wfo_neg = int(sum(1 for c in fold_calmars if c < 0)) else: wfo_score = 0.0 wfo_min = -99.0 wfo_neg = len(WFO_FOLDS) if sigs["time"].max() >= pd.Timestamp(OOS_START): oos = calculate_metrics(sigs, score_start=OOS_START, min_trades=1) # 0 OOS trades is neutral for a long-only strategy — replace penalty sentinel with 0.0. # IS 0-trade solutions are genuinely degenerate (optimizer found params that never trade); # OOS 0 trades just means conditions weren't met in the evaluation window. if int(oos.get("Total Trades", 0)) == 0: for _k in ("Calmar Ratio", "Sortino Ratio", "Sharpe Ratio"): if oos.get(_k) == -10.0: oos[_k] = 0.0 else: oos = {k: np.nan for k in METRIC_COLS} # OOS regime composition — characterises the OOS window so 0-trade outcomes can be # interpreted correctly (long-only sitting out a bear market is correct behavior). oos_bear_pct = oos_neutral_pct = oos_bull_pct = np.nan if "mvrv_regime" in sigs.columns: oos_sigs = sigs[sigs["time"] >= pd.Timestamp(OOS_START)] if len(oos_sigs) > 0: n = len(oos_sigs) oos_bear_pct = float((oos_sigs["mvrv_regime"] == -1).sum() / n * 100) oos_neutral_pct = float((oos_sigs["mvrv_regime"] == 0).sum() / n * 100) oos_bull_pct = float((oos_sigs["mvrv_regime"] == 1).sum() / n * 100) result: dict[str, Any] = {k: params[k] for k in PARAM_COLS} result.update(metrics) result.update( { "Composite": comp, "Min_Trades_Floor": get_min_trades(years), "Sub1_Calmar": sub1.get("Calmar Ratio", -10.0), "Sub1_Sortino": sub1.get("Sortino Ratio", -10.0), "Sub1_Trades": sub1.get("Total Trades", 0), "Sub2_Calmar": sub2.get("Calmar Ratio", -10.0), "Sub2_Sortino": sub2.get("Sortino Ratio", -10.0), "Sub2_Trades": sub2.get("Total Trades", 0), "WFO_Score": wfo_score, "WFO_Min_Fold": wfo_min, "WFO_Neg_Folds": wfo_neg, "WFO_Fold_Calmars": json.dumps([round(c, 6) for c in fold_calmars]), "OOS_Total Trades": oos.get("Total Trades", np.nan), "OOS_Total P&L %": oos.get("Total P&L %", np.nan), "OOS_Sortino Ratio": oos.get("Sortino Ratio", np.nan), "OOS_Calmar Ratio": oos.get("Calmar Ratio", np.nan), "OOS_Max Drawdown %": oos.get("Max Drawdown %", np.nan), "OOS_Bear_Pct": oos_bear_pct, "OOS_Neutral_Pct": oos_neutral_pct, "OOS_Bull_Pct": oos_bull_pct, "Fragile_0.05": fragile_count(train_sigs["activation_score"], params, 0.05), "Fragile_1.0": fragile_count(train_sigs["activation_score"], params, 1.0), "source": params.get("source", "candidate"), } ) return result def init_worker(tf: str, asset: str) -> None: global _WORKER_TF, _WORKER_CACHE, ASSET _WORKER_TF = tf _WORKER_CACHE = {} ASSET = asset def evaluate_worker(params: dict[str, Any]) -> dict[str, Any]: if _WORKER_TF is None or _WORKER_CACHE is None: raise RuntimeError("sweep worker was not initialized") return evaluate(_WORKER_TF, params, _WORKER_CACHE) def rank_tuple(row: dict[str, Any]) -> tuple[Any, ...]: fragile_ok = int(as_float(row.get("Fragile_0.05", 0)) == 0) composite = as_float(row.get("Composite", 0.0)) pnl_dd = as_float(row.get("P&L/DD Ratio", 0.0)) drawdown = abs(as_float(row.get("Max Drawdown %", -999.0), -999.0)) sub_ok = int(as_float(row.get("Sub1_Calmar", -10.0)) > 0) + int(as_float(row.get("Sub2_Calmar", -10.0)) > 0) oos_sortino = as_float(row.get("OOS_Sortino Ratio", -10.0), -10.0) return ( int(composite > 0), int(pnl_dd > 0), fragile_ok, sub_ok, pnl_dd, -drawdown, -int(as_float(row.get("WFO_Neg_Folds", 99))), as_float(row.get("WFO_Min_Fold", -99.0)), as_float(row.get("WFO_Score", 0.0)), oos_sortino, as_float(row.get("Calmar Ratio", -10.0)), composite, as_float(row.get("Total P&L %", 0.0)), ) def write_winner(tf: str, row: dict[str, Any]) -> None: path = winner_path(tf) ordered = {k: row.get(k, "") for k in PARAM_COLS + METRIC_COLS} pd.DataFrame([ordered]).to_csv(path, index=False) def promotable(best: dict[str, Any], current: dict[str, Any] | None, stored_winner: dict[str, Any] | None = None) -> bool: if current is None or best.get("source") == "current": return False if as_float(best.get("P&L/DD Ratio", 0.0)) <= 0: return False if as_float(best.get("Fragile_0.05", 99)) != 0: return False if as_float(best.get("Sub1_Calmar", -10.0)) <= 0 or as_float(best.get("Sub2_Calmar", -10.0)) <= 0: return False if int(as_float(best.get("WFO_Neg_Folds", 99))) != 0: return False # Use the higher of re-evaluated and stored Calmar/P&L-DD as the comparison floor. # If the current winner's artifact was overwritten by a training run that preceded this # sweep, cur_eval reflects new weights rather than the original winner's, producing a # falsely-degraded baseline that lets a weaker candidate through. The stored CSV # values are authoritative; taking the max ensures we never lower the bar due to # an overwritten artifact. ref_pnl_dd = max( as_float(current.get("P&L/DD Ratio", 0.0)), as_float((stored_winner or {}).get("P&L/DD Ratio", 0.0)), ) ref_calmar = max( as_float(current.get("Calmar Ratio", -10.0)), as_float((stored_winner or {}).get("Calmar Ratio", -10.0)), ) if as_float(best.get("P&L/DD Ratio", 0.0)) < ref_pnl_dd: return False if as_float(best.get("Calmar Ratio", -10.0)) < ref_calmar: return False # OOS Sortino protection: block promotion if new candidate degrades OOS performance # materially vs the current winner. This is protective (not selective) — we are NOT # choosing based on OOS; we are blocking a step backward on the real-money period. # Gate is skipped when: either side has too few OOS trades to compare meaningfully, # or OOS is deeply bear-dominated (long-only naturally trades less in bear markets). best_oos_trades = int(as_float(best.get("OOS_Total Trades"), 0.0)) ref_oos_trades = int(max( as_float((stored_winner or {}).get("OOS_Total Trades"), 0.0), as_float(current.get("OOS_Total Trades"), 0.0), )) oos_bear_pct = as_float(best.get("OOS_Bear_Pct"), 0.0) oos_gate_applies = ( best_oos_trades >= OOS_MIN_TRADES and ref_oos_trades >= OOS_MIN_TRADES and oos_bear_pct < OOS_BEAR_PCT_SKIP ) if oos_gate_applies: best_oos_sortino = as_float(best.get("OOS_Sortino Ratio"), -10.0) ref_oos_sortino = max( as_float(current.get("OOS_Sortino Ratio"), -10.0), as_float((stored_winner or {}).get("OOS_Sortino Ratio"), -10.0), ) if best_oos_sortino < ref_oos_sortino - OOS_SORTINO_DELTA: return False return True def partition_candidate_ranges(total: int, workers: int) -> list[tuple[int, int]]: """Return balanced, contiguous ranges covering every candidate exactly once.""" if total <= 0: return [] shard_count = min(max(1, workers), total) base, extra = divmod(total, shard_count) ranges = [] start = 0 for ordinal in range(shard_count): end = start + base + (1 if ordinal < extra else 0) ranges.append((start, end)) start = end return ranges def evaluate_candidates_sequential( tf: str, candidates: list[dict[str, Any]], show_progress: bool = True, ) -> list[dict[str, Any]]: cache: dict[str, tuple[pd.DataFrame, np.ndarray, np.ndarray, np.ndarray]] = {} results = [] for idx, params in enumerate(candidates, 1): results.append(evaluate(tf, params, cache)) if show_progress and idx % 5000 == 0: print(f"{tf}: evaluated {idx:,}/{len(candidates):,}", flush=True) return results def _write_pickle_atomic(path: Path, payload: dict[str, Any]) -> None: temp_path = path.with_suffix(path.suffix + ".tmp") with temp_path.open("wb") as handle: pickle.dump(payload, handle, protocol=pickle.HIGHEST_PROTOCOL) temp_path.replace(path) def run_internal_shard(tf: str, input_path: Path, output_path: Path) -> None: global ASSET with input_path.open("rb") as handle: payload = pickle.load(handle) if payload.get("version") != 1 or payload.get("tf") != tf: raise ValueError("invalid sweep shard metadata") candidates = payload.get("candidates") if not isinstance(candidates, list) or payload.get("end", 0) - payload.get("start", 0) != len(candidates): raise ValueError("invalid sweep shard candidate range") ASSET = str(payload["asset"]) results = evaluate_candidates_sequential(tf, candidates, show_progress=False) _write_pickle_atomic( output_path, { "version": 1, "asset": ASSET, "tf": tf, "ordinal": payload["ordinal"], "start": payload["start"], "end": payload["end"], "results": results, }, ) def evaluate_candidates_subprocess( tf: str, candidates: list[dict[str, Any]], workers: int, ) -> tuple[list[dict[str, Any]], int]: ranges = partition_candidate_ranges(len(candidates), workers) if len(ranges) <= 1: return evaluate_candidates_sequential(tf, candidates), 1 print(f"{tf}: retrying with {len(ranges)} subprocess shards", flush=True) with tempfile.TemporaryDirectory(prefix=f"mlp-sweep-{ASSET}-{tf}-") as temp_dir: temp_root = Path(temp_dir) specs = [] processes: list[tuple[subprocess.Popen, Any, Path]] = [] try: for ordinal, (start, end) in enumerate(ranges): input_path = temp_root / f"shard-{ordinal:04d}.input.pkl" output_path = temp_root / f"shard-{ordinal:04d}.output.pkl" log_path = temp_root / f"shard-{ordinal:04d}.log" _write_pickle_atomic( input_path, { "version": 1, "asset": ASSET, "tf": tf, "ordinal": ordinal, "start": start, "end": end, "candidates": candidates[start:end], }, ) log_handle = log_path.open("wb") process = subprocess.Popen( [ sys.executable, str(Path(__file__).resolve()), "--asset", ASSET, "--_internal-shard", "--_shard-tf", tf, "--_shard-input", str(input_path), "--_shard-output", str(output_path), ], cwd=REPO, stdout=log_handle, stderr=subprocess.STDOUT, ) specs.append((ordinal, start, end, output_path, log_path)) processes.append((process, log_handle, log_path)) for process, log_handle, log_path in processes: returncode = process.wait() log_handle.close() if returncode != 0: tail = log_path.read_text(encoding="utf-8", errors="replace")[-4000:] raise RuntimeError(f"sweep shard failed ({returncode}): {tail}") except Exception: for process, log_handle, _ in processes: if process.poll() is None: process.terminate() process.wait() if not log_handle.closed: log_handle.close() raise merged: list[dict[str, Any]] = [] expected_start = 0 for ordinal, start, end, output_path, _ in specs: if start != expected_start or not output_path.exists(): raise RuntimeError("sweep shard coverage is incomplete") with output_path.open("rb") as handle: output = pickle.load(handle) expected = (1, ASSET, tf, ordinal, start, end) actual = ( output.get("version"), output.get("asset"), output.get("tf"), output.get("ordinal"), output.get("start"), output.get("end"), ) shard_results = output.get("results") if actual != expected or not isinstance(shard_results, list) or len(shard_results) != end - start: raise RuntimeError("invalid sweep shard output") merged.extend(shard_results) expected_start = end if expected_start != len(candidates) or len(merged) != len(candidates): raise RuntimeError("sweep shard merge did not preserve candidate coverage") return merged, len(ranges) def evaluate_candidates(tf: str, candidates: list[dict[str, Any]], workers: int) -> tuple[list[dict[str, Any]], int]: if workers == 1: print(f"{tf}: evaluating {len(candidates):,} candidates sequentially", flush=True) return evaluate_candidates_sequential(tf, candidates), 1 print(f"{tf}: evaluating {len(candidates):,} candidates with {workers} worker processes", flush=True) chunk_size = max(1, min(250, len(candidates) // max(1, workers * 8))) mp_context = mp.get_context("spawn") results = [] try: with concurrent.futures.ProcessPoolExecutor( max_workers=workers, mp_context=mp_context, initializer=init_worker, initargs=(tf, ASSET), ) as pool: for idx, result in enumerate(pool.map(evaluate_worker, candidates, chunksize=chunk_size), 1): results.append(result) if idx % 5000 == 0: print(f"{tf}: evaluated {idx:,}/{len(candidates):,}", flush=True) except (OSError, PermissionError) as exc: print(f"{tf}: worker pool unavailable ({exc})", flush=True) try: return evaluate_candidates_subprocess(tf, candidates, workers) except Exception as shard_exc: print(f"{tf}: subprocess sharding unavailable ({shard_exc}); falling back to sequential evaluation", flush=True) return evaluate_candidates(tf, candidates, 1) return results, workers def run_tf(tf: str, samples: int, seed: int, top_k: int, promote: bool, extra_weights: list[str] | None = None, workers: int = 1, worker_fraction: float = DEFAULT_WORKER_FRACTION, force_promote: bool = False, wide: bool = False, entry_lo: float | None = None, entry_hi: float | None = None, exit_lo: float | None = None, exit_hi: float | None = None) -> dict[str, Any]: candidates = generate_candidates(tf, samples, seed, extra_weights, wide=wide, entry_lo=entry_lo, entry_hi=entry_hi, exit_lo=exit_lo, exit_hi=exit_hi) current = load_winner(tf) # Preserve stored metrics before normalise_param() strips them. They serve as the # authoritative quality floor if the winner's artifact was overwritten during training. stored_winner = dict(current) if current else None if current: current = normalise_param(current) current["source"] = "current" candidates.insert(0, current) resolved_workers = resolve_worker_count(workers, len(candidates), worker_fraction) results, actual_workers = evaluate_candidates(tf, candidates, resolved_workers) results.sort(key=rank_tuple, reverse=True) best_overall = results[0] cur_eval = next((r for r in results if r.get("source") == "current"), None) # Warn when the winner's artifact was overwritten before the sweep ran: cur_eval # uses the new artifact weights, so its Calmar will differ from stored_winner's. if cur_eval is not None and stored_winner is not None: stored_cal = as_float(stored_winner.get("Calmar Ratio", -10.0)) eval_cal = as_float(cur_eval.get("Calmar Ratio", -10.0)) if stored_cal > 0 and eval_cal < stored_cal * 0.5: print( f" [WARN] {tf}: cur_eval Calmar={eval_cal:.3f} is <50% of stored winner " f"Calmar={stored_cal:.3f} — winner artifact was likely overwritten during " f"training; using stored metrics as promotion floor.", flush=True ) # Pick best gate-passing candidate for promotion. The globally top-ranked row # often has WFO_Neg_Folds > 0 and fails promotable(), leaving gate-passing rows # stranded below it. Since results is already sorted by rank_tuple, the first # gate-passing candidate here is the highest-ranked one. def _gate_passes(row: dict[str, Any]) -> bool: return ( row.get("source") != "current" and as_float(row.get("P&L/DD Ratio", 0.0)) > 0 and as_float(row.get("Fragile_0.05", 99)) == 0 and as_float(row.get("Sub1_Calmar", -10.0)) > 0 and as_float(row.get("Sub2_Calmar", -10.0)) > 0 and int(as_float(row.get("WFO_Neg_Folds", 99))) == 0 ) gate_passing = [r for r in results if _gate_passes(r)] best = gate_passing[0] if gate_passing else best_overall can_promote = promotable(best, cur_eval, stored_winner) # Diagnostic: report when OOS gate specifically is the reason promotion is blocked. if not can_promote and gate_passing and cur_eval is not None: best_oos_t = int(as_float(best.get("OOS_Total Trades"), 0.0)) ref_oos_t = int(max( as_float((stored_winner or {}).get("OOS_Total Trades"), 0.0), as_float(cur_eval.get("OOS_Total Trades"), 0.0), )) bear_pct = as_float(best.get("OOS_Bear_Pct"), 0.0) if best_oos_t >= OOS_MIN_TRADES and ref_oos_t >= OOS_MIN_TRADES and bear_pct < OOS_BEAR_PCT_SKIP: best_oos_sor = as_float(best.get("OOS_Sortino Ratio"), -10.0) ref_oos_sor = max( as_float(cur_eval.get("OOS_Sortino Ratio"), -10.0), as_float((stored_winner or {}).get("OOS_Sortino Ratio"), -10.0), ) if best_oos_sor < ref_oos_sor - OOS_SORTINO_DELTA: print( f" [OOS BLOCK] {tf}: new OOS_Sortino={best_oos_sor:.3f} < " f"ref={ref_oos_sor:.3f} − {OOS_SORTINO_DELTA} " f"(trades: best={best_oos_t}, ref={ref_oos_t}, bear={bear_pct:.0f}%)", flush=True, ) # --force-promote: write best gate-passing row even when no prior winner exists. # Intended for first-ever run on a new asset where promotable() always returns False. force_writing = force_promote and gate_passing and cur_eval is None SWEEP_DIR.mkdir(parents=True, exist_ok=True) out = SWEEP_DIR / f"optimization_sweep_strategy_mlp_scores_{ASSET}_{tf}.csv" pd.DataFrame(results[:top_k]).to_csv(out, index=False) if (promote and can_promote) or force_writing: write_winner(tf, best) return { "timeframe": tf, "evaluated": len(candidates), "workers": actual_workers, "sweep_file": str(out.relative_to(REPO)), "promoted": bool((promote and can_promote) or force_writing), "promotable": bool(can_promote), "current_calmar": cur_eval.get("Calmar Ratio") if cur_eval else np.nan, "current_pnl_dd": cur_eval.get("P&L/DD Ratio") if cur_eval else np.nan, "current_wfo_min": cur_eval.get("WFO_Min_Fold") if cur_eval else np.nan, "current_oos_sortino": cur_eval.get("OOS_Sortino Ratio") if cur_eval else np.nan, "best_source": best.get("source"), "best_weights": best.get("mlp_weights_file"), "best_entry": best.get("i_long_entry_activation_threshold"), "best_exit": best.get("i_long_exit_activation_threshold"), "best_conf": best.get("i_long_exit_activation_confirmation_threshold"), "best_trail": best.get("i_trailing_stop_threshold"), "best_suppress_bear": best.get("i_mvrv_suppress_bear"), "best_trades": best.get("Total Trades"), "best_pnl": best.get("Total P&L %"), "best_drawdown": best.get("Max Drawdown %"), "best_calmar": best.get("Calmar Ratio"), "best_pnl_dd": best.get("P&L/DD Ratio"), "best_composite": best.get("Composite"), "best_sub1_calmar": best.get("Sub1_Calmar"), "best_sub2_calmar": best.get("Sub2_Calmar"), "best_wfo_score": best.get("WFO_Score"), "best_wfo_min": best.get("WFO_Min_Fold"), "best_wfo_neg": best.get("WFO_Neg_Folds"), "best_oos_trades": best.get("OOS_Total Trades"), "best_oos_pnl": best.get("OOS_Total P&L %"), "best_oos_sortino": best.get("OOS_Sortino Ratio"), "best_fragile_0_05": best.get("Fragile_0.05"), "best_fragile_1_0": best.get("Fragile_1.0"), # overall top-ranked row (may differ from best when it failed gate checks) "overall_wfo_neg": best_overall.get("WFO_Neg_Folds"), "overall_calmar": best_overall.get("Calmar Ratio"), "overall_pnl_dd": best_overall.get("P&L/DD Ratio"), } def main() -> None: parser = argparse.ArgumentParser(description="Run deeper MLP threshold sweeps") parser.add_argument("--asset", default=_DEFAULT_ASSET, help="Asset identifier, e.g. COINBASE_ETHUSD (default: COINBASE_BTCUSD)") parser.add_argument("--timeframes", nargs="+", default=TFS, choices=TFS) parser.add_argument("--samples", type=int, default=50000, help="Approximate candidates per timeframe") parser.add_argument("--seed", type=int, default=20260618) parser.add_argument("--top-k", type=int, default=2000) parser.add_argument("--extra-weights", nargs="*", default=[], help="Additional MLP artifact JSONs to include as candidate weight files for every selected timeframe.") parser.add_argument("--workers", type=int, default=0, help="Worker processes for candidate evaluation. Use 0 for auto, " "currently 50%% of logical CPUs capped by candidate count; " "use 1 for sequential evaluation.") parser.add_argument("--worker-fraction", type=float, default=DEFAULT_WORKER_FRACTION, help="Logical CPU fraction used when --workers=0.") parser.add_argument("--promote", action="store_true", help="Write improved winner CSVs") parser.add_argument("--force-promote", action="store_true", help="Write best gate-passing candidate even when no prior winner exists " "(use for first-ever run on a new asset). Implies --promote.") parser.add_argument("--wide", action="store_true", help="Double the threshold search ranges: entry ±600, exit/conf ±750, trail 0-100.") parser.add_argument("--entry-lo", type=float, default=None, help="Override entry threshold lower bound.") parser.add_argument("--entry-hi", type=float, default=None, help="Override entry threshold upper bound.") parser.add_argument("--exit-lo", type=float, default=None, help="Override exit threshold lower bound.") parser.add_argument("--exit-hi", type=float, default=None, help="Override exit threshold upper bound.") parser.add_argument("--_internal-shard", action="store_true", help=argparse.SUPPRESS) parser.add_argument("--_shard-tf", default=None, help=argparse.SUPPRESS) parser.add_argument("--_shard-input", default=None, help=argparse.SUPPRESS) parser.add_argument("--_shard-output", default=None, help=argparse.SUPPRESS) args = parser.parse_args() global ASSET ASSET = args.asset if args._internal_shard: if not args._shard_tf or not args._shard_input or not args._shard_output: parser.error("internal shard mode requires tf, input, and output") run_internal_shard(args._shard_tf, Path(args._shard_input), Path(args._shard_output)) return REPORT_DIR.mkdir(parents=True, exist_ok=True) summaries = [] for offset, tf in enumerate(args.timeframes): print(f"\n=== {tf}: {args.samples:,} requested samples ===", flush=True) summaries.append(run_tf(tf, args.samples, args.seed + offset * 1009, args.top_k, args.promote or args.force_promote, args.extra_weights, args.workers, args.worker_fraction, force_promote=args.force_promote, wide=args.wide, entry_lo=args.entry_lo, entry_hi=args.entry_hi, exit_lo=args.exit_lo, exit_hi=args.exit_hi)) summary_path = REPORT_DIR / "mlp_deep_sweep_summary.csv" pd.DataFrame(summaries).to_csv(summary_path, index=False) print(f"\nSummary written: {summary_path.relative_to(REPO)}") print(pd.DataFrame(summaries).to_string(index=False)) if __name__ == "__main__": main()