removed strategy stuff from here
This commit is contained in:
Vasily.onl
parent
9629d3090b
commit
f67b6b8ebd
206
main.py
206
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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user