""" Validate Strategy Performance This script runs a single backtest of a specific strategy configuration on a given dataset. It is useful for verifying the performance of a strategy with specific parameters (often found via optimization) and visualizing the trade list. Usage: python strategies/validate_strategy.py --data --strategy_file [--show-trades] Arguments: --data: Path to the OHLCV CSV data file. --strategy_file: Name of the strategy Python file in the strategies/ folder (e.g., strategy_rsi_gaussian.py). --show-trades: If present, prints the list of executed trades to the console. """ import pandas as pd import os import numpy as np import argparse import importlib.util import sys # Add project root to sys.path to allow imports from 'strategies' package sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from config import SCORE_START, TRAIN_END, TRAIN_START def main(): parser = argparse.ArgumentParser(description="Validate Strategy Performance") parser.add_argument("--show-trades", action="store_true", help="Show the list of trades") parser.add_argument("--data", type=str, default='results/TV_Export.csv', help="Path to the data CSV file") parser.add_argument("--strategy_file", type=str, default="strategy_activation_scores.py", help="Name of the strategy Python file (must be in strategies/ folder)") parser.add_argument("--params_file", type=str, help="Path to a CSV or JSON file with parameters to override defaults") parser.add_argument("--export", type=str, help="Path to export the signals CSV (e.g. strategies/DEBUG.csv)") parser.add_argument("--canonical-metrics", action=argparse.BooleanOptionalAction, default=True, help="Also print strategy.calculate_metrics output, matching optimizer scoring (default: true)") parser.add_argument("--score-start", type=str, default=SCORE_START, help=f"Start date for canonical metrics scoring window (default: {SCORE_START})") parser.add_argument("--regime-suppress", choices=["bull", "bear", "none"], default="none", help="Suppress entries outside the given regime. 'bull' = only enter when zscore >= threshold; " "'bear' = only enter when zscore < threshold; 'none' = no suppression (default)") parser.add_argument("--regime-threshold", type=float, default=15.0, help="MVRV zscore percentile threshold for regime classification (default: 15.0). " "zscore >= threshold = bull; zscore < threshold = bear") args = parser.parse_args() # Path to the CSV file provided in context csv_path = args.data if not os.path.exists(csv_path): print(f"Error: File not found at {csv_path}") return # Dynamic Strategy Import strategy_path = os.path.join(os.path.dirname(__file__), args.strategy_file) if not os.path.exists(strategy_path): print(f"Error: Strategy file not found at {strategy_path}") return spec = importlib.util.spec_from_file_location("strategy_module", strategy_path) strategy_signals = importlib.util.module_from_spec(spec) sys.modules["strategy_module"] = strategy_signals spec.loader.exec_module(strategy_signals) print(f"Loading data from {csv_path}...") df = pd.read_csv(csv_path) # Ensure column names are lower case and stripped of whitespace df.columns = df.columns.str.lower().str.strip() df['time'] = pd.to_datetime(df['time'], utc=True).dt.tz_localize(None) # Pre-filter to the backtesting date range BEFORE generating signals so that # crossunder state does not carry over from warm-up bars outside this window. start_date = pd.Timestamp(TRAIN_START) end_date = pd.Timestamp(TRAIN_END) df = df.loc[(df['time'] >= start_date) & (df['time'] <= end_date)].copy().reset_index(drop=True) # Load parameters if provided strategy_params = {} if args.params_file and os.path.exists(args.params_file): import json print(f"Loading parameters from {args.params_file}...") if args.params_file.endswith('.csv'): # Assume it's an optimization result, take the first row (winner) df_params = pd.read_csv(args.params_file) strategy_params = df_params.iloc[0].to_dict() print(f"Loaded parameters: {strategy_params}") elif args.params_file.endswith('.json'): with open(args.params_file) as f: strategy_params = json.load(f) print(f"Loaded parameters: {strategy_params}") signal_params = dict(strategy_params) if args.strategy_file == "strategy_mlp_scores.py": signal_params["_pine_time_start"] = args.score_start signal_params["_pine_time_end"] = TRAIN_END print("Generating signals...") signals = strategy_signals.generate_signals(df, **signal_params) # Runtime regime suppression: suppress entries outside the designated regime. # Uses the 'zscore' column (0-100 percentile MVRV scale, pre-computed in the data CSV). # bull = zscore >= threshold (default 15); bear = zscore < threshold. if args.regime_suppress != "none": zscore_col = "zscore" if zscore_col not in signals.columns: print(f"[WARNING] --regime-suppress requested but '{zscore_col}' column not found in data; suppression skipped.") else: thr = args.regime_threshold if args.regime_suppress == "bull": # Only allow entries when in bull regime (zscore >= thr) bear_mask = signals[zscore_col] < thr suppressed = signals["execute_entry"] & bear_mask signals.loc[bear_mask, "execute_entry"] = False else: # bear # Only allow entries when in bear regime (zscore < thr) bull_mask = signals[zscore_col] >= thr suppressed = signals["execute_entry"] & bull_mask signals.loc[bull_mask, "execute_entry"] = False n_suppressed = suppressed.sum() print(f"[regime-suppress={args.regime_suppress}] threshold={thr} suppressed {n_suppressed} entry signals") canonical_metrics = None canonical_error = None if args.canonical_metrics: if hasattr(strategy_signals, "calculate_metrics"): try: canonical_metrics = strategy_signals.calculate_metrics(signals.copy(), score_start=args.score_start) except Exception as exc: canonical_error = exc else: canonical_error = RuntimeError( f"{args.strategy_file} does not expose calculate_metrics()" ) # Create a display dataframe for the events display_events = signals[signals['execute_entry'] | signals['execute_exit']].copy() display_events['signal_type'] = '' display_events.loc[display_events['execute_entry'], 'signal_type'] = 'ENTRY LONG' display_events.loc[display_events['execute_exit'], 'signal_type'] = 'EXIT LONG' if args.show_trades: print("\n--- First 50 Trades ---") # Dynamically select columns to display based on what's available cols_to_show = ['time', 'signal_type', 'close'] possible_cols = ['hband', 'filt', 'stoch_rsi', 'activation_score'] for col in possible_cols: if col in display_events.columns: cols_to_show.append(col) print(display_events[cols_to_show].head(50).to_string(index=False)) # --- Performance Metrics --- COMMISSION_RATE = 0.005 # 0.5% per side, matching TradingView default initial_capital = 100.0 equity = initial_capital peak_equity = initial_capital max_drawdown_pct = 0.0 trades_pnl = [] in_position = False entry_price = 0.0 total_trades = 0 profitable_trades = 0 daily_equity_curve = [] # MAE/MFE per-trade tracking trade_mae_list = [] # Max Adverse Excursion: worst close-to-entry drawdown per trade trade_mfe_list = [] # Max Favorable Excursion: best close-to-entry gain per trade current_mae = 0.0 current_mfe = 0.0 # Iterate through ALL bars to calculate drawdown correctly (Max Adverse Excursion) for index, row in signals.iterrows(): # 1. Handle Entry if row['execute_entry']: in_position = True entry_price = row['close'] total_trades += 1 current_mae = 0.0 current_mfe = 0.0 # We assume entry at close, so no drawdown exposure on the entry bar itself daily_equity_curve.append(equity) continue # 2. Handle Open Position (Drawdown Check) if in_position: # Calculate Floating Equity using the LOW of the bar to capture Max Drawdown # This simulates the worst-case equity during the holding period floating_pnl_pct = (row['low'] - entry_price) / entry_price floating_equity = equity * (1 + floating_pnl_pct) drawdown = (peak_equity - floating_equity) / peak_equity if drawdown > max_drawdown_pct: max_drawdown_pct = drawdown # Update MAE/MFE using close price (consistent with signal logic) close_ret = (row['close'] - entry_price) / entry_price if close_ret < current_mae: current_mae = close_ret if close_ret > current_mfe: current_mfe = close_ret # 3. Handle Exit if row['execute_exit']: in_position = False exit_price = row['close'] # Record MAE/MFE for this trade trade_mae_list.append(current_mae) trade_mfe_list.append(current_mfe) # Calculate Realized P&L including 0.5% commission per side effective_entry = entry_price * (1 + COMMISSION_RATE) effective_exit = exit_price * (1 - COMMISSION_RATE) pnl_pct = (effective_exit - effective_entry) / effective_entry trades_pnl.append(pnl_pct) if pnl_pct > 0: profitable_trades += 1 # Update Realized Equity equity *= (1 + pnl_pct) if equity > peak_equity: peak_equity = equity daily_equity_curve.append(equity) else: # Mark to market at close for daily equity curve floating_pnl_pct_close = (row['close'] - entry_price) / entry_price floating_equity_close = equity * (1 + floating_pnl_pct_close) daily_equity_curve.append(floating_equity_close) else: # Not in position, equity is flat daily_equity_curve.append(equity) total_pnl_pct = (equity - initial_capital) / initial_capital * 100 profitable_pct = (profitable_trades / total_trades * 100) if total_trades > 0 else 0.0 pct_in_market = signals['position'].mean() * 100 # --- Sharpe & Sortino Ratios --- equity_series = pd.Series(daily_equity_curve) returns = equity_series.pct_change().dropna() sharpe_ratio = 0.0 sortino_ratio = 0.0 if len(returns) > 0: mean_return = returns.mean() std_return = returns.std() downside_deviation = np.sqrt(np.mean(np.minimum(0, returns)**2)) if std_return != 0: sharpe_ratio = (mean_return / std_return) * np.sqrt(365) if downside_deviation != 0: sortino_ratio = (mean_return / downside_deviation) * np.sqrt(365) calmar_ratio = -10.0 n_years = len(df) / 365.25 total_return = total_pnl_pct / 100.0 if n_years > 0 and (1 + total_return) > 1e-9 and max_drawdown_pct > 0.001: annualized_return_pct = ((1 + total_return) ** (1.0 / n_years) - 1.0) * 100.0 calmar_ratio = annualized_return_pct / (max_drawdown_pct * 100.0) avg_mae = np.mean(trade_mae_list) * 100 if trade_mae_list else 0.0 avg_mfe = np.mean(trade_mfe_list) * 100 if trade_mfe_list else 0.0 worst_mae = min(trade_mae_list) * 100 if trade_mae_list else 0.0 best_mfe = max(trade_mfe_list) * 100 if trade_mfe_list else 0.0 print("\n" + "="*30) print("Trade-walk diagnostics") print(f"Data File: {csv_path}") print(f"Date Range: {start_date.date()} to {end_date.date()}") if args.regime_suppress != "none": print(f"Regime Filter: {args.regime_suppress} (zscore threshold={args.regime_threshold})") print(f"Total P&L: {total_pnl_pct:.2f}%") print(f"Max Equity Drawdown: {max_drawdown_pct * 100:.2f}%") print(f"Total Trades: {total_trades}") print(f"Profitable Trades: {profitable_pct:.2f}% ({profitable_trades}/{total_trades})") print(f"% Time In Market: {pct_in_market:.1f}%") print(f"Avg MAE: {avg_mae:.2f}% (worst: {worst_mae:.2f}%)") print(f"Avg MFE: {avg_mfe:.2f}% (best: {best_mfe:.2f}%)") print(f"Sharpe Ratio: {sharpe_ratio:.2f}") print(f"Sortino Ratio: {sortino_ratio:.2f}") print(f"Calmar Ratio: {calmar_ratio:.2f}") print("="*30) if args.canonical_metrics: print("\n" + "="*30) print("Canonical optimizer metrics") print(f"Scoring Window: {args.score_start} to {end_date.date()}") print("Source: strategy.calculate_metrics(signals, score_start=...)") if canonical_metrics is not None: print(f"Total P&L: {canonical_metrics.get('Total P&L %', 0.0):.2f}%") print(f"Max Equity Drawdown: {abs(canonical_metrics.get('Max Drawdown %', 0.0)):.2f}%") print(f"Total Trades: {int(canonical_metrics.get('Total Trades', 0))}") print(f"% Time In Market: {canonical_metrics.get('% In Market', 0.0):.1f}%") print(f"Sharpe Ratio: {canonical_metrics.get('Sharpe Ratio', 0.0):.2f}") print(f"Sortino Ratio: {canonical_metrics.get('Sortino Ratio', 0.0):.2f}") print(f"Calmar Ratio: {canonical_metrics.get('Calmar Ratio', 0.0):.2f}") print(f"P&L/DD Ratio: {canonical_metrics.get('P&L/DD Ratio', 0.0):.2f}") else: print(f"Unavailable: {canonical_error}") print("="*30) if args.export: signals.to_csv(args.export) print(f"Signals exported to {args.export}") if __name__ == "__main__": main()