diff --git a/configs/config_bbrs.json b/configs/config_bbrs.json index 6184216..d9bfbc5 100644 --- a/configs/config_bbrs.json +++ b/configs/config_bbrs.json @@ -3,7 +3,6 @@ "stop_date": "2025-03-15", "initial_usd": 10000, "timeframes": ["1min"], - "stop_loss_pcts": [0.05], "strategies": [ { "name": "bbrs", @@ -17,7 +16,8 @@ "sideways_rsi_threshold": [40, 60], "sideways_bb_multiplier": 1.8, "strategy_name": "MarketRegimeStrategy", - "SqueezeStrategy": true + "SqueezeStrategy": true, + "stop_loss_pct": 0.05 } } ], diff --git a/configs/config_combined.json b/configs/config_combined.json index cb20b38..274afac 100644 --- a/configs/config_combined.json +++ b/configs/config_combined.json @@ -3,12 +3,14 @@ "stop_date": "2025-03-15", "initial_usd": 10000, "timeframes": ["15min"], - "stop_loss_pcts": [0.04], "strategies": [ { "name": "default", "weight": 0.6, - "params": {} + "params": { + "timeframe": "15min", + "stop_loss_pct": 0.03 + } }, { "name": "bbrs", @@ -17,7 +19,13 @@ "bb_width": 0.05, "bb_period": 20, "rsi_period": 14, - "strategy_name": "MarketRegimeStrategy" + "trending_rsi_threshold": [30, 70], + "trending_bb_multiplier": 2.5, + "sideways_rsi_threshold": [40, 60], + "sideways_bb_multiplier": 1.8, + "strategy_name": "MarketRegimeStrategy", + "SqueezeStrategy": true, + "stop_loss_pct": 0.05 } } ], diff --git a/configs/config_default.json b/configs/config_default.json index e547c20..dd0bf23 100644 --- a/configs/config_default.json +++ b/configs/config_default.json @@ -3,12 +3,14 @@ "stop_date": "2025-03-15", "initial_usd": 10000, "timeframes": ["15min"], - "stop_loss_pcts": [0.03, 0.05], "strategies": [ { "name": "default", "weight": 1.0, - "params": {} + "params": { + "timeframe": "15min", + "stop_loss_pct": 0.03 + } } ], "combination_rules": { diff --git a/configs/config_default_5min.json b/configs/config_default_5min.json index 9b64437..7c31018 100644 --- a/configs/config_default_5min.json +++ b/configs/config_default_5min.json @@ -3,13 +3,13 @@ "stop_date": "2024-01-31", "initial_usd": 10000, "timeframes": ["5min"], - "stop_loss_pcts": [0.03, 0.05], "strategies": [ { "name": "default", "weight": 1.0, "params": { - "timeframe": "5min" + "timeframe": "5min", + "stop_loss_pct": 0.03 } } ], diff --git a/main.py b/main.py index 37bbfe5..df0f752 100644 --- a/main.py +++ b/main.py @@ -38,183 +38,169 @@ def strategy_manager_exit(backtester: Backtest, df_index: int): """Strategy Manager exit function""" return backtester.strategy_manager.get_exit_signal(backtester, df_index) -def process_timeframe_data(min1_df, df, stop_loss_pcts, rule_name, initial_usd, strategy_config=None, debug=False): - """Process the entire timeframe with all stop loss values using Strategy Manager""" +def process_timeframe_data(data_1min, timeframe, config, debug=False): + """Process a timeframe using Strategy Manager with configuration""" results_rows = [] trade_rows = [] - for stop_loss_pct in stop_loss_pcts: - # Create and initialize strategy manager - if strategy_config: - # Use provided strategy configuration - strategy_manager = create_strategy_manager(strategy_config) - else: - # Default to single default strategy for backward compatibility - default_strategy_config = { - "strategies": [ - { - "name": "default", - "weight": 1.0, - "params": {"stop_loss_pct": stop_loss_pct} - } - ], - "combination_rules": { - "entry": "any", - "exit": "any", - "min_confidence": 0.5 - } - } - strategy_manager = create_strategy_manager(default_strategy_config) - - # Inject stop_loss_pct into all strategy params if not present - for strategy in strategy_manager.strategies: - if "stop_loss_pct" not in strategy.params: - strategy.params["stop_loss_pct"] = stop_loss_pct - - # Get the primary timeframe from the first strategy for backtester setup - primary_strategy = strategy_manager.strategies[0] - primary_timeframe = primary_strategy.get_timeframes()[0] - - # For BBRS strategy, it works with 1-minute data directly and handles internal resampling - # For other strategies, use their preferred timeframe - if primary_strategy.name == "bbrs": - # BBRS strategy processes 1-minute data and outputs signals on its internal timeframes - # Use 1-minute data for backtester working dataframe - working_df = min1_df.copy() - else: - # Other strategies specify their preferred timeframe - # Create backtester working data from the primary strategy's primary timeframe - temp_backtester = type('temp', (), {})() - temp_backtester.original_df = min1_df - - # Let the primary strategy resample the data to get the working dataframe - primary_strategy._resample_data(min1_df) - working_df = primary_strategy.get_primary_timeframe_data() - - # Prepare working dataframe for backtester (ensure timestamp column) - working_df_for_backtest = working_df.copy().reset_index() - if 'index' in working_df_for_backtest.columns: - working_df_for_backtest = working_df_for_backtest.rename(columns={'index': 'timestamp'}) - - # Initialize backtest with strategy manager initialization - backtester = Backtest(initial_usd, working_df_for_backtest, working_df_for_backtest, strategy_manager_init) - - # Store original min1_df for strategy processing - backtester.original_df = min1_df - - # Attach strategy manager to backtester and initialize - backtester.strategy_manager = strategy_manager - strategy_manager.initialize(backtester) + # Extract values from config + initial_usd = config['initial_usd'] + strategy_config = { + "strategies": config['strategies'], + "combination_rules": config['combination_rules'] + } - # Run backtest with strategy manager functions - results = backtester.run( - strategy_manager_entry, - strategy_manager_exit, - debug - ) + # Create and initialize strategy manager + if not strategy_config: + logging.error("No strategy configuration provided") + return results_rows, trade_rows + + strategy_manager = create_strategy_manager(strategy_config) + + # Get the primary timeframe from the first strategy for backtester setup + primary_strategy = strategy_manager.strategies[0] + primary_timeframe = primary_strategy.get_timeframes()[0] + + # For BBRS strategy, it works with 1-minute data directly and handles internal resampling + # For other strategies, use their preferred timeframe + if primary_strategy.name == "bbrs": + # BBRS strategy processes 1-minute data and outputs signals on its internal timeframes + # Use 1-minute data for backtester working dataframe + working_df = data_1min.copy() + else: + # Other strategies specify their preferred timeframe + # Let the primary strategy resample the data to get the working dataframe + primary_strategy._resample_data(data_1min) + working_df = primary_strategy.get_primary_timeframe_data() + + # Prepare working dataframe for backtester (ensure timestamp column) + working_df_for_backtest = working_df.copy().reset_index() + if 'index' in working_df_for_backtest.columns: + working_df_for_backtest = working_df_for_backtest.rename(columns={'index': 'timestamp'}) + + # Initialize backtest with strategy manager initialization + backtester = Backtest(initial_usd, working_df_for_backtest, working_df_for_backtest, strategy_manager_init) + + # Store original min1_df for strategy processing + backtester.original_df = data_1min + + # Attach strategy manager to backtester and initialize + backtester.strategy_manager = strategy_manager + strategy_manager.initialize(backtester) - n_trades = results["n_trades"] - trades = results.get('trades', []) - wins = [1 for t in trades if t['exit'] is not None and t['exit'] > t['entry']] - n_winning_trades = len(wins) - total_profit = sum(trade['profit_pct'] for trade in trades) - total_loss = sum(-trade['profit_pct'] for trade in trades if trade['profit_pct'] < 0) - win_rate = n_winning_trades / n_trades if n_trades > 0 else 0 - avg_trade = total_profit / n_trades if n_trades > 0 else 0 - profit_ratio = total_profit / total_loss if total_loss > 0 else float('inf') - cumulative_profit = 0 - max_drawdown = 0 - peak = 0 + # Run backtest with strategy manager functions + results = backtester.run( + strategy_manager_entry, + strategy_manager_exit, + debug + ) - for trade in trades: - cumulative_profit += trade['profit_pct'] + n_trades = results["n_trades"] + trades = results.get('trades', []) + wins = [1 for t in trades if t['exit'] is not None and t['exit'] > t['entry']] + n_winning_trades = len(wins) + total_profit = sum(trade['profit_pct'] for trade in trades) + total_loss = sum(-trade['profit_pct'] for trade in trades if trade['profit_pct'] < 0) + win_rate = n_winning_trades / n_trades if n_trades > 0 else 0 + avg_trade = total_profit / n_trades if n_trades > 0 else 0 + profit_ratio = total_profit / total_loss if total_loss > 0 else float('inf') + cumulative_profit = 0 + max_drawdown = 0 + peak = 0 - if cumulative_profit > peak: - peak = cumulative_profit - drawdown = peak - cumulative_profit + for trade in trades: + cumulative_profit += trade['profit_pct'] - if drawdown > max_drawdown: - max_drawdown = drawdown + if cumulative_profit > peak: + peak = cumulative_profit + drawdown = peak - cumulative_profit - final_usd = initial_usd + if drawdown > max_drawdown: + max_drawdown = drawdown - for trade in trades: - final_usd *= (1 + trade['profit_pct']) + final_usd = initial_usd - total_fees_usd = sum(trade.get('fee_usd', 0.0) for trade in trades) + for trade in trades: + final_usd *= (1 + trade['profit_pct']) - # Update row to include timeframe information - row = { - "timeframe": f"{rule_name}({primary_timeframe})", # Show actual timeframe used + total_fees_usd = sum(trade.get('fee_usd', 0.0) for trade in trades) + + # Get stop_loss_pct from the first strategy for reporting + # In multi-strategy setups, strategies can have different stop_loss_pct values + stop_loss_pct = primary_strategy.params.get("stop_loss_pct", "N/A") + + # Update row to include timeframe information + row = { + "timeframe": f"{timeframe}({primary_timeframe})", # Show actual timeframe used + "stop_loss_pct": stop_loss_pct, + "n_trades": n_trades, + "n_stop_loss": sum(1 for trade in trades if 'type' in trade and trade['type'] == 'STOP_LOSS'), + "win_rate": win_rate, + "max_drawdown": max_drawdown, + "avg_trade": avg_trade, + "total_profit": total_profit, + "total_loss": total_loss, + "profit_ratio": profit_ratio, + "initial_usd": initial_usd, + "final_usd": final_usd, + "total_fees_usd": total_fees_usd, + } + results_rows.append(row) + + for trade in trades: + trade_rows.append({ + "timeframe": f"{timeframe}({primary_timeframe})", "stop_loss_pct": stop_loss_pct, - "n_trades": n_trades, - "n_stop_loss": sum(1 for trade in trades if 'type' in trade and trade['type'] == 'STOP_LOSS'), - "win_rate": win_rate, - "max_drawdown": max_drawdown, - "avg_trade": avg_trade, - "total_profit": total_profit, - "total_loss": total_loss, - "profit_ratio": profit_ratio, - "initial_usd": initial_usd, - "final_usd": final_usd, - "total_fees_usd": total_fees_usd, - } - results_rows.append(row) - - for trade in trades: - trade_rows.append({ - "timeframe": f"{rule_name}({primary_timeframe})", - "stop_loss_pct": stop_loss_pct, - "entry_time": trade.get("entry_time"), - "exit_time": trade.get("exit_time"), - "entry_price": trade.get("entry"), - "exit_price": trade.get("exit"), - "profit_pct": trade.get("profit_pct"), - "type": trade.get("type"), - "fee_usd": trade.get("fee_usd"), - }) - - # Log strategy summary - strategy_summary = strategy_manager.get_strategy_summary() - logging.info(f"Timeframe: {rule_name}({primary_timeframe}), Stop Loss: {stop_loss_pct}, " - f"Trades: {n_trades}, Strategies: {[s['name'] for s in strategy_summary['strategies']]}") - - if debug: - # Plot after each backtest run - try: - # Check if any strategy has processed_data for universal plotting - processed_data = None - for strategy in strategy_manager.strategies: - if hasattr(backtester, 'processed_data') and backtester.processed_data is not None: - processed_data = backtester.processed_data - break - - if processed_data is not None and not processed_data.empty: - # Format strategy data with actual executed trades for universal plotting - formatted_data = BacktestCharts.format_strategy_data_with_trades(processed_data, results) - # Plot using universal function - BacktestCharts.plot_data(formatted_data) + "entry_time": trade.get("entry_time"), + "exit_time": trade.get("exit_time"), + "entry_price": trade.get("entry"), + "exit_price": trade.get("exit"), + "profit_pct": trade.get("profit_pct"), + "type": trade.get("type"), + "fee_usd": trade.get("fee_usd"), + }) + + # Log strategy summary + strategy_summary = strategy_manager.get_strategy_summary() + logging.info(f"Timeframe: {timeframe}({primary_timeframe}), Stop Loss: {stop_loss_pct}, " + f"Trades: {n_trades}, Strategies: {[s['name'] for s in strategy_summary['strategies']]}") + + if debug: + # Plot after each backtest run + try: + # Check if any strategy has processed_data for universal plotting + processed_data = None + for strategy in strategy_manager.strategies: + if hasattr(backtester, 'processed_data') and backtester.processed_data is not None: + processed_data = backtester.processed_data + break + + if processed_data is not None and not processed_data.empty: + # Format strategy data with actual executed trades for universal plotting + formatted_data = BacktestCharts.format_strategy_data_with_trades(processed_data, results) + # Plot using universal function + BacktestCharts.plot_data(formatted_data) + else: + # Fallback to meta_trend plot if available + if "meta_trend" in backtester.strategies: + meta_trend = backtester.strategies["meta_trend"] + # Use the working dataframe for plotting + BacktestCharts.plot(working_df, meta_trend) else: - # Fallback to meta_trend plot if available - if "meta_trend" in backtester.strategies: - meta_trend = backtester.strategies["meta_trend"] - # Use the working dataframe for plotting - BacktestCharts.plot(working_df, meta_trend) - else: - print("No plotting data available") - except Exception as e: - print(f"Plotting failed: {e}") + print("No plotting data available") + except Exception as e: + print(f"Plotting failed: {e}") return results_rows, trade_rows def process(timeframe_info, debug=False): - """Process a single (timeframe, stop_loss_pct) combination with strategy config""" - rule, data_1min, stop_loss_pct, initial_usd, strategy_config = timeframe_info + """Process a single timeframe with strategy config""" + timeframe, data_1min, config = timeframe_info - # Pass the original 1-minute data - strategies will handle their own timeframe resampling + # Pass the essential data and full config results_rows, all_trade_rows = process_timeframe_data( - data_1min, data_1min, [stop_loss_pct], rule, initial_usd, strategy_config, debug=debug + data_1min, timeframe, config, debug=debug ) return results_rows, all_trade_rows @@ -272,68 +258,25 @@ if __name__ == "__main__": parser.add_argument("config", type=str, nargs="?", help="Path to config JSON file.") args = parser.parse_args() - # Default values (from config.json) - default_config = { - "start_date": "2025-03-01", - "stop_date": datetime.datetime.today().strftime('%Y-%m-%d'), - "initial_usd": 10000, - "timeframes": ["15min"], - "stop_loss_pcts": [0.03], - "strategies": [ - { - "name": "default", - "weight": 1.0, - "params": {} - } - ], - "combination_rules": { - "entry": "any", - "exit": "any", - "min_confidence": 0.5 - } - } - - if args.config: - with open(args.config, 'r') as f: + # Use config_default.json as fallback if no config provided + config_file = args.config or "configs/config_default.json" + + try: + with open(config_file, 'r') as f: config = json.load(f) - elif not debug: - print("No config file provided. Please enter the following values (press Enter to use default):") - - start_date = input(f"Start date [{default_config['start_date']}]: ") or default_config['start_date'] - stop_date = input(f"Stop date [{default_config['stop_date']}]: ") or default_config['stop_date'] - - initial_usd_str = input(f"Initial USD [{default_config['initial_usd']}]: ") or str(default_config['initial_usd']) - initial_usd = float(initial_usd_str) - - timeframes_str = input(f"Timeframes (comma separated) [{', '.join(default_config['timeframes'])}]: ") or ','.join(default_config['timeframes']) - timeframes = [tf.strip() for tf in timeframes_str.split(',') if tf.strip()] - - stop_loss_pcts_str = input(f"Stop loss pcts (comma separated) [{', '.join(str(x) for x in default_config['stop_loss_pcts'])}]: ") or ','.join(str(x) for x in default_config['stop_loss_pcts']) - stop_loss_pcts = [float(x.strip()) for x in stop_loss_pcts_str.split(',') if x.strip()] - - config = { - 'start_date': start_date, - 'stop_date': stop_date, - 'initial_usd': initial_usd, - 'timeframes': timeframes, - 'stop_loss_pcts': stop_loss_pcts, - 'strategies': default_config['strategies'], - 'combination_rules': default_config['combination_rules'] - } - else: - config = default_config + print(f"Using config: {config_file}") + except FileNotFoundError: + print(f"Error: Config file '{config_file}' not found.") + print("Available configs: configs/config_default.json, configs/config_bbrs.json, configs/config_combined.json") + exit(1) + except json.JSONDecodeError as e: + print(f"Error: Invalid JSON in config file '{config_file}': {e}") + exit(1) start_date = config['start_date'] stop_date = config['stop_date'] initial_usd = config['initial_usd'] timeframes = config['timeframes'] - stop_loss_pcts = config['stop_loss_pcts'] - - # Extract strategy configuration - strategy_config = { - "strategies": config.get('strategies', default_config['strategies']), - "combination_rules": config.get('combination_rules', default_config['combination_rules']) - } timestamp = datetime.datetime.now().strftime("%Y_%m_%d_%H_%M") @@ -351,11 +294,10 @@ if __name__ == "__main__": f"Initial USD\t{initial_usd}" ] - # Create tasks for each (timeframe, stop_loss_pct) combination + # Create tasks for each timeframe tasks = [ - (name, data_1min, stop_loss_pct, initial_usd, strategy_config) + (name, data_1min, config) for name in timeframes - for stop_loss_pct in stop_loss_pcts ] if debug: