From 4f98924295dd636fc908397f6b5664ffd4b5ba7e Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Fri, 11 Jul 2025 01:57:21 +0800 Subject: [PATCH] add more stuff --- .DS_Store | Bin 10244 -> 10244 bytes ChanLun.py | 14 +- strategies/BB9033.py | 280 +++++++++++++++++++++++++ strategies/ChanLun_BTC_30.py | 219 +++++++++++++------ strategies/bollinger_atr_strategy.pine | 136 ++++++++++++ 5 files changed, 572 insertions(+), 77 deletions(-) create mode 100644 strategies/BB9033.py create mode 100644 strategies/bollinger_atr_strategy.pine diff --git a/.DS_Store b/.DS_Store index 8504d65df196ca66845da4e2998753cd71bc1530..f2cbfe54ffb67215df8d33cbc8fd0b9bdb5511d5 100644 GIT binary patch delta 134 zcmZn(XbG6$ÄU^hRb#%3OYS?oMX<;4X_Ir&Kp3=F#_I|vqPN>x{znwjY+7#kSY z>L^qj8ktz=C>R=87}eHta)>JHTL;Bw=j7()_e{PmD9_kCxldeykr7Br@RS7?<>ln( Vr2~y(+`L{qjD0h^!h8%Hf&gu%Cu0Br delta 57 zcmZn(XbG6$OBU^hRb+GZYsS?ruC#mPBI`T02vn>R{?v2Sc(V%*HGFds>L@ DataFrame: + """ + 计算技术指标 + """ + dataframe_3 = resample_to_interval(dataframe, self.get_ticker_indicator() * self.time3) + + # 计算ATR（用于止损计算） + dataframe_3['atr'] = ta.ATR(dataframe_3, timeperiod=self.atr_length) + + # 计算布林带（使用标准方法：移动平均线 ± 标准差倍数） + bb_upper, bb_middle, bb_lower = ta.BBANDS(dataframe_3['close'], timeperiod=self.bb_length, nbdevup=self.atr_multiplier, nbdevdn=self.atr_multiplier, matype=0) + dataframe_3['bb_upper'] = bb_upper + dataframe_3['bb_lower'] = bb_lower + + # 计算突破条件（与Pine Script保持一致） + # 确保所有用于计算的数据都不是NaN + valid_data = ( + dataframe_3['close'].notna() & + dataframe_3['bb_upper'].notna() & + dataframe_3['bb_lower'].notna() & + dataframe_3['close'].shift(1).notna() & + dataframe_3['bb_upper'].shift(1).notna() & + dataframe_3['bb_lower'].shift(1).notna() + ) + + dataframe_3['break_above_upper'] = ( + (dataframe_3['close'] > dataframe_3['bb_upper']) & + (dataframe_3['close'].shift(1) <= dataframe_3['bb_upper'].shift(1)) & + valid_data + ) + dataframe_3['break_below_lower'] = ( + (dataframe_3['close'] < dataframe_3['bb_lower']) & + (dataframe_3['close'].shift(1) >= dataframe_3['bb_lower'].shift(1)) & + valid_data + ) + dataframe = resampled_merge(dataframe, dataframe_3) + return dataframe + + def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + """ + Based on TA indicators, populates the entry trend columns + """ + break_below_lower = 'resample_{}_break_below_lower'.format(self.get_ticker_indicator()*self.time3) + break_above_upper = 'resample_{}_break_above_upper'.format(self.get_ticker_indicator()*self.time3) + + # 做多条件：价格跌破下轨 + dataframe.loc[ + ( + (dataframe[break_below_lower] == True) & # 价格跌破下轨，明确检查True值 + (dataframe[break_below_lower].notna()) # 确保不是NaN + ), + 'enter_long'] = 1 + + # 做空条件：价格突破上轨 + dataframe.loc[ + ( + (dataframe[break_above_upper] == True) & # 价格突破上轨，明确检查True值 + (dataframe[break_above_upper].notna()) # 确保不是NaN + ), + 'enter_short'] = 1 + + return dataframe + + def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: + """ + Based on TA indicators, populates the exit trend columns + """ + bb_upper_str = 'resample_{}_bb_upper'.format(self.get_ticker_indicator()*self.time3) + bb_lower_str = 'resample_{}_bb_lower'.format(self.get_ticker_indicator()*self.time3) + + # 多头止盈：价格突破上轨（与Pine Script一致） + dataframe.loc[ + ( + (dataframe['close'] > dataframe[bb_upper_str]) & # 当前价格突破上轨 + (dataframe[bb_upper_str].notna()) & # 确保布林带上轨不是NaN + (dataframe['close'].notna()) # 确保收盘价不是NaN + ), + 'exit_long'] = 1 + + # 空头止盈：价格跌破下轨（与Pine Script一致） + dataframe.loc[ + ( + (dataframe['close'] < dataframe[bb_lower_str]) & # 当前价格跌破下轨 + (dataframe[bb_lower_str].notna()) & # 确保布林带下轨不是NaN + (dataframe['close'].notna()) # 确保收盘价不是NaN + ), + 'exit_short'] = 1 + + return dataframe + + def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, + current_rate: float, current_profit: float, + after_fill: bool, **kwargs) -> float: + """ + 自定义止损逻辑：使用开单时记录的ATR止损价格 + """ + + # 检查是否有存储的止损价格 + trade_id = str(trade.id) + if trade_id not in self.trade_stop_prices: + return self.stoploss + + # 获取最新的K线数据 + dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) + if len(dataframe) < 2: + return self.stoploss + + # 使用上一个K线的收盘价 + last_close = dataframe.iloc[-2]['close'] # 上一个完整K线的收盘价 + + stop_price = self.trade_stop_prices[trade_id] + + if trade.is_short: + # 空头止损：上一个K线收盘价超过止损价格时触发止损 + if last_close >= stop_price: + stop_loss_pct = -abs((last_close - stop_price) / last_close) + else: + stop_loss_pct = 1.0 # 不触发止损 + else: + # 多头止损：上一个K线收盘价低于止损价格时触发止损 + if last_close <= stop_price: + stop_loss_pct = -abs((stop_price - last_close) / last_close) + else: + stop_loss_pct = 1.0 # 不触发止损 + + # 确保止损不会比默认止损更宽松 + return max(stop_loss_pct, self.stoploss) + + def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, + time_in_force: str, current_time: datetime, entry_tag: str, + side: str, **kwargs) -> bool: + """ + 确认交易进场 + """ + + # 获取最新数据进行最终确认 + dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) + + if len(dataframe) == 0: + return False + + latest_candle = dataframe.iloc[-1] + + # 使用重采样后的字段名 + bb_upper_str = 'resample_{}_bb_upper'.format(self.get_ticker_indicator() * self.time3) + bb_lower_str = 'resample_{}_bb_lower'.format(self.get_ticker_indicator() * self.time3) + atr_str = 'resample_{}_atr'.format(self.get_ticker_indicator() * self.time3) + + # 确保技术指标有效 + if (np.isnan(latest_candle[bb_upper_str]) or + np.isnan(latest_candle[bb_lower_str]) or + np.isnan(latest_candle[atr_str])): + return False + + return True + + def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, + **kwargs) -> None: + """ + 当订单填充时的回调函数 + 在开仓时记录基于开单时ATR的止损价格 + """ + + # 处理开仓订单（包括做多和做空） + if (order.ft_order_side == 'buy' or order.ft_order_side == 'sell') and trade.is_open: + # 获取开仓时的数据 + dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) + + if len(dataframe) == 0: + return + + # 获取开仓时的ATR值 + atr_str = 'resample_{}_atr'.format(self.get_ticker_indicator() * self.time3) + + # 找到最接近开仓时间的K线 + open_candle = dataframe.iloc[-1] # 使用最新的K线作为开仓时的数据 + atr_value = open_candle[atr_str] + + if not np.isnan(atr_value) and atr_value > 0: + # 计算止损价格并存储 + atr_stop_distance = self.atr_stop_multiplier * atr_value + + if trade.is_short: + # 空头止损：入场价 + ATR止损距离 + stop_price = trade.open_rate + atr_stop_distance + else: + # 多头止损：入场价 - ATR止损距离 + stop_price = trade.open_rate - atr_stop_distance + + # 使用trade_id作为key存储止损价格 + self.trade_stop_prices[str(trade.id)] = stop_price + + #logger.info(f"交易 {trade.id} 开仓，记录止损价格: {stop_price}, ATR: {atr_value}, 开仓价: {trade.open_rate}") + #logger.info(f"{current_time} {pair} {trade.open_rate} {stop_price} {atr_value}") + def trade_exit(self, pair: str, trade: Trade, order: Order, current_time: datetime, + **kwargs) -> None: + """ + 当交易退出时的回调函数 + 清理存储的止损价格记录 + """ + trade_id = str(trade.id) + if trade_id in self.trade_stop_prices: + del self.trade_stop_prices[trade_id] + logger.info(f"交易 {trade.id} 已关闭，清理止损价格记录") + + def get_ticker_indicator(self): + return int(self.timeframe[:-1]) \ No newline at end of file diff --git a/strategies/ChanLun_BTC_30.py b/strategies/ChanLun_BTC_30.py index 18dbdac..ec7aaac 100644 --- a/strategies/ChanLun_BTC_30.py +++ b/strategies/ChanLun_BTC_30.py @@ -22,7 +22,7 @@ logger = logging.getLogger(__name__) # freqtrade plot-dataframe --strategy ChanLun_BTC_30 --datadir user_data/data/binance -c ./user_data/ChanLun_SOL_30.json --timerange=20250309- # freqtrade trade -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies -# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies --timerange=20250520- +# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies --timerange=20250510-20250520 # freqtrade download-data -c ./user_data/Chan/config/ChanLun_BTC_30.json -t 1m --pairs BTC/USDT:USDT --timerange=20250405- # freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy ChanLun_BTC_30 --strategy-path ./user_data/Chan/strategies -c ./user_data/Chan/config/ChanLun_BTC_30.json -e 200 --timerange=20250201-20250401 @@ -37,20 +37,20 @@ class ChanLun_BTC_30(IStrategy): # 30m and 1h minimal_roi = { - "0": 0.60, + "0": 0.15, "360": 0.2, "640": 0.1, "1200": 0 } # 5m and 15m - minimal_roi = { + minimal_roi_1 = { "0": 0.1, "60": 0.05, "120": 0.02, "240": 0 } # 15m and 30m - minimal_roi = { + minimal_roi_1 = { "0": 0.1, "240": 0.05, "480": 0.03, @@ -64,15 +64,16 @@ class ChanLun_BTC_30(IStrategy): } can_short = True lev = 1.0 - stoploss = -0.01 - #use_custom_stoploss = True + stoploss = -0.2 # 设置为很大的负值，让custom_stoploss来控制 + use_custom_stoploss = False # 启用自定义止损 trailing_stop = False trailing_stop_positive = 0.025 trailing_stop_positive_offset = 0.045 trailing_only_offset_is_reached = False - #position_adjustment_enable = True + # 启用仓位调整功能以支持分批止盈 + position_adjustment_enable = True startup_candle_count = 780 time5 = 5 @@ -80,16 +81,18 @@ class ChanLun_BTC_30(IStrategy): time30 = 30 time60 = 60 time4h = 240 - time30 = 60 + time30 = 3 last_time = datetime.now() chan = ChanLun() chanpy = ChanPY() classifier = ChanLunClassifier(None) last_order = None last_trade = None + def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # resample our dataframes + dataframe_3 = resample_to_interval(dataframe, self.get_ticker_indicator() * 3) dataframe_5 = resample_to_interval(dataframe, self.get_ticker_indicator() * 5) dataframe_15 = resample_to_interval(dataframe, self.get_ticker_indicator() * 15) dataframe_30 = resample_to_interval(dataframe, self.get_ticker_indicator() * 30) @@ -104,6 +107,7 @@ class ChanLun_BTC_30(IStrategy): #dataframe_1w = resample_to_interval(dataframe, self.get_ticker_indicator() * 10080) #dataframe_1m = resample_to_interval(dataframe, self.get_ticker_indicator() * 43200) dataframe = self.add_indicators(dataframe) + dataframe_3 = self.add_indicators(dataframe_3) dataframe_5 = self.add_indicators(dataframe_5) dataframe_15 = self.add_indicators(dataframe_15) dataframe_30 = self.add_indicators(dataframe_30) @@ -113,9 +117,9 @@ class ChanLun_BTC_30(IStrategy): #self.chan.plot_dual(dataframe_5, dataframe_30) chanpy_state = self.chanpy.get_bsp_state(dataframe_5) dataframe_5['chanpy_state'] = chanpy_state - state_list = self.chan.get_klc_state_list(dataframe_60) - dataframe_60['state'] = state_list - dataframe_60['fx'] = state_list + state_list = self.chan.get_klc_state_list(dataframe_3) + dataframe_3['state'] = state_list + dataframe_3['fx'] = state_list #bi_list_1 = self.chan.get_bi_list(dataframe) #bi_list_5 = self.chan.get_bi_list(dataframe_5) #bi_list_15 = self.chan.get_bi_list(dataframe_15) @@ -130,6 +134,7 @@ class ChanLun_BTC_30(IStrategy): self.print_seg(dataframe_5) print("-------------------------------------------------------------------------------") self.last_time = datetime.now() + dataframe = resampled_merge(dataframe, dataframe_3) dataframe = resampled_merge(dataframe, dataframe_5) #dataframe = resampled_merge(dataframe, dataframe_15) #dataframe = resampled_merge(dataframe, dataframe_30) @@ -157,12 +162,17 @@ class ChanLun_BTC_30(IStrategy): bb120 = ta.BBANDS(df, timeperiod=120, nbdevup=3.0, nbdevdn=3.0, matype=0) bb30 = ta.BBANDS(df, timeperiod=90, nbdevup=3.0, nbdevdn=3.0, matype=0) bb302 = ta.BBANDS(df, timeperiod=90, nbdevup=2.0, nbdevdn=2.0, matype=0) + + # 计算布林带中轨（移动平均线） + bb30_middle = ta.SMA(df, timeperiod=90) + # 手动计算布林带 %B 指标 (BBP) # %B = (Price - Lower Band) / (Upper Band - Lower Band) bbp365 = (df['close'] - bb365['lowerband']) / (bb365['upperband'] - bb365['lowerband']) bbp120 = (df['close'] - bb120['lowerband']) / (bb120['upperband'] - bb120['lowerband']) bbp30 = (df['close'] - bb30['lowerband']) / (bb30['upperband'] - bb30['lowerband']) bbp302 = (df['close'] - bb302['lowerband']) / (bb302['upperband'] - bb302['lowerband']) + df['atr'] = ta.ATR(df, timeperiod=14) df['bbup365'] = bb365['upperband'] df['bblow365'] = bb365['lowerband'] df['bbp365'] = bbp365 @@ -171,6 +181,7 @@ class ChanLun_BTC_30(IStrategy): df['bbp120'] = bbp120 df['bbup30'] = bb30['upperband'] df['bblow30'] = bb30['lowerband'] + df['bbmiddle30'] = bb30_middle # 添加bb30中轨 df['bbp30'] = bbp30 df['bbup302'] = bb302['upperband'] df['bblow302'] = bb302['lowerband'] @@ -215,6 +226,129 @@ class ChanLun_BTC_30(IStrategy): new_exitprice = proposed_rate - 50 return new_exitprice + def adjust_trade_position(self, trade: Trade, current_time: datetime, + current_rate: float, current_profit: float, + min_stake: Optional[float], max_stake: float, + current_entry_rate: float, current_exit_rate: float, + current_entry_profit: float, current_exit_profit: float, + **kwargs) -> Optional[float]: + """ + 基于布林带的分批止盈逻辑 + """ + # 获取当前数据 + dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) + if dataframe is None or len(dataframe) == 0: + return None + + last_candle = dataframe.iloc[-1] + + # 获取布林带数据 + bb30_middle = last_candle['bbmiddle30'] + bb30_upper = last_candle['bbup30'] + bb30_lower = last_candle['bblow30'] + bb302_upper = last_candle['bbup302'] + bb302_lower = last_candle['bblow302'] + + # 获取交易的状态标记 + first_tp_triggered = trade.get_custom_data(key="first_tp_triggered", default=False) + second_tp_triggered = trade.get_custom_data(key="second_tp_triggered", default=False) + + if trade.is_short: + # 做空逻辑 + if not first_tp_triggered and current_rate <= bb30_middle: + # 第一次止盈：价格跌到bb30中轨，止盈50% + logger.info(f"做空第一次止盈触发：价格{current_rate} <= BB30中轨{bb30_middle}") + trade.set_custom_data(key="first_tp_triggered", value=True) + trade.set_custom_data(key="new_stoploss", value=trade.open_rate) # 设置止损为开仓价 + return -(trade.amount * 0.5) # 减少50%仓位 + + elif first_tp_triggered and not second_tp_triggered and current_rate <= bb302_lower: + # 第二次止盈：继续跌到bb302下轨，止盈剩余仓位的60% + logger.info(f"做空第二次止盈触发：价格{current_rate} <= BB302下轨{bb302_lower}") + trade.set_custom_data(key="second_tp_triggered", value=True) + trade.set_custom_data(key="new_stoploss", value=bb30_middle) # 移动止损到bb30中轨 + remaining_amount = trade.amount * 0.5 # 剩余50% + return -(remaining_amount * 0.6) # 减少剩余仓位的60% + + else: + # 做多逻辑 + if not first_tp_triggered and current_rate >= bb30_middle: + # 第一次止盈：价格涨到bb30中轨，止盈50% + logger.info(f"做多第一次止盈触发：价格{current_rate} >= BB30中轨{bb30_middle}") + trade.set_custom_data(key="first_tp_triggered", value=True) + trade.set_custom_data(key="new_stoploss", value=trade.open_rate) # 设置止损为开仓价 + return -(trade.amount * 0.5) # 减少50%仓位 + + elif first_tp_triggered and not second_tp_triggered and current_rate >= bb302_upper: + # 第二次止盈：继续涨到bb302上轨，止盈剩余仓位的60% + logger.info(f"做多第二次止盈触发：价格{current_rate} >= BB302上轨{bb302_upper}") + trade.set_custom_data(key="second_tp_triggered", value=True) + trade.set_custom_data(key="new_stoploss", value=bb30_middle) # 移动止损到bb30中轨 + remaining_amount = trade.amount * 0.5 # 剩余50% + return -(remaining_amount * 0.6) # 减少剩余仓位的60% + + return None + + def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, + current_rate: float, current_profit: float, after_fill: bool, + **kwargs) -> float | None: + """ + 动态止损逻辑 + """ + # 检查是否有自定义的新止损价格（分批止盈后的动态止损） + new_stoploss_price = trade.get_custom_data(key="new_stoploss") + if new_stoploss_price: + logger.info(f"使用动态止损价格: {new_stoploss_price}") + return stoploss_from_absolute(new_stoploss_price, current_rate, is_short=trade.is_short) + + + # 如果没有ATR数据，使用固定的5%止损作为备用 + logger.warning(f"未找到开仓时ATR数据，使用默认5%止损") + return -0.05 + + def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, + current_profit: float, **kwargs): + """ + 自定义退出逻辑 - 处理最终止盈条件 + """ + # 获取当前数据 + dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) + if dataframe is None or len(dataframe) == 0: + return None + + last_candle = dataframe.iloc[-1] + + # 获取布林带数据 + bb30_upper = last_candle['bbup30'] + bb30_lower = last_candle['bblow30'] + + # 检查是否已经触发过前两次止盈 + first_tp_triggered = trade.get_custom_data(key="first_tp_triggered", default=False) + second_tp_triggered = trade.get_custom_data(key="second_tp_triggered", default=False) + + if trade.is_short: + # 做空：如果价格跌到bb30下轨，全部止盈 + if first_tp_triggered and second_tp_triggered and current_rate <= bb30_lower: + logger.info(f"做空最终止盈触发：价格{current_rate} <= BB30下轨{bb30_lower}") + return "short_final_tp_bb30_lower" + else: + # 做多：如果价格涨到bb30上轨，全部止盈 + if first_tp_triggered and second_tp_triggered and current_rate >= bb30_upper: + logger.info(f"做多最终止盈触发：价格{current_rate} >= BB30上轨{bb30_upper}") + return "long_final_tp_bb30_upper" + + # 原有退出逻辑 + if trade.is_short: + last_high = trade.get_custom_data(key="entry_candle_high") + if last_high and current_rate > last_high: + return "Relay Top FX exit" + else: + last_low = trade.get_custom_data(key="entry_candle_low") + if last_low and current_rate < last_low: + return "Relay Bottom FX exit" + + return None + def confirm_trade_entry1(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: str | None, side: str, **kwargs) -> bool: @@ -248,36 +382,8 @@ class ChanLun_BTC_30(IStrategy): return stoploss_from_absolute(last_low, current_rate, is_short=trade.is_short) # return maximum stoploss value, keeping current stoploss price unchanged return None - def custom_exit1(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, - current_profit: float, **kwargs): - #dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) - #last_candle = dataframe.iloc[-1].squeeze() - """ - # Above 20% profit, sell when rsi < 80 - if current_profit > 0.2: - if last_candle["rsi"] < 80: - return "rsi_below_80" - # Between 2% and 10%, sell if EMA-long above EMA-short - if 0.02 < current_profit < 0.1: - if last_candle["emalong"] > last_candle["emashort"]: - return "ema_long_below_80" - - # Sell any positions at a loss if they are held for more than one day. - if current_profit < 0.0 and (current_time - trade.open_date_utc).days >= 1: - return "unclog" - """ - if trade.is_short: - last_high = trade.get_custom_data(key="entry_candle_high") - if last_high and current_rate > last_high: - #print(trade.open_date, last_high, current_rate, "Relay Top FX exit") - return "Relay Top FX exit" - else: - last_low = trade.get_custom_data(key="entry_candle_low") - if last_low and current_rate < last_low: - #print(trade.open_date, last_low, current_rate, "Relay Bottom FX exit") - return "Relay Bottom FX exit" - def order_filled1(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None: + def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None: """ Called right after an order fills. Will be called for all order types (entry, exit, stoploss, position adjustment). @@ -290,33 +396,12 @@ class ChanLun_BTC_30(IStrategy): # Obtain pair dataframe (just to show how to access it) dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() - ema5 = 'resample_{}_ema5'.format(self.get_ticker_indicator()*self.time30) - ema10 = 'resample_{}_ema10'.format(self.get_ticker_indicator()*self.time30) - ema26 = 'resample_{}_ema26'.format(self.get_ticker_indicator()*self.time30) - ema52 = 'resample_{}_ema52'.format(self.get_ticker_indicator()*self.time30) - #print(last_candle[ema5], last_candle[ema10], last_candle[ema26], last_candle[ema52]) - #print(last_candle['close']) - klc_list = self.chan.get_klc_list(resample_to_interval(dataframe, self.get_ticker_indicator() * self.time30)) - bi_list = self.chan.cal_bi_list(klc_list) - if self.last_order is None: - if trade.is_short and klc_list[-2].last_top_klc: - if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side): - last_high = klc_list[-2].last_top_klc.high - print(klc_list[-2].start_time, "--------------------------------", order.order_date, order.side, last_high) - trade.set_custom_data(key="entry_candle_high", value=last_high) - else: - if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side) and klc_list[-2].last_bottom_klc: - last_low = klc_list[-2].last_bottom_klc.low - trade.set_custom_data(key="entry_candle_low", value=last_low) - print(klc_list[-2].start_time, "--------------------------------", order.order_date, order.side, last_low) - #print(trade.open_date, trade.close_date, last_high, last_low, order.ft_order_side, klc_list[-2].start_time, klc_list[-2].end_time) - self.last_order = order - else: - if self.last_order.side != order.side: - self.last_order = None - trade.set_custom_data(key="entry_candle_high", value=None) - trade.set_custom_data(key="entry_candle_low", value=None) - self.last_trade = trade + + # 保存开仓时的ATR值用于止损计算 + if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side): + entry_atr = last_candle['atr'] + trade.set_custom_data(key="entry_atr", value=entry_atr) + logger.info(f"保存开仓时ATR值: {entry_atr}") return None def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: state_str = 'resample_{}_state'.format(self.get_ticker_indicator()*self.time30) diff --git a/strategies/bollinger_atr_strategy.pine b/strategies/bollinger_atr_strategy.pine new file mode 100644 index 0000000..6f6decb --- /dev/null +++ b/strategies/bollinger_atr_strategy.pine @@ -0,0 +1,136 @@ +//@version=6 +strategy("布林带ATR反转策略", shorttitle="BBB_ATR", overlay=true, default_qty_type=strategy.percent_of_equity, default_qty_value=100, calc_on_every_tick=true) + +// 输入参数 +bb_length = input.int(90, "布林带长度", minval=10, maxval=200) +atr_multiplier = input.float(3.0, "ATR乘数(轨道)", minval=1.0, maxval=10.0, step=0.1) +atr_stop_multiplier = input.float(3.0, "ATR乘数(止损)", minval=0.5, maxval=5.0, step=0.1) +atr_length = input.int(9, "ATR计算周期", minval=5, maxval=50) + +// 显示设置 +show_bands = input.bool(true, "显示布林带") +show_signals = input.bool(true, "显示信号") + +// 计算移动平均线(中线) +bb_middle = ta.sma(close, bb_length) + +// 计算ATR +atr_value = ta.atr(atr_length) + +// 计算上下轨 +bb_upper = bb_middle + (atr_multiplier * atr_value) +bb_lower = bb_middle - (atr_multiplier * atr_value) + +// 显示布林带 +plot(show_bands ? bb_middle : na, "中线", color=color.blue, linewidth=2) +plot(show_bands ? bb_upper : na, "上轨", color=color.red, linewidth=2) +plot(show_bands ? bb_lower : na, "下轨", color=color.green, linewidth=2) + + +// 交易条件 +// 做空条件：价格突破上轨 +short_condition = close > bb_upper and close[1] <= bb_upper[1] + +// 做多条件：价格跌破下轨 +long_condition = close < bb_lower and close[1] >= bb_lower[1] + +// 做空止盈条件：价格跌破下轨 +short_take_profit = close < bb_lower + +// 做多止盈条件：价格突破上轨 +long_take_profit = close > bb_upper + +// 记录入场价格和止损位 +var float long_entry_price = na +var float short_entry_price = na +var float long_stop_loss = na +var float short_stop_loss = na + +// 执行交易逻辑 +if strategy.position_size == 0 + if long_condition + strategy.entry("做多", strategy.long) + long_entry_price := close + long_stop_loss := close - (atr_stop_multiplier * atr_value) + + if short_condition + strategy.entry("做空", strategy.short) + short_entry_price := close + short_stop_loss := close + (atr_stop_multiplier * atr_value) + +// 多头仓位管理 +if strategy.position_size > 0 + // 止盈：价格突破上轨 + if long_take_profit + strategy.close("做多", comment="多头止盈") + long_entry_price := na + long_stop_loss := na + // 止损：价格跌破止损位 + else if close <= long_stop_loss + strategy.close("做多", comment="多头止损") + long_entry_price := na + long_stop_loss := na + +// 空头仓位管理 +if strategy.position_size < 0 + // 止盈：价格跌破下轨 + if short_take_profit + strategy.close("做空", comment="空头止盈") + short_entry_price := na + short_stop_loss := na + // 止损：价格突破止损位 + else if close >= short_stop_loss + strategy.close("做空", comment="空头止损") + short_entry_price := na + short_stop_loss := na + +// 显示信号 +if show_signals + if long_condition and strategy.position_size == 0 + label.new(bar_index, low, "做多", color=color.green, style=label.style_label_up, size=size.normal, textcolor=color.white) + if short_condition and strategy.position_size == 0 + label.new(bar_index, high, "做空", color=color.red, style=label.style_label_down, size=size.normal, textcolor=color.white) + if long_take_profit and strategy.position_size > 0 + label.new(bar_index, high, "多头止盈", color=color.green, style=label.style_label_down, size=size.small, textcolor=color.white) + if short_take_profit and strategy.position_size < 0 + label.new(bar_index, low, "空头止盈", color=color.red, style=label.style_label_up, size=size.small, textcolor=color.white) + +// 显示止损线 +plot(strategy.position_size > 0 and not na(long_stop_loss) ? long_stop_loss : na, "多头止损", color=color.red, style=plot.style_linebr, linewidth=1) +plot(strategy.position_size < 0 and not na(short_stop_loss) ? short_stop_loss : na, "空头止损", color=color.red, style=plot.style_linebr, linewidth=1) + + +// 信息表格 +if barstate.islast + var table info_table = table.new(position.top_right, 2, 10, bgcolor=color.white, border_width=1) + table.cell(info_table, 0, 0, "布林带ATR反转策略", text_color=color.black, bgcolor=color.gray) + table.cell(info_table, 1, 0, "", text_color=color.black, bgcolor=color.gray) + + table.cell(info_table, 0, 1, "布林带长度", text_color=color.black) + table.cell(info_table, 1, 1, str.tostring(bb_length), text_color=color.black) + + table.cell(info_table, 0, 2, "ATR轨道乘数", text_color=color.black) + table.cell(info_table, 1, 2, str.tostring(atr_multiplier), text_color=color.black) + + table.cell(info_table, 0, 3, "ATR止损乘数", text_color=color.black) + table.cell(info_table, 1, 3, str.tostring(atr_stop_multiplier), text_color=color.black) + + table.cell(info_table, 0, 4, "当前ATR", text_color=color.black) + table.cell(info_table, 1, 4, str.tostring(math.round(atr_value, 4)), text_color=color.black) + + table.cell(info_table, 0, 5, "上轨价位", text_color=color.black) + table.cell(info_table, 1, 5, str.tostring(math.round(bb_upper, 2)), text_color=color.black) + + table.cell(info_table, 0, 6, "下轨价位", text_color=color.black) + table.cell(info_table, 1, 6, str.tostring(math.round(bb_lower, 2)), text_color=color.black) + + table.cell(info_table, 0, 7, "当前价格", text_color=color.black) + table.cell(info_table, 1, 7, str.tostring(math.round(close, 2)), text_color=color.black) + + table.cell(info_table, 0, 8, "仓位状态", text_color=color.black) + position_text = strategy.position_size > 0 ? "多头" : strategy.position_size < 0 ? "空头" : "空仓" + table.cell(info_table, 1, 8, position_text, text_color=color.black) + + table.cell(info_table, 0, 9, "价格位置", text_color=color.black) + price_position = close > bb_upper ? "上轨之上" : close < bb_lower ? "下轨之下" : "轨道之间" + table.cell(info_table, 1, 9, price_position, text_color=color.black) \ No newline at end of file