MonteWalk

Instructions

Implementation Reference

• tools/backtesting.py:85-170 (handler)

  Core handler implementing the MA crossover backtest logic, fetching data from yfinance or CoinGecko, computing signals, returns, Sharpe, drawdown, and optionally visualizing equity curve.
  def run_backtest(symbol: str, fast_ma: int, slow_ma: int, start_date: str = "2020-01-01", end_date: str = "2023-12-31", visualize: bool = False) -> str: """ Backtests a Moving Average Crossover strategy. Args: symbol: Ticker symbol fast_ma: Fast moving average period slow_ma: Slow moving average period start_date: Backtest start date end_date: Backtest end date visualize: If True, returns equity curve chart """ try: logger.info(f"Starting backtest for {symbol} (Fast: {fast_ma}, Slow: {slow_ma}) from {start_date} to {end_date}") # Use yfinance directly instead of get_price df = _fetch_data(symbol, start_date, end_date) if df.empty: logger.warning(f"Backtest failed: No data for {symbol}") return f"No data found for {symbol}" # Strategy Logic df['Fast_MA'] = ta.sma(df['Close'], length=fast_ma) df['Slow_MA'] = ta.sma(df['Close'], length=slow_ma) # Signals df['Signal'] = 0 df.loc[df['Fast_MA'] > df['Slow_MA'], 'Signal'] = 1 df['Position'] = df['Signal'].diff() # Calculate Returns df['Market_Return'] = df['Close'].pct_change() df['Strategy_Return'] = df['Market_Return'] * df['Signal'].shift(1) # Equity Curves df['Strategy_Equity'] = (1 + df['Strategy_Return']).cumprod() df['BuyHold_Equity'] = (1 + df['Market_Return']).cumprod() # Metrics total_return = (1 + df['Strategy_Return']).prod() - 1 buy_hold_return = (1 + df['Market_Return']).prod() - 1 # Max Drawdown cum_ret = (1 + df['Strategy_Return']).cumprod() peak = cum_ret.expanding(min_periods=1).max() dd = (cum_ret / peak) - 1 max_dd = dd.min() # Sharpe Ratio (assuming 0% risk free for simplicity or use config) sharpe = df['Strategy_Return'].mean() / df['Strategy_Return'].std() * np.sqrt(252) result = ( f"Backtest Results for {symbol} ({start_date} to {end_date}):\n" f"Strategy: MA Crossover ({fast_ma}/{slow_ma})\n" f"------------------------------------------------\n" f"Total Return: {total_return:.2%}\n" f"Buy & Hold Return: {buy_hold_return:.2%}\n" f"Sharpe Ratio: {sharpe:.2f}\n" f"Max Drawdown: {max_dd:.2%}" ) if visualize: try: from tools.visualizer import plot_line df_clean = df.dropna(subset=['Strategy_Equity', 'BuyHold_Equity']) chart = plot_line( { 'x': list(range(len(df_clean))), 'y': [df_clean['Strategy_Equity'].tolist(), df_clean['BuyHold_Equity'].tolist()] }, x_label="Days", y_label="Equity (Initial = $1)", title=f"Backtest: {symbol} MA({fast_ma}/{slow_ma})", labels=["Strategy", "Buy & Hold"] ) result += f"\n\n{chart}" except Exception as e: logger.error(f"Error generating backtest chart: {e}") result += f"\n(Visualization error: {str(e)})" logger.info(f"Backtest completed for {symbol}. Return: {total_return:.2%}") return result except Exception as e: logger.error(f"Backtest failed for {symbol}: {e}", exc_info=True) return f"Error running backtest: {str(e)}"
• server.py:385-388 (registration)

  MCP tool registration of run_backtest in the server.py MCP server under the 'Backtesting' category.
  register_tools( [run_backtest, walk_forward_analysis], "Backtesting" )
• tools/backtesting.py:73-84 (helper)

  Key helper function that fetches historical price data, supporting both stocks (yfinance) and cryptocurrencies (CoinGecko API).
  def _fetch_data(symbol: str, start: str, end: str): """Fetch data from either yfinance (stocks) or CoinGecko (crypto).""" # Try to find CoinGecko ID coin_id = _get_coingecko_id(symbol) if coin_id: logger.info(f"Recognized {symbol} as crypto (CoinGecko ID: {coin_id})") return _fetch_crypto_data(coin_id, start, end) else: # Fall back to stock data logger.info(f"Treating {symbol} as stock symbol") return yf.download(symbol, start=start, end=end, progress=False)
• app.py:290-290 (registration)

  Tool grouping in app.py Gradio UI toolbox under 'Backtesting' category.
  "Backtesting": [run_backtest, walk_forward_analysis],
• tools/backtesting.py:16-47 (helper)

  Helper to resolve cryptocurrency symbols to CoinGecko IDs, using cached common mappings and fallback search.
  def _get_coingecko_id(symbol: str) -> Optional[str]: """ Dynamically search for CoinGecko ID using the search API. Falls back to common mappings for performance. """ symbol = symbol.upper() # Try common mappings first (fast) common_map = { 'BTC': 'bitcoin', 'ETH': 'ethereum', 'SHIB': 'shiba-inu', 'SOL': 'solana', 'XRP': 'ripple', 'ADA': 'cardano', 'DOGE': 'dogecoin', 'USDC': 'usd-coin', 'USDT': 'tether', } if symbol in common_map: return common_map[symbol] # Dynamic search for other symbols try: results = cg.search(query=symbol) if results.get('coins'): return results['coins'][0]['id'] except Exception as e: logger.debug(f"CoinGecko search failed for {symbol}: {e}") return None