#!/usr/bin/env python3 """ QuanterLab — Grid Search Parameter Optimizer Generated by QuanterLab (https://quanterlab.com) on 2026-05-13 19:18 Runs a 50x50 parameter sweep over: X-axis: level_covariance (0.0001 to 1) Y-axis: trend_covariance (1e-05 to 0.1) Outputs: PNG heatmap report + CSV results + interactive HTML 3D surfaces Ticker: NFLX Method: kalman_trend NOTE: This code uses yfinance (unadjusted close) as its data source. The QuanterLab platform uses FMP (also unadjusted close). Minor differences may occur due to data provider rounding or delayed updates. The strategy logic and engine are identical. """ # ============================================================================ # IMPORTS # ============================================================================ import pandas as pd import numpy as np from scipy import stats from statsmodels.regression.linear_model import OLS from statsmodels.tools import add_constant from statsmodels.tsa.stattools import adfuller from datetime import datetime, timedelta import warnings import sys import time warnings.filterwarnings('ignore') try: from arch import arch_model HAS_ARCH = True except ImportError: HAS_ARCH = False print("[WARNING] arch library not installed. GARCH sigma mode will fall back to constant.") print(" Install with: pip install arch") try: import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt from matplotlib.colors import TwoSlopeNorm HAS_MPL = True except ImportError: HAS_MPL = False print("[WARNING] matplotlib not installed. PNG report will be skipped.") print(" Install with: pip install matplotlib") try: from sklearn.preprocessing import StandardScaler from hmmlearn import hmm as hmm_module HAS_HMM = True except ImportError: HAS_HMM = False # ============================================================================ # CONFIGURATION — mirrors ALL side panel settings at time of export # ============================================================================ TICKER = 'NFLX' TIMEFRAME = '1h' LOOKBACK_DAYS = 365 DURATION = '1y' DIRECTION = 'long' INITIAL_CAPITAL = 100000 METHOD = 'kalman_trend' # Risk management (Section 3) SPREAD_BPS = 0 COMMISSION_PER_TRADE = 0.0 MARKET_IMPACT_ENABLED = False MARKET_IMPACT_K = 0.1 LEVERAGE = 1.0 # Only applies to pairs trading STOP_LOSS_PCT = None TAKE_PROFIT_PCT = None MAX_HOLDING_DAYS = None # Position sizing POSITION_SIZING = 'full_equity' POSITION_SIZING_MODE = 'fixed' RISK_PER_TRADE_PCT = 10 / 100 # as fraction RISK_BUDGET_K = 0.02 # Sweep parameters X_PARAM = 'level_covariance' Y_PARAM = 'trend_covariance' X_VALUES = np.linspace(0.0001, 1, 50).tolist() Y_VALUES = np.linspace(1e-05, 0.1, 50).tolist() # Fixed parameters (not being swept) FIXED_PARAMS = {'observation_covariance': 0.401, 'trend_threshold': 1, 'trend_exit_z': 0} # HMM Regime Filter HMM_ENABLED = False HMM_N_REGIMES = 3 HMM_TRAINING_WINDOW = 252 HMM_RETRAIN_INTERVAL = 63 HMM_ALLOWED_REGIMES = [0, 1] HMM_USE_VIX = True if HMM_ENABLED and not HAS_HMM: print("[WARNING] HMM enabled but hmmlearn not installed. Regime filter will be skipped.") print(" Install with: pip install hmmlearn scikit-learn") # Out-of-Sample Reserve OOS_PCT = 0 # ============================================================================ # DATA LOADING (yfinance) # ============================================================================ def load_data(ticker, days, timeframe='1d'): try: import yfinance as yf except ImportError: print("yfinance not installed. Install with: pip install yfinance") return pd.DataFrame() interval_map = {'1d': '1d', '1h': '1h', '2h': '1h', '4h': '1h', '6h': '1h'} interval = interval_map.get(timeframe, '1d') YF_LIMITS = {'1h': 729, '1d': 36500} max_days = YF_LIMITS.get(interval, 36500) actual_days = min(days, max_days) if days > max_days: print(f" [yfinance limit] {interval} data capped at {max_days} days (requested {days}d)") for attempt in range(3): try: end = datetime.now() start = end - timedelta(days=actual_days) df = yf.download(ticker, start=start, end=end, interval=interval, progress=False, auto_adjust=False) if isinstance(df.columns, pd.MultiIndex): df.columns = df.columns.get_level_values(0) df.columns = [c.lower() for c in df.columns] df = df[['open', 'high', 'low', 'close', 'volume']] if df.empty or len(df) < 10: raise ValueError(f"Insufficient data ({len(df)} bars)") # Resample 1h -> 2h/4h/6h if needed (yfinance doesn't support these natively) if timeframe in ('2h', '4h', '6h') and interval == '1h': df = df.resample(timeframe).agg({ 'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum' }).dropna() return df except Exception as e: if attempt < 2: reduced = int(actual_days * 0.8) print(f" [Retry {attempt+1}/3] {ticker} download failed: {e}. Retrying with {reduced}d...") actual_days = reduced else: print(f" [FAILED] Could not load {ticker} after 3 attempts: {e}") return pd.DataFrame() _BUFFER_DAYS = 33 # Extra bars for lookback warm-up print(f"Loading data for {TICKER}...") df = load_data(TICKER, LOOKBACK_DAYS + _BUFFER_DAYS, TIMEFRAME) if df.empty: print("No data. Exiting.") sys.exit(1) # Compute target trading bars: load without buffer to see how many bars the user's duration covers # For resampled timeframes (2h/4h), yfinance's 1h cap may cause both fetches to return # the same data. In that case, estimate buffer bars from calendar days and bar frequency. _df_target = load_data(TICKER, LOOKBACK_DAYS, TIMEFRAME) _TARGET_BARS = len(_df_target) if not _df_target.empty else len(df) del _df_target # If buffer was eliminated by yfinance cap, estimate buffer bars from calendar days if _TARGET_BARS >= len(df): _BARS_PER_DAY = {'1h': 6.5, '2h': 3.25, '4h': 1.625, '6h': 1.083, '1d': 1.0} _bpd = _BARS_PER_DAY.get(TIMEFRAME, 1.0) _estimated_buffer_bars = max(int(_BUFFER_DAYS * _bpd), 30) _TARGET_BARS = max(len(df) - _estimated_buffer_bars, int(len(df) * 0.5)) print(f" [yfinance cap] Buffer collapsed; estimated {_estimated_buffer_bars} buffer bars from {_BUFFER_DAYS}d") print(f" Loaded {len(df)} bars (includes {_BUFFER_DAYS}d lookback buffer)") print(f" Target trading window: {_TARGET_BARS} bars (user's {LOOKBACK_DAYS}d duration)") # ============================================================================ # HMM REGIME COMPUTATION (MATCHES platform stochastic_engine._compute_hmm_regime) # ============================================================================ def compute_hmm_regimes(df_input, vix_data=None): """Compute HMM regime labels using expanding-window retraining. Returns numpy array of regime labels aligned to df_input index (by position).""" if not HMM_ENABLED or not HAS_HMM: return np.zeros(len(df_input), dtype=int) n_regimes = HMM_N_REGIMES training_window = HMM_TRAINING_WINDOW retrain_interval = HMM_RETRAIN_INTERVAL close = df_input['close'].astype(float) features = pd.DataFrame(index=df_input.index) features['log_return'] = np.log(close / close.shift(1)) features['realized_vol'] = features['log_return'].rolling(20).std() * np.sqrt(252) # Add VIX feature if available if HMM_USE_VIX and vix_data is not None and len(vix_data) > 0: try: vix_aligned = vix_data.reindex(df_input.index, method='ffill') features['vix_change'] = vix_aligned.pct_change() print(f" HMM: VIX feature added ({features['vix_change'].dropna().shape[0]} bars)") except Exception as e: print(f" HMM: VIX alignment failed: {e}") features = features.dropna() n = len(features) if n < max(60, training_window): print(f" HMM: insufficient data ({n} bars, need {training_window})") return np.zeros(len(df_input), dtype=int) # Map feature indices to df indices feature_to_df = {} for fi, idx in enumerate(features.index): pos = df_input.index.get_loc(idx) feature_to_df[fi] = pos hmm_regime = np.zeros(len(df_input), dtype=int) # Expanding-window training with periodic retraining current_model = None current_scaler = None last_train_idx = 0 for i in range(n): # Check if we need to train/retrain need_train = False if current_model is None and i >= training_window - 1: need_train = True elif current_model is not None and (i - last_train_idx) >= retrain_interval: need_train = True if need_train: train_end = i + 1 train_start = max(0, train_end - training_window * 2) # Cap at 2x window X_train = features.values[train_start:train_end] try: scaler_new = StandardScaler() X_scaled = scaler_new.fit_transform(X_train) # BIC-based model selection with 3 restarts best_model = None best_bic = np.inf for restart in range(3): try: model = hmm_module.GaussianHMM( n_components=n_regimes, covariance_type='full', n_iter=100, tol=1e-3, random_state=42 + restart * 7, verbose=False) model.fit(X_scaled) log_lik = model.score(X_scaled) n_params = n_regimes * (n_regimes - 1) + n_regimes * 2 + n_regimes * 3 + (n_regimes - 1) bic = -2 * log_lik + n_params * np.log(len(X_scaled)) if bic < best_bic: best_bic = bic best_model = model except Exception: continue if best_model is not None: current_model = best_model current_scaler = scaler_new last_train_idx = i except Exception: pass # Predict with current model if current_model is not None and i >= training_window - 1: try: x = features.values[i:i+1] x_scaled = current_scaler.transform(x) regime = current_model.predict(x_scaled)[0] df_idx = feature_to_df[i] hmm_regime[df_idx] = int(regime) except Exception: pass # Sort regimes by mean return (highest mean return = regime 0) regime_returns = {} for r in range(n_regimes): mask = np.array([hmm_regime[feature_to_df.get(fi, 0)] == r for fi in range(len(features))]) if mask.sum() > 0: regime_returns[r] = features['log_return'].values[mask].mean() else: regime_returns[r] = 0.0 sorted_regimes = sorted(regime_returns.keys(), key=lambda r: regime_returns[r], reverse=True) mapping = {old: new for new, old in enumerate(sorted_regimes)} original = hmm_regime.copy() for i in range(len(hmm_regime)): hmm_regime[i] = mapping.get(original[i], original[i]) dist = np.bincount(hmm_regime, minlength=n_regimes) print(f" HMM computed: {n_regimes} regimes, distribution={dist}") return hmm_regime # Load VIX data for HMM regime detection (if enabled) _vix_data = None if HMM_ENABLED and HAS_HMM and HMM_USE_VIX: print("Loading VIX data for HMM...") try: _df_vix = load_data('^VIX', LOOKBACK_DAYS + _BUFFER_DAYS, '1d') if not _df_vix.empty: _vix_data = _df_vix['close'] print(f" VIX loaded: {len(_vix_data)} bars") else: print(" VIX data empty, proceeding without VIX feature") except Exception as _e: print(f" VIX load failed: {_e}, proceeding without VIX feature") # Compute HMM regimes once (shared across all sweep iterations) _hmm_regimes = None if HMM_ENABLED and HAS_HMM: print("Computing HMM regimes...") _hmm_regimes = compute_hmm_regimes(df, _vix_data) print(f" HMM regime computation complete") elif HMM_ENABLED: print("HMM skipped (hmmlearn not installed)") # OOS Data Split _oos_split_idx = None _df_full = None _full_target_bars = _TARGET_BARS _full_hmm_regimes = _hmm_regimes.copy() if _hmm_regimes is not None else None if OOS_PCT > 0: # Split on TARGET window, not total bars — buffer is for indicator warm-up only _oos_target_bars = int(_TARGET_BARS * (1 - OOS_PCT / 100)) _oos_held_out = _TARGET_BARS - _oos_target_bars _oos_split_idx = len(df) - _oos_held_out _df_full = df.copy() df = df.iloc[:_oos_split_idx].copy() _TARGET_BARS = _oos_target_bars if _hmm_regimes is not None: _hmm_regimes = _hmm_regimes[:_oos_split_idx] print(f"OOS Reserve: {OOS_PCT}% ({_oos_held_out} bars held out)") print(f" In-sample: {_oos_split_idx} bars ({_oos_split_idx - _oos_target_bars} buffer + {_oos_target_bars} target), OOS: {_oos_held_out} bars") print(f" Adjusted target bars: {_TARGET_BARS}") # ============================================================================ # BACKTEST ENGINE (Kalman Trend Filter) — platform parity version # ============================================================================ # Matches stochastic_engine._compute_kalman_trend: # - timeframe dt rescaling (F[0,1] and Q scaled so daily=1.0) # - Joseph-form P update (preserves PSD under round-off) # - P0 = I (weakly informative, preserves pre-hardening signal behaviour) def run_single_backtest(df_input, params, target_bars=None, hmm_regimes=None): """Run a single Kalman Trend backtest. Returns metrics dict.""" level_cov = float(params.get('level_covariance', 0.01)) trend_cov = float(params.get('trend_covariance', 0.001)) obs_cov = float(params.get('observation_covariance', 1.0)) trend_threshold = float(params.get('trend_threshold', 1.0)) trend_exit_z = float(params.get('trend_exit_z', 0.0)) prices = df_input['close'].values n = len(prices) if n < 50: return {'sharpe': 0, 'total_return': 0, 'max_drawdown': 0, 'trades': 0, 'win_rate': 0, 'profit_factor': 0} # --- Timeframe rescaling (daily = 1.0, intraday assumes 6.5h trading day) --- _TF_DT = { '1d': 1.0, 'daily': 1.0, '1D': 1.0, '1h': 1.0 / 6.5, 'hourly': 1.0 / 6.5, '60m': 1.0 / 6.5, '30m': 0.5 / 6.5, '15m': 0.25 / 6.5, '5m': 5.0 / (6.5 * 60), '1m': 1.0 / (6.5 * 60), '2h': 2.0 / 6.5, '4h': 4.0 / 6.5, '6h': 6.0 / 6.5, '1w': 5.0, 'weekly': 5.0, '1mo': 21.0, 'monthly': 21.0, } dt = _TF_DT.get(TIMEFRAME, 1.0) # Kalman filter: state = [level, slope] state = np.array([prices[0], 0.0]) P = np.eye(2) # weakly informative prior F = np.array([[1.0, dt], [0.0, 1.0]]) H = np.array([[1.0, 0.0]]) Q = np.array([[level_cov * dt, 0.0], [0.0, trend_cov * dt]]) R_mat = np.array([[obs_cov]]) kft_signal = np.full(n, np.nan) _I2 = np.eye(2) for i in range(n): state_pred = F @ state P_pred = F @ P @ F.T + Q innovation = prices[i] - H @ state_pred S = H @ P_pred @ H.T + R_mat if S[0, 0] < 1e-15: S[0, 0] = 1e-15 K = P_pred @ H.T @ np.linalg.inv(S) state = state_pred + (K @ innovation).flatten() # Joseph-form P update — preserves PSD under round-off IKH = _I2 - K @ H P = IKH @ P_pred @ IKH.T + K @ R_mat @ K.T kft_signal[i] = np.clip(state[1] / np.sqrt(P[1, 1]), -5.0, 5.0) if P[1, 1] > 1e-10 else 0.0 # Generate entry/exit signals entry_signal = np.zeros(n, dtype=bool) exit_signal = np.zeros(n, dtype=bool) if DIRECTION == 'long': entry_signal = kft_signal > trend_threshold exit_signal = kft_signal < -trend_exit_z elif DIRECTION == 'short': entry_signal = kft_signal < -trend_threshold exit_signal = kft_signal > trend_exit_z else: entry_signal = np.abs(kft_signal) > trend_threshold exit_signal = np.abs(kft_signal) < trend_exit_z # HMM regime filter if hmm_regimes is not None and HMM_ENABLED: allowed = np.isin(hmm_regimes, HMM_ALLOWED_REGIMES) entry_signal = entry_signal & allowed # Simulate trades capital = INITIAL_CAPITAL position = 0 entry_price = 0.0 shares = 0 holding_days = 0 just_exited = False trades_list = [] equity = [capital] # Pin trading window if target_bars is not None and target_bars > 0: trade_start = max(0, n - target_bars) else: trade_start = 0 for i in range(trade_start, n): price = prices[i] just_exited = False # Exit check if position != 0: exit_trade = False holding_days += 1 if exit_signal[i]: exit_trade = True if STOP_LOSS_PCT: if DIRECTION == 'long' and price <= entry_price * (1 - STOP_LOSS_PCT): exit_trade = True elif DIRECTION == 'short' and price >= entry_price * (1 + STOP_LOSS_PCT): exit_trade = True if TAKE_PROFIT_PCT: if DIRECTION == 'long' and price >= entry_price * (1 + TAKE_PROFIT_PCT): exit_trade = True elif DIRECTION == 'short' and price <= entry_price * (1 - TAKE_PROFIT_PCT): exit_trade = True if MAX_HOLDING_DAYS and holding_days >= MAX_HOLDING_DAYS: exit_trade = True # HMM: force exit if regime leaves allowed set if hmm_regimes is not None and HMM_ENABLED: if hmm_regimes[i] not in HMM_ALLOWED_REGIMES: exit_trade = True if exit_trade: cost = (SPREAD_BPS / 10000) * shares * price if DIRECTION == 'long': pnl = (price - entry_price) * shares - cost else: pnl = (entry_price - price) * shares - cost capital += pnl trades_list.append(pnl) position, shares, holding_days = 0, 0, 0 just_exited = True # Entry check if position == 0 and not just_exited and entry_signal[i]: if np.isnan(kft_signal[i]): equity.append(equity[-1]) continue entry_price = price shares = int(capital / entry_price) if entry_price > 0 else 0 if shares > 0: cost = (SPREAD_BPS / 10000) * shares * entry_price capital -= cost position = 1 holding_days = 0 # Track equity eq = capital if position and shares > 0: if DIRECTION == 'long': eq += (price - entry_price) * shares else: eq += (entry_price - price) * shares equity.append(eq) # Force close if position and shares > 0: price = prices[-1] cost = (SPREAD_BPS / 10000) * shares * price if DIRECTION == 'long': pnl = (price - entry_price) * shares - cost else: pnl = (entry_price - price) * shares - cost capital += pnl trades_list.append(pnl) # Metrics total_return = (capital / INITIAL_CAPITAL - 1) * 100 n_trades = len(trades_list) if n_trades == 0: return {'sharpe': float('nan'), 'total_return': 0, 'max_drawdown': 0, 'trades': 0, 'win_rate': 0, 'profit_factor': 0} wins = [t for t in trades_list if t > 0] losses = [t for t in trades_list if t <= 0] win_rate = len(wins) / n_trades * 100 eq_arr = np.array(equity) running_max = np.maximum.accumulate(eq_arr) drawdowns = (eq_arr - running_max) / running_max * 100 max_dd = abs(float(np.min(drawdowns))) # Sharpe if len(equity) > 1: returns = np.diff(eq_arr) / eq_arr[:-1] if np.std(returns) > 0: sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252) else: sharpe = 0 else: sharpe = 0 # Profit factor gross_profit = sum(wins) if wins else 0 gross_loss = abs(sum(losses)) if losses else 0 pf = gross_profit / gross_loss if gross_loss > 0 else (99.99 if gross_profit > 0 else 0) # Penalize very few trades if n_trades < 3: sharpe *= 0.5 return { 'sharpe': round(sharpe, 3), 'total_return': round(total_return, 2), 'max_drawdown': round(max_dd, 2), 'trades': n_trades, 'win_rate': round(win_rate, 1), 'profit_factor': round(pf, 2) } # ============================================================================ # GRID SEARCH SWEEP # ============================================================================ def run_sweep(): """Run {len(X_VALUES)}x{len(Y_VALUES)} parameter grid search.""" total = len(X_VALUES) * len(Y_VALUES) print(f"\nStarting {len(X_VALUES)}x{len(Y_VALUES)} = {total} sweep...") print(f" X: {X_PARAM} = {X_VALUES[0]:.2f} to {X_VALUES[-1]:.2f}") print(f" Y: {Y_PARAM} = {Y_VALUES[0]:.2f} to {Y_VALUES[-1]:.2f}") print(f" Fixed: {FIXED_PARAMS}") print() sharpe_grid = np.zeros((len(Y_VALUES), len(X_VALUES))) return_grid = np.zeros((len(Y_VALUES), len(X_VALUES))) drawdown_grid = np.zeros((len(Y_VALUES), len(X_VALUES))) trades_grid = np.zeros((len(Y_VALUES), len(X_VALUES))) pf_grid = np.zeros((len(Y_VALUES), len(X_VALUES))) completed = 0 t0 = time.time() for yi, yv in enumerate(Y_VALUES): for xi, xv in enumerate(X_VALUES): params = dict(FIXED_PARAMS) params[X_PARAM] = int(round(xv)) if X_PARAM in {'lookback', 'signal_persistence', 'rolling_hedge_window', 'rolling_mean_window'} else round(xv, 4) params[Y_PARAM] = int(round(yv)) if Y_PARAM in {'lookback', 'signal_persistence', 'rolling_hedge_window', 'rolling_mean_window'} else round(yv, 4) try: metrics = run_single_backtest(df.copy(), params, target_bars=_TARGET_BARS, hmm_regimes=_hmm_regimes) sharpe_grid[yi, xi] = metrics.get('sharpe', 0) return_grid[yi, xi] = metrics.get('total_return', 0) drawdown_grid[yi, xi] = metrics.get('max_drawdown', 0) trades_grid[yi, xi] = metrics.get('trades', 0) pf_grid[yi, xi] = metrics.get('profit_factor', 0) except Exception as e: print(f" Error at {X_PARAM}={xv:.2f}, {Y_PARAM}={yv:.2f}: {e}") completed += 1 if completed % 10 == 0 or completed == total: elapsed = time.time() - t0 eta = (elapsed / completed) * (total - completed) if completed > 0 else 0 print(f" {completed}/{total} ({100*completed/total:.0f}%) — ETA: {eta:.0f}s", end='\r') print(f"\nSweep complete in {time.time()-t0:.1f}s") # Find best parameters by multiple metrics best_idx = np.unravel_index(np.nanargmax(sharpe_grid), sharpe_grid.shape) best_y, best_x = Y_VALUES[best_idx[0]], X_VALUES[best_idx[1]] # Best by Return ret_idx = np.unravel_index(np.nanargmax(return_grid), return_grid.shape) # Best by Profit Factor (cap at 999 to avoid inf) pf_capped = np.where(np.isfinite(pf_grid), pf_grid, -1) pf_idx = np.unravel_index(np.nanargmax(pf_capped), pf_capped.shape) # Best by Min Drawdown (lowest drawdown among configs with positive return) dd_search = np.where(return_grid > 0, drawdown_grid, np.inf) if np.all(np.isinf(dd_search)): dd_idx = np.unravel_index(np.nanargmin(drawdown_grid), drawdown_grid.shape) else: dd_idx = np.unravel_index(np.nanargmin(dd_search), dd_search.shape) def _fmt_row(label, idx): return (f" {label:<16} {X_VALUES[idx[1]]:>8.2f} {Y_VALUES[idx[0]]:>8.2f} " f"{sharpe_grid[idx]:>7.3f} {return_grid[idx]:>8.2f}% " f"{drawdown_grid[idx]:>7.2f}% {int(trades_grid[idx]):>6} {pf_grid[idx]:>6.2f}") print(f"\n==========================================================================================") print(f"BEST PARAMETERS BY METRIC") print(f"==========================================================================================") print(f" {'Optimized For':<16} {X_PARAM:>8} {Y_PARAM:>8} {'Sharpe':>7} {'Return':>9} {'Max DD':>8} {'Trades':>6} {'PF':>6}") print(f" {'-'*84}") print(_fmt_row("Best Sharpe", best_idx)) print(_fmt_row("Best Return", ret_idx)) print(_fmt_row("Best PF", pf_idx)) print(_fmt_row("Min Drawdown", dd_idx)) print(f"==========================================================================================") # OOS Validation: run best params on full data but ONLY trade in OOS window # The filter/indicator warms up on IS data, but metrics are pure OOS oos_metrics = None if OOS_PCT > 0 and _oos_split_idx is not None and _df_full is not None: print(f"\n============================================================") print("OUT-OF-SAMPLE VALIDATION (pure OOS — no IS contamination)") print(f"============================================================") best_params = dict(FIXED_PARAMS) int_params = {'lookback', 'signal_persistence', 'rolling_hedge_window', 'rolling_mean_window'} best_params[X_PARAM] = int(round(best_x)) if X_PARAM in int_params else round(best_x, 4) best_params[Y_PARAM] = int(round(best_y)) if Y_PARAM in int_params else round(best_y, 4) try: # Pin trading window to OOS-only bars: # trade_start = len(full_data) - oos_bars = oos_split_idx # This lets the indicator warm up on IS, but only trades in OOS _oos_target_bars = len(_df_full) - _oos_split_idx oos_only_metrics = run_single_backtest( _df_full.copy(), best_params, target_bars=_oos_target_bars, hmm_regimes=_full_hmm_regimes) is_metrics = { 'sharpe': sharpe_grid[best_idx], 'total_return': return_grid[best_idx], 'max_drawdown': drawdown_grid[best_idx], 'trades': int(trades_grid[best_idx]), 'profit_factor': pf_grid[best_idx] } header = f" {'Metric':<20} {'In-Sample':>12} {'OOS Only':>14}" print(header) print(f" {'-'*46}") for key in ['sharpe', 'total_return', 'max_drawdown', 'trades', 'profit_factor']: is_val = is_metrics.get(key, 0) oos_val = oos_only_metrics.get(key, 0) if key == 'trades': print(f" {key:<20} {int(is_val):>12} {int(oos_val):>14}") elif key == 'sharpe': print(f" {key:<20} {is_val:>12.3f} {oos_val:>14.3f}") else: print(f" {key:<20} {is_val:>12.2f} {oos_val:>14.2f}") print(f"============================================================") oos_metrics = oos_only_metrics except Exception as oos_err: print(f" OOS validation failed: {oos_err}") return sharpe_grid, return_grid, drawdown_grid, trades_grid, pf_grid, oos_metrics # ============================================================================ # ROBUSTNESS SCORECARD # ============================================================================ def _flood_fill(mask, start): """Flood-fill from start position, return set of (y,x) coords in contiguous region.""" from collections import deque ny, nx = mask.shape visited = set() queue = deque([start]) visited.add(start) while queue: cy, cx = queue.popleft() for dy, dx in [(-1,0),(1,0),(0,-1),(0,1)]: ny2, nx2 = cy+dy, cx+dx if 0 <= ny2 < ny and 0 <= nx2 < nx and (ny2, nx2) not in visited and mask[ny2, nx2]: visited.add((ny2, nx2)) queue.append((ny2, nx2)) return visited def _find_largest_zone(sharpe, threshold): """Find the largest contiguous region where sharpe >= threshold. Returns (zone_coords_set, zone_avg_sharpe).""" mask = sharpe >= threshold ny, nx = sharpe.shape visited_global = set() best_zone = set() best_avg = 0.0 for y in range(ny): for x in range(nx): if mask[y, x] and (y, x) not in visited_global: zone = _flood_fill(mask, (y, x)) visited_global |= zone if len(zone) > len(best_zone): zone_vals = [float(sharpe[cy, cx]) for cy, cx in zone] best_zone = zone best_avg = sum(zone_vals) / len(zone_vals) return best_zone, best_avg def _zone_bounds(zone_coords, x_values, y_values): """Get parameter range boundaries from a set of (y,x) index coordinates.""" if not zone_coords: return None ys = [c[0] for c in zone_coords] xs = [c[1] for c in zone_coords] return { X_PARAM: [round(float(x_values[min(xs)]), 4), round(float(x_values[max(xs)]), 4)], Y_PARAM: [round(float(y_values[min(ys)]), 4), round(float(y_values[max(ys)]), 4)] } def compute_robustness_scorecard(sharpe, returns, drawdown, trades, pf): """Analyse the grid surface for robustness signals. Two-layer analysis: A) PEAK — evaluates the single best Sharpe point and its neighbourhood. B) ZONE — finds the largest contiguous region with Sharpe > 0.5 (the "broad tradeable area"), even if it doesn't contain the peak. The overall score blends both: a sharp fragile peak with a broad zone elsewhere still gets a useful verdict.""" scorecard = {} ny, nx = sharpe.shape total_cells = nx * ny best_idx = np.unravel_index(np.nanargmax(sharpe), sharpe.shape) best_sharpe = float(sharpe[best_idx]) scorecard['best_sharpe'] = round(best_sharpe, 3) # ── A. PEAK ANALYSIS — neighbourhood of the single best point ────── by, bx = best_idx neighbours = [] for dy in range(-1, 2): for dx in range(-1, 2): if dy == 0 and dx == 0: continue ny2, nx2 = by + dy, bx + dx if 0 <= ny2 < ny and 0 <= nx2 < nx and not np.isnan(sharpe[ny2, nx2]): neighbours.append(float(sharpe[ny2, nx2])) if neighbours: scorecard['peak_neighbour_avg'] = round(float(np.mean(neighbours)), 3) scorecard['peak_neighbour_drop'] = round((1 - np.mean(neighbours) / best_sharpe) * 100, 1) if best_sharpe > 0 else 100 else: scorecard['peak_neighbour_avg'] = 0 scorecard['peak_neighbour_drop'] = 100 # Peak plateau: contiguous region around peak ≥80% of peak if best_sharpe > 0: peak_threshold = best_sharpe * 0.80 peak_plateau = _flood_fill(sharpe >= peak_threshold, best_idx) scorecard['peak_plateau_cells'] = len(peak_plateau) scorecard['peak_plateau_pct'] = round(len(peak_plateau) / total_cells * 100, 1) scorecard['peak_region'] = _zone_bounds(peak_plateau, X_VALUES, Y_VALUES) else: scorecard['peak_plateau_cells'] = 0 scorecard['peak_plateau_pct'] = 0 scorecard['peak_region'] = {X_PARAM: [float(X_VALUES[bx]), float(X_VALUES[bx])], Y_PARAM: [float(Y_VALUES[by]), float(Y_VALUES[by])]} # ── B. ZONE ANALYSIS — largest contiguous "tradeable" region ──────── # Use adaptive threshold: max(0.5, median of positive cells) positive_vals = sharpe[sharpe > 0] if len(positive_vals) > 0: zone_threshold = max(0.5, float(np.median(positive_vals))) else: zone_threshold = 0.5 scorecard['zone_threshold'] = round(zone_threshold, 3) zone_coords, zone_avg = _find_largest_zone(sharpe, zone_threshold) scorecard['zone_cells'] = len(zone_coords) scorecard['zone_pct'] = round(len(zone_coords) / total_cells * 100, 1) scorecard['zone_avg_sharpe'] = round(zone_avg, 3) scorecard['zone_region'] = _zone_bounds(zone_coords, X_VALUES, Y_VALUES) # Zone coverage in parameter space (grid-density invariant) # Measures what fraction of the X and Y parameter RANGES the zone spans x_range = float(X_VALUES[-1] - X_VALUES[0]) if len(X_VALUES) > 1 else 1 y_range = float(Y_VALUES[-1] - Y_VALUES[0]) if len(Y_VALUES) > 1 else 1 if zone_coords and x_range > 0 and y_range > 0: zr = scorecard['zone_region'] x_key = list(zr.keys())[0] y_key = list(zr.keys())[1] zone_x_span = (zr[x_key][1] - zr[x_key][0]) / x_range zone_y_span = (zr[y_key][1] - zr[y_key][0]) / y_range # Geometric mean of X and Y coverage — rewards zones that span both axes scorecard['zone_coverage'] = round(float((zone_x_span * zone_y_span) ** 0.5 * 100), 1) else: scorecard['zone_coverage'] = 0 # Check if peak is inside the zone scorecard['peak_in_zone'] = best_idx in zone_coords # Zone internal consistency: std of Sharpe within the zone if zone_coords: zone_sharpes = [float(sharpe[cy, cx]) for cy, cx in zone_coords] scorecard['zone_internal_std'] = round(float(np.std(zone_sharpes)), 3) # Zone avg metrics for the recommended region zone_rets = [float(returns[cy, cx]) for cy, cx in zone_coords if not np.isnan(returns[cy, cx])] zone_dds = [float(drawdown[cy, cx]) for cy, cx in zone_coords if not np.isnan(drawdown[cy, cx])] scorecard['zone_avg_return'] = round(float(np.mean(zone_rets)), 2) if zone_rets else 0 scorecard['zone_avg_dd'] = round(float(np.mean(zone_dds)), 2) if zone_dds else 0 else: scorecard['zone_internal_std'] = 0 scorecard['zone_avg_return'] = 0 scorecard['zone_avg_dd'] = 0 # ── C. SURFACE QUALITY — normalised gradient smoothness ───────────── # Normalise gradients per step: divide by the parameter step size so # wide ranges (lookback 5-500) don't unfairly inflate the gradient. grad_y = np.abs(np.diff(sharpe, axis=0)) # change per Y-step grad_x = np.abs(np.diff(sharpe, axis=1)) # change per X-step # Normalise by Sharpe range so gradient is relative, not absolute sharpe_range = float(np.nanmax(sharpe) - np.nanmin(sharpe)) if sharpe_range > 0: norm_grad_y = grad_y / sharpe_range norm_grad_x = grad_x / sharpe_range avg_gradient = float(np.nanmean(np.concatenate([norm_grad_y.ravel(), norm_grad_x.ravel()]))) else: avg_gradient = 0.0 scorecard['avg_gradient'] = round(avg_gradient, 4) # ── D. OVERALL SCORE — blended from peak + zone ──────────────────── score = 0 # Zone breadth (max 35 pts) — uses parameter-space coverage, grid-density invariant # zone_coverage is geometric mean of X/Y span as % of total range score += min(35, scorecard['zone_coverage'] * 0.7) # Zone consistency (max 20 pts) — low internal std = stable zone if scorecard['zone_avg_sharpe'] > 0: cv = scorecard['zone_internal_std'] / scorecard['zone_avg_sharpe'] # coefficient of variation score += max(0, 20 - cv * 40) # Peak stability (max 15 pts) — neighbourhood doesn't collapse score += max(0, 15 - scorecard['peak_neighbour_drop'] * 0.5) # Gradient smoothness (max 15 pts) — normalised gradient, expect ~0.05 for smooth score += max(0, 15 - avg_gradient * 100) # Bonus: peak inside zone means peak is part of a broad region (max 15 pts) if scorecard['peak_in_zone']: score += 15 scorecard['robustness_score'] = int(min(100, max(0, score))) if scorecard['robustness_score'] >= 70: scorecard['verdict'] = 'ROBUST — Broad stable zone, suitable for walk-forward' elif scorecard['robustness_score'] >= 40: scorecard['verdict'] = 'MODERATE — Tradeable zone exists but validate with walk-forward' else: scorecard['verdict'] = 'FRAGILE — No broad stable zone found' return scorecard def print_robustness_scorecard(sc): """Pretty-print the robustness scorecard to console.""" print(f"\n======================================================================") print("ROBUSTNESS SCORECARD") print(f"======================================================================") print(f" Overall Score: {sc['robustness_score']}/100") print(f" Verdict: {sc['verdict']}") print(f" {'-'*66}") # Peak analysis print(f" PEAK ANALYSIS (best Sharpe = {sc['best_sharpe']:.3f}):") print(f" Neighbourhood avg: {sc['peak_neighbour_avg']:.3f} ({sc['peak_neighbour_drop']:.1f}% drop)") print(f" Peak plateau: {sc['peak_plateau_cells']} cells ({sc['peak_plateau_pct']}% of grid)") if sc.get('peak_region'): for param, bounds in sc['peak_region'].items(): print(f" Peak range {param}: {bounds[0]:.4f} — {bounds[1]:.4f}") print() # Zone analysis print(f" ZONE ANALYSIS (threshold = {sc['zone_threshold']:.3f}):") print(f" Largest zone: {sc['zone_cells']} cells ({sc['zone_pct']}% of grid)") print(f" Parameter coverage: {sc['zone_coverage']}% of param space (grid-density invariant)") print(f" Zone avg Sharpe: {sc['zone_avg_sharpe']:.3f} (std: {sc['zone_internal_std']:.3f})") print(f" Zone avg Return: {sc['zone_avg_return']:.1f}%") print(f" Zone avg Max DD: {sc['zone_avg_dd']:.1f}%") print(f" Peak inside zone: {'YES' if sc['peak_in_zone'] else 'NO — peak is an isolated spike'}") if sc.get('zone_region'): for param, bounds in sc['zone_region'].items(): print(f" Zone range {param}: {bounds[0]:.4f} — {bounds[1]:.4f}") print(f" {'-'*66}") print(f" Surface smoothness: {sc['avg_gradient']:.4f} avg gradient") print(f"======================================================================") # ============================================================================ # PNG REPORT GENERATION # ============================================================================ def generate_report(sharpe, returns, drawdown, trades, pf, oos_metrics=None, scorecard=None): """Generate 3x3 report: 3D surfaces (top row) + 2D heatmaps (middle) + summary (bottom).""" if not HAS_MPL: print("Skipping PNG (matplotlib not installed)") return from mpl_toolkits.mplot3d import Axes3D x_labels = [f"{v:.1f}" if not isinstance(v, int) else str(v) for v in X_VALUES] y_labels = [f"{v:.1f}" if not isinstance(v, int) else str(v) for v in Y_VALUES] fig = plt.figure(figsize=(22, 20)) fig.patch.set_facecolor('#0d1117') fig.suptitle(f'SC001STCB Grid Search — {{TICKER}} ({METHOD})', color='white', fontsize=16, fontweight='bold', y=0.98) X_mesh, Y_mesh = np.meshgrid(X_VALUES, Y_VALUES) # ── Row 1: 3D Surfaces ────────────────────────────────────────────── surfaces_3d = [ (sharpe, 'Sharpe Ratio — 3D Surface', 'RdYlGn'), (returns, 'Total Return (%) — 3D Surface', 'RdYlGn'), (drawdown, 'Max Drawdown (%) — 3D Surface', 'RdYlGn_r'), ] for idx, (grid, title, cmap) in enumerate(surfaces_3d): ax = fig.add_subplot(3, 3, idx + 1, projection='3d') ax.set_facecolor('#0a0a0a') ax.xaxis.pane.fill = False ax.yaxis.pane.fill = False ax.zaxis.pane.fill = False ax.xaxis.pane.set_edgecolor('#333333') ax.yaxis.pane.set_edgecolor('#333333') ax.zaxis.pane.set_edgecolor('#333333') surf = ax.plot_surface(X_mesh, Y_mesh, grid, cmap=cmap, alpha=0.85, edgecolor='none', antialiased=True, rstride=1, cstride=1) # Contour projection on base z_offset = np.nanmin(grid) - (np.nanmax(grid) - np.nanmin(grid)) * 0.15 ax.contourf(X_mesh, Y_mesh, grid, zdir='z', offset=z_offset, cmap=cmap, alpha=0.3) ax.set_zlim(z_offset, np.nanmax(grid) * 1.05 if np.nanmax(grid) > 0 else np.nanmax(grid) * 0.95) ax.set_xlabel(X_PARAM.replace('_', ' ').title(), color='#8b949e', fontsize=8, labelpad=8) ax.set_ylabel(Y_PARAM.replace('_', ' ').title(), color='#8b949e', fontsize=8, labelpad=8) ax.set_title(title, color='white', fontsize=10, fontweight='bold', pad=4) ax.tick_params(colors='#8b949e', labelsize=7) ax.view_init(elev=25, azim=-45) fig.colorbar(surf, ax=ax, shrink=0.5, pad=0.08) # ── Row 2: 2D Heatmaps ───────────────────────────────────────────── heatmaps = [ (sharpe, 'Sharpe Ratio', 'RdYlGn', True), (returns, 'Total Return (%)', 'RdYlGn', True), (drawdown, 'Max Drawdown (%)', 'RdYlGn_r', False), ] for idx, (grid, title, cmap, diverging) in enumerate(heatmaps): ax = fig.add_subplot(3, 3, idx + 4) ax.set_facecolor('#161b22') if diverging and np.nanmin(grid) < 0 < np.nanmax(grid): norm = TwoSlopeNorm(vmin=np.nanmin(grid), vcenter=0, vmax=np.nanmax(grid)) else: norm = None im = ax.imshow(grid, cmap=cmap, aspect='auto', norm=norm, interpolation='nearest') fig.colorbar(im, ax=ax, fraction=0.046, pad=0.04) ax.set_xticks(range(len(x_labels))) ax.set_xticklabels(x_labels, rotation=45, fontsize=6, color='#8b949e') ax.set_yticks(range(len(y_labels))) ax.set_yticklabels(y_labels, fontsize=6, color='#8b949e') ax.set_xlabel(X_PARAM.replace('_', ' ').title(), color='#8b949e', fontsize=9) ax.set_ylabel(Y_PARAM.replace('_', ' ').title(), color='#8b949e', fontsize=9) ax.set_title(title, color='white', fontsize=10, fontweight='bold', pad=8) # Annotate cells (only for grids <= 15x15) if grid.shape[0] <= 15 and grid.shape[1] <= 15: for i in range(grid.shape[0]): for j in range(grid.shape[1]): val = grid[i, j] if np.isnan(val): continue text = f"{val:.2f}" if abs(val) < 100 else f"{val:.0f}" brightness = (val - np.nanmin(grid)) / (np.nanmax(grid) - np.nanmin(grid) + 1e-10) color = 'black' if brightness > 0.5 else 'white' ax.text(j, i, text, ha='center', va='center', fontsize=5, color=color) best = np.unravel_index(np.nanargmax(grid) if 'Drawdown' not in title else np.nanargmin(grid), grid.shape) ax.add_patch(plt.Rectangle((best[1]-0.5, best[0]-0.5), 1, 1, fill=False, edgecolor='cyan', linewidth=2)) # ── Row 3: Trade Count + Profit Factor + Summary ──────────────────── extras = [ (trades, 'Trade Count', 'YlOrRd', False), (pf, 'Profit Factor', 'RdYlGn', True), ] for idx, (grid, title, cmap, diverging) in enumerate(extras): ax = fig.add_subplot(3, 3, idx + 7) ax.set_facecolor('#161b22') norm = None if diverging and np.nanmin(grid) < 0 < np.nanmax(grid): norm = TwoSlopeNorm(vmin=np.nanmin(grid), vcenter=0, vmax=np.nanmax(grid)) im = ax.imshow(grid, cmap=cmap, aspect='auto', norm=norm, interpolation='nearest') fig.colorbar(im, ax=ax, fraction=0.046, pad=0.04) ax.set_xticks(range(len(x_labels))) ax.set_xticklabels(x_labels, rotation=45, fontsize=6, color='#8b949e') ax.set_yticks(range(len(y_labels))) ax.set_yticklabels(y_labels, fontsize=6, color='#8b949e') ax.set_xlabel(X_PARAM.replace('_', ' ').title(), color='#8b949e', fontsize=9) ax.set_ylabel(Y_PARAM.replace('_', ' ').title(), color='#8b949e', fontsize=9) ax.set_title(title, color='white', fontsize=10, fontweight='bold', pad=8) if grid.shape[0] <= 15 and grid.shape[1] <= 15: for i in range(grid.shape[0]): for j in range(grid.shape[1]): val = grid[i, j] if np.isnan(val): continue text = f"{val:.2f}" if abs(val) < 100 else f"{val:.0f}" brightness = (val - np.nanmin(grid)) / (np.nanmax(grid) - np.nanmin(grid) + 1e-10) color = 'black' if brightness > 0.5 else 'white' ax.text(j, i, text, ha='center', va='center', fontsize=5, color=color) best = np.unravel_index(np.nanargmax(grid) if 'Drawdown' not in title else np.nanargmin(grid), grid.shape) ax.add_patch(plt.Rectangle((best[1]-0.5, best[0]-0.5), 1, 1, fill=False, edgecolor='cyan', linewidth=2)) # Summary text ax = fig.add_subplot(3, 3, 9) ax.set_facecolor('#161b22') ax.axis('off') best_sharpe_idx = np.unravel_index(np.nanargmax(sharpe), sharpe.shape) ret_idx = np.unravel_index(np.nanargmax(returns), returns.shape) pf_c = np.where(np.isfinite(pf), pf, -1) pf_idx = np.unravel_index(np.nanargmax(pf_c), pf_c.shape) dd_s = np.where(returns > 0, drawdown, np.inf) dd_idx = np.unravel_index(np.nanargmin(dd_s), dd_s.shape) if not np.all(np.isinf(dd_s)) else np.unravel_index(np.nanargmin(drawdown), drawdown.shape) def _r(label, idx): return f"{label:<12} {X_VALUES[idx[1]]:>6.2f} {Y_VALUES[idx[0]]:>6.2f} {sharpe[idx]:>6.3f} {returns[idx]:>7.1f}% {drawdown[idx]:>6.1f}% {pf[idx]:>5.2f}" summary = f"""BEST PARAMETERS BY METRIC ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ {'Target':<12} {X_PARAM[:6]:>6} {Y_PARAM[:6]:>6} {'Sharpe':>6} {'Return':>8} {'MaxDD':>7} {'PF':>5} {_r('Sharpe', best_sharpe_idx)} {_r('Return', ret_idx)} {_r('PF', pf_idx)} {_r('Min DD', dd_idx)} Grid: {len(X_VALUES)}x{len(Y_VALUES)} = {len(X_VALUES)*len(Y_VALUES)} runs Direction: {DIRECTION} | Spread: {SPREAD_BPS} BPS Fixed: {FIXED_PARAMS}""" if HMM_ENABLED: summary += f"\n\nHMM: {HMM_N_REGIMES} regimes, allowed={HMM_ALLOWED_REGIMES}" if oos_metrics: summary += f"\n\nPure OOS ({OOS_PCT}% held out):" summary += f"\n Sharpe: {oos_metrics['sharpe']:.3f} Return: {oos_metrics['total_return']:.2f}%" summary += f"\n Max DD: {oos_metrics['max_drawdown']:.2f}% Trades: {oos_metrics['trades']}" if scorecard: sc = scorecard summary += f"\n\nRobustness: {sc['robustness_score']}/100 — {sc['verdict'].split(' — ')[0]}" summary += f"\n Peak plateau: {sc['peak_plateau_cells']} cells ({sc['peak_plateau_pct']}%)" summary += f"\n Peak drop: {sc['peak_neighbour_drop']:.1f}%" summary += f"\n Zone: {sc['zone_coverage']}% param coverage, avg Sharpe {sc['zone_avg_sharpe']:.2f}" summary += f"\n Peak in zone: {'YES' if sc['peak_in_zone'] else 'NO'}" if sc.get('zone_region'): for p, b in sc['zone_region'].items(): summary += f"\n {p}: {b[0]:.4f} — {b[1]:.4f}" ax.text(0.05, 0.5, summary, transform=ax.transAxes, fontsize=8, color='#e6edf3', fontfamily='monospace', verticalalignment='center', bbox=dict(boxstyle='round,pad=0.5', facecolor='#21262d', edgecolor='#30363d')) plt.tight_layout(rect=[0, 0, 1, 0.95]) fname = f"{TICKER}_{METHOD}_grid_search.png" plt.savefig(fname, dpi=150, facecolor='#0d1117', bbox_inches='tight') print(f"Report saved: {fname}") plt.show() # Also save CSV csv_fname = f"{TICKER}_{METHOD}_grid_search.csv" rows = [] for yi, yv in enumerate(Y_VALUES): for xi, xv in enumerate(X_VALUES): rows.append({ X_PARAM: xv, Y_PARAM: yv, 'sharpe': sharpe[yi, xi], 'return_pct': returns[yi, xi], 'max_drawdown': drawdown[yi, xi], 'trades': int(trades[yi, xi]), 'profit_factor': pf[yi, xi] }) pd.DataFrame(rows).to_csv(csv_fname, index=False) print(f"CSV saved: {csv_fname}") def generate_interactive_report(sharpe, returns, drawdown, trades, pf, oos_metrics=None, scorecard=None): """Generate interactive HTML report with Plotly 3D surfaces — rotatable, zoomable.""" import json x_vals = [round(v, 4) for v in X_VALUES] y_vals = [round(v, 4) for v in Y_VALUES] x_label = X_PARAM.replace('_', ' ').title() y_label = Y_PARAM.replace('_', ' ').title() best_idx = np.unravel_index(np.nanargmax(sharpe), sharpe.shape) best_x = X_VALUES[best_idx[1]] best_y = Y_VALUES[best_idx[0]] ret_idx = np.unravel_index(np.nanargmax(returns), returns.shape) pf_c = np.where(np.isfinite(pf), pf, -1) pf_idx = np.unravel_index(np.nanargmax(pf_c), pf_c.shape) dd_s = np.where(returns > 0, drawdown, np.inf) dd_idx = np.unravel_index(np.nanargmin(dd_s), dd_s.shape) if not np.all(np.isinf(dd_s)) else np.unravel_index(np.nanargmin(drawdown), drawdown.shape) # Convert numpy arrays to JSON-safe nested lists def to_list(arr): return [[round(float(v), 4) if np.isfinite(v) else None for v in row] for row in arr] def _row(label, idx): return f"{label:<12} {X_VALUES[idx[1]]:>8.2f} {Y_VALUES[idx[0]]:>8.2f} {sharpe[idx]:>7.3f} {returns[idx]:>7.1f}% {drawdown[idx]:>6.1f}% {int(trades[idx]):>5} {pf[idx]:>5.2f}" # Build summary text — multi-metric table summary_best = ( f"{'Target':<12} {x_label[:8]:>8} {y_label[:8]:>8} {'Sharpe':>7} {'Return':>8} {'MaxDD':>7} {'Trds':>5} {'PF':>5}\n" f"{'-'*68}\n" f"{_row('Sharpe', best_idx)}\n" f"{_row('Return', ret_idx)}\n" f"{_row('PF', pf_idx)}\n" f"{_row('Min DD', dd_idx)}" ) summary_settings = ( f"Direction: {DIRECTION}\n" f"Sizing: {POSITION_SIZING} ({POSITION_SIZING_MODE})\n" f"Risk/Trade: {RISK_PER_TRADE_PCT*100:.0f}%\n" f"Spread: {SPREAD_BPS} BPS\n" f"SL: {STOP_LOSS_PCT} TP: {TAKE_PROFIT_PCT}\n" f"Max Hold: {MAX_HOLDING_DAYS}\n" f"Fixed: {FIXED_PARAMS}" ) if HMM_ENABLED: summary_settings += f"\nHMM: {HMM_N_REGIMES} regimes, allowed={HMM_ALLOWED_REGIMES}" if oos_metrics: summary_best += ( f"\n\nPure OOS Results ({OOS_PCT}% held out):\n" f"Sharpe: {oos_metrics['sharpe']:.3f}\n" f"Return: {oos_metrics['total_return']:.2f}%\n" f"Max DD: {oos_metrics['max_drawdown']:.2f}%\n" f"Trades: {oos_metrics['trades']}" ) grid_info = f"{len(X_VALUES)}x{len(Y_VALUES)} = {len(X_VALUES)*len(Y_VALUES)} combinations | Direction: {DIRECTION} | {datetime.now().strftime('%Y-%m-%d %H:%M')}" # HTML template with placeholder substitution (avoids f-string brace escaping) # Double braces needed because this is inside the outer code f-string template html_template = """ SC001STCB Grid Search

