import pandas as pd import numpy as np from datetime import datetime, timedelta from typing import Dict, List, Tuple, Optional import requests from dataclasses import dataclass import matplotlib.pyplot as plt from scipy import stats from itertools import product import warnings warnings.filterwarnings('ignore') @dataclass class StrategyConfig: """Configuration for EURUSD TSMOM Strategy""" api_key: str = "YOUR_FMP_API_KEY" symbol: str = "EURUSD" benchmark: str = "SPY" start_date: str = "2019-01-01" end_date: str = "2025-12-31" initial_capital: float = 100000.0 vol_target: float = 0.15 vol_lookback: int = 63 tsmom_lookbacks: List[int] = None sma_fast_periods: List[int] = None sma_slow_periods: List[int] = None vol_estimators: List[str] = None commission_pct: float = 0.00002 slippage_pct: float = 0.00005 initial_train_days: int = 504 validation_days: int = 126 test_days: int = 63 refit_frequency_days: int = 63 feature_warmup_days: int = 252 max_leverage: float = 2.0 def __post_init__(self): if self.tsmom_lookbacks is None: self.tsmom_lookbacks = [21, 63, 126, 252] if self.sma_fast_periods is None: self.sma_fast_periods = [10, 20, 50] if self.sma_slow_periods is None: self.sma_slow_periods = [50, 100, 200] if self.vol_estimators is None: self.vol_estimators = ['close_to_close', 'parkinson', 'garman_klass', 'rogers_satchell', 'yang_zhang'] class FMPDataFetcher: """Fetch forex data from Financial Modeling Prep API""" def __init__(self, api_key: str): self.api_key = api_key self.base_url = "https://financialmodelingprep.com/api/v3" def fetch_forex_data(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame: """Fetch forex historical data""" url = f"{self.base_url}/historical-price-full/{symbol}" params = { 'from': start_date, 'to': end_date, 'apikey': self.api_key } response = requests.get(url, params=params) if response.status_code != 200: raise ValueError(f"API request failed: {response.status_code}") data = response.json() if 'historical' not in data: raise ValueError(f"No data returned for {symbol}") df = pd.DataFrame(data['historical']) df['date'] = pd.to_datetime(df['date']) df = df.sort_values('date').reset_index(drop=True) df = df.rename(columns={ 'open': 'Open', 'high': 'High', 'low': 'Low', 'close': 'Close', 'adjClose': 'Adj Close', 'volume': 'Volume' }) if 'Adj Close' not in df.columns: df['Adj Close'] = df['Close'] df = df.set_index('date') return df[['Open', 'High', 'Low', 'Close', 'Adj Close']] class VolatilityEstimators: """Multiple volatility estimation methods""" @staticmethod def close_to_close(df: pd.DataFrame, window: int) -> pd.Series: """Standard close-to-close volatility""" returns = np.log(df['Adj Close'] / df['Adj Close'].shift(1)) vol = returns.rolling(window).std() * np.sqrt(252) return vol.shift(1) @staticmethod def parkinson(df: pd.DataFrame, window: int) -> pd.Series: """Parkinson volatility using high-low range""" hl = np.log(df['High'] / df['Low']) var = (1 / (4 * np.log(2))) * (hl ** 2) vol = np.sqrt(var.rolling(window).mean() * 252) return vol.shift(1) @staticmethod def garman_klass(df: pd.DataFrame, window: int) -> pd.Series: """Garman-Klass volatility""" hl = np.log(df['High'] / df['Low']) co = np.log(df['Adj Close'] / df['Open']) var = 0.5 * (hl ** 2) - (2 * np.log(2) - 1) * (co ** 2) vol = np.sqrt(var.rolling(window).mean() * 252) return vol.shift(1) @staticmethod def rogers_satchell(df: pd.DataFrame, window: int) -> pd.Series: """Rogers-Satchell volatility""" hc = np.log(df['High'] / df['Adj Close']) ho = np.log(df['High'] / df['Open']) lc = np.log(df['Low'] / df['Adj Close']) lo = np.log(df['Low'] / df['Open']) var = hc * ho + lc * lo vol = np.sqrt(var.rolling(window).mean() * 252) return vol.shift(1) @staticmethod def yang_zhang(df: pd.DataFrame, window: int) -> pd.Series: """Yang-Zhang volatility""" oc = np.log(df['Open'] / df['Adj Close'].shift(1)) var_o = oc.rolling(window).var() co = np.log(df['Adj Close'] / df['Open']) var_c = co.rolling(window).var() hc = np.log(df['High'] / df['Adj Close']) ho = np.log(df['High'] / df['Open']) lc = np.log(df['Low'] / df['Adj Close']) lo = np.log(df['Low'] / df['Open']) var_rs = (hc * ho + lc * lo).rolling(window).mean() k = 0.34 / (1.34 + (window + 1) / (window - 1)) vol = np.sqrt(var_o + k * var_c + (1 - k) * var_rs) * np.sqrt(252) return vol.shift(1) @classmethod def compute_all(cls, df: pd.DataFrame, window: int) -> Dict[str, pd.Series]: """Compute all volatility estimators""" return { 'close_to_close': cls.close_to_close(df, window), 'parkinson': cls.parkinson(df, window), 'garman_klass': cls.garman_klass(df, window), 'rogers_satchell': cls.rogers_satchell(df, window), 'yang_zhang': cls.yang_zhang(df, window) } class TSMOMStrategy: """Time Series Momentum Strategy with multiple volatility estimators""" def __init__(self, config: StrategyConfig): self.config = config self.data_fetcher = FMPDataFetcher(config.api_key) self.vol_estimators = VolatilityEstimators() def fetch_data(self) -> Tuple[pd.DataFrame, pd.DataFrame]: """Fetch forex and benchmark data with warmup period""" warmup_start = (pd.to_datetime(self.config.start_date) - timedelta(days=self.config.feature_warmup_days)).strftime('%Y-%m-%d') print(f"Fetching {self.config.symbol} data from {warmup_start} to {self.config.end_date}...") forex_data = self.data_fetcher.fetch_forex_data( self.config.symbol, warmup_start, self.config.end_date ) print(f"Fetching {self.config.benchmark} data...") benchmark_data = self.data_fetcher.fetch_forex_data( self.config.benchmark, warmup_start, self.config.end_date ) return forex_data, benchmark_data def compute_features(self, df: pd.DataFrame) -> pd.DataFrame: """Compute all features with strict T-1 availability""" features = df.copy() vol_dict = self.vol_estimators.compute_all(df, self.config.vol_lookback) for name, vol in vol_dict.items(): features[f'vol_{name}'] = vol for lookback in self.config.tsmom_lookbacks: mom = (df['Adj Close'] / df['Adj Close'].shift(lookback)) - 1 features[f'tsmom_{lookback}'] = mom.shift(1) for period in set(self.config.sma_fast_periods + self.config.sma_slow_periods): sma = df['Adj Close'].rolling(period).mean() features[f'sma_{period}'] = sma.shift(1) return features def generate_signals(self, features: pd.DataFrame, tsmom_lookback: int, sma_fast: Optional[int], sma_slow: Optional[int], vol_estimator: str) -> pd.DataFrame: """Generate trading signals for specific parameter combination""" signals = pd.DataFrame(index=features.index) tsmom_signal = np.sign(features[f'tsmom_{tsmom_lookback}']) if sma_fast is not None and sma_slow is not None: sma_signal = np.sign(features[f'sma_{sma_fast}'] - features[f'sma_{sma_slow}']) combined_signal = tsmom_signal * sma_signal else: combined_signal = tsmom_signal vol = features[f'vol_{vol_estimator}'] vol_scalar = self.config.vol_target / vol vol_scalar = vol_scalar.clip(0, self.config.max_leverage) signals['raw_signal'] = combined_signal signals['position'] = combined_signal * vol_scalar signals['volatility'] = vol return signals def backtest_strategy(self, df: pd.DataFrame, signals: pd.DataFrame, start_idx: int, end_idx: int) -> pd.DataFrame: """Backtest with corrected P&L calculation and trade counting""" df_bt = df.iloc[start_idx:end_idx].copy() signals_bt = signals.iloc[start_idx:end_idx].copy() results = pd.DataFrame(index=df_bt.index) results['target_position'] = signals_bt['position'] portfolio_value = self.config.initial_capital position_size = 0.0 equity_curve = [] positions_held = [] trade_log = [] # Minimum notional for trade counting MIN_TRADE_NOTIONAL = 100.0 for i, (date, row) in enumerate(df_bt.iterrows()): prev_position = position_size target_position = results.loc[date, 'target_position'] if np.isnan(target_position): target_position = 0.0 # STEP 1: Calculate gap P&L using OLD position if i > 0: prev_close = df_bt['Adj Close'].iloc[i-1] current_open = row['Open'] gap_return = (current_open / prev_close) - 1 gap_pnl = prev_position * prev_close * gap_return portfolio_value += gap_pnl # STEP 2: Check if we need to trade trade_occurred = False if i == 0 or not np.isclose(target_position, prev_position, atol=1e-8): # Size new position using updated portfolio value and Close[T-1] if i > 0: sizing_price = df_bt['Adj Close'].iloc[i-1] else: sizing_price = row['Open'] position_value = portfolio_value * target_position new_position_size = position_value / sizing_price if sizing_price > 0 else 0 # Calculate transaction costs execution_price = row['Open'] notional_trade = abs(new_position_size - prev_position) * execution_price # Only count as trade if above minimum threshold if notional_trade >= MIN_TRADE_NOTIONAL: commission = notional_trade * self.config.commission_pct slippage = notional_trade * self.config.slippage_pct total_costs = commission + slippage portfolio_value -= total_costs position_size = new_position_size trade_occurred = True trade_log.append({ 'date': date, 'prev_position': prev_position, 'new_position': position_size, 'execution_price': execution_price, 'costs': total_costs, 'gap_pnl': gap_pnl if i > 0 else 0, 'notional': notional_trade }) else: # Below threshold: update position without costs position_size = new_position_size # STEP 3: Calculate INTRADAY P&L (Open[T] → Close[T]) using current position if i > 0: current_close = row['Adj Close'] current_open = row['Open'] intraday_return = (current_close / current_open) - 1 intraday_pnl = position_size * current_open * intraday_return portfolio_value += intraday_pnl equity_curve.append(portfolio_value) positions_held.append(position_size) results['portfolio_value'] = equity_curve results['position_held'] = positions_held results['returns'] = results['portfolio_value'].pct_change() return results, trade_log def calculate_robust_metrics(self, returns: pd.Series, benchmark_returns: pd.Series, trades: List[Dict]) -> Dict: """Calculate comprehensive performance and robustness metrics""" returns = returns.dropna() benchmark_returns = benchmark_returns.dropna() common_idx = returns.index.intersection(benchmark_returns.index) returns = returns.loc[common_idx] benchmark_returns = benchmark_returns.loc[common_idx] total_return = (1 + returns).prod() - 1 ann_return = (1 + total_return) ** (252 / len(returns)) - 1 ann_vol = returns.std() * np.sqrt(252) sharpe = ann_return / ann_vol if ann_vol > 0 else 0 cumulative = (1 + returns).cumprod() running_max = cumulative.expanding().max() drawdown = (cumulative - running_max) / running_max max_dd = drawdown.min() calmar = ann_return / abs(max_dd) if max_dd != 0 else 0 downside_returns = returns[returns < 0] downside_std = downside_returns.std() * np.sqrt(252) sortino = ann_return / downside_std if downside_std > 0 else 0 win_rate = (returns > 0).sum() / len(returns) if len(returns) > 0 else 0 bench_total_return = (1 + benchmark_returns).prod() - 1 bench_ann_return = (1 + bench_total_return) ** (252 / len(benchmark_returns)) - 1 bench_ann_vol = benchmark_returns.std() * np.sqrt(252) bench_sharpe = bench_ann_return / bench_ann_vol if bench_ann_vol > 0 else 0 covariance = returns.cov(benchmark_returns) bench_variance = benchmark_returns.var() beta = covariance / bench_variance if bench_variance > 0 else 0 alpha = ann_return - (bench_ann_return * beta) n_obs = len(returns) skew = returns.skew() kurt = returns.kurtosis() dsr = sharpe * np.sqrt(1 - (skew * sharpe / np.sqrt(n_obs)) + ((kurt - 1) / 4) * (sharpe ** 2 / n_obs)) sr_std = np.sqrt((1 + 0.5 * sharpe ** 2) / n_obs) psr = stats.norm.cdf((sharpe - 0) / sr_std) return { 'total_return': total_return, 'annual_return': ann_return, 'annual_volatility': ann_vol, 'sharpe_ratio': sharpe, 'sortino_ratio': sortino, 'calmar_ratio': calmar, 'max_drawdown': max_dd, 'win_rate': win_rate, 'deflated_sharpe': dsr, 'probabilistic_sharpe': psr, 'beta': beta, 'alpha': alpha, 'benchmark_return': bench_ann_return, 'benchmark_sharpe': bench_sharpe, 'num_trades': len(trades), 'num_observations': n_obs } def walk_forward_optimization(self, df: pd.DataFrame, features: pd.DataFrame, benchmark_df: pd.DataFrame) -> Dict: """Walk-forward expanding window optimization - NO LOOK-AHEAD""" print("\n" + "="*80) print("WALK-FORWARD EXPANDING WINDOW OPTIMIZATION") print("="*80) valid_start = pd.to_datetime(self.config.start_date) df = df[df.index >= valid_start].copy() features = features[features.index >= valid_start].copy() benchmark_df = benchmark_df[benchmark_df.index >= valid_start].copy() param_combinations = [] for tsmom_lb in self.config.tsmom_lookbacks: for vol_est in self.config.vol_estimators: param_combinations.append({ 'tsmom_lookback': tsmom_lb, 'sma_fast': None, 'sma_slow': None, 'vol_estimator': vol_est, 'name': f'TSMOM{tsmom_lb}_{vol_est}' }) for sma_fast, sma_slow in product(self.config.sma_fast_periods, self.config.sma_slow_periods): if sma_fast < sma_slow: for vol_est in self.config.vol_estimators: param_combinations.append({ 'tsmom_lookback': tsmom_lb, 'sma_fast': sma_fast, 'sma_slow': sma_slow, 'vol_estimator': vol_est, 'name': f'TSMOM{tsmom_lb}_SMA{sma_fast}x{sma_slow}_{vol_est}' }) print(f"\nTotal parameter combinations to test: {len(param_combinations)}") wf_results = [] train_start_idx = 0 while train_start_idx + self.config.initial_train_days < len(df): train_end_idx = train_start_idx + self.config.initial_train_days val_end_idx = min(train_end_idx + self.config.validation_days, len(df)) test_end_idx = min(val_end_idx + self.config.test_days, len(df)) if val_end_idx >= len(df): break train_dates = (df.index[train_start_idx], df.index[train_end_idx-1]) val_dates = (df.index[train_end_idx], df.index[val_end_idx-1]) print(f"\n{'='*80}") print(f"Walk-Forward Block: Train {train_dates[0].date()} to {train_dates[1].date()}") print(f" Validate {val_dates[0].date()} to {val_dates[1].date()}") if test_end_idx < len(df): test_dates = (df.index[val_end_idx], df.index[test_end_idx-1]) print(f" Test {test_dates[0].date()} to {test_dates[1].date()}") print(f"{'='*80}") # CRITICAL FIX: Only use features up to train_end for signal generation features_train = features.iloc[:train_end_idx].copy() best_sharpe = -np.inf best_params = None for params in param_combinations: # Generate signals using ONLY training period features signals_train = self.generate_signals( features_train, params['tsmom_lookback'], params['sma_fast'], params['sma_slow'], params['vol_estimator'] ) results, _ = self.backtest_strategy(df, signals_train, train_start_idx, train_end_idx) sharpe = (results['returns'].mean() / results['returns'].std() * np.sqrt(252) if results['returns'].std() > 0 else -np.inf) if sharpe > best_sharpe: best_sharpe = sharpe best_params = params print(f"\nBest parameters on training: {best_params['name']}") print(f"Training Sharpe: {best_sharpe:.3f}") # CRITICAL FIX: Generate validation signals using features up to val_end features_val = features.iloc[:val_end_idx].copy() signals_val = self.generate_signals( features_val, best_params['tsmom_lookback'], best_params['sma_fast'], best_params['sma_slow'], best_params['vol_estimator'] ) val_results, val_trades = self.backtest_strategy(df, signals_val, train_end_idx, val_end_idx) val_sharpe = (val_results['returns'].mean() / val_results['returns'].std() * np.sqrt(252) if val_results['returns'].std() > 0 else 0) bench_rets = benchmark_df['Adj Close'].iloc[train_end_idx:val_end_idx].pct_change() val_metrics = self.calculate_robust_metrics(val_results['returns'], bench_rets, val_trades) print(f"Validation Sharpe: {val_sharpe:.3f}") print(f"Validation Return: {val_metrics['annual_return']:.2%}") print(f"Walk-Forward Efficiency: {(val_sharpe / best_sharpe * 100):.1f}%") wf_results.append({ 'train_period': train_dates, 'val_period': val_dates, 'best_params': best_params, 'train_sharpe': best_sharpe, 'val_sharpe': val_sharpe, 'val_metrics': val_metrics, 'wfe': val_sharpe / best_sharpe if best_sharpe > 0 else 0 }) train_start_idx += self.config.refit_frequency_days return { 'wf_results': wf_results, 'param_combinations': param_combinations } def run_full_backtest(self, df: pd.DataFrame, features: pd.DataFrame, benchmark_df: pd.DataFrame, best_params: Dict) -> Tuple[pd.DataFrame, List, Dict]: """Run full backtest with best parameters""" print(f"\n{'='*80}") print("FULL PERIOD BACKTEST WITH BEST PARAMETERS") print(f"{'='*80}") print(f"Strategy: {best_params['name']}") signals = self.generate_signals( features, best_params['tsmom_lookback'], best_params['sma_fast'], best_params['sma_slow'], best_params['vol_estimator'] ) valid_start = pd.to_datetime(self.config.start_date) start_idx = df.index.get_loc(df[df.index >= valid_start].index[0]) end_idx = len(df) results, trades = self.backtest_strategy(df, signals, start_idx, end_idx) bench_rets = benchmark_df['Adj Close'].pct_change() metrics = self.calculate_robust_metrics(results['returns'], bench_rets, trades) return results, trades, metrics def plot_results(self, results: pd.DataFrame, benchmark_df: pd.DataFrame, metrics: Dict, wf_results: List[Dict]): """Comprehensive visualization""" fig = plt.figure(figsize=(16, 12)) ax1 = plt.subplot(3, 2, 1) strategy_equity = results['portfolio_value'] / self.config.initial_capital bench_equity = (1 + benchmark_df['Adj Close'].pct_change()).cumprod() bench_equity = bench_equity.reindex(strategy_equity.index, method='ffill') ax1.plot(strategy_equity.index, strategy_equity, label='Strategy', linewidth=2) ax1.plot(bench_equity.index, bench_equity, label=self.config.benchmark, linewidth=2, alpha=0.7) ax1.set_title('Equity Curve Comparison', fontsize=12, fontweight='bold') ax1.set_ylabel('Cumulative Return') ax1.legend() ax1.grid(True, alpha=0.3) ax2 = plt.subplot(3, 2, 2) cumulative = strategy_equity running_max = cumulative.expanding().max() drawdown = (cumulative - running_max) / running_max ax2.fill_between(drawdown.index, drawdown, 0, alpha=0.3, color='red') ax2.plot(drawdown.index, drawdown, color='red', linewidth=1) ax2.set_title('Strategy Drawdown', fontsize=12, fontweight='bold') ax2.set_ylabel('Drawdown') ax2.grid(True, alpha=0.3) ax2.yaxis.set_major_formatter(plt.FuncFormatter(lambda y, _: '{:.0%}'.format(y))) ax3 = plt.subplot(3, 2, 3) rolling_sharpe = (results['returns'].rolling(126).mean() / results['returns'].rolling(126).std() * np.sqrt(252)) ax3.plot(rolling_sharpe.index, rolling_sharpe, linewidth=2) ax3.axhline(y=0, color='black', linestyle='--', alpha=0.3) ax3.set_title('Rolling 6M Sharpe Ratio', fontsize=12, fontweight='bold') ax3.set_ylabel('Sharpe Ratio') ax3.grid(True, alpha=0.3) ax4 = plt.subplot(3, 2, 4) ax4.plot(results.index, results['position_held'], linewidth=1) ax4.axhline(y=0, color='black', linestyle='--', alpha=0.3) ax4.set_title('Position Exposure', fontsize=12, fontweight='bold') ax4.set_ylabel('Position Size') ax4.grid(True, alpha=0.3) ax5 = plt.subplot(3, 2, 5) ax5.hist(results['returns'].dropna(), bins=50, alpha=0.7, edgecolor='black') ax5.axvline(x=0, color='red', linestyle='--', linewidth=2) ax5.set_title('Daily Returns Distribution', fontsize=12, fontweight='bold') ax5.set_xlabel('Daily Return') ax5.set_ylabel('Frequency') ax5.grid(True, alpha=0.3) ax6 = plt.subplot(3, 2, 6) if wf_results: blocks = [f"Block {i+1}" for i in range(len(wf_results))] wfe_values = [r['wfe'] for r in wf_results] colors = ['green' if w >= 0.5 else 'orange' if w >= 0.3 else 'red' for w in wfe_values] ax6.bar(blocks, wfe_values, color=colors, alpha=0.7, edgecolor='black') ax6.axhline(y=1.0, color='black', linestyle='--', linewidth=2, label='Perfect WFE=1.0') ax6.axhline(y=0.5, color='orange', linestyle=':', linewidth=1.5, label='Acceptable WFE=0.5') ax6.set_title('Walk-Forward Efficiency by Block', fontsize=12, fontweight='bold') ax6.set_ylabel('WFE Ratio') ax6.legend() ax6.grid(True, alpha=0.3, axis='y') plt.setp(ax6.xaxis.get_majorticklabels(), rotation=45, ha='right') plt.tight_layout() plt.show() print(f"\n{'='*80}") print("PERFORMANCE SUMMARY") print(f"{'='*80}") print(f"Total Return: {metrics['total_return']:>10.2%}") print(f"Annual Return: {metrics['annual_return']:>10.2%}") print(f"Annual Volatility: {metrics['annual_volatility']:>10.2%}") print(f"Sharpe Ratio: {metrics['sharpe_ratio']:>10.3f}") print(f"Sortino Ratio: {metrics['sortino_ratio']:>10.3f}") print(f"Calmar Ratio: {metrics['calmar_ratio']:>10.3f}") print(f"Max Drawdown: {metrics['max_drawdown']:>10.2%}") print(f"Win Rate: {metrics['win_rate']:>10.2%}") print(f"\nROBUSTNESS METRICS") print(f"Deflated Sharpe: {metrics['deflated_sharpe']:>10.3f}") print(f"Probabilistic Sharpe: {metrics['probabilistic_sharpe']:>10.3f}") print(f"\nBENCHMARK COMPARISON ({self.config.benchmark})") print(f"Strategy Return: {metrics['annual_return']:>10.2%}") print(f"Benchmark Return: {metrics['benchmark_return']:>10.2%}") print(f"Alpha: {metrics['alpha']:>10.2%}") print(f"Beta: {metrics['beta']:>10.3f}") print(f"Strategy Sharpe: {metrics['sharpe_ratio']:>10.3f}") print(f"Benchmark Sharpe: {metrics['benchmark_sharpe']:>10.3f}") print(f"\nTRADING ACTIVITY") print(f"Number of Trades: {metrics['num_trades']:>10.0f}") print(f"Observations: {metrics['num_observations']:>10.0f}") print(f"{'='*80}") def main(): """Main execution function""" config = StrategyConfig( symbol="EURUSD", benchmark="SPY", start_date="2020-01-01", end_date="2024-12-31", initial_capital=100000.0, vol_target=0.15, vol_lookback=63, tsmom_lookbacks=[21, 63, 126], sma_fast_periods=[10, 20, 50], sma_slow_periods=[50, 100, 200], vol_estimators=['close_to_close', 'parkinson', 'garman_klass', 'rogers_satchell', 'yang_zhang'], commission_pct=0.00002, slippage_pct=0.00005, initial_train_days=504, validation_days=126, test_days=63, refit_frequency_days=63, feature_warmup_days=252, max_leverage=2.0 ) strategy = TSMOMStrategy(config) forex_data, benchmark_data = strategy.fetch_data() print("\nComputing features...") features = strategy.compute_features(forex_data) valid_start = pd.to_datetime(config.start_date) features_valid = features[features.index >= valid_start] missing_pct = features_valid.isna().sum() / len(features_valid) * 100 print("\nFeature completeness (after warmup):") for col in features_valid.columns: if missing_pct[col] > 0: print(f" {col}: {100-missing_pct[col]:.1f}% complete") wf_output = strategy.walk_forward_optimization(forex_data, features, benchmark_data) if wf_output['wf_results']: avg_params_performance = {} for result in wf_output['wf_results']: param_name = result['best_params']['name'] if param_name not in avg_params_performance: avg_params_performance[param_name] = [] avg_params_performance[param_name].append(result['val_sharpe']) best_param_name = max(avg_params_performance, key=lambda k: np.mean(avg_params_performance[k])) best_params = next(p for p in wf_output['param_combinations'] if p['name'] == best_param_name) print(f"\n{'='*80}") print(f"BEST OVERALL PARAMETERS: {best_param_name}") print(f"Average Validation Sharpe: {np.mean(avg_params_performance[best_param_name]):.3f}") print(f"{'='*80}") else: best_params = wf_output['param_combinations'][0] results, trades, metrics = strategy.run_full_backtest( forex_data, features, benchmark_data, best_params ) strategy.plot_results(results, benchmark_data, metrics, wf_output['wf_results']) return strategy, results, trades, metrics, wf_output if __name__ == "__main__": strategy, results, trades, metrics, wf_output = main()

Time Series Momentum on EUR/USD: Advanced TSMOM with Multi-Estimator Volatility Targeting

Subscribe to our premium content to read the rest.

Keep Reading

AlgoEdge Insights