walk_forward_analysis
Optimizes moving average crossover strategies by training parameters on historical data and testing performance on subsequent periods to validate trading models.
Instructions
Performs Walk Forward Analysis on MA Crossover. Optimizes (Fast, Slow) on Train, tests on Test.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| symbol | Yes | ||
| start_date | No | 2020-01-01 | |
| end_date | No | 2023-12-31 | |
| train_months | No | ||
| test_months | No |
Implementation Reference
- tools/backtesting.py:172-245 (handler)The core handler function implementing walk-forward analysis: fetches data, creates rolling train/test windows, optimizes MA crossover parameters on train data, applies to test periods, and summarizes out-of-sample performance.def walk_forward_analysis(symbol: str, start_date: str = "2020-01-01", end_date: str = "2023-12-31", train_months: int = 12, test_months: int = 3) -> str: """ Performs Walk Forward Analysis on MA Crossover. Optimizes (Fast, Slow) on Train, tests on Test. """ df = _fetch_data(symbol, start_date, end_date) if df.empty: return "No data found." close = df['Close'] if isinstance(close, pd.DataFrame): close = close.iloc[:, 0] # Generate windows # Simplified: Iterate by index assuming daily data # 21 days/month train_len = train_months * 21 test_len = test_months * 21 step = test_len results = [] # Parameter Grid fast_params = [10, 20, 50] slow_params = [50, 100, 200] current_idx = 0 while current_idx + train_len + test_len < len(df): train_data = close.iloc[current_idx : current_idx + train_len] test_data = close.iloc[current_idx + train_len : current_idx + train_len + test_len] # Optimize on Train best_perf = -np.inf best_params = (0, 0) for f in fast_params: for s in slow_params: if f >= s: continue # Vectorized backtest on train fast_ma = train_data.rolling(window=f).mean() slow_ma = train_data.rolling(window=s).mean() signal = (fast_ma > slow_ma).astype(int).shift(1) ret = train_data.pct_change() * signal perf = ret.sum() # Simple sum of returns if perf > best_perf: best_perf = perf best_params = (f, s) # Test on Test Data f, s = best_params fast_ma_test = test_data.rolling(window=f).mean() slow_ma_test = test_data.rolling(window=s).mean() signal_test = (fast_ma_test > slow_ma_test).astype(int).shift(1) ret_test = test_data.pct_change() * signal_test test_perf = ret_test.sum() results.append({ "Period": f"{test_data.index[0].date()} to {test_data.index[-1].date()}", "Best Params": best_params, "Test Return": test_perf }) current_idx += step # Format Output output = ["Walk Forward Analysis Results:"] total_ret = 0 for r in results: output.append(f"[{r['Period']}] Params: {r['Best Params']}, Return: {r['Test Return']:.2%}") total_ret += r['Test Return'] output.append(f"Total Walk Forward Return: {total_ret:.2%}") return "\n".join(output)
- server.py:384-388 (registration)Registers walk_forward_analysis as an MCP tool in the 'Backtesting' category using the register_tools helper.register_tools( [run_backtest, walk_forward_analysis], "Backtesting" )
- app.py:290-290 (registration)Includes walk_forward_analysis in the tools_map dictionary under 'Backtesting' category for Gradio UI and MCP server integration."Backtesting": [run_backtest, walk_forward_analysis],