diff --git a/bitmart/三分之一策略-5分钟交易.py b/bitmart/三分之一策略-5分钟交易.py index 66143cc..4a52d23 100644 --- a/bitmart/三分之一策略-5分钟交易.py +++ b/bitmart/三分之一策略-5分钟交易.py @@ -16,6 +16,15 @@ BitMart 三分之一策略 — 5分钟K线交易 - 同一根5分钟K线内只交易一次 4. 同根K线多空都触及时，用开盘价与触发价距离判断先后。 + +5. 反手信号（基于上一根K线高低点+影线幅度）： + - 持空反手做多：当前K线涨到上一根K线最高价，且上一根K线上影线涨幅 > 0.01% -> 平空反手开多 + - 持多反手做空：当前K线跌到上一根K线最低价，且上一根K线下影线跌幅 > 0.01% -> 平多反手开空 + （上影线 = high - max(open,close)，上影线涨幅 = (上影线/开盘价)*100；下影线同理） + +6. 反手信号二（基于上一根开盘价+影线幅度）： + - 持多反手做空：上一根K线上影线涨幅 > 0.01%，且当前K线跌到上一根开盘价 -> 平多反手开空 + - 持空反手做多：上一根K线下影线跌幅 > 0.01%，且当前K线涨到上一根开盘价 -> 平空反手开多 """ import random import time @@ -71,6 +80,7 @@ class BitmartOneThirdStrategy: self.last_trigger_kline_id = None self.last_trigger_direction = None self.last_trade_kline_id = None + self.min_shadow_pct = 0.01 # 反手信号：上/下影线幅度需 > 0.01% # ========================= 三分之一策略核心 ========================= @@ -104,6 +114,81 @@ class BitmartOneThirdStrategy: short_trigger = p_close - body / 3 return long_trigger, short_trigger + def get_upper_shadow(self, candle): + """上影线 = high - max(open, close)""" + o, c, h = float(candle['open']), float(candle['close']), float(candle['high']) + return h - max(o, c) + + def get_lower_shadow(self, candle): + """下影线 = min(open, close) - low""" + o, c, l = float(candle['open']), float(candle['close']), float(candle['low']) + return min(o, c) - l + + def upper_shadow_pct(self, candle): + """上影线涨幅 = (上影线/开盘价)*100""" + o = float(candle['open']) + if o <= 0: + return 0.0 + return self.get_upper_shadow(candle) / o * 100 + + def lower_shadow_pct(self, candle): + """下影线跌幅 = (下影线/开盘价)*100""" + o = float(candle['open']) + if o <= 0: + return 0.0 + return self.get_lower_shadow(candle) / o * 100 + + def check_reverse_by_prev_high_low(self, kline_data): + """ + 反手信号：持空时当前K线涨到上一根最高价且上一根上影线涨幅>0.01% -> 反手做多； + 持多时当前K线跌到上一根最低价且上一根下影线跌幅>0.01% -> 反手做空。 + 使用紧邻的上一根K线（kline_data[-2]），非“有效实体”前一根。 + 返回：(方向 'long'|'short', 上一根K线) 或 (None, None) + """ + if len(kline_data) < 2: + return None, None + curr = kline_data[-1] + prev = kline_data[-2] + curr_high = float(curr['high']) + curr_low = float(curr['low']) + prev_high = float(prev['high']) + prev_low = float(prev['low']) + + # 持空反手做多：当前涨到上一根最高价，且上一根上影线涨幅 > 0.01% + if self.start == -1: + if curr_high >= prev_high and self.upper_shadow_pct(prev) > self.min_shadow_pct: + return 'long', prev + # 持多反手做空：当前跌到上一根最低价，且上一根下影线跌幅 > 0.01% + if self.start == 1: + if curr_low <= prev_low and self.lower_shadow_pct(prev) > self.min_shadow_pct: + return 'short', prev + return None, None + + def check_reverse_by_prev_open(self, kline_data): + """ + 反手信号二：持多时上一根上影线涨幅>0.01%且当前K线跌到上一根开盘价 -> 反手做空； + 持空时上一根下影线跌幅>0.01%且当前K线涨到上一根开盘价 -> 反手做多。 + 使用紧邻的上一根K线（kline_data[-2]）。 + 返回：(方向 'long'|'short', 上一根K线) 或 (None, None) + """ + if len(kline_data) < 2: + return None, None + curr = kline_data[-1] + prev = kline_data[-2] + curr_high = float(curr['high']) + curr_low = float(curr['low']) + prev_open = float(prev['open']) + + # 持多反手做空：上一根上影线涨幅>0.01%，当前跌到上一根开盘价 + if self.start == 1: + if self.upper_shadow_pct(prev) > self.min_shadow_pct and curr_low <= prev_open: + return 'short', prev + # 持空反手做多：上一根下影线跌幅>0.01%，当前涨到上一根开盘价 + if self.start == -1: + if self.lower_shadow_pct(prev) > self.min_shadow_pct and curr_high >= prev_open: + return 'long', prev + return None, None + def check_realtime_trigger(self, kline_data): """ 检查当前5分钟K线是否触发信号。 @@ -383,6 +468,45 @@ class BitmartOneThirdStrategy: if not self.get_position_status(): logger.warning("获取仓位失败，使用缓存") + curr_kline_id = curr['id'] + + # 优先检测：反手信号一（涨到上根最高/跌到上根最低 + 上/下影线>0.01%） + rev_dir, rev_prev = self.check_reverse_by_prev_high_low(kline_data) + rev_type = "一" + if not rev_dir: + # 反手信号二（跌到/涨到上根开盘价 + 上/下影线>0.01%） + rev_dir, rev_prev = self.check_reverse_by_prev_open(kline_data) + rev_type = "二" + if rev_dir and self.last_trade_kline_id != curr_kline_id: + balance = self.get_available_balance() + trade_size = (balance or 0) * self.risk_percent + if rev_prev is not None: + up_pct = self.upper_shadow_pct(rev_prev) + low_pct = self.lower_shadow_pct(rev_prev) + logger.info(f"{'='*50}") + if rev_type == "一": + logger.info(f"🔄 反手信号{rev_type}({rev_dir})：当前K线触及上一根{'最高' if rev_dir == 'long' else '最低'}价") + else: + logger.info(f"🔄 反手信号{rev_type}({rev_dir})：当前K线触及上一根开盘价 O={rev_prev['open']:.2f}") + logger.info(f" 上一根 上影线涨幅={up_pct:.3f}% 下影线跌幅={low_pct:.3f}%") + logger.info(f"{'='*50}") + if rev_dir == 'long': + logger.info("📈 平空反手开多") + self.平仓() + time.sleep(1) + self.开单(marketPriceLongOrder=1, size=trade_size) + else: + logger.info("📉 平多反手开空") + self.平仓() + time.sleep(1) + self.开单(marketPriceLongOrder=-1, size=trade_size) + self.last_trade_kline_id = curr_kline_id + self.last_trigger_kline_id = curr_kline_id + self.last_trigger_direction = rev_dir + self.get_position_status() + time.sleep(self.check_interval) + continue + direction, trigger_price, valid_prev, curr_kline = self.check_realtime_trigger(kline_data) if direction: