23423423

bitmart/回测-三分之一策略.py

@@ -1,27 +1,29 @@
 """
-量化交易回测系统 - 三分之一回归策略（优化版）
+量化交易回测系统 - 三分之一回归策略（双向触发版）
 
 ========== 策略规则 ==========
 
-1. 开多条件：
-   - 找到实体>=0.1的前一根K线（如果前一根实体<0.1，继续往前找）
-   - 前一根是阴线（close < open）
-   - 当前K线的最高价（包括影线）涨到前一根阴线实体的 1/3 处
-   - 即：当前high >= prev_close + (prev_open - prev_close) / 3
-
-2. 平多/开空条件：
-   - 找到实体>=0.1的前一根K线（如果前一根实体<0.1，继续往前找）
-   - 前一根是阳线（close > open）
-   - 当前K线的最低价（包括影线）跌到前一根阳线实体的 1/3 处
-   - 即：当前low <= prev_close - (prev_close - prev_open) / 3
+1. 触发价格计算（基于有效的前一根K线，实体>=0.1）：
+   - 做多触发价格 = 前一根收盘价 + 实体/3（向上突破1/3）
+   - 做空触发价格 = 前一根收盘价 - 实体/3（向下突破1/3）
+   
+2. 信号触发条件（无论前一根是阴线还是阳线）：
+   - 当前K线最高价 >= 做多触发价格 → 做多信号
+   - 当前K线最低价 <= 做空触发价格 → 做空信号
 
 3. 执行逻辑：
-   - 做多时遇到开空信号 -> 平多并反手开空
-   - 做空时遇到开多信号 -> 平空并反手开多
+   - 做多时遇到做空信号 -> 平多并反手开空
+   - 做空时遇到做多信号 -> 平空并反手开多
+   - 如果同时触发两个方向，以距离开盘价更近的方向优先
 
 4. 实体过滤：
    - 如果前一根K线的实体部分（|open - close|）< 0.1，继续往前查找
-   - 直到找到实体>=0.1的K线，再用那根K线来判断1/3位置
+   - 直到找到实体>=0.1的K线，再用那根K线来计算触发价格
+
+示例：
+   前一根K线：开盘3200，收盘3100（阴线，实体=100）
+   - 做多触发价格 = 3100 + 100/3 = 3133.33（价格反弹到这里做多）
+   - 做空触发价格 = 3100 - 100/3 = 3066.67（价格继续下跌到这里做空）
 """
 
 import datetime
@@ -37,7 +39,7 @@ try:
     import plotly.graph_objects as go
 except Exception:
     go = None
-from models.bitmart_klines import BitMartETH15M
+from models.bitmart_klines import BitMartETH5M
 
 # 配置中文字体
 import matplotlib.font_manager as fm
@@ -111,64 +113,80 @@ def find_valid_prev_bar(all_data: List[Dict], current_idx: int, min_body_size: f
     return None, None
 
 
-def get_one_third_level(prev):
+def get_one_third_levels(prev):
     """
-    计算前一根K线实体的 1/3 回归位置
-    返回：(触发价格, 方向)
-    - 如果前一根是阴线：返回向上1/3价格，方向为 'long'
-    - 如果前一根是阳线：返回向下1/3价格，方向为 'short'
+    计算前一根K线实体的 1/3 双向触发价格
+    返回：(做多触发价格, 做空触发价格)
+    
+    无论阴线还是阳线：
+    - 做多触发价格 = 收盘价 + 实体/3（向上突破1/3）
+    - 做空触发价格 = 收盘价 - 实体/3（向下突破1/3）
     """
     p_open = float(prev['open'])
     p_close = float(prev['close'])
     
-    if is_bearish(prev):  # 阴线，向上回归
-        # 阴线实体 = open - close
-        body = p_open - p_close
-        trigger_price = p_close + body / 3  # 从低点涨 1/3
-        return trigger_price, 'long'
+    body = abs(p_open - p_close)
     
-    elif is_bullish(prev):  # 阳线，向下回归
-        # 阳线实体 = close - open
-        body = p_close - p_open
-        trigger_price = p_close - body / 3  # 从高点跌 1/3
-        return trigger_price, 'short'
+    if body < 0.001:  # 十字星，忽略
+        return None, None
     
-    return None, None
+    # 双向触发价格
+    long_trigger = p_close + body / 3   # 向上1/3触发做多
+    short_trigger = p_close - body / 3  # 向下1/3触发做空
+    
+    return long_trigger, short_trigger
 
 
 def check_trigger(all_data: List[Dict], current_idx: int, min_body_size: float = 0.1):
     """
-    检查当前K线是否触发了交易信号
+    检查当前K线是否触发了交易信号（双向检测）
     返回：(方向, 触发价格, 有效前一根K线索引) 或 (None, None, None)
-    规则：考虑影线部分（high/low），因为实际交易中价格会到达影线位置
+    
+    规则：
+    - 当前K线高点 >= 做多触发价格 → 做多信号
+    - 当前K线低点 <= 做空触发价格 → 做空信号
+    - 如果同时触发两个方向，以先触发的为准（这里简化为优先做空，因为下跌更快）
     """
     if current_idx <= 0:
         return None, None, None
     
     curr = all_data[current_idx]
     
-    # 查找实体>=0.1的前一根K线
+    # 查找实体>=min_body_size的前一根K线
     valid_prev_idx, prev = find_valid_prev_bar(all_data, current_idx, min_body_size)
     
     if prev is None:
         return None, None, None
     
-    trigger_price, direction = get_one_third_level(prev)
+    long_trigger, short_trigger = get_one_third_levels(prev)
     
-    if trigger_price is None:
+    if long_trigger is None:
         return None, None, None
     
-    # 使用影线部分（high/low）来判断，因为实际交易中价格会到达这些位置
+    # 使用影线部分（high/low）来判断
     c_high = float(curr['high'])
     c_low = float(curr['low'])
     
-    # 做多：前一根阴线，当前K线的最高价（包括影线）达到触发价格
-    if direction == 'long' and c_high >= trigger_price:
-        return 'long', trigger_price, valid_prev_idx
+    # 检测是否触发
+    long_triggered = c_high >= long_trigger
+    short_triggered = c_low <= short_trigger
     
-    # 做空：前一根阳线，当前K线的最低价（包括影线）达到触发价格
-    if direction == 'short' and c_low <= trigger_price:
-        return 'short', trigger_price, valid_prev_idx
+    # 如果两个方向都触发，需要判断哪个先触发
+    # 简化处理：比较触发价格距离开盘价的远近，更近的先触发
+    if long_triggered and short_triggered:
+        c_open = float(curr['open'])
+        dist_to_long = abs(long_trigger - c_open)
+        dist_to_short = abs(short_trigger - c_open)
+        if dist_to_short <= dist_to_long:
+            return 'short', short_trigger, valid_prev_idx
+        else:
+            return 'long', long_trigger, valid_prev_idx
+    
+    if short_triggered:
+        return 'short', short_trigger, valid_prev_idx
+    
+    if long_triggered:
+        return 'long', long_trigger, valid_prev_idx
     
     return None, None, None
 
@@ -200,7 +218,7 @@ def backtest_one_third_strategy(dates: List[str]):
     all_data: List[Dict] = []
     
     for d in dates:
-        day_data = get_data_by_date(BitMartETH15M, d)
+        day_data = get_data_by_date(BitMartETH5M, d)
         all_data.extend(day_data)
     
     logger.info(f"总共查询了 {len(dates)} 天，获取到 {len(all_data)} 条K线数据")

交易/bitmart-三分之一策略交易.py

@@ -1,23 +1,25 @@
 """
-BitMart 三分之一回归策略交易
-使用5分钟K线周期
+BitMart 三分之一回归策略交易（双向触发版）
+使用5分钟K线周期，实时监测
 
 策略规则：
-1. 开多条件：
-   - 找到实体>=0.1的前一根K线（如果前一根实体<0.1，继续往前找）
-   - 前一根是阴线（close < open）
-   - 当前K线的最高价（包括影线）涨到前一根阴线实体的 1/3 处
-   - 即：当前high >= prev_close + (prev_open - prev_close) / 3
-
-2. 平多/开空条件：
-   - 找到实体>=0.1的前一根K线
-   - 前一根是阳线（close > open）
-   - 当前K线的最低价（包括影线）跌到前一根阳线实体的 1/3 处
-   - 即：当前low <= prev_close - (prev_close - prev_open) / 3
+1. 触发价格计算（基于有效的前一根K线，实体>=0.1）：
+   - 做多触发价格 = 前一根收盘价 + 实体/3（向上突破1/3）
+   - 做空触发价格 = 前一根收盘价 - 实体/3（向下突破1/3）
+   
+2. 信号触发条件（无论前一根是阴线还是阳线）：
+   - 当前K线最高价 >= 做多触发价格 → 做多信号
+   - 当前K线最低价 <= 做空触发价格 → 做空信号
 
 3. 执行逻辑：
-   - 做多时遇到开空信号 -> 平多并反手开空
-   - 做空时遇到开多信号 -> 平空并反手开多
+   - 做多时遇到做空信号 -> 平多并反手开空
+   - 做空时遇到做多信号 -> 平空并反手开多
+   - 如果同时触发两个方向，以距离开盘价更近的方向优先
+
+示例：
+   前一根K线：开盘3200，收盘3100（阴线，实体=100）
+   - 做多触发价格 = 3100 + 100/3 = 3133.33（价格反弹到这里做多）
+   - 做空触发价格 = 3100 - 100/3 = 3066.67（价格继续下跌到这里做空）
 """
 
 import time
@@ -102,35 +104,37 @@ class BitmartOneThirdStrategy:
 
         return None, None
 
-    def get_one_third_level(self, prev):
+    def get_one_third_levels(self, prev):
         """
-        计算前一根K线实体的 1/3 回归位置
-        返回：(触发价格, 方向)
-        - 如果前一根是阴线：返回向上1/3价格，方向为 'long'
-        - 如果前一根是阳线：返回向下1/3价格，方向为 'short'
+        计算前一根K线实体的 1/3 双向触发价格
+        返回：(做多触发价格, 做空触发价格)
+        
+        无论阴线还是阳线：
+        - 做多触发价格 = 收盘价 + 实体/3（向上突破1/3）
+        - 做空触发价格 = 收盘价 - 实体/3（向下突破1/3）
         """
         p_open = float(prev['open'])
         p_close = float(prev['close'])
 
-        if self.is_bearish(prev):  # 阴线，向上回归
-            # 阴线实体 = open - close
-            body = p_open - p_close
-            trigger_price = p_close + body / 3  # 从低点涨 1/3
-            return trigger_price, 'long'
+        body = abs(p_open - p_close)
 
-        elif self.is_bullish(prev):  # 阳线，向下回归
-            # 阳线实体 = close - open
-            body = p_close - p_open
-            trigger_price = p_close - body / 3  # 从高点跌 1/3
-            return trigger_price, 'short'
+        if body < 0.001:  # 十字星，忽略
+            return None, None
 
-        return None, None
+        # 双向触发价格
+        long_trigger = p_close + body / 3   # 向上1/3触发做多
+        short_trigger = p_close - body / 3  # 向下1/3触发做空
+
+        return long_trigger, short_trigger
 
     def check_trigger(self, all_data, current_idx):
         """
-        检查当前K线是否触发了交易信号
+        检查当前K线是否触发了交易信号（双向检测）
         返回：(方向, 触发价格, 有效前一根K线索引) 或 (None, None, None)
-        规则：考虑影线部分（high/low），因为实际交易中价格会到达影线位置
+        
+        规则：
+        - 当前K线高点 >= 做多触发价格 → 做多信号
+        - 当前K线低点 <= 做空触发价格 → 做空信号
         """
         if current_idx <= 0:
             return None, None, None
@@ -143,28 +147,40 @@ class BitmartOneThirdStrategy:
         if prev is None:
             return None, None, None
 
-        trigger_price, direction = self.get_one_third_level(prev)
+        long_trigger, short_trigger = self.get_one_third_levels(prev)
 
-        if trigger_price is None:
+        if long_trigger is None:
             return None, None, None
 
         # 使用影线部分（high/low）来判断
         c_high = float(curr['high'])
         c_low = float(curr['low'])
 
-        # 做多：前一根阴线，当前K线的最高价（包括影线）达到触发价格
-        if direction == 'long' and c_high >= trigger_price:
-            return 'long', trigger_price, valid_prev_idx
+        # 检测是否触发
+        long_triggered = c_high >= long_trigger
+        short_triggered = c_low <= short_trigger
 
-        # 做空：前一根阳线，当前K线的最低价（包括影线）达到触发价格
-        if direction == 'short' and c_low <= trigger_price:
-            return 'short', trigger_price, valid_prev_idx
+        # 如果两个方向都触发，判断哪个先触发
+        if long_triggered and short_triggered:
+            c_open = float(curr['open'])
+            dist_to_long = abs(long_trigger - c_open)
+            dist_to_short = abs(short_trigger - c_open)
+            if dist_to_short <= dist_to_long:
+                return 'short', short_trigger, valid_prev_idx
+            else:
+                return 'long', long_trigger, valid_prev_idx
+
+        if short_triggered:
+            return 'short', short_trigger, valid_prev_idx
+
+        if long_triggered:
+            return 'long', long_trigger, valid_prev_idx
 
         return None, None, None
 
     def check_realtime_trigger(self, kline_data):
         """
-        实时检测当前K线是否触发信号
+        实时检测当前K线是否触发信号（双向检测）
         基于已收盘的K线计算触发价格，用当前正在形成的K线判断
         返回：(方向, 触发价格, 有效前一根K线, 当前K线) 或 (None, None, None, None)
         """
@@ -176,34 +192,54 @@ class BitmartOneThirdStrategy:
         curr_kline_id = curr['id']
 
         # 从倒数第二根开始往前找有效K线（已收盘的K线）
-        # current_idx 设为 len-1，但我们从 len-2 开始找有效的前一根
         valid_prev_idx, prev = self.find_valid_prev_bar(kline_data, len(kline_data) - 1, self.min_body_size)
 
         if prev is None:
             return None, None, None, None
 
-        trigger_price, direction = self.get_one_third_level(prev)
+        long_trigger, short_trigger = self.get_one_third_levels(prev)
 
-        if trigger_price is None:
-            return None, None, None, None
-
-        # 检查是否在同一根K线内已经触发过相同方向
-        if self.last_trigger_kline_id == curr_kline_id and self.last_trigger_direction == direction:
+        if long_trigger is None:
             return None, None, None, None
 
         # 使用当前K线的实时高低点来判断
         c_high = float(curr['high'])
         c_low = float(curr['low'])
 
-        # 做多：前一根阴线，当前K线的最高价达到触发价格
-        if direction == 'long' and c_high >= trigger_price:
-            return 'long', trigger_price, prev, curr
+        # 检测是否触发
+        long_triggered = c_high >= long_trigger
+        short_triggered = c_low <= short_trigger
 
-        # 做空：前一根阳线，当前K线的最低价达到触发价格
-        if direction == 'short' and c_low <= trigger_price:
-            return 'short', trigger_price, prev, curr
+        # 确定触发方向
+        direction = None
+        trigger_price = None
 
-        return None, None, None, None
+        if long_triggered and short_triggered:
+            # 两个方向都触发，判断哪个先（距离开盘价更近的先触发）
+            c_open = float(curr['open'])
+            dist_to_long = abs(long_trigger - c_open)
+            dist_to_short = abs(short_trigger - c_open)
+            if dist_to_short <= dist_to_long:
+                direction = 'short'
+                trigger_price = short_trigger
+            else:
+                direction = 'long'
+                trigger_price = long_trigger
+        elif short_triggered:
+            direction = 'short'
+            trigger_price = short_trigger
+        elif long_triggered:
+            direction = 'long'
+            trigger_price = long_trigger
+
+        if direction is None:
+            return None, None, None, None
+
+        # 检查是否在同一根K线内已经触发过相同方向
+        if self.last_trigger_kline_id == curr_kline_id and self.last_trigger_direction == direction:
+            return None, None, None, None
+
+        return direction, trigger_price, prev, curr
 
     # ========================= BitMart API 函数 =========================