代码结构优化

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: