walk_forward_analysis
Optimizes moving average crossover strategies using walk-forward analysis to validate trading parameters across training and testing periods.
Instructions
Performs Walk Forward Analysis on MA Crossover. Optimizes (Fast, Slow) on Train, tests on Test.
Input 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 |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- tools/backtesting.py:172-245 (handler)Main handler function implementing walk-forward analysis for MA crossover strategy. Fetches data, creates rolling train/test windows, optimizes MA parameters on train data, evaluates on test data, and compiles results.
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)MCP tool registration of walk_forward_analysis (with run_backtest) in the 'Backtesting' category via register_tools function.
register_tools( [run_backtest, walk_forward_analysis], "Backtesting" ) - app.py:289-291 (registration)Inclusion of walk_forward_analysis in the Gradio UI toolbox under 'Backtesting' category (supports MCP server mode).
"Risk Management": [portfolio_risk, var, max_drawdown, monte_carlo_simulation], "Backtesting": [run_backtest, walk_forward_analysis], "Technical Analysis": [compute_indicators, rolling_stats, get_technical_summary], - tools/backtesting.py:172-176 (schema)Type hints and docstring defining input parameters (symbol, dates, train/test months) and str output for the tool.
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. """