diff --git a/main.py b/main.py index 6aa47c7..37bbfe5 100644 --- a/main.py +++ b/main.py @@ -24,212 +24,6 @@ logging.basicConfig( ] ) -def default_init_strategy(backtester: Backtest): - """Calculate meta trend - """ - supertrends = Supertrends(backtester.df, verbose=False) - - supertrend_results_list = supertrends.calculate_supertrend_indicators() - trends = [st['results']['trend'] for st in supertrend_results_list] - trends_arr = np.stack(trends, axis=1) - meta_trend = np.where((trends_arr[:,0] == trends_arr[:,1]) & (trends_arr[:,1] == trends_arr[:,2]), - trends_arr[:,0], 0) - - backtester.strategies["meta_trend"] = meta_trend - -def bbrs_init_strategy(backtester: Backtest): - """BBRs entry strategy initialization - just setup basic structure""" - # Initialize empty strategies - backtester.strategies["buy_signals"] = pd.Series(False, index=range(len(backtester.df))) - backtester.strategies["sell_signals"] = pd.Series(False, index=range(len(backtester.df))) - return backtester - -def run_bbrs_strategy_processing(backtester: Backtest, original_df): - """Run the actual strategy processing after backtest is initialized""" - config_strategy = { - "bb_width": 0.05, - "bb_period": 20, - "rsi_period": 14, - "trending": { - "rsi_threshold": [30, 70], - "bb_std_dev_multiplier": 2.5, - }, - "sideways": { - "rsi_threshold": [40, 60], - "bb_std_dev_multiplier": 1.8, - }, - "strategy_name": "MarketRegimeStrategy", # "MarketRegimeStrategy", # CryptoTradingStrategy - "SqueezeStrategy": True - } - - strategy = Strategy(config=config_strategy, logging=logging) - processed_data = strategy.run(original_df, config_strategy["strategy_name"]) - print(f"processed_data: {processed_data.head()}") - - # Store processed data for plotting - backtester.processed_data = processed_data - - if processed_data.empty: - # If strategy processing failed, create empty signals aligned with backtest DataFrame - buy_condition = pd.Series(False, index=range(len(backtester.df))) - sell_condition = pd.Series(False, index=range(len(backtester.df))) - else: - # Get original signals from processed data - buy_signals_raw = processed_data.get('BuySignal', pd.Series(False, index=processed_data.index)).astype(bool) - sell_signals_raw = processed_data.get('SellSignal', pd.Series(False, index=processed_data.index)).astype(bool) - - # Get the DatetimeIndex from the original 1-minute data - original_datetime_index = original_df.index - - # Reindex signals from 15-minute to 1-minute resolution using forward-fill - # This maps each 15-minute signal to the corresponding 1-minute timestamps - buy_signals_1min = buy_signals_raw.reindex(original_datetime_index, method='ffill').fillna(False) - sell_signals_1min = sell_signals_raw.reindex(original_datetime_index, method='ffill').fillna(False) - - # Convert to integer index to match backtest DataFrame - buy_condition = pd.Series(buy_signals_1min.values, index=range(len(buy_signals_1min))) - sell_condition = pd.Series(sell_signals_1min.values, index=range(len(sell_signals_1min))) - - # Ensure same length as backtest DataFrame (should be same now, but safety check) - if len(buy_condition) != len(backtester.df): - target_length = len(backtester.df) - if len(buy_condition) > target_length: - buy_condition = buy_condition[:target_length] - sell_condition = sell_condition[:target_length] - else: - # Pad with False if shorter - buy_values = buy_condition.values - sell_values = sell_condition.values - buy_values = np.pad(buy_values, (0, target_length - len(buy_values)), constant_values=False) - sell_values = np.pad(sell_values, (0, target_length - len(sell_values)), constant_values=False) - buy_condition = pd.Series(buy_values, index=range(target_length)) - sell_condition = pd.Series(sell_values, index=range(target_length)) - - backtester.strategies["buy_signals"] = buy_condition - backtester.strategies["sell_signals"] = sell_condition - # backtester.strategies["buy_signals"] = sell_condition - # backtester.strategies["sell_signals"] = buy_condition - - print(f"buy_signals length: {len(backtester.strategies['buy_signals'])}, backtest df length: {len(backtester.df)}") - -def bbrs_entry_strategy(backtester: Backtest, df_index): - """BBRs entry strategy - Entry when buy signal is true - """ - return backtester.strategies["buy_signals"].iloc[df_index] - -def bbrs_exit_strategy(backtester: Backtest, df_index): - """BBRs exit strategy - Exit when sell signal is true or stop loss is triggered - """ - if backtester.strategies["sell_signals"].iloc[df_index]: - return "SELL_SIGNAL", backtester.df.iloc[df_index]['close'] - - # Check for stop loss using BBRs-specific stop loss strategy - stop_loss_result, sell_price = bbrs_stop_loss_strategy(backtester) - if stop_loss_result: - backtester.strategies["current_trade_min1_start_idx"] = \ - backtester.current_trade_min1_start_idx - return "STOP_LOSS", sell_price - - return None, None - -def bbrs_stop_loss_strategy(backtester: Backtest): - """BBRs stop loss strategy - Calculate stop loss price based on 5% loss - Find the first min1 candle that is below the stop loss price - If the stop loss price is below the open price, use the open price as the stop loss price - """ - # Use 5% stop loss as requested - stop_loss_pct = 0.05 - stop_price = backtester.entry_price * (1 - stop_loss_pct) - - # Use the original min1 dataframe that has datetime index - min1_df = backtester.original_df if hasattr(backtester, 'original_df') else backtester.min1_df - min1_index = min1_df.index - - # Find candles from entry time to current time - start_candidates = min1_index[min1_index >= backtester.entry_time] - if len(start_candidates) == 0: - return False, None - - backtester.current_trade_min1_start_idx = start_candidates[0] - end_candidates = min1_index[min1_index <= backtester.current_date] - - if len(end_candidates) == 0: - print("Warning: no end candidate here. Need to be checked") - return False, None - backtester.current_min1_end_idx = end_candidates[-1] - - # Get the slice of data between entry and current time - min1_slice = min1_df.loc[backtester.current_trade_min1_start_idx:backtester.current_min1_end_idx] - - # Check if any candle's low price hits the stop loss - if (min1_slice['low'] <= stop_price).any(): - stop_candle = min1_slice[min1_slice['low'] <= stop_price].iloc[0] - - # If the candle opened below stop price, use open price; otherwise use stop price - if stop_candle['open'] < stop_price: - sell_price = stop_candle['open'] - else: - sell_price = stop_price - return True, sell_price - - return False, None - -def default_entry_strategy(backtester: Backtest, df_index): - """Entry strategy - Entry when meta trend is 1 - """ - return backtester.strategies["meta_trend"][df_index - 1] != 1 and backtester.strategies["meta_trend"][df_index] == 1 - -def stop_loss_strategy(backtester: Backtest): - """Stop loss strategy - Calculate stop loss price - Find the first min1 candle that is below the stop loss price - If the stop loss price is below the open price, use the open price as the stop loss price - """ - stop_price = backtester.entry_price * (1 - backtester.strategies["stop_loss_pct"]) - - min1_index = backtester.min1_df.index - start_candidates = min1_index[min1_index >= backtester.entry_time] - backtester.current_trade_min1_start_idx = start_candidates[0] - end_candidates = min1_index[min1_index <= backtester.current_date] - - if len(end_candidates) == 0: - print("Warning: no end candidate here. Need to be checked") - return False, None - backtester.current_min1_end_idx = end_candidates[-1] - - min1_slice = backtester.min1_df.loc[backtester.current_trade_min1_start_idx:backtester.current_min1_end_idx] - - # print(f"lowest low in that range: {min1_slice['low'].min()}, count: {len(min1_slice)}") - # print(f"slice start: {min1_slice.index[0]}, slice end: {min1_slice.index[-1]}") - - if (min1_slice['low'] <= stop_price).any(): - stop_candle = min1_slice[min1_slice['low'] <= stop_price].iloc[0] - - if stop_candle['open'] < stop_price: - sell_price = stop_candle['open'] - else: - sell_price = stop_price - return True, sell_price - - return False, None - -def default_exit_strategy(backtester: Backtest, df_index): - if backtester.strategies["meta_trend"][df_index - 1] != 1 and \ - backtester.strategies["meta_trend"][df_index] == -1: - return "META_TREND_EXIT_SIGNAL", None - - stop_loss_result, sell_price = stop_loss_strategy(backtester) - if stop_loss_result: - backtester.strategies["current_trade_min1_start_idx"] = \ - backtester.min1_df.index[backtester.min1_df.index <= backtester.current_date][-1] - return "STOP_LOSS", sell_price - - return None, None - def strategy_manager_init(backtester: Backtest): """Strategy Manager initialization function""" # This will be called by Backtest.__init__, but actual initialization