From d81b9eced0cca8bffe607110e91cb4e721080397 Mon Sep 17 00:00:00 2001 From: ddrwode <34234@3来 34> Date: Tue, 3 Feb 2026 10:19:56 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BB=A3=E7=A0=81=E7=BB=93=E6=9E=84=E4=BC=98?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- 交易/bitmart-五分之一策略交易.py | 94 +++++--------------------------- 1 file changed, 14 insertions(+), 80 deletions(-) diff --git a/交易/bitmart-五分之一策略交易.py b/交易/bitmart-五分之一策略交易.py index e97861d..7f7d882 100644 --- a/交易/bitmart-五分之一策略交易.py +++ b/交易/bitmart-五分之一策略交易.py @@ -1,6 +1,6 @@ """ BitMart 五分之一策略交易（3分钟精准版） -使用3分钟K线周期计算触发价格，实时监测；同根K线内多空都触及时用1分钟K线判断先后 +使用3分钟K线周期计算触发价格，实时监测；同根K线内多空都触及时用开盘价距离判断先后 策略规则（与 bitmart/回测数据-五分之一策略-3分钟精准版.py 完全一致）： 1. 触发价格计算（基于有效的前一根3分钟K线，实体>=0.1）： @@ -16,10 +16,9 @@ BitMart 五分之一策略交易（3分钟精准版） - 做空时遇到做多信号 -> 平空并反手开多 - 同一根3分钟K线内只交易一次 -4. 精准判断（使用1分钟K线）： +4. 同根触发判断（无需1分钟K线）： - 当一根3分钟K线同时触及做多和做空价格时 - - 使用该3分钟K线对应的3根1分钟K线来判断哪个方向先被触发 - - 使交易更贴近真实成交顺序 + - 使用该3分钟K线开盘价与触发价的距离判断先后 """ import random import time @@ -108,65 +107,10 @@ class BitmartOneFifthStrategy: short_trigger = p_close - body / 5 return long_trigger, short_trigger - def get_1m_bars_for_3m_bar(self, bar_3m): - """获取当前3分钟K线对应的3根1分钟K线（用于同根内多空都触发时判断先后）""" - try: - start_ts = int(bar_3m['id']) - end_ts = start_ts + 3 * 60 # 秒 - response = self.contractAPI.get_kline( - contract_symbol=self.contract_symbol, - step=1, - start_time=start_ts, - end_time=end_ts - )[0] - if response.get('code') != 1000: - return [] - data = response.get('data', []) - out = [] - for k in data: - out.append({ - 'id': int(k["timestamp"]), - 'open': float(k["open_price"]), - 'high': float(k["high_price"]), - 'low': float(k["low_price"]), - 'close': float(k["close_price"]) - }) - out.sort(key=lambda x: x['id']) - return out - except Exception as e: - logger.warning(f"获取1分钟K线失败: {e}") - return [] - - def determine_trigger_order_by_1m(self, bars_1m, long_trigger, short_trigger): - """ - 使用1分钟K线判断在一根3分钟周期内，先触发做多还是做空。 - 按时间顺序遍历每根1分钟K线，先触及哪个方向则返回该方向； - 若同一根1分钟K线内两个方向都触及，用开盘价距离判断。 - - 返回：'long', 'short' 或 None - """ - if not bars_1m: - return None - for bar in bars_1m: - high = float(bar['high']) - low = float(bar['low']) - open_price = float(bar['open']) - long_triggered = high >= long_trigger - short_triggered = low <= short_trigger - if long_triggered and not short_triggered: - return 'long' - if short_triggered and not long_triggered: - return 'short' - if long_triggered and short_triggered: - dist_to_long = abs(long_trigger - open_price) - dist_to_short = abs(short_trigger - open_price) - return 'short' if dist_to_short <= dist_to_long else 'long' - return None - def check_realtime_trigger(self, kline_data): """ 检查当前3分钟K线是否触发交易信号（与回测逻辑完全一致） - 若同时触发多空，则用该3分钟内的1分钟K线判断先后顺序。 + 若同时触发多空，则用开盘价距离判断先后顺序（不请求1分钟K线）。 参数: kline_data: K线数据列表 @@ -197,26 +141,17 @@ class BitmartOneFifthStrategy: direction = None trigger_price = None - # 如果同时触发多空，用1分钟K线判断先后顺序（与回测逻辑一致） + # 如果同时触发多空，用开盘价距离判断先后顺序（避免请求1分钟K线） if both_triggered: - bars_1m = self.get_1m_bars_for_3m_bar(curr) - if bars_1m: - direction = self.determine_trigger_order_by_1m( - bars_1m, long_trigger, short_trigger - ) - if direction: - trigger_price = long_trigger if direction == 'long' else short_trigger - # 如果1分钟K线无法判断，使用开盘价距离判断 - if direction is None: - 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 + 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 @@ -380,7 +315,6 @@ class BitmartOneFifthStrategy: return False direction_str = "做多" if marketPriceLongOrder == 1 else "做空" logger.info(f"执行{direction_str}操作，金额: {size}") - size = 25 try: if marketPriceLongOrder == -1: self.click_safe('x://button[normalize-space(text()) ="市价"]')