""" 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 strategies.mlp_comparison import ComparisonSnapshot, require_replay, fingerprint from strategies.mlp_accounting import window_metrics, COMMISSION_RATE from strategies.mlp_execution_stability import assess_stability, stability_assessed, stability_passes, stability_periods, VERSION as STABILITY_VERSION 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" _COMPARISON: ComparisonSnapshot | None = None _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", # Persisted metrics are provenance; future comparisons replay their inputs. "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: raise ValueError(f"Existing winner has no replayable row: {path}") 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"] = as_bool(row.get("i_use_long_exit_confirmation", True)) 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]: numeric = ['i_long_entry_activation_threshold', 'i_long_exit_activation_threshold', 'i_long_exit_activation_confirmation_threshold', 'i_trailing_stop_threshold', 'i_regime_window', 'i_regime_entry_min_score', 'i_exit_score_window', 'i_entry_score_window'] for key in numeric: if key in p and not np.isfinite(float(p[key])): raise ValueError(f'Nonfinite parameter: {key}') return { "mlp_weights_file": str(Path(p["mlp_weights_file"]).resolve()), "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": as_bool(p.get("i_use_long_exit_confirmation", True)), "i_use_long_entry_confirmation": as_bool(p.get("i_use_long_entry_confirmation", False)), "i_regime_window": int(as_float(p.get("i_regime_window", 0))), "i_regime_entry_min_score": as_float(p.get("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]]: current = load_winner(tf) weights = set(extra_weights or []) if current: weights.add(current['mlp_weights_file']) if Path(artifact_path(tf)).exists(): weights.add(artifact_path(tf)) if tf == '1D' and ASSET == 'COINBASE_BTCUSD' and Path(artifact_path('12H')).exists(): weights.add(artifact_path('12H')) weight_choices = sorted({str(Path(path).resolve()) for path in weights}) entry_lo = entry_lo if entry_lo is not None else (-600. if wide else -300.) entry_hi = entry_hi if entry_hi is not None else (600. if wide else 300.) exit_lo = exit_lo if exit_lo is not None else (-750. if wide else -300.) exit_hi = exit_hi if exit_hi is not None else (750. if wide else 400.) trail_hi = 100. if wide else 50. # Every artifact receives the same additional threshold vectors. A single # frozen configuration per artifact remains visible separately. Historical # optimization budgets are unknown and are not made equal by this search. if not weight_choices: raise ValueError('No artifacts available') budget = max(0, samples // len(weight_choices)) rng = np.random.default_rng(seed) templates = [] for _ in range(budget): templates.append(dict( i_long_entry_activation_threshold=float(rng.choice(np.arange(entry_lo, entry_hi, 5.))), i_long_exit_activation_threshold=float(rng.choice(np.arange(exit_lo, exit_hi, 10.))), i_long_exit_activation_confirmation_threshold=float(rng.choice(np.arange(exit_lo, exit_hi, 10.))), i_trailing_stop_threshold=float(rng.choice(np.arange(0., trail_hi, 5.))), i_use_long_exit_confirmation=True, i_use_long_entry_confirmation=False, i_regime_window=0, i_regime_entry_min_score=-1000., 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])))) balanced = [] for weights in weight_choices: frozen = current if current and str(Path(current['mlp_weights_file']).resolve()) == weights else recommended_from_artifact(weights) balanced.append(dict(normalise_param(frozen), source='frozen', Additional_Threshold_Budget=len(templates))) for template in templates: balanced.append(dict(template, mlp_weights_file=weights, source='equal_tuning', Additional_Threshold_Budget=len(templates))) return balanced def load_scored_frame(tf: str, weights_path: str) -> tuple[pd.DataFrame, np.ndarray, np.ndarray, np.ndarray]: full = _COMPARISON.frame() if _COMPARISON else pd.read_csv(data_path(tf)) full.columns = full.columns.str.lower().str.strip() unit = 's' if pd.api.types.is_numeric_dtype(full['time']) else None full["time"] = pd.to_datetime(full["time"], unit=unit, utc=True).dt.tz_localize(None) full = full.loc[full["time"] >= TRAIN_START].copy().reset_index(drop=True) art = _COMPARISON.artifact(weights_path) if _COMPARISON else load_mlp_artifact(weights_path) art_cols = art.get("feature_cols") or [] if _COMPARISON and art_cols != json.loads(_COMPARISON.manifest_json)["contract"]["feature_schema"]: raise ValueError("Artifact feature schema differs from incumbent comparison schema") 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." ) if _COMPARISON: required = list(dict.fromkeys((art_cols or list(_FEATURE_COLS)) + ['close', 'high', 'stoch_peak_norm'])) missing = [c for c in required if c not in full] if missing: raise ValueError(f'Missing comparison features: {missing}') measured = full.loc[full.time >= pd.Timestamp(SCORE_START), required] if not np.isfinite(measured.to_numpy(dtype=float)).all(): raise ValueError('Nonfinite comparison prices/features') if (full[['close', 'high']] <= 0).any().any(): raise ValueError('Nonpositive comparison prices') 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"]) if _COMPARISON and not np.isfinite(scored.activation_score).all(): raise ValueError("Nonfinite model scores") return scored, full["stoch_peak_norm"].to_numpy(np.float64) if _COMPARISON else 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) + pd.Timedelta(days=1))] 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"] cache_key = (_COMPARISON.fingerprint, weights, _COMPARISON.artifact_hashes[weights]) if _COMPARISON else weights if cache_key not in cache: try: cache[cache_key] = load_scored_frame(tf, weights) except ValueError as exc: print(f" [SKIP] {os.path.basename(weights)}: {exc}", flush=True) cache[cache_key] = _STALE_SENTINEL if cache[cache_key] 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")}) result.update(Evaluation_Status='failed', Comparison_Fingerprint=_COMPARISON.fingerprint if _COMPARISON else '') return result scored, stoch_peak, close_arr, high_arr = cache[cache_key] if _COMPARISON and as_bool(params.get("i_mvrv_suppress_bear", False)): if "mvrv_regime" not in scored or not np.isfinite(scored.mvrv_regime).all(): raise ValueError("Missing or nonfinite MVRV execution gate") 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) + pd.Timedelta(days=1)].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) # Keep prior history so the first measured interval retains carry-in P&L # and entry fees. The accounting layer applies the lower scoring bound. sub2 = calculate_metrics(train_sigs, score_start=SUBPERIOD_SPLIT) fold_calmars = [] all_periods = [] for _, oos_start, oos_end in WFO_FOLDS: fold_window = train_sigs[train_sigs["time"] < pd.Timestamp(oos_end) + pd.Timedelta(days=1)].copy() fold_metrics = calculate_metrics(fold_window, score_start=oos_start, min_trades=WFO_MIN_OOS_TRADES) period = dict(start=oos_start, end=oos_end, eligible=int(fold_metrics.get('Total Trades', 0)) >= WFO_MIN_OOS_TRADES) try: period.update(window_metrics(fold_window, start=oos_start)) except ValueError: period['coverage'] = 'empty' all_periods.append(period) 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) for name, start, end in [('IS', SCORE_START, TRAIN_END), ('Sub1', SCORE_START, str(pd.Timestamp(SUBPERIOD_SPLIT) - pd.Timedelta(nanoseconds=1))), ('Sub2', SUBPERIOD_SPLIT, TRAIN_END), ('OOS', OOS_START, None)]: period = dict(name=name, start=start, end=end) try: period.update(window_metrics(sigs, start=start, end=end)) except ValueError: period['coverage'] = 'empty' all_periods.append(period) result: dict[str, Any] = {k: params[k] for k in PARAM_COLS} if _COMPARISON: bar = pd.Timedelta(hours={'4H':4, '6H':6, '8H':8, '12H':12, '1D':24}[tf]) covered = (scored.time.min() <= pd.Timestamp(SCORE_START) and scored.time.max() >= pd.Timestamp(TRAIN_END) + pd.Timedelta(days=1) - bar and (scored.time >= pd.Timestamp(OOS_START)).sum() >= 2) result.update(Comparison_Fingerprint=_COMPARISON.fingerprint, Artifact_SHA256=_COMPARISON.artifact_hashes[weights], Evaluation_Status='complete' if covered else 'incomplete') if _COMPARISON: result['Evaluation_Key'] = fingerprint(dict(context=_COMPARISON.fingerprint, artifact=result['Artifact_SHA256'], params={k: params[k] for k in PARAM_COLS})) result['Period_Measurements'] = json.dumps(all_periods, default=str) result['Additional_Threshold_Budget'] = params.get('Additional_Threshold_Budget', 0) 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"), } ) result['Stability_Status'] = 'unassessed_ineligible' if _COMPARISON and result['Evaluation_Status'] == 'complete' and ( params.get('source') == 'current' or core_gates_pass(result)): report = assess_stability(scored, signal_params, stoch_peak, close_arr, high_arr, _score_to_signals, stability_periods()) result.update(Stability_Status='passed' if report['passed'] else 'failed', Stability_Report=json.dumps(report, sort_keys=True), Stability_Evaluation_Key=result['Evaluation_Key'], Stability_Context=_COMPARISON.fingerprint) return result def init_worker(tf: str, asset: str, snapshot=None) -> None: global _WORKER_TF, _WORKER_CACHE, ASSET, _COMPARISON _COMPARISON = snapshot if snapshot: snapshot.validate_sources(REPO) _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, ...]: stability_ok = int(stability_passes(row)) 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(row.get("Evaluation_Status", "complete") == "complete"), int(composite > 0), int(pnl_dd > 0), stability_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 core_gates_pass(row: dict[str, Any]) -> bool: return (as_float(row.get('P&L/DD Ratio', 0.)) > 0 and as_float(row.get('Sub1_Calmar', -10.)) > 0 and as_float(row.get('Sub2_Calmar', -10.)) > 0 and int(as_float(row.get('WFO_Neg_Folds', 99))) == 0) 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 not stability_assessed(current): return False if as_float(best.get("P&L/DD Ratio", 0.0)) <= 0: return False if not stability_passes(best): 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 if (not best.get('Comparison_Fingerprint') or best.get('Comparison_Fingerprint') != current.get('Comparison_Fingerprint') or best.get('Evaluation_Status') != 'complete' or current.get('Evaluation_Status') != 'complete'): return False ref_pnl_dd = as_float(current.get('P&L/DD Ratio'), 0.) ref_calmar = as_float(current.get('Calmar Ratio'), -10.) 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(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 = as_float(current.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, _COMPARISON 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"]) _COMPARISON = payload.get("snapshot") if _COMPARISON: _COMPARISON.validate_sources(REPO) 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], "snapshot": _COMPARISON, }, ) 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, _COMPARISON), ) 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, expected_comparison_fingerprint: str | None = None, expected_incumbent_identity: str | 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 original numbers only as historical provenance. stored_winner = dict(current) if current else None if current: current = normalise_param(current) current["source"] = "current" candidates.insert(0, current) global _COMPARISON _COMPARISON = capture_comparison(tf, [p['mlp_weights_file'] for p in candidates]) if expected_comparison_fingerprint and expected_comparison_fingerprint != _COMPARISON.fingerprint: raise ValueError('Comparison context changed since adaptive baseline') 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) if stored_winner is not None: require_replay(cur_eval or {}) if expected_incumbent_identity and cur_eval['Evaluation_Key'] != expected_incumbent_identity: raise ValueError('Incumbent identity changed since adaptive baseline') repeated = require_replay(evaluate(tf, current, {})) if json.dumps(cur_eval, sort_keys=True) != json.dumps(repeated, sort_keys=True): raise ValueError('Incumbent replay is not reproducible') _COMPARISON.validate_sources(REPO) # 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 row.get("Evaluation_Status") == "complete" and as_float(row.get("P&L/DD Ratio", 0.0)) > 0 and stability_passes(row) 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(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 = as_float(cur_eval.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 = bool(force_promote and gate_passing and stored_winner 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: # Paths written to the winner must still denote the evaluated bytes. if data_path(tf).read_bytes() != _COMPARISON.data: raise ValueError('Comparison data changed before promotion') for path, raw in _COMPARISON.artifacts: if Path(path).read_bytes() != raw: raise ValueError('Comparison artifact changed before promotion') if load_winner(tf) != stored_winner: raise ValueError('Incumbent changed before promotion') _COMPARISON.validate_sources(REPO) write_winner(tf, best) return { "timeframe": tf, "Comparison_Fingerprint": _COMPARISON.fingerprint, "comparison_manifest": _COMPARISON.manifest_json, "stored_calmar_provenance": (stored_winner or {}).get('Calmar Ratio'), "historical_discrepancy": 'Unresolved; stored metrics excluded from numerical comparison', "historical_search_budget": 'Unknown and unequal; equal additional trials do not erase historical selection', "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"), "best_stability_status": best.get("Stability_Status", "unassessed"), "current_stability_status": cur_eval.get("Stability_Status", "unassessed") if cur_eval else None, "stability_assessed_candidates": sum(r.get('Stability_Status') in ('passed', 'failed') for r in results), # 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 capture_comparison(tf, weights): winner = load_winner(tf) schema_path = winner["mlp_weights_file"] if winner else weights[0] schema = json.loads(Path(schema_path).read_bytes())["feature_cols"] return ComparisonSnapshot.capture(data_path(tf), weights, repo=REPO, asset=ASSET, tf=tf, contract=dict(feature_schema=schema, commission=COMMISSION_RATE, train_start=TRAIN_START, score_start=SCORE_START, train_end=TRAIN_END, oos_start=OOS_START, split=SUBPERIOD_SPLIT, folds=WFO_FOLDS, min_fold_trades=WFO_MIN_OOS_TRADES, min_valid_folds=WFO_MIN_VALID_FOLDS, execution_stability=STABILITY_VERSION, execution='close decisions; next interval exposure/entry fee; additive fees; terminal mark', coverage='complete IS bounds; at least two retained OOS bars; finite measured features; retain gaps; drop unconfirmed last row')) def replay_incumbent(asset, tf): global ASSET, _COMPARISON previous_asset, previous_snapshot = ASSET, _COMPARISON try: ASSET = asset winner = load_winner(tf) if winner is None: raise ValueError(f'No incumbent for {asset} {tf}') params = normalise_param(winner) params['source'] = 'current' _COMPARISON = capture_comparison(tf, [params['mlp_weights_file']]) row = require_replay(evaluate(tf, params, {})) again = require_replay(evaluate(tf, params, {})) if row != again: # NaN diagnostics compare unequal; serialized values are stable. if json.dumps(row, sort_keys=True) != json.dumps(again, sort_keys=True): raise ValueError('Incumbent replay is not reproducible') _COMPARISON.validate_sources(REPO) return row finally: ASSET, _COMPARISON = previous_asset, previous_snapshot def run_tf(*args, **kwargs): global _COMPARISON previous = _COMPARISON try: return _run_tf(*args, **kwargs) finally: _COMPARISON = previous 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("--expected-comparison-fingerprint") parser.add_argument("--expected-incumbent-identity") 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, expected_comparison_fingerprint=args.expected_comparison_fingerprint, expected_incumbent_identity=args.expected_incumbent_identity)) 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()