""" VectorBT Provider - Slim Orchestrator Delegates to specialized modules: - vbt_strategies: 20+ strategy implementations - vbt_indicators: All VBT indicators + custom indicators - vbt_portfolio: Portfolio construction (from_signals, from_orders, stops, sizing) - vbt_metrics: Full metrics extraction (100+ metrics) - vbt_optimization: Vectorized parameter optimization + walk-forward """ import sys import json from typing import Dict, Any from pathlib import Path # Setup paths for both direct execution and package import _SCRIPT_DIR = Path(__file__).parent _BACKTESTING_DIR = _SCRIPT_DIR.parent # Add parent dirs to sys.path so absolute imports work when run directly if str(_BACKTESTING_DIR) not in sys.path: sys.path.insert(0, str(_BACKTESTING_DIR)) if str(_SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(_SCRIPT_DIR)) from base.base_provider import ( BacktestingProviderBase, BacktestResult, PerformanceMetrics, Trade, EquityPoint, BacktestStatistics, json_response, parse_json_input ) class VectorBTProvider(BacktestingProviderBase): """ VectorBT Provider - Ultra-fast vectorized backtesting. Thin orchestrator that delegates to specialized modules for strategies, portfolio construction, metrics, and optimization. """ def __init__(self): super().__init__() @property def name(self) -> str: return "VectorBT" @property def version(self) -> str: try: import vectorbt as vbt return vbt.__version__ except Exception: return "pure-numpy" @property def capabilities(self) -> Dict[str, Any]: return { 'backtesting': True, 'optimization': True, 'liveTrading': False, 'research': True, 'multiAsset': ['equity', 'crypto', 'forex', 'futures'], 'indicators': True, 'customStrategies': True, 'maxConcurrentBacktests': 50, 'vectorization': True, 'walkForward': True, 'randomBenchmarking': True, 'stopLoss': True, 'takeProfit': True, 'positionSizing': True, } def initialize(self, config: Dict[str, Any]) -> Dict[str, Any]: """Initialize VectorBT provider.""" self.config = config vbt = self._import_vbt() return self._create_success_result(f'VectorBT {vbt.__version__} ready') def test_connection(self) -> Dict[str, Any]: """Test VectorBT availability.""" vbt = self._import_vbt() return self._create_success_result(f'VectorBT {vbt.__version__} is available') def run_backtest(self, request: Dict[str, Any]) -> Dict[str, Any]: """Run backtest using VectorBT with full feature support.""" import pandas as pd import numpy as np import sys backtest_id = self._generate_id() logs = [] # === COMPREHENSIVE DEBUG LOGGING === print(f'[PY-BT] === run_backtest START ===', file=sys.stderr) print(f'[PY-BT] Request keys: {sorted(request.keys())}', file=sys.stderr) print(f'[PY-BT] strategy: {request.get("strategy")}', file=sys.stderr) print(f'[PY-BT] startDate: {request.get("startDate")}', file=sys.stderr) print(f'[PY-BT] endDate: {request.get("endDate")}', file=sys.stderr) print(f'[PY-BT] initialCapital: {request.get("initialCapital")}', file=sys.stderr) print(f'[PY-BT] assets: {request.get("assets")}', file=sys.stderr) print(f'[PY-BT] commission: {request.get("commission")}', file=sys.stderr) print(f'[PY-BT] slippage: {request.get("slippage")}', file=sys.stderr) print(f'[PY-BT] positionSizing: {request.get("positionSizing")}', file=sys.stderr) print(f'[PY-BT] positionSizeValue: {request.get("positionSizeValue")}', file=sys.stderr) print(f'[PY-BT] stopLoss: {request.get("stopLoss")}', file=sys.stderr) print(f'[PY-BT] takeProfit: {request.get("takeProfit")}', file=sys.stderr) print(f'[PY-BT] trailingStop: {request.get("trailingStop")}', file=sys.stderr) try: vbt = self._import_vbt() except Exception as e: print(f'[PY-BT] FAILED to import vbt: {e}', file=sys.stderr) return self._create_error_result(str(e)) try: import vbt_strategies as strat import vbt_portfolio as pf import vbt_metrics as metrics print(f'[PY-BT] Imports OK: strat, pf, metrics', file=sys.stderr) logs.append(f'{self._current_timestamp()}: Starting VectorBT backtest {backtest_id}') # --- Extract request parameters --- strategy = request.get('strategy', {}) start_date = request.get('startDate') end_date = request.get('endDate') initial_capital = request.get('initialCapital', 100000) parameters = strategy.get('params') or strategy.get('parameters', {}) strategy_type = strategy.get('type', 'sma_crossover') # Accept UI's flat symbols[] OR legacy assets[].symbol symbols = request.get('symbols') or [a.get('symbol') for a in request.get('assets', []) if a.get('symbol')] if not symbols: return self._create_error_result('No symbols provided') print(f'[PY-BT] Extracted: strategy_type={strategy_type}, params={parameters}', file=sys.stderr) print(f'[PY-BT] Date range: {start_date} to {end_date}, capital={initial_capital}', file=sys.stderr) # --- Download market data via yfinance --- interval = request.get('interval', '1d') multi_asset = len(symbols) > 1 print(f'[PY-BT] Symbols: {symbols}, multi_asset={multi_asset}, interval={interval}', file=sys.stderr) logs.append(f'{self._current_timestamp()}: Downloading data for {symbols}') close_data, using_synthetic = self._load_market_data( symbols, start_date, end_date, interval=interval, multi_asset=multi_asset ) import pandas as pd # --- Multi-asset portfolio path --- if multi_asset and isinstance(close_data, pd.DataFrame) and close_data.shape[1] > 1: close_df = close_data col_symbols = list(close_df.columns) n_assets = len(col_symbols) # Weighted allocation: use portfolio weights if provided, else equal-weight req_weights = request.get('weights', None) if req_weights and len(req_weights) == n_assets: capital_alloc = [initial_capital * float(w) for w in req_weights] else: capital_alloc = [initial_capital / n_assets] * n_assets print(f'[PY-BT] Multi-asset: {n_assets} symbols, alloc={[f"${c:.0f}" for c in capital_alloc]}', file=sys.stderr) logs.append(f'{self._current_timestamp()}: Multi-asset portfolio: {col_symbols}, ' f'allocation: {[f"${c:.0f}" for c in capital_alloc]}') all_trades = [] per_symbol_equity = {} combined_equity_values = None for i, sym_col in enumerate(col_symbols): sym_series = close_df[sym_col].dropna() if len(sym_series) < 5: logs.append(f'{self._current_timestamp()}: Skipping {sym_col}: insufficient data') continue sym_capital = capital_alloc[i] sym_series = pd.Series(sym_series.values.astype(float), index=sym_series.index, name='Close') if strategy_type == 'code': sym_portfolio = self._run_custom_code(vbt, strategy, sym_series, sym_capital) else: entries, exits = strat.build_strategy_signals(vbt, strategy_type, sym_series, parameters) sym_portfolio = pf.build_portfolio(vbt, sym_series, entries, exits, sym_capital, request) sym_equity = sym_portfolio.value() per_symbol_equity[sym_col] = sym_equity if combined_equity_values is None: combined_equity_values = sym_equity.copy() else: common_idx = combined_equity_values.index.intersection(sym_equity.index) combined_equity_values = combined_equity_values.loc[common_idx] + sym_equity.loc[common_idx] sym_trades = self._parse_trades(sym_portfolio, [sym_col]) all_trades.extend(sym_trades) n_sym_trades = len(sym_trades) logs.append(f'{self._current_timestamp()}: {sym_col}: {n_sym_trades} trades, ' f'return: {sym_portfolio.total_return() * 100:.2f}%') if combined_equity_values is None or len(combined_equity_values) == 0: return self._create_error_result('No valid data for any symbol in the portfolio') # Build combined portfolio from merged equity combined_portfolio = pf.SimplePortfolio.from_equity_series( combined_equity_values, initial_capital ) portfolio = combined_portfolio close_series = close_df.iloc[:, 0].dropna() close_series = pd.Series(close_series.values.astype(float), index=close_series.index, name='Close') trades_list = all_trades logs.append(f'{self._current_timestamp()}: Portfolio combined: {len(all_trades)} total trades, ' f'final value: {combined_equity_values.iloc[-1]:.2f}') else: # --- Single-asset path --- close_series = close_data if isinstance(close_data, pd.Series) else close_data if isinstance(close_series, pd.DataFrame): close_series = close_series.iloc[:, 0].dropna() close_series = pd.Series(close_series.values.astype(float), index=close_series.index, name='Close') print(f'[PY-BT] Data loaded: {len(close_series)} bars, synthetic={using_synthetic}', file=sys.stderr) logs.append(f'{self._current_timestamp()}: Data: {len(close_series)} bars, ' f'synthetic={using_synthetic}') if strategy_type == 'code': portfolio = self._run_custom_code( vbt, strategy, close_series, initial_capital ) else: logs.append(f'{self._current_timestamp()}: Building signals for strategy: {strategy_type}, params: {parameters}') entries, exits = strat.build_strategy_signals( vbt, strategy_type, close_series, parameters ) n_entries = int(entries.sum()) if hasattr(entries, 'sum') else 0 n_exits = int(exits.sum()) if hasattr(exits, 'sum') else 0 logs.append(f'{self._current_timestamp()}: Signals: {n_entries} entries, {n_exits} exits') portfolio = pf.build_portfolio( vbt, close_series, entries, exits, initial_capital, request ) trades_list = self._parse_trades(portfolio, symbols) n_trades = len(trades_list) if multi_asset else ( len(portfolio.trades.records_readable) if hasattr(portfolio.trades, 'records_readable') else 0 ) logs.append(f'{self._current_timestamp()}: Backtest completed: {n_trades} trades, ' f'final value: {portfolio.final_value():.2f}, ' f'return: {portfolio.total_return() * 100:.2f}%') # --- Extract comprehensive metrics --- risk_free_rate = request.get('riskFreeRate', 0.0) all_metrics = metrics.extract_full_metrics( portfolio, initial_capital, close_series, vbt, risk_free_rate=risk_free_rate ) # --- Build result data structures --- perf = all_metrics['performance'] stats = all_metrics['statistics'] extended = all_metrics['extended_stats'] performance = PerformanceMetrics( total_return=perf['totalReturn'], annualized_return=perf['annualizedReturn'], sharpe_ratio=perf['sharpeRatio'], sortino_ratio=perf['sortinoRatio'], max_drawdown=perf['maxDrawdown'], win_rate=perf['winRate'], loss_rate=perf['lossRate'], profit_factor=perf['profitFactor'], volatility=perf['volatility'], calmar_ratio=perf['calmarRatio'], total_trades=perf['totalTrades'], winning_trades=perf['winningTrades'], losing_trades=perf['losingTrades'], average_win=perf['averageWin'], average_loss=perf['averageLoss'], largest_win=perf['largestWin'], largest_loss=perf['largestLoss'], average_trade_return=perf['averageTradeReturn'], expectancy=perf['expectancy'], ) statistics = BacktestStatistics( start_date=stats['startDate'], end_date=stats['endDate'], initial_capital=stats['initialCapital'], final_capital=stats['finalCapital'], total_fees=stats['totalFees'], total_slippage=stats['totalSlippage'], total_trades=stats['totalTrades'], winning_days=stats['winningDays'], losing_days=stats['losingDays'], average_daily_return=stats['averageDailyReturn'], best_day=stats['bestDay'], worst_day=stats['worstDay'], consecutive_wins=stats['consecutiveWins'], consecutive_losses=stats['consecutiveLosses'], ) # Parse trades (multi-asset already built trades_list above) if not multi_asset: trades_list = self._parse_trades(portfolio, symbols) # Build equity curve equity = self._build_equity_curve(portfolio, initial_capital) # Assemble result result = BacktestResult( id=backtest_id, status='completed', performance=performance, trades=trades_list, equity=equity, statistics=statistics, logs=logs, start_time=self._current_timestamp(), end_time=self._current_timestamp(), ) result_dict = result.to_dict() result_dict['using_synthetic_data'] = using_synthetic result_dict['dataSource'] = getattr(self, '_broker_data_source', '') or 'yahoo' result_dict['extended_stats'] = extended if using_synthetic: result_dict['synthetic_data_warning'] = ( 'WARNING: This backtest used SYNTHETIC (fake) data because real market data ' 'could not be loaded. Install yfinance (pip install yfinance) and ensure ' 'internet connectivity for real results. These results have NO financial meaning.' ) logs.append(f'{self._current_timestamp()}: *** SYNTHETIC DATA WARNING: Results are based on fake data ***') # Include detailed analysis sub-results result_dict['trade_analysis'] = all_metrics.get('trade_analysis', {}) result_dict['drawdown_analysis'] = all_metrics.get('drawdown_analysis', {}) result_dict['returns_analysis'] = all_metrics.get('returns_analysis', {}) result_dict['risk_metrics'] = all_metrics.get('risk_metrics', {}) # --- Advanced Metrics (benchmark, rolling, monthly) --- self._add_advanced_metrics( result_dict, portfolio, request, initial_capital, start_date, end_date ) return {'success': True, 'message': 'Backtest completed', 'data': result_dict} except Exception as e: self._error(f'Backtest {backtest_id} failed', e) import traceback tb = traceback.format_exc() logs.append(f'{self._current_timestamp()}: Error: {str(e)}') logs.append(f'{self._current_timestamp()}: Traceback: {tb}') error_result = self._create_error_result(f'VectorBT backtest failed: {str(e)}') error_result['data'] = error_result.get('data', {}) if isinstance(error_result.get('data'), dict): error_result['data']['logs'] = logs else: error_result['logs'] = logs return error_result def optimize(self, request: Dict[str, Any]) -> Dict[str, Any]: """Run parameter optimization. Accepts the UI's arg shape: - symbols (flat list) ← UI sends this - strategy: {type, params} ← strategy.params holds the seed values - paramRanges: {: {min,max,step}} ← UI's ranges (also accepts `parameters`) - optimizeObjective / objective - optimizeMethod / method - maxIterations """ try: vbt = self._import_vbt() except Exception as e: return self._create_error_result(str(e)) try: import vbt_optimization as opt strategy = request.get('strategy', {}) strategy_type = strategy.get('type', 'sma_crossover') # Accept both UI's `paramRanges` and legacy `parameters` / strategy.params seeds parameters = (request.get('paramRanges') or request.get('parameters') or strategy.get('params', {})) objective = request.get('optimizeObjective') or request.get('objective', 'sharpe') method = request.get('optimizeMethod') or request.get('method', 'grid') initial_capital = request.get('initialCapital', 100000) start_date = request.get('startDate') end_date = request.get('endDate') max_iterations = request.get('maxIterations', 500) # Accept UI's flat symbols OR legacy assets list symbols = request.get('symbols') or [a.get('symbol') for a in request.get('assets', []) if a.get('symbol')] if not symbols: return {'success': False, 'error': 'No symbols provided'} close_series, using_synthetic = self._load_market_data(self._normalize_symbols(symbols), start_date, end_date) result = opt.optimize( vbt, close_series, strategy_type, parameters, objective, initial_capital, method, max_iterations, request ) # opt.optimize returns its own success flag — surface failures if not result.get('success', True): return {'success': False, 'error': result.get('error', 'Optimization failed')} result['usingSyntheticData'] = using_synthetic return {'success': True, 'data': result} except Exception as e: self._error('Optimization failed', e) return {'success': False, 'error': f'Optimization failed: {str(e)}'} def walk_forward(self, request: Dict[str, Any]) -> Dict[str, Any]: """Walk-forward optimization. Accepts UI's arg shape: - symbols, strategy.{type,params} - paramRanges (or parameters or strategy.params) - wfSplits / nSplits, wfTrainRatio / trainRatio """ try: vbt = self._import_vbt() except Exception as e: return self._create_error_result(str(e)) try: import vbt_optimization as opt strategy = request.get('strategy', {}) strategy_type = strategy.get('type', 'sma_crossover') parameters = (request.get('paramRanges') or request.get('parameters') or strategy.get('params', {})) objective = request.get('optimizeObjective') or request.get('objective', 'sharpe') initial_capital = request.get('initialCapital', 100000) start_date = request.get('startDate') end_date = request.get('endDate') n_splits = request.get('wfSplits') or request.get('nSplits', 5) train_ratio = request.get('wfTrainRatio') or request.get('trainRatio', 0.7) symbols = request.get('symbols') or [a.get('symbol') for a in request.get('assets', []) if a.get('symbol')] if not symbols: return {'success': False, 'error': 'No symbols provided'} close_series, using_synthetic = self._load_market_data(self._normalize_symbols(symbols), start_date, end_date) result = opt.walk_forward_optimize( vbt, close_series, strategy_type, parameters, objective, initial_capital, n_splits, train_ratio, request ) if not result.get('success', True): return {'success': False, 'error': result.get('error', 'Walk-forward failed')} result['usingSyntheticData'] = using_synthetic return {'success': True, 'data': result} except Exception as e: self._error('Walk-forward optimization failed', e) return {'success': False, 'error': f'Walk-forward failed: {str(e)}'} def get_strategies(self, request: Dict[str, Any]) -> Dict[str, Any]: """Return available strategy catalog.""" import vbt_strategies as strat return {'success': True, 'data': strat.get_strategy_catalog()} def get_indicators(self, request: Dict[str, Any]) -> Dict[str, Any]: """Return indicator catalog normalized to {indicators: {Category: [{id, name}]}}.""" import vbt_indicators as ind raw = ind.get_indicator_catalog() grouped = {} for cat in raw.get('categories', []): cat_name = cat.get('name', 'Other') grouped[cat_name] = [ {'id': it['id'], 'name': it.get('name', it['id'])} for it in cat.get('indicators', []) ] return {'indicators': grouped} def get_command_options(self, request: Dict[str, Any]) -> Dict[str, Any]: """Return provider-specific option lists for all command dropdowns. Names here MUST match what the Python dispatchers below actually accept — these strings are sent back unchanged in execute() requests. """ return { 'success': True, 'data': { 'position_sizing_methods': ['percent', 'fixed', 'kelly', 'vol_target', 'risk'], 'optimize_objectives': ['sharpe', 'sortino', 'calmar', 'return', 'omega'], 'optimize_methods': ['grid', 'random'], 'label_types': ['FIXLB', 'MEANLB', 'LEXLB', 'TRENDLB', 'BOLB'], 'splitter_types': ['RollingSplitter', 'ExpandingSplitter', 'PurgedKFoldSplitter'], 'signal_generators': ['RAND', 'RANDX', 'RANDNX', 'RPROB', 'RPROBX'], 'indicator_signal_modes': ['crossover_signals', 'threshold_signals', 'breakout_signals', 'mean_reversion_signals', 'signal_filter'], 'returns_analysis_types': ['returns_stats', 'drawdowns', 'ranges', 'rolling'], }, } def get_historical_data(self, request: Dict[str, Any]) -> list: """Get historical data using yfinance.""" try: import yfinance as yf import pandas as pd symbols = request.get('symbols', []) normalized = self._normalize_symbols(symbols) start_date = request.get('startDate') end_date = request.get('endDate') timeframe = request.get('timeframe', 'daily') data = yf.download( normalized if len(normalized) > 1 else normalized[0], start=start_date, end=end_date, progress=False ) if data is None or data.empty: return [] result = [] for i, symbol in enumerate(symbols): norm_sym = normalized[i] if i < len(normalized) else symbol try: if len(normalized) == 1: symbol_data = data elif isinstance(data.columns, pd.MultiIndex): # MultiIndex: extract per-symbol OHLCV symbol_data = data.xs(norm_sym, axis=1, level=1) if norm_sym in data.columns.get_level_values(1) else data else: symbol_data = data bars = [] for date, row in symbol_data.iterrows(): bars.append({ 'date': date.isoformat(), 'open': float(row.get('Open', 0) if hasattr(row, 'get') else 0), 'high': float(row.get('High', 0) if hasattr(row, 'get') else 0), 'low': float(row.get('Low', 0) if hasattr(row, 'get') else 0), 'close': float(row.get('Close', 0) if hasattr(row, 'get') else 0), 'volume': int(row.get('Volume', 0) if hasattr(row, 'get') else 0), }) result.append({'symbol': symbol, 'timeframe': timeframe, 'data': bars}) except Exception: result.append({'symbol': symbol, 'timeframe': timeframe, 'data': []}) return result except Exception as e: self._error('Data fetch failed', e) raise def generate_signals(self, request: Dict[str, Any]) -> Dict[str, Any]: """Generate trading signals using vbt_signals module.""" try: import numpy as np import pandas as pd import vbt_signals as sig generator_type = request.get('generatorType', 'RAND') symbols = request.get('symbols', ['SPY']) start_date = request.get('startDate') end_date = request.get('endDate') params = request.get('params', {}) # Load market data to get a close series close_series, using_synthetic = self._load_market_data( self._normalize_symbols(symbols), start_date, end_date ) gen_map = { 'RAND': lambda: sig.RAND(close_series, n=params.get('n', 10), seed=params.get('seed')), 'RANDX': lambda: sig.RANDX(close_series, n=params.get('n', 10), seed=params.get('seed')), 'RANDNX': lambda: sig.RANDNX(close_series, n=params.get('n', 10), seed=params.get('seed'), min_hold=params.get('min_hold', 1), max_hold=params.get('max_hold', 20)), 'RPROB': lambda: sig.RPROB(close_series, entry_prob=params.get('entry_prob', 0.1), seed=params.get('seed')), 'RPROBX': lambda: sig.RPROBX(close_series, entry_prob=params.get('entry_prob', 0.1), exit_prob=params.get('exit_prob', 0.1), seed=params.get('seed')), 'RPROBCX': lambda: sig.RPROBCX(close_series, entry_prob=params.get('entry_prob', 0.1), exit_prob=params.get('exit_prob', 0.1), cooldown=params.get('cooldown', 5), seed=params.get('seed')), 'RPROBNX': lambda: sig.RPROBNX(close_series, n=params.get('n', 10), entry_prob=params.get('entry_prob', 0.1), exit_prob=params.get('exit_prob', 0.2), seed=params.get('seed')), } result_data = {'generatorType': generator_type, 'totalBars': len(close_series), 'usingSyntheticData': using_synthetic} if generator_type in gen_map: output = gen_map[generator_type]() if isinstance(output, tuple): entries, exits = output result_data['entryCount'] = int(entries.sum()) if hasattr(entries, 'sum') else int(np.sum(entries)) result_data['exitCount'] = int(exits.sum()) if hasattr(exits, 'sum') else int(np.sum(exits)) result_data['entries'] = [str(d) for d in entries[entries].index[:50]] if hasattr(entries, 'index') else [] result_data['exits'] = [str(d) for d in exits[exits].index[:50]] if hasattr(exits, 'index') else [] else: entries = output result_data['entryCount'] = int(entries.sum()) if hasattr(entries, 'sum') else int(np.sum(entries)) result_data['entries'] = [str(d) for d in entries[entries].index[:50]] if hasattr(entries, 'index') else [] elif generator_type in ('STX', 'STCX', 'OHLCSTX', 'OHLCSTCX'): # Stop-based generators need entry signals first entries_base = sig.RPROB(close_series, entry_prob=0.05, seed=42) if generator_type == 'STX': exits = sig.STX(close_series, entries_base, stop_loss=params.get('stop_loss'), take_profit=params.get('take_profit')) elif generator_type == 'STCX': exits = sig.STCX(close_series, entries_base, stop_loss=params.get('stop_loss'), take_profit=params.get('take_profit'), trailing_stop=params.get('trailing_stop')) else: exits = sig.STX(close_series, entries_base, stop_loss=params.get('stop_loss'), take_profit=params.get('take_profit')) result_data['baseEntryCount'] = int(entries_base.sum()) if hasattr(entries_base, 'sum') else 0 result_data['exitCount'] = int(exits.sum()) if hasattr(exits, 'sum') else int(np.sum(exits)) result_data['exits'] = [str(d) for d in exits[exits].index[:50]] if hasattr(exits, 'index') else [] else: return {'success': False, 'error': f'Unknown generator type: {generator_type}'} # Apply clean_signals post-processing if requested clean = request.get('params', {}).get('clean', False) or request.get('clean', False) if clean and 'entryCount' in result_data: try: cleaned_entries, cleaned_exits = sig.clean_signals( entries if isinstance(entries, pd.Series) else pd.Series(entries, index=close_series.index), exits if isinstance(exits, pd.Series) else pd.Series(exits, index=close_series.index) if 'exits' in dir() else pd.Series(False, index=close_series.index) ) result_data['cleanedEntryCount'] = int(cleaned_entries.sum()) result_data['cleanedExitCount'] = int(cleaned_exits.sum()) result_data['cleaned'] = True except Exception: result_data['cleaned'] = False return {'success': True, 'data': result_data} except Exception as e: return {'success': False, 'error': str(e), 'traceback': __import__('traceback').format_exc()} def generate_labels(self, request: Dict[str, Any]) -> Dict[str, Any]: """Generate ML labels using vbt_labels module.""" try: import numpy as np import pandas as pd import vbt_labels as lb label_type = request.get('labelType', 'FIXLB') symbols = request.get('symbols', ['SPY']) start_date = request.get('startDate') end_date = request.get('endDate') params = request.get('params', {}) close_series, using_synthetic = self._load_market_data( self._normalize_symbols(symbols), start_date, end_date ) label_map = { 'FIXLB': lambda: lb.FIXLB(close_series, horizon=params.get('horizon', 5), threshold=params.get('threshold', 0.0)), 'MEANLB': lambda: lb.MEANLB(close_series, window=params.get('window', 20), threshold=params.get('threshold', 1.0)), 'LEXLB': lambda: lb.LEXLB(close_series, window=params.get('window', 5)), 'TRENDLB': lambda: lb.TRENDLB(close_series, window=params.get('window', 20), threshold=params.get('threshold', 0.0)), 'BOLB': lambda: lb.BOLB(close_series, window=params.get('window', 20), alpha=params.get('alpha', 2.0)), } if label_type not in label_map: return {'success': False, 'error': f'Unknown label type: {label_type}'} labels = label_map[label_type]() # Build distribution summary unique, counts = np.unique(labels.dropna().values if hasattr(labels, 'dropna') else labels[~np.isnan(labels)], return_counts=True) distribution = {str(int(v)): int(c) for v, c in zip(unique, counts)} result_data = { 'labelType': label_type, 'totalBars': len(close_series), 'labeledBars': int(labels.notna().sum()) if hasattr(labels, 'notna') else int(np.sum(~np.isnan(labels))), 'usingSyntheticData': using_synthetic, 'distribution': distribution, 'sampleLabels': [ {'date': str(labels.index[i]), 'label': int(labels.iloc[i])} for i in range(min(50, len(labels))) if not pd.isna(labels.iloc[i]) ], } return {'success': True, 'data': result_data} except Exception as e: return {'success': False, 'error': str(e), 'traceback': __import__('traceback').format_exc()} def generate_splits(self, request: Dict[str, Any]) -> Dict[str, Any]: """Generate cross-validation splits using vbt_splitters module.""" try: import numpy as np import pandas as pd import vbt_splitters as sp splitter_type = request.get('splitterType', 'RollingSplitter') symbols = request.get('symbols', []) start_date = request.get('startDate') end_date = request.get('endDate') total_bars = request.get('totalBars') params = request.get('params', {}) # Build an index - from market data if available, else synthetic if symbols and start_date and end_date: close_series, using_synthetic = self._load_market_data( self._normalize_symbols(symbols), start_date, end_date ) index = close_series.index elif total_bars: index = pd.date_range(start=start_date or '2020-01-01', periods=total_bars, freq='B') else: index = pd.date_range(start='2020-01-01', end='2024-01-01', freq='B') # Accept both snake_case Python names AND the UI's camelCase aliases. # The UI exposes generic WINDOW LENGTH / STEP SIZE inputs — for the # test window we derive a sensible split (1/3 of window) so users get # non-empty splits without having to expose every internal knob. window_len = params.get('window_len') or params.get('windowLength') or 60 step = params.get('step') or params.get('stepSize') or max(1, window_len // 4) test_len = params.get('test_len') or params.get('testLength') or max(5, window_len // 3) splitter_map = { 'RollingSplitter': lambda: sp.RollingSplitter( window_len=window_len, test_len=test_len, step=step, ), 'ExpandingSplitter': lambda: sp.ExpandingSplitter( min_len=params.get('min_len') or params.get('windowLength') or window_len, test_len=test_len, step=step, ), 'PurgedKFoldSplitter': lambda: sp.PurgedKFoldSplitter( n_splits=params.get('n_splits', 5), purge_len=params.get('purge_len', 5), embargo_len=params.get('embargo_len', 5), ), } # Back-compat alias — older UI may send the short form if splitter_type == 'PurgedKFold': splitter_type = 'PurgedKFoldSplitter' # RangeSplitter: user supplies date ranges directly if splitter_type == 'RangeSplitter': ranges_raw = params.get('ranges', []) if not ranges_raw: return {'success': False, 'error': 'RangeSplitter requires "ranges" array with train/test date ranges'} splitter = sp.RangeSplitter(ranges=[(r['trainStart'], r['trainEnd'], r['testStart'], r['testEnd']) for r in ranges_raw]) splits = [] for i, (train_mask, test_mask) in enumerate(splitter.split(index)): train_idx = np.where(train_mask)[0] if hasattr(train_mask, '__len__') else train_mask test_idx = np.where(test_mask)[0] if hasattr(test_mask, '__len__') else test_mask splits.append({ 'fold': i, 'trainStart': str(index[train_idx[0]]) if len(train_idx) > 0 else '', 'trainEnd': str(index[train_idx[-1]]) if len(train_idx) > 0 else '', 'trainSize': len(train_idx), 'testStart': str(index[test_idx[0]]) if len(test_idx) > 0 else '', 'testEnd': str(index[test_idx[-1]]) if len(test_idx) > 0 else '', 'testSize': len(test_idx), }) return {'success': True, 'data': { 'splitterType': splitter_type, 'totalBars': len(index), 'nSplits': len(splits), 'indexStart': str(index[0]), 'indexEnd': str(index[-1]), 'splits': splits, }} if splitter_type not in splitter_map: return {'success': False, 'error': f'Unknown splitter type: {splitter_type}'} splitter = splitter_map[splitter_type]() splits = [] for i, (train_mask, test_mask) in enumerate(splitter.split(index)): # Splitters yield boolean masks — convert to positional indices. train_pos = np.where(np.asarray(train_mask))[0] test_pos = np.where(np.asarray(test_mask))[0] if len(train_pos) == 0 or len(test_pos) == 0: continue splits.append({ 'fold': i, 'trainStart': str(index[train_pos[0]]), 'trainEnd': str(index[train_pos[-1]]), 'trainSize': int(len(train_pos)), 'testStart': str(index[test_pos[0]]), 'testEnd': str(index[test_pos[-1]]), 'testSize': int(len(test_pos)), }) result_data = { 'splitterType': splitter_type, 'totalBars': len(index), 'nSplits': len(splits), 'indexStart': str(index[0]), 'indexEnd': str(index[-1]), 'splits': splits, } return {'success': True, 'data': result_data} except Exception as e: return {'success': False, 'error': str(e), 'traceback': __import__('traceback').format_exc()} def analyze_returns(self, request: Dict[str, Any]) -> Dict[str, Any]: """Analyze returns using vbt_returns (ReturnsAccessor) + vbt_generic (Drawdowns, Ranges).""" try: import numpy as np import pandas as pd import vbt_returns as ret import vbt_generic as gen analysis_type = request.get('analysisType', 'returns_stats') symbols = request.get('symbols', ['SPY']) start_date = request.get('startDate') end_date = request.get('endDate') # UI sends benchmarkSymbol; legacy callers pass benchmark benchmark = request.get('benchmark') or request.get('benchmarkSymbol') or '' params = dict(request.get('params') or {}) # UI surfaces a "ROLLING WINDOW" spinbox at the top level — fold it into params. if 'rollingWindow' in request and 'window' not in params: params['window'] = request['rollingWindow'] close_series, using_synthetic = self._load_market_data( self._normalize_symbols(symbols), start_date, end_date ) returns = close_series.pct_change().dropna() # Load benchmark if provided benchmark_rets = None if benchmark and benchmark.strip(): try: bench_close, _ = self._load_market_data( self._normalize_symbols([benchmark]), start_date, end_date ) benchmark_rets = bench_close.pct_change().dropna() # Align lengths common_idx = returns.index.intersection(benchmark_rets.index) returns = returns.loc[common_idx] benchmark_rets = benchmark_rets.loc[common_idx] except Exception: benchmark_rets = None result_data = { 'analysisType': analysis_type, 'totalBars': len(close_series), 'returnBars': len(returns), 'usingSyntheticData': using_synthetic, } if analysis_type == 'returns_stats': risk_free = params.get('risk_free', 0.0) n_trials = params.get('n_trials', 1) omega_threshold = params.get('omega_threshold', 0.0) acc = ret.ReturnsAccessor(returns, benchmark_rets=benchmark_rets) stats = acc.stats() result_data['stats'] = {k: float(v) if isinstance(v, (int, float, np.floating, np.integer)) else str(v) for k, v in stats.items()} # Add extra metrics try: result_data['stats']['Deflated Sharpe'] = float(acc.deflated_sharpe_ratio(n_trials=n_trials)) except Exception: pass try: result_data['stats']['Up Capture'] = float(acc.up_capture()) result_data['stats']['Down Capture'] = float(acc.down_capture()) result_data['stats']['Up/Down Ratio'] = float(acc.up_down_ratio()) except Exception: pass # Cumulative returns as sample cum = acc.cumulative() result_data['cumulativeReturns'] = [ {'date': str(cum.index[i]), 'value': float(cum.iloc[i])} for i in range(0, len(cum), max(1, len(cum) // 100)) ] elif analysis_type == 'drawdowns': equity = (1 + returns).cumprod() dd = gen.Drawdowns.from_ts(equity) result_data['stats'] = {k: float(v) if isinstance(v, (int, float, np.floating, np.integer)) else str(v) for k, v in dd.stats().items()} result_data['totalDrawdowns'] = dd.count result_data['maxDrawdown'] = float(dd.max_drawdown()) result_data['avgDrawdown'] = float(dd.avg_drawdown()) result_data['activeDrawdown'] = float(dd.active_drawdown()) result_data['activeDuration'] = int(dd.active_duration()) # Records records_df = dd.records_readable if records_df is not None and len(records_df) > 0: result_data['records'] = records_df.head(50).to_dict('records') # Drawdown series (sampled) dd_series = dd.drawdown() result_data['drawdownSeries'] = [ {'date': str(dd_series.index[i]), 'value': float(dd_series.iloc[i])} for i in range(0, len(dd_series), max(1, len(dd_series) // 100)) ] elif analysis_type == 'ranges': threshold = params.get('threshold', 0.0) rng = gen.Ranges.from_ts(returns, threshold=threshold) result_data['stats'] = {k: float(v) if isinstance(v, (int, float, np.floating, np.integer)) else str(v) for k, v in rng.stats().items()} result_data['totalRanges'] = rng.count result_data['avgDuration'] = float(rng.avg_duration()) result_data['maxDuration'] = int(rng.max_duration()) result_data['coverage'] = float(rng.coverage()) records_df = rng.records_readable if records_df is not None and len(records_df) > 0: result_data['records'] = records_df.head(50).to_dict('records') elif analysis_type == 'rolling': window = params.get('window', 252) risk_free = params.get('risk_free', 0.0) metric = params.get('metric', 'sharpe') acc = ret.ReturnsAccessor(returns, benchmark_rets=benchmark_rets) rolling_map = { 'total': lambda: acc.rolling_total(window), 'annualized': lambda: acc.rolling_annualized(window), 'volatility': lambda: acc.rolling_annualized_volatility(window), 'sharpe': lambda: acc.rolling_sharpe_ratio(window, risk_free), 'sortino': lambda: acc.rolling_sortino_ratio(window, risk_free), 'calmar': lambda: acc.rolling_calmar_ratio(window), 'omega': lambda: acc.rolling_omega_ratio(window), 'info_ratio': lambda: acc.rolling_information_ratio(window), 'downside_risk': lambda: acc.rolling_downside_risk(window), } if metric not in rolling_map: return {'success': False, 'error': f'Unknown rolling metric: {metric}'} series = rolling_map[metric]() series = series.dropna() result_data['metric'] = metric result_data['window'] = window result_data['dataPoints'] = len(series) result_data['series'] = [ {'date': str(series.index[i]), 'value': float(series.iloc[i])} for i in range(0, len(series), max(1, len(series) // 200)) ] else: return {'success': False, 'error': f'Unknown analysis type: {analysis_type}'} return {'success': True, 'data': result_data} except Exception as e: return {'success': False, 'error': str(e), 'traceback': __import__('traceback').format_exc()} def _calculate_signal_metrics(self, entries, exits, close_series): """Calculate performance metrics for generated signals.""" import numpy as np metrics = { 'totalSignals': int(entries.sum()), 'signalDensity': float(entries.sum() / len(close_series) * 100), # % of bars with signals } # Simple forward-looking performance (next bar return after entry) if entries.sum() > 0: entry_indices = np.where(entries)[0] forward_returns = [] holding_periods = [] for entry_idx in entry_indices: if entry_idx < len(close_series) - 1: entry_price = close_series.iloc[entry_idx] # Find next exit or end of data exit_idx = None for idx in range(entry_idx + 1, len(exits)): if exits.iloc[idx]: exit_idx = idx break if exit_idx is None: exit_idx = len(close_series) - 1 exit_price = close_series.iloc[exit_idx] ret = (exit_price - entry_price) / entry_price * 100 forward_returns.append(ret) holding_periods.append(exit_idx - entry_idx) if forward_returns: returns_array = np.array(forward_returns) metrics['avgReturn'] = float(np.mean(returns_array)) metrics['medianReturn'] = float(np.median(returns_array)) metrics['winRate'] = float(np.sum(returns_array > 0) / len(returns_array) * 100) metrics['bestTrade'] = float(np.max(returns_array)) metrics['worstTrade'] = float(np.min(returns_array)) metrics['avgHoldingPeriod'] = float(np.mean(holding_periods)) metrics['profitFactor'] = float( np.sum(returns_array[returns_array > 0]) / abs(np.sum(returns_array[returns_array < 0])) ) if np.sum(returns_array < 0) != 0 else float('inf') return metrics def indicator_signals(self, request: Dict[str, Any]) -> Dict[str, Any]: """Generate indicator-based signals (crossover, threshold, breakout, mean_reversion, filter).""" try: import numpy as np import pandas as pd import vbt_indicators as ind mode = request.get('mode', 'crossover_signals') symbols = request.get('symbols', ['SPY']) start_date = request.get('startDate') end_date = request.get('endDate') indicator = request.get('indicator', 'rsi') params = request.get('params', {}) close_series, using_synthetic = self._load_market_data( self._normalize_symbols(symbols), start_date, end_date ) result_data = { 'mode': mode, 'totalBars': len(close_series), 'usingSyntheticData': using_synthetic, 'symbol': symbols[0] if symbols else 'Unknown', 'dateRange': { 'start': str(close_series.index[0]), 'end': str(close_series.index[-1]) } } if mode == 'crossover_signals': fast_ind = params.get('fast_indicator', 'ma') fast_period = params.get('fast_period', 10) slow_ind = params.get('slow_indicator', 'ma') slow_period = params.get('slow_period', 20) calc_map = { 'ma': lambda s, p: ind.calculate_ma(None, s, p, ewm=False), 'ema': lambda s, p: ind.calculate_ma(None, s, p, ewm=True) } fast_line = calc_map.get(fast_ind, calc_map['ma'])(close_series, fast_period) slow_line = calc_map.get(slow_ind, calc_map['ma'])(close_series, slow_period) entries, exits = ind.generate_crossover_signals(fast_line, slow_line, close_series.index) # Add performance metrics metrics = self._calculate_signal_metrics(entries, exits, close_series) result_data.update(metrics) # Add signal details with context result_data['entryCount'] = int(entries.sum()) result_data['exitCount'] = int(exits.sum()) result_data['entrySignals'] = [ { 'date': str(d), 'price': float(close_series.loc[d]), 'fastMA': float(fast_line[close_series.index.get_loc(d)]), 'slowMA': float(slow_line[close_series.index.get_loc(d)]), } for d in entries[entries].index[:50] ] result_data['exitSignals'] = [ { 'date': str(d), 'price': float(close_series.loc[d]), 'fastMA': float(fast_line[close_series.index.get_loc(d)]), 'slowMA': float(slow_line[close_series.index.get_loc(d)]), } for d in exits[exits].index[:50] ] elif mode == 'threshold_signals': period = params.get('period', 14) lower = params.get('lower', 30) upper = params.get('upper', 70) # Need to get full OHLCV for some indicators import yfinance as yf normalized_symbols = self._normalize_symbols(symbols) raw_data = yf.download( normalized_symbols if len(normalized_symbols) > 1 else normalized_symbols[0], start=start_date, end=end_date, progress=False ) high_series = raw_data['High'].values.flatten() if 'High' in raw_data.columns else close_series low_series = raw_data['Low'].values.flatten() if 'Low' in raw_data.columns else close_series ind_calc = { 'rsi': lambda s, p: ind.calculate_rsi(None, s, p), 'cci': lambda s, p: ind.calculate_cci(high_series, low_series, s, p), 'williams_r': lambda s, p: ind.calculate_williams_r(high_series, low_series, s, p), 'stoch': lambda s, p: ind.calculate_stoch(None, high_series, low_series, s, p)['k'] } calc_fn = ind_calc.get(indicator, ind_calc['rsi']) values = calc_fn(close_series, period) entries, exits = ind.generate_threshold_signals(values, lower, upper, close_series.index) # Add performance metrics metrics = self._calculate_signal_metrics(entries, exits, close_series) result_data.update(metrics) # Add signal details with context result_data['entryCount'] = int(entries.sum()) result_data['exitCount'] = int(exits.sum()) result_data['entrySignals'] = [ { 'date': str(d), 'price': float(close_series.loc[d]), 'indicatorValue': float(values[close_series.index.get_loc(d)]), 'threshold': lower, } for d in entries[entries].index[:50] ] result_data['exitSignals'] = [ { 'date': str(d), 'price': float(close_series.loc[d]), 'indicatorValue': float(values[close_series.index.get_loc(d)]), 'threshold': upper, } for d in exits[exits].index[:50] ] elif mode == 'breakout_signals': channel = params.get('channel', 'donchian') period = params.get('period', 20) # Get full OHLCV for Keltner import yfinance as yf normalized_symbols = self._normalize_symbols(symbols) raw_data = yf.download( normalized_symbols if len(normalized_symbols) > 1 else normalized_symbols[0], start=start_date, end=end_date, progress=False ) high_series = raw_data['High'].values.flatten() if 'High' in raw_data.columns else close_series low_series = raw_data['Low'].values.flatten() if 'Low' in raw_data.columns else close_series if channel == 'donchian': ch = ind.calculate_donchian(close_series, period) upper_ch = ch['upper'] if isinstance(ch, dict) else ch[0] lower_ch = ch['lower'] if isinstance(ch, dict) else ch[1] elif channel == 'bbands': bb = ind.calculate_bbands(None, close_series, period) upper_ch = bb['upper'] if isinstance(bb, dict) else bb[0] lower_ch = bb['lower'] if isinstance(bb, dict) else bb[2] else: ch = ind.calculate_keltner(high_series, low_series, close_series, ema_period=period, atr_period=10, multiplier=2.0) upper_ch = ch['upper'] if isinstance(ch, dict) else ch[0] lower_ch = ch['lower'] if isinstance(ch, dict) else ch[2] entries, exits = ind.generate_breakout_signals(close_series.values, upper_ch, lower_ch, close_series.index) result_data['entryCount'] = int(entries.sum()) result_data['exitCount'] = int(exits.sum()) result_data['entries'] = [str(d) for d in entries[entries].index[:50]] result_data['exits'] = [str(d) for d in exits[exits].index[:50]] elif mode == 'mean_reversion_signals': period = params.get('period', 20) z_entry = params.get('z_entry', 2.0) z_exit = params.get('z_exit', 0.0) zscore = ind.calculate_zscore(close_series, period) entries, exits = ind.generate_mean_reversion_signals(zscore, z_entry, z_exit, close_series.index) result_data['entryCount'] = int(entries.sum()) result_data['exitCount'] = int(exits.sum()) result_data['entries'] = [str(d) for d in entries[entries].index[:50]] result_data['exits'] = [str(d) for d in exits[exits].index[:50]] elif mode == 'signal_filter': base_indicator = params.get('base_indicator', 'rsi') base_period = params.get('base_period', 14) filter_indicator = params.get('filter_indicator', 'adx') filter_period = params.get('filter_period', 14) filter_threshold = params.get('filter_threshold', 25) filter_type = params.get('filter_type', 'above') # Get full OHLCV for indicators that need it import yfinance as yf normalized_symbols = self._normalize_symbols(symbols) raw_data = yf.download( normalized_symbols if len(normalized_symbols) > 1 else normalized_symbols[0], start=start_date, end=end_date, progress=False ) high_series = raw_data['High'].values.flatten() if 'High' in raw_data.columns else close_series low_series = raw_data['Low'].values.flatten() if 'Low' in raw_data.columns else close_series # Compute base indicator and generate threshold signals base_calc = { 'rsi': lambda s, p: ind.calculate_rsi(None, s, p), 'cci': lambda s, p: ind.calculate_cci(high_series, low_series, s, p), 'williams_r': lambda s, p: ind.calculate_williams_r(high_series, low_series, s, p), 'momentum': lambda s, p: ind.calculate_momentum(s, p), 'stoch': lambda s, p: ind.calculate_stoch(None, high_series, low_series, s, p)['k'], 'macd': lambda s, p: ind.calculate_macd(None, s, p, p * 2, 9)['macd'] } base_fn = base_calc.get(base_indicator, base_calc['rsi']) base_values = base_fn(close_series, base_period) entries, exits = ind.generate_threshold_signals(base_values, 30, 70, close_series.index) # Compute filter indicator filter_calc = { 'adx': lambda s, p: ind.calculate_adx(s, high=high_series, low=low_series, period=p), 'atr': lambda s, p: ind.calculate_atr(None, high_series, low_series, s, p), 'mstd': lambda s, p: ind.calculate_mstd(None, s, p), 'zscore': lambda s, p: ind.calculate_zscore(s, p), 'rsi': lambda s, p: ind.calculate_rsi(None, s, p) } filter_fn = filter_calc.get(filter_indicator, filter_calc['adx']) filter_values = filter_fn(close_series, filter_period) filtered_entries, filtered_exits = ind.apply_signal_filter( entries, exits, filter_values, filter_threshold, filter_type ) result_data['originalEntryCount'] = int(entries.sum()) result_data['filteredEntryCount'] = int(filtered_entries.sum()) result_data['exitCount'] = int(filtered_exits.sum()) result_data['entries'] = [str(d) for d in filtered_entries[filtered_entries].index[:50]] result_data['exits'] = [str(d) for d in filtered_exits[filtered_exits].index[:50]] else: return {'success': False, 'error': f'Unknown indicator signal mode: {mode}'} return {'success': True, 'data': result_data} except Exception as e: return {'success': False, 'error': str(e), 'traceback': __import__('traceback').format_exc()} def indicator_param_sweep(self, request: Dict[str, Any]) -> Dict[str, Any]: """Run indicator parameter sweep using IndicatorFactory.run_combs().""" import sys print("[INDICATOR_SWEEP] === STARTING ===", flush=True) sys.stdout.flush() # TEST: Return immediately to see if function is being called # return {'success': True, 'data': {'indicator': 'TEST', 'totalCombinations': 999, 'usingSyntheticData': False, 'results': []}} try: import numpy as np import pandas as pd import vbt_indicators as ind # UI alias: `sma` is the same as `ma` in this provider's indicator catalog indicator = request.get('indicator', 'rsi') if indicator == 'sma': indicator = 'ma' symbols = request.get('symbols', ['SPY']) start_date = request.get('startDate') end_date = request.get('endDate') # Accept either paramRanges (keyed-by-name dict) OR paramRange (singular # min/max/step from the UI). For the singular form we sweep the # indicator's primary period-style argument. primary_param_by_indicator = { 'momentum': 'lookback', 'macd': 'fast', 'stoch': 'k_period', 'obv': None, 'vwap': None, } param_ranges = request.get('paramRanges') if not param_ranges: pr = request.get('paramRange') if pr: primary = primary_param_by_indicator.get(indicator, 'period') if primary: param_ranges = {primary: pr} else: return {'success': False, 'error': f'Indicator {indicator} has no sweep parameter'} else: param_ranges = {} print(f"[INDICATOR_SWEEP] Got request: indicator={indicator}, symbols={symbols}", flush=True) print(f"[INDICATOR_SWEEP] param_ranges={param_ranges}", flush=True) sys.stdout.flush() # Load full OHLCV data instead of just close normalized_symbols = self._normalize_symbols(symbols) using_synthetic = False try: import yfinance as yf raw_data = yf.download( normalized_symbols if len(normalized_symbols) > 1 else normalized_symbols[0], start=start_date, end=end_date, progress=False ) if raw_data is None or raw_data.empty: raise ValueError(f'No data returned for {normalized_symbols}') # Extract OHLCV columns if len(normalized_symbols) == 1: # Single symbol - yfinance returns multi-index columns, need to flatten close_series = (raw_data['Close'].values if 'Close' in raw_data else raw_data['close'].values).flatten() high_series = (raw_data['High'].values if 'High' in raw_data else raw_data['high'].values).flatten() low_series = (raw_data['Low'].values if 'Low' in raw_data else raw_data['low'].values).flatten() volume_series = (raw_data['Volume'].values if 'Volume' in raw_data else raw_data['volume'].values).flatten() else: # Multi-symbol: take first symbol close_series = raw_data['Close'].iloc[:, 0].values high_series = raw_data['High'].iloc[:, 0].values low_series = raw_data['Low'].iloc[:, 0].values volume_series = raw_data['Volume'].iloc[:, 0].values except Exception as e: print(f"[INDICATOR_SWEEP] Data load failed: {e}, using synthetic") using_synthetic = True # Generate synthetic OHLCV close_series = self._generate_synthetic_data(symbols, start_date, end_date).values high_series = close_series * 1.02 # High = close * 1.02 low_series = close_series * 0.98 # Low = close * 0.98 volume_series = np.random.randint(1000000, 10000000, len(close_series)) # Build parameter lists from ranges param_lists = {} for name, rng in param_ranges.items(): mn, mx, st = rng.get('min', 5), rng.get('max', 50), rng.get('step', 5) param_lists[name] = list(range(int(mn), int(mx) + 1, int(st))) if st >= 1 else list(np.arange(mn, mx + st, st)) print(f"[INDICATOR_SWEEP] indicator={indicator}") print(f"[INDICATOR_SWEEP] param_ranges={param_ranges}") print(f"[INDICATOR_SWEEP] param_lists={param_lists}") print(f"[INDICATOR_SWEEP] close_series length={len(close_series)}") # Map indicator to factory-compatible function # Signatures from python -c inspection: # Single-price indicators (work with close only): # calculate_ma(vbt, close, period, ewm=False) # calculate_rsi(vbt, close, period=14) # calculate_mstd(vbt, close, period, ewm=False) # calculate_zscore(close, period=20) # calculate_momentum(close, lookback=20) # calculate_donchian(close, period=20) -> dict # calculate_bbands(vbt, close, period=20, alpha=2.0) -> dict # calculate_obv(vbt, close, volume) - needs volume # Multi-price indicators (need high/low/close): # calculate_atr(vbt, high, low, close, period=14) # calculate_cci(high, low, close, period=20) # calculate_williams_r(high, low, close, period=14) # calculate_keltner(high, low, close, ema_period=20, atr_period=10, multiplier=2.0) -> dict # calculate_stoch(vbt, high, low, close, k_period=14, d_period=3) -> dict # calculate_macd(vbt, close, fast_period=12, slow_period=26, signal_period=9) -> dict # calculate_adx(close, high=None, low=None, period=14) -> dict # Now we have full OHLCV data, support ALL indicators calc_map = { # Moving averages (close only) 'ma': lambda period: ind.calculate_ma(None, close_series, period, ewm=False), 'ema': lambda period: ind.calculate_ma(None, close_series, period, ewm=True), 'mstd': lambda period: ind.calculate_mstd(None, close_series, period, ewm=False), # Oscillators (close only) 'rsi': lambda period: ind.calculate_rsi(None, close_series, period), 'momentum': lambda lookback: ind.calculate_momentum(close_series, lookback), # Statistical (close only) 'zscore': lambda period: ind.calculate_zscore(close_series, period), # Channels/Bands (close only) 'donchian': lambda period: ind.calculate_donchian(close_series, period), 'bbands': lambda period: ind.calculate_bbands(None, close_series, period, alpha=2.0), 'macd': lambda fast, slow, signal: ind.calculate_macd(None, close_series, fast, slow, signal), 'adx': lambda period: ind.calculate_adx(close_series, high=high_series, low=low_series, period=period), # Multi-price indicators (need high/low/close) 'atr': lambda period: ind.calculate_atr(None, high_series, low_series, close_series, period), 'cci': lambda period: ind.calculate_cci(high_series, low_series, close_series, period), 'williams_r': lambda period: ind.calculate_williams_r(high_series, low_series, close_series, period), 'keltner': lambda period: ind.calculate_keltner(high_series, low_series, close_series, ema_period=period, atr_period=10, multiplier=2.0), 'stoch': lambda k_period, d_period: ind.calculate_stoch(None, high_series, low_series, close_series, k_period, d_period), # Volume indicators 'obv': lambda: ind.calculate_obv(None, close_series, volume_series), 'vwap': lambda: ind.calculate_vwap(high_series, low_series, close_series, volume_series), } if indicator not in calc_map: return {'success': False, 'error': f'Indicator {indicator} not supported for param sweep'} print(f"[INDICATOR_SWEEP] Creating factory with param_names={list(param_lists.keys())}") # Create factory - the lambdas now capture OHLCV data internally # So we don't pass inputs to run_combs, only params factory_fn = ind.IndicatorFactory.from_custom_func( calc_map[indicator], input_names=[], # No inputs needed, captured in lambda param_names=list(param_lists.keys()), output_names=['output'], short_name=indicator, ) print(f"[INDICATOR_SWEEP] Calling run_combs with param_ranges={param_lists}") # Call run_combs without inputs since lambdas capture OHLCV results = factory_fn.run_combs(param_ranges=param_lists) print(f"[INDICATOR_SWEEP] run_combs returned {len(results)} results") # Summarize results summaries = [] for r in results[:200]: # Limit output params_dict = r.get('params', {}) output = r.get('output') summary = {'params': params_dict} # Output can be a dict (from IndicatorFactory) or array if output is not None: # IndicatorFactory wraps output in dict with 'output' key if isinstance(output, dict) and 'output' in output: output = output['output'] # Now output is the actual indicator result (array or dict) if isinstance(output, dict): # Multi-output indicator (MACD, BBands, Stoch) # Calculate stats for each component summary['outputs'] = {} for component_name, component_array in output.items(): if component_array is not None and hasattr(component_array, '__len__') and len(component_array) > 0: vals = np.array(component_array, dtype=float) vals = vals[~np.isnan(vals)] if len(vals) > 0: summary['outputs'][component_name] = { 'mean': float(np.mean(vals)), 'std': float(np.std(vals)), 'min': float(np.min(vals)), 'max': float(np.max(vals)), 'last': float(vals[-1]) } elif hasattr(output, '__len__') and len(output) > 0: # Single-output indicator (MA, RSI, ATR, etc.) vals = np.array(output, dtype=float) vals = vals[~np.isnan(vals)] if len(vals) > 0: summary['mean'] = float(np.mean(vals)) summary['std'] = float(np.std(vals)) summary['min'] = float(np.min(vals)) summary['max'] = float(np.max(vals)) summary['last'] = float(vals[-1]) summaries.append(summary) result_data = { 'indicator': indicator, 'totalCombinations': len(results), 'usingSyntheticData': using_synthetic, 'results': summaries, } print(f"[INDICATOR_SWEEP] Returning: {result_data}") return {'success': True, 'data': result_data} except Exception as e: print(f"[INDICATOR_SWEEP] ERROR: {e}") import traceback traceback.print_exc() return {'success': False, 'error': str(e), 'traceback': __import__('traceback').format_exc()} def labels_to_signals(self, request: Dict[str, Any]) -> Dict[str, Any]: """Generate labels then convert to trading signals using labels_to_signals().""" try: import numpy as np import pandas as pd import vbt_labels as lb label_type = request.get('labelType', 'FIXLB') symbols = request.get('symbols', ['SPY']) start_date = request.get('startDate') end_date = request.get('endDate') params = request.get('params', {}) entry_label = request.get('entryLabel', 1) exit_label = request.get('exitLabel', -1) close_series, using_synthetic = self._load_market_data( self._normalize_symbols(symbols), start_date, end_date ) label_map = { 'FIXLB': lambda: lb.FIXLB(close_series, horizon=params.get('horizon', 5), threshold=params.get('threshold', 0.0)), 'MEANLB': lambda: lb.MEANLB(close_series, window=params.get('window', 20), threshold=params.get('threshold', 1.0)), 'LEXLB': lambda: lb.LEXLB(close_series, window=params.get('window', 5)), 'TRENDLB': lambda: lb.TRENDLB(close_series, window=params.get('window', 20), threshold=params.get('threshold', 0.0)), 'BOLB': lambda: lb.BOLB(close_series, window=params.get('window', 20), alpha=params.get('alpha', 2.0)), } if label_type not in label_map: return {'success': False, 'error': f'Unknown label type: {label_type}'} labels = label_map[label_type]() entries, exits = lb.labels_to_signals(labels, entry_label=entry_label, exit_label=exit_label) result_data = { 'labelType': label_type, 'entryLabel': entry_label, 'exitLabel': exit_label, 'totalBars': len(close_series), 'usingSyntheticData': using_synthetic, 'entryCount': int(entries.sum()), 'exitCount': int(exits.sum()), 'entries': [str(d) for d in entries[entries].index[:50]], 'exits': [str(d) for d in exits[exits].index[:50]], } return {'success': True, 'data': result_data} except Exception as e: return {'success': False, 'error': str(e), 'traceback': __import__('traceback').format_exc()} def calculate_indicator(self, request: Dict[str, Any]) -> Dict[str, Any]: """Compute one indicator over a symbol's close series and return a value series. The UI's Indicators page picks one of the IDs from get_indicators() and a date range; we return enough samples to plot a small preview. """ try: import numpy as np import pandas as pd import vbt_indicators as ind indicator = request.get('indicator', 'sma') if indicator == 'sma': indicator = 'ma' symbols = request.get('symbols', ['SPY']) start_date = request.get('startDate') end_date = request.get('endDate') params = request.get('params', {}) or {} period = int(params.get('period', 14)) close_series, using_synthetic = self._load_market_data( self._normalize_symbols(symbols), start_date, end_date ) # Try to grab full OHLCV for indicators that need high/low/volume. high = low = volume = None try: import yfinance as yf normalized = self._normalize_symbols(symbols) raw = yf.download(normalized if len(normalized) > 1 else normalized[0], start=start_date, end=end_date, progress=False) if raw is not None and not raw.empty: high = raw['High'].values.flatten() if 'High' in raw.columns else None low = raw['Low'].values.flatten() if 'Low' in raw.columns else None volume = raw['Volume'].values.flatten() if 'Volume' in raw.columns else None except Exception: pass if high is None: high = close_series.values * 1.02 low = close_series.values * 0.98 volume = np.random.randint(1_000_000, 10_000_000, len(close_series)) close_arr = close_series.values calc_map = { 'sma': lambda: ind.calculate_ma(None, close_arr, period, ewm=False), 'ma': lambda: ind.calculate_ma(None, close_arr, period, ewm=False), 'ema': lambda: ind.calculate_ma(None, close_arr, period, ewm=True), 'mstd':lambda: ind.calculate_mstd(None, close_arr, period, ewm=False), 'rsi': lambda: ind.calculate_rsi(None, close_arr, period), 'momentum': lambda: ind.calculate_momentum(close_arr, period), 'zscore': lambda: ind.calculate_zscore(close_arr, period), 'donchian': lambda: ind.calculate_donchian(close_arr, period), 'bbands': lambda: ind.calculate_bbands(None, close_arr, period, alpha=2.0), 'macd': lambda: ind.calculate_macd(None, close_arr, 12, 26, 9), 'adx': lambda: ind.calculate_adx(close_arr, high=high, low=low, period=period), 'atr': lambda: ind.calculate_atr(None, high, low, close_arr, period), 'cci': lambda: ind.calculate_cci(high, low, close_arr, period), 'williams_r': lambda: ind.calculate_williams_r(high, low, close_arr, period), 'keltner': lambda: ind.calculate_keltner(high, low, close_arr, ema_period=period, atr_period=10, multiplier=2.0), 'stochastic': lambda: ind.calculate_stoch(None, high, low, close_arr, period, 3), 'stoch': lambda: ind.calculate_stoch(None, high, low, close_arr, period, 3), 'obv': lambda: ind.calculate_obv(None, close_arr, volume), 'vwap': lambda: ind.calculate_vwap(high, low, close_arr, volume), } if indicator not in calc_map: return {'success': False, 'error': f'Unknown indicator: {indicator}'} out = calc_map[indicator]() # Indicators return either an array (single output) or dict (multi-output). def sample_series(arr): arr = np.asarray(arr, dtype=float) idx = close_series.index # Down-sample to ~200 points for the UI stride = max(1, len(arr) // 200) return [ {'date': str(idx[i]), 'value': (None if np.isnan(arr[i]) else float(arr[i]))} for i in range(0, len(arr), stride) ] result_data = { 'indicator': indicator, 'period': period, 'symbol': symbols[0] if symbols else '', 'totalBars': len(close_arr), 'usingSyntheticData': using_synthetic, } if isinstance(out, dict): # Multi-output indicator: emit one series per component series = {} stats = {} for name, comp in out.items(): if comp is None or not hasattr(comp, '__len__'): continue series[name] = sample_series(comp) vals = np.asarray(comp, dtype=float) vals = vals[~np.isnan(vals)] if len(vals): stats[name] = { 'mean': float(np.mean(vals)), 'std': float(np.std(vals)), 'min': float(np.min(vals)), 'max': float(np.max(vals)), 'last': float(vals[-1]), } result_data['series'] = series result_data['stats'] = stats else: vals = np.asarray(out, dtype=float) clean = vals[~np.isnan(vals)] result_data['series'] = sample_series(vals) if len(clean): result_data['stats'] = { 'mean': float(np.mean(clean)), 'std': float(np.std(clean)), 'min': float(np.min(clean)), 'max': float(np.max(clean)), 'last': float(clean[-1]), } return {'success': True, 'data': result_data} except Exception as e: return {'success': False, 'error': str(e), 'traceback': __import__('traceback').format_exc()} # ======================================================================== # Private Helpers # ======================================================================== def _import_vbt(self): """Import VectorBT or return a stub (indicators/portfolio are now pure numpy/pandas).""" try: import vectorbt as vbt # Verify it's a usable install (some partial installs lack __version__) _ = vbt.__version__ return vbt except (ImportError, AttributeError): # All indicators and portfolio logic are now pure numpy/pandas, # so vectorbt is no longer required. Return a simple namespace stub. import types stub = types.SimpleNamespace(__version__='pure-numpy') return stub @staticmethod def _canon_symbol(sym): """Canonical key for matching injected broker candles: uppercase, strip an EXCHANGE: prefix and any .NS/.BO suffix. 'NSE:RELIANCE.NS' -> 'RELIANCE'.""" s = (sym or "").strip().upper() if ":" in s: parts = s.split(":") s = parts[1] if len(parts) > 1 and parts[1] else parts[0] for suf in (".NS", ".BO"): if s.endswith(suf): s = s[: -len(suf)] return s def _load_from_broker_candles(self, symbols, multi_asset): """Build a close Series (single symbol) or DataFrame (multi) from candles the C++ host pre-fetched from a broker and stashed on self._broker_candles. Returns None when the injected candles do not FULLY cover the requested symbols (the caller then falls back to yfinance). Multi-symbol is all-or-nothing to avoid mixing broker + yfinance rows. """ import pandas as pd candles_map = getattr(self, "_broker_candles", None) or {} if not candles_map: return None by_canon = {self._canon_symbol(k): v for k, v in candles_map.items()} def series_for(sym): rows = by_canon.get(self._canon_symbol(sym)) if not rows: return None idx = pd.to_datetime([int(r["t"]) for r in rows], unit="s") closes = [float(r["c"]) for r in rows] s = pd.Series(closes, index=idx, name="Close").sort_index() s = s[~s.index.duplicated(keep="last")] return s if len(s) >= 5 else None if multi_asset and len(symbols) > 1: cols = {} for sym in symbols: s = series_for(sym) if s is None: return None # all-or-nothing cols[self._canon_symbol(sym)] = s df = pd.DataFrame(cols).dropna() return df.round(4) if len(df) >= 5 else None s = series_for(symbols[0]) if symbols else None return s.round(4) if s is not None else None def _load_market_data(self, symbols: list, start_date: str, end_date: str, interval: str = '1d', multi_asset: bool = False): """ Load market data via yfinance, falling back to synthetic if unavailable. Args: symbols: List of ticker symbols start_date: Start date string end_date: End date string interval: Data interval (1m, 5m, 15m, 1h, 4h, 1d, 1wk) multi_asset: If True and multiple symbols, return DataFrame with all columns Returns: (close_series_or_df, using_synthetic) tuple """ import pandas as pd import numpy as np using_synthetic = False # ── Broker-sourced data injection ──────────────────────────────────── # The C++ host pre-fetches candles from a connected broker and passes # them in args["brokerCandles"] (stashed on self in main()). When they # fully cover the requested symbols, use them as REAL data and skip # yfinance entirely. Any gap → fall through to the yfinance path below. injected = self._load_from_broker_candles(symbols, multi_asset) if injected is not None: return injected, False # --- Normalize symbols for yfinance compatibility --- normalized_symbols = self._normalize_symbols(symbols) try: import yfinance as yf # Try downloading with normalized symbols raw_data = yf.download( normalized_symbols if len(normalized_symbols) > 1 else normalized_symbols[0], start=start_date, end=end_date, interval=interval, progress=False ) if raw_data is None or (hasattr(raw_data, 'empty') and raw_data.empty): raise ValueError(f'No data returned for {normalized_symbols}') close_data = self._extract_close_column(raw_data, normalized_symbols) if close_data is None or (hasattr(close_data, 'empty') and close_data.empty): raise ValueError(f'No Close data for {normalized_symbols}') if isinstance(close_data, pd.DataFrame): if multi_asset and close_data.shape[1] > 1: close_data = close_data.dropna() if len(close_data) < 5: raise ValueError(f'Insufficient data: only {len(close_data)} bars for {normalized_symbols}') close_data = close_data.round(4) return close_data, using_synthetic else: close_data = close_data.iloc[:, 0].dropna() if len(close_data) < 5: raise ValueError(f'Insufficient data: only {len(close_data)} bars for {normalized_symbols}') close_series = pd.Series( close_data.values.astype(float).flatten(), index=close_data.index, name='Close' ) close_series = close_series.round(4) except ImportError: print('[VBT] WARNING: yfinance not installed - using SYNTHETIC data. ' 'Results will NOT reflect real market conditions. ' 'Install yfinance: pip install yfinance', file=sys.stderr) using_synthetic = True close_series = self._generate_synthetic_data(symbols, start_date, end_date) except Exception as e: print(f'[VBT] WARNING: Data download failed for {normalized_symbols}: {e} - ' f'using SYNTHETIC data. Results will NOT reflect real market conditions.', file=sys.stderr) using_synthetic = True close_series = self._generate_synthetic_data(symbols, start_date, end_date) return close_series, using_synthetic @staticmethod def _normalize_symbols(symbols: list) -> list: """ Clean up symbol list. Symbols are expected to already include correct exchange suffixes (e.g., PIDILITIND.NS, AAPL) as resolved at portfolio-add time. This method just strips whitespace and uppercases. """ normalized = [] for sym in symbols: sym = sym.strip().upper() if sym: normalized.append(sym) return normalized if normalized else symbols @staticmethod def _extract_close_column(raw_data, symbols: list): """ Robustly extract Close price data from yfinance output. Handles: - Single-symbol download (plain columns: Open, High, Low, Close, ...) - Multi-symbol download with MultiIndex columns (Price, Ticker) - Newer yfinance versions that changed column structure """ import pandas as pd columns = raw_data.columns # Case 1: Plain columns (single symbol download) if isinstance(columns, pd.Index) and not isinstance(columns, pd.MultiIndex): if 'Close' in columns: return raw_data['Close'] # Fallback to first column return raw_data.iloc[:, 0] if len(columns) > 0 else None # Case 2: MultiIndex columns (multi-symbol download) if isinstance(columns, pd.MultiIndex): level_0_vals = columns.get_level_values(0).unique().tolist() level_1_vals = columns.get_level_values(1).unique().tolist() # yfinance 0.2.31+: columns are (Price, Ticker) e.g. ('Close', 'AAPL') if 'Close' in level_0_vals: close_data = raw_data['Close'] if isinstance(close_data, pd.Series): return close_data # DataFrame with ticker columns return close_data.dropna(how='all') # Some versions: columns are (Ticker, Price) if 'Close' in level_1_vals: close_data = raw_data.xs('Close', axis=1, level=1) if isinstance(close_data, pd.Series): return close_data return close_data.dropna(how='all') # Fallback: take first level-0 group first_group = level_0_vals[0] return raw_data[first_group] # Fallback return raw_data.iloc[:, 0] if len(raw_data.columns) > 0 else None @staticmethod def _generate_synthetic_data(symbols: list, start_date: str, end_date: str): """Generate synthetic price data as fallback. WARNING: This produces fake data and should only be used when real market data is unavailable. Results from synthetic data have no financial meaning. Uses a deterministic seed based on symbol name (not Python's hash() which is randomized across runs). """ import pandas as pd import numpy as np dates = pd.date_range(start=start_date, end=end_date, freq='B') if len(dates) == 0: dates = pd.date_range(start='2023-01-01', periods=252, freq='B') # Use deterministic seed: sum of char codes (NOT hash() which is randomized per-run) sym = symbols[0] if symbols else 'DEFAULT' seed = sum(ord(c) for c in sym) % (2**31) np.random.seed(seed) price = 100.0 prices = [] for _ in range(len(dates)): price *= (1 + np.random.normal(0.0003, 0.015)) prices.append(price) return pd.Series(prices, index=dates, name='Close', dtype=float) def _run_custom_code(self, vbt, strategy: dict, close_series, initial_capital: float): """Execute custom strategy code.""" import pandas as pd import numpy as np code = strategy.get('code', {}) source = code.get('source', '') if isinstance(code, dict) else str(code or '') if not source: raise ValueError('Custom strategy requires code') context = { 'vbt': vbt, 'pd': pd, 'np': np, 'data': close_series, 'initial_capital': initial_capital, } exec(source, context) portfolio = context.get('portfolio') if portfolio is None: raise ValueError('Custom strategy must create a "portfolio" variable') return portfolio def _parse_trades(self, portfolio, symbols: list) -> list: """Parse trades from VBT portfolio into Trade objects.""" import numpy as np trades = [] symbol = symbols[0] if symbols else 'UNKNOWN' try: if hasattr(portfolio, 'trades') and hasattr(portfolio.trades, 'records_readable'): records = portfolio.trades.records_readable value_index = portfolio.value().index for idx in range(len(records)): row = records.iloc[idx] # Get entry/exit dates from index entry_idx = int(row.get('Entry Idx', 0)) exit_idx = int(row.get('Exit Idx', entry_idx)) entry_date = str(value_index[entry_idx]).split(' ')[0] if entry_idx < len(value_index) else '' exit_date = str(value_index[exit_idx]).split(' ')[0] if exit_idx < len(value_index) else '' pnl = float(row.get('PnL', 0)) entry_price = float(row.get('Entry Price', 0)) exit_price = float(row.get('Exit Price', 0)) size = float(row.get('Size', 0)) fees = float(row.get('Entry Fees', 0)) + float(row.get('Exit Fees', 0)) ret = float(row.get('Return', 0)) # Determine exit reason exit_reason = 'signal' if 'Status' in records.columns: status = str(row.get('Status', '')) if 'StopLoss' in status: exit_reason = 'stop_loss' elif 'TakeProfit' in status: exit_reason = 'take_profit' trades.append(Trade( id=f'trade_{idx}', symbol=symbol, entry_date=entry_date, side='long', quantity=size, entry_price=entry_price, commission=fees, slippage=0, exit_date=exit_date, exit_price=exit_price, pnl=pnl, pnl_percent=ret, holding_period=max(0, exit_idx - entry_idx), exit_reason=exit_reason, )) elif hasattr(portfolio, 'trades') and hasattr(portfolio.trades, 'records'): # Fallback to raw records records = portfolio.trades.records for idx, trade in enumerate(records): pnl = float(trade.get('pnl', 0) if hasattr(trade, 'get') else getattr(trade, 'pnl', 0)) trades.append(Trade( id=f'trade_{idx}', symbol=symbol, entry_date=str(getattr(trade, 'entry_idx', 0)), side='long', quantity=float(getattr(trade, 'size', 0)), entry_price=float(getattr(trade, 'entry_price', 0)), commission=float(getattr(trade, 'fees', 0)), slippage=0, exit_date=str(getattr(trade, 'exit_idx', 0)), exit_price=float(getattr(trade, 'exit_price', 0)), pnl=pnl, pnl_percent=float(getattr(trade, 'return', 0)), holding_period=0, exit_reason='signal', )) except Exception as e: self._error('Failed to parse trades', e) return trades def _build_equity_curve(self, portfolio, initial_capital: float) -> list: """Build equity curve with returns and drawdown.""" equity = [] try: value_series = portfolio.value() peak = initial_capital for date, value in value_series.items(): returns = (value - initial_capital) / initial_capital if initial_capital > 0 else 0 peak = max(peak, value) drawdown = (value - peak) / peak if peak > 0 else 0 equity.append(EquityPoint( date=str(date).split(' ')[0], equity=float(value), returns=float(returns), drawdown=float(drawdown), )) except Exception as e: self._error('Failed to build equity curve', e) return equity def _add_advanced_metrics( self, result_dict: dict, portfolio, request: dict, initial_capital: float, start_date: str, end_date: str ): """Add advanced metrics, benchmark overlay, monthly returns, rolling metrics.""" import numpy as np try: from base.advanced_metrics import calculate_all as calc_advanced equity_values = portfolio.value().values.astype(float) dates_list = [str(d).split(' ')[0] for d in portfolio.value().index] # Load benchmark benchmark_symbol = request.get('benchmark', '') benchmark_normalized = None if benchmark_symbol: benchmark_normalized = self._load_benchmark( benchmark_symbol, start_date, end_date, len(equity_values) ) # Add benchmark to equity curve if benchmark_normalized is not None and 'equity' in result_dict: for i, point in enumerate(result_dict['equity']): if i < len(benchmark_normalized): point['benchmark'] = float(benchmark_normalized[i] * initial_capital) # Benchmark in equity scale for alpha/beta benchmark_equity = None if benchmark_normalized is not None: benchmark_equity = benchmark_normalized * equity_values[0] advanced = calc_advanced( equity_values, benchmark_series=benchmark_equity, risk_free_rate=0.0, dates=dates_list, ) result_dict['advanced_metrics'] = advanced result_dict['monthly_returns'] = advanced.pop('monthlyReturns', []) result_dict['rolling_metrics'] = { 'rollingSharpe': advanced.pop('rollingSharpe', []), 'rollingVolatility': advanced.pop('rollingVolatility', []), 'rollingDrawdown': advanced.pop('rollingDrawdown', []), } # Random benchmark p-value if request.get('randomBenchmark', False): try: import vbt_portfolio as pf_mod vbt = self._import_vbt() close_series = portfolio.close if close_series is not None: rand_stats = pf_mod.get_random_benchmark_stats( vbt, close_series, initial_capital, n_trials=100 ) strategy_return = float(portfolio.total_return()) # P-value: % of random strategies that beat this one rand_stats['strategyReturn'] = strategy_return rand_stats['pValue'] = float( np.mean(np.array([rand_stats['mean']]) > strategy_return) ) result_dict['random_benchmark'] = rand_stats except Exception: pass except Exception as adv_err: self._error('Advanced metrics failed (non-fatal)', adv_err) def _load_benchmark(self, symbol: str, start_date: str, end_date: str, target_len: int): """Load benchmark equity series normalized to start at 1.0.""" import numpy as np import pandas as pd try: import yfinance as yf # Normalize benchmark symbol too norm_sym = self._normalize_symbols([symbol]) bench_symbol = norm_sym[0] if norm_sym else symbol raw = yf.download(bench_symbol, start=start_date, end=end_date, progress=False) bench_data = self._extract_close_column(raw, [bench_symbol]) if isinstance(bench_data, pd.DataFrame): bench_data = bench_data.iloc[:, 0] if bench_data is not None and len(bench_data) > 0: values = np.round(bench_data.values.astype(float).flatten(), 4) normalized = values / values[0] if len(normalized) != target_len: indices = np.linspace(0, len(normalized) - 1, target_len).astype(int) normalized = normalized[indices] return normalized except Exception as e: self._error(f'Benchmark load failed for {symbol}', e) return None # ============================================================================ # CLI Entry Point # ============================================================================ def main(): import io import os import warnings # Suppress all warnings (they can corrupt stdout JSON) warnings.filterwarnings('ignore') os.environ['PYTHONWARNINGS'] = 'ignore' # Support both stdin (--stdin flag) and argv modes use_stdin = '--stdin' in sys.argv if use_stdin: if len(sys.argv) < 2: print(json_response({'success': False, 'error': 'Usage: vectorbt_provider.py --stdin'})) return command = sys.argv[1] json_args = sys.stdin.read() else: if len(sys.argv) < 3: print(json_response({'success': False, 'error': 'Usage: vectorbt_provider.py '})) return command = sys.argv[1] json_args = sys.argv[2] # Redirect stdout to a buffer during execution so that any stray # print() calls from libraries (numpy, pandas, yfinance, vectorbt) # don't corrupt the JSON output. We capture them and only emit the # final JSON response on the real stdout. real_stdout = sys.stdout captured = io.StringIO() try: sys.stdout = captured print(f'[PY-MAIN] command={command}, args_len={len(json_args)}', file=sys.stderr) args = parse_json_input(json_args) print(f'[PY-MAIN] parsed args keys: {sorted(args.keys()) if isinstance(args, dict) else type(args)}', file=sys.stderr) provider = VectorBTProvider() provider._broker_candles = args.get("brokerCandles") or {} provider._broker_data_source = args.get("brokerDataSource", "") if command == 'test_import': vbt = provider._import_vbt() result_str = json_response({'success': True, 'version': vbt.__version__}) elif command == 'test_connection': result = provider.test_connection() result_str = json_response(result) elif command == 'initialize': result = provider.initialize(args) result_str = json_response(result) elif command == 'run_backtest': print(f'[PY-MAIN] Calling run_backtest...', file=sys.stderr) result = provider.run_backtest(args) print(f'[PY-MAIN] run_backtest returned, success={result.get("success")}', file=sys.stderr) if result.get('data'): perf = result['data'].get('performance') if isinstance(result.get('data'), dict) else None if perf: print(f'[PY-MAIN] result performance (BEFORE json_response): {perf}', file=sys.stderr) result_str = json_response(result) print(f'[PY-MAIN] json_response length: {len(result_str)}', file=sys.stderr) print(f'[PY-MAIN] json_response (first 500): {result_str[:500]}', file=sys.stderr) elif command == 'optimize': result = provider.optimize(args) result_str = json_response(result) elif command == 'walk_forward': result = provider.walk_forward(args) result_str = json_response(result) elif command == 'get_strategies': result = provider.get_strategies(args) result_str = json_response(result) elif command == 'get_indicators': result = provider.get_indicators(args) result_str = json_response(result) elif command == 'get_command_options': result = provider.get_command_options(args) result_str = json_response(result) elif command == 'generate_signals': result = provider.generate_signals(args) result_str = json_response(result) elif command == 'generate_labels': result = provider.generate_labels(args) result_str = json_response(result) elif command == 'generate_splits': result = provider.generate_splits(args) result_str = json_response(result) elif command == 'get_historical_data': result = provider.get_historical_data(args) result_str = json_response({'success': True, 'data': result}) elif command == 'analyze_returns': result = provider.analyze_returns(args) result_str = json_response(result) elif command == 'indicator_signals': result = provider.indicator_signals(args) result_str = json_response(result) elif command == 'calculate_indicator': result = provider.calculate_indicator(args) result_str = json_response(result) elif command == 'indicator_param_sweep': print("[MAIN] About to call indicator_param_sweep") sys.stdout.flush() result = provider.indicator_param_sweep(args) print(f"[MAIN] indicator_param_sweep returned: {result}") sys.stdout.flush() result_str = json_response(result) elif command == 'labels_to_signals': result = provider.labels_to_signals(args) result_str = json_response(result) else: result_str = json_response({'success': False, 'error': f'Unknown command: {command}'}) # Emit only the JSON on the real stdout sys.stdout = real_stdout print(result_str) except Exception as e: sys.stdout = real_stdout # Dump captured output to stderr for debugging stray_output = captured.getvalue() if stray_output: print(stray_output, file=sys.stderr) print(json_response({ 'success': False, 'error': str(e), 'traceback': __import__('traceback').format_exc() })) if __name__ == '__main__': main()