Sharpe Ratio
Total Return (%)
Max Drawdown (%)
Trade Count
Profit Factor

BEST PARAMETERS BY METRIC


        

        

    








"""

    # Inject all data as a single JSON object — avoids multiple placeholder replacements
    data_dict = {
        'x': x_vals, 'y': y_vals,
        'xl': x_label, 'yl': y_label,
        'title': f"{TICKER} ({METHOD})",
        'info': grid_info,
        'best': summary_best,
        'settings': summary_settings,
        'sharpe': to_list(sharpe),
        'ret': to_list(returns),
        'dd': to_list(drawdown),
        'trades': to_list(trades),
        'pf': to_list(pf),
    }
    if scorecard:
        data_dict['scorecard'] = scorecard
    # numpy types are not JSON-serializable; convert to native Python types
    def _json_safe(obj):
        if isinstance(obj, (np.integer,)):
            return int(obj)
        if isinstance(obj, (np.floating,)):
            return float(obj)
        if isinstance(obj, np.ndarray):
            return obj.tolist()
        raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
    data_obj = json.dumps(data_dict, default=_json_safe)
    html = html_template.replace('__JSON_DATA__', data_obj)

    html_fname = f"{TICKER}_{METHOD}_grid_search.html"
    with open(html_fname, 'w', encoding='utf-8') as f:
        f.write(html)
    print(f"Interactive report saved: {html_fname}")
    print(f"  Open in browser for rotatable 3D surfaces")


# ============================================================================
# MAIN
# ============================================================================
def save_walkforward_handoff(scorecard, oos_metrics):
    """Save JSON handoff file for Walk-Forward module import."""
    import json as _json

    # Build zone ranges from scorecard
    zone_ranges = scorecard.get('zone_region', {})
    peak_ranges = scorecard.get('peak_region', {})

    # Best params by metric (from grid)
    best_idx = np.unravel_index(np.nanargmax(sharpe_grid) if 'sharpe_grid' in dir() else 0,
                                 sharpe_grid.shape if 'sharpe_grid' in dir() else (1,1))

    handoff = {
        'version': '1.0',
        'generated': datetime.now().strftime('%Y-%m-%d %H:%M'),
        'ticker': TICKER,
        'ticker_b': getattr(sys.modules[__name__], 'TICKER_B', ''),
        'method': METHOD,
        'timeframe': TIMEFRAME,
        'duration_days': LOOKBACK_DAYS,
        'direction': DIRECTION,
        'initial_capital': INITIAL_CAPITAL,
        'spread_bps': SPREAD_BPS,
        'position_sizing': POSITION_SIZING,
        'fixed_params': FIXED_PARAMS,
        'sweep': {
            'x_param': X_PARAM,
            'y_param': Y_PARAM,
            'x_min': float(X_VALUES[0]),
            'x_max': float(X_VALUES[-1]),
            'y_min': float(Y_VALUES[0]),
            'y_max': float(Y_VALUES[-1]),
            'x_points': len(X_VALUES),
            'y_points': len(Y_VALUES)
        },
        'robustness': {
            'score': int(scorecard.get('robustness_score', 0)),
            'verdict': scorecard.get('verdict', ''),
            'zone_ranges': {k: [float(v[0]), float(v[1])] for k, v in zone_ranges.items()},
            'peak_ranges': {k: [float(v[0]), float(v[1])] for k, v in peak_ranges.items()},
            'zone_avg_sharpe': float(scorecard.get('zone_avg_sharpe', 0)),
            'zone_coverage': float(scorecard.get('zone_coverage', 0)),
            'peak_in_zone': bool(scorecard.get('peak_in_zone', False))
        },
        'oos': {
            'enabled': OOS_PCT > 0,
            'pct': OOS_PCT,
            'sharpe': float(oos_metrics.get('sharpe', 0)) if oos_metrics else None,
            'total_return': float(oos_metrics.get('total_return', 0)) if oos_metrics else None,
            'max_drawdown': float(oos_metrics.get('max_drawdown', 0)) if oos_metrics else None
        },
        'walkforward_recommended': (lambda: {
            # Use zone ranges if they actually narrow the space (< 90% coverage),
            # otherwise fall back to peak ranges which are always narrower
            'x_param': X_PARAM,
            'y_param': Y_PARAM,
            'x_min': float((peak_ranges if float(scorecard.get('zone_coverage', 100)) >= 90 else zone_ranges).get(X_PARAM, [X_VALUES[0], X_VALUES[-1]])[0]),
            'x_max': float((peak_ranges if float(scorecard.get('zone_coverage', 100)) >= 90 else zone_ranges).get(X_PARAM, [X_VALUES[0], X_VALUES[-1]])[1]),
            'y_min': float((peak_ranges if float(scorecard.get('zone_coverage', 100)) >= 90 else zone_ranges).get(Y_PARAM, [Y_VALUES[0], Y_VALUES[-1]])[0]),
            'y_max': float((peak_ranges if float(scorecard.get('zone_coverage', 100)) >= 90 else zone_ranges).get(Y_PARAM, [Y_VALUES[0], Y_VALUES[-1]])[1]),
            'source': 'peak_plateau' if float(scorecard.get('zone_coverage', 100)) >= 90 else 'zone'
        })()
    }

    fname = f'{TICKER}_{METHOD}_walkforward_config.json'
    with open(fname, 'w') as f:
        _json.dump(handoff, f, indent=2)
    print(f"\nWalk-forward handoff saved: {fname}")
    print(f"  Import this file in the Walk-Forward module on QuanterLab")
    return fname


if __name__ == '__main__':
    sharpe, returns, drawdown, trades, pf, oos_metrics = run_sweep()
    scorecard = compute_robustness_scorecard(sharpe, returns, drawdown, trades, pf)
    print_robustness_scorecard(scorecard)
    generate_report(sharpe, returns, drawdown, trades, pf, oos_metrics, scorecard)
    generate_interactive_report(sharpe, returns, drawdown, trades, pf, oos_metrics, scorecard)
    save_walkforward_handoff(scorecard, oos_metrics)