diff --git a/交易/bitmart-五分之一策略交易.py b/交易/bitmart-五分之一策略交易.py index 5527dc4..e97861d 100644 --- a/交易/bitmart-五分之一策略交易.py +++ b/交易/bitmart-五分之一策略交易.py @@ -1,15 +1,15 @@ """ -BitMart 五分之一回归策略交易（精准版） +BitMart 五分之一策略交易（3分钟精准版） 使用3分钟K线周期计算触发价格，实时监测；同根K线内多空都触及时用1分钟K线判断先后 -策略规则（与 bitmart/回测-三分之一策略-精准版.py 一致）： -1. 触发价格计算（基于有效的前一根K线，实体>=0.1）： +策略规则（与 bitmart/回测数据-五分之一策略-3分钟精准版.py 完全一致）： +1. 触发价格计算（基于有效的前一根3分钟K线，实体>=0.1）： - 做多触发价格 = 收盘价 + 实体/5（从收盘价往上涨1/5） - 做空触发价格 = 收盘价 - 实体/5（从收盘价往下跌1/5） 2. 信号触发条件： - - 当前K线最高价 >= 做多触发价格 → 做多信号 - - 当前K线最低价 <= 做空触发价格 → 做空信号 + - 当前3分钟K线最高价 >= 做多触发价格 → 做多信号 + - 当前3分钟K线最低价 <= 做空触发价格 → 做空信号 3. 执行逻辑： - 做多时遇到做空信号 -> 平多并反手开空 @@ -17,7 +17,9 @@ BitMart 五分之一回归策略交易（精准版） - 同一根3分钟K线内只交易一次 4. 精准判断（使用1分钟K线）： - - 当当前3分钟K线同时触及做多和做空价格时，拉取该3分钟对应的3根1分钟K线判断哪个方向先被触发 + - 当一根3分钟K线同时触及做多和做空价格时 + - 使用该3分钟K线对应的3根1分钟K线来判断哪个方向先被触发 + - 使交易更贴近真实成交顺序 """ import random import time @@ -71,18 +73,6 @@ class BitmartOneFifthStrategy: self.last_trigger_kline_id = None self.last_trigger_direction = None self.last_trade_kline_id = None - - # 反手信号当前价格容差（美元） - # 当检测到反手信号时，当前价格必须在触发价附近才执行 - # 避免"先涨后跌"或"先跌后涨"的情况下错误开仓 - self.reverse_price_tolerance = 2.0 # 2美元容差 - - # 基于开仓价格的反手信号参数 - # 记录开仓时使用的前一根K线实体大小（用于计算反手触发价） - self.entry_prev_body = None # 开仓时前一根K线的实体大小 - self.entry_price = None # 开仓价格（用于计算反手触发价） - self.entry_kline_id = None # 开仓时的K线ID（用于判断是否在同一根K线内） - self.first_minute_reverse_executed = False # 当前K线是否已执行过第一分钟反手 # ========================= 五分之一策略核心 ========================= @@ -149,8 +139,11 @@ class BitmartOneFifthStrategy: def determine_trigger_order_by_1m(self, bars_1m, long_trigger, short_trigger): """ - 用1分钟K线判断在3分钟周期内先触发做多还是做空 - 返回 'long', 'short' 或 None + 使用1分钟K线判断在一根3分钟周期内，先触发做多还是做空。 + 按时间顺序遍历每根1分钟K线，先触及哪个方向则返回该方向； + 若同一根1分钟K线内两个方向都触及，用开盘价距离判断。 + + 返回：'long', 'short' 或 None """ if not bars_1m: return None @@ -158,116 +151,25 @@ class BitmartOneFifthStrategy: high = float(bar['high']) low = float(bar['low']) open_price = float(bar['open']) - long_ok = high >= long_trigger - short_ok = low <= short_trigger - if long_ok and not short_ok: + long_triggered = high >= long_trigger + short_triggered = low <= short_trigger + if long_triggered and not short_triggered: return 'long' - if short_ok and not long_ok: + if short_triggered and not long_triggered: return 'short' - if long_ok and short_ok: - d_long = abs(long_trigger - open_price) - d_short = abs(short_trigger - open_price) - return 'short' if d_short < d_long else 'long' + 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_first_minute_reverse_signal(self, curr_kline, kline_data): + def check_realtime_trigger(self, kline_data): """ - 检测基于开仓价格的反手信号（只在3分钟K线的第一分钟有效） - - 反手规则： - - 空仓反手开多：开空仓后，价格涨到 开仓价格 + 前一根K线实体/5 → 平空开多 - - 多仓反手开空：开多仓后，价格跌到 开仓价格 - 前一根K线实体/5 → 平多开空 - - :param curr_kline: 当前3分钟K线数据 - :param kline_data: 所有K线数据（用于获取前一根K线实体） - :return: (方向, 触发价格) 或 (None, None) - """ - # 检查是否有持仓 - if self.start == 0: - return None, None - - curr_kline_id = curr_kline['id'] - - # 如果K线切换了，重置第一分钟反手标记 - if self.entry_kline_id != curr_kline_id: - self.first_minute_reverse_executed = False - self.entry_kline_id = curr_kline_id # 更新当前K线ID - - # 检查当前K线是否已执行过第一分钟反手 - if self.first_minute_reverse_executed: - return None, None - - # 获取开仓价格（如果没有记录，使用API返回的开仓均价） - entry_price = self.entry_price - if entry_price is None and self.open_avg_price: - entry_price = float(self.open_avg_price) - if entry_price is None: - return None, None - - # 获取前一根有效K线的实体大小 - valid_prev_idx, valid_prev = self.find_valid_prev_bar(kline_data, len(kline_data) - 1, self.min_body_size) - if valid_prev is None: - return None, None - prev_body = self.get_body_size(valid_prev) - - # 计算反手触发价格 - reverse_offset = prev_body / 5 - - # 获取第一分钟K线 - bars_1m = self.get_1m_bars_for_3m_bar(curr_kline) - if not bars_1m or len(bars_1m) < 1: - return None, None - - first_1m = bars_1m[0] # 第一分钟K线 - first_1m_high = float(first_1m['high']) - first_1m_low = float(first_1m['low']) - first_1m_close = float(first_1m['close']) - - if self.start == -1: - # 持有空仓，检测反手开多信号 - # 反手触发价 = 开仓价格 + 前一根实体/5 - reverse_long_trigger = entry_price + reverse_offset - - # 检查第一分钟是否触及反手触发价 - if first_1m_high >= reverse_long_trigger: - # 第一分钟高点触及反手触发价，并且当前价格在触发价附近 - if first_1m_close >= reverse_long_trigger - self.reverse_price_tolerance: - logger.info(f"🔄 第一分钟反手信号检测：持空仓，第一分钟高点{first_1m_high:.2f}>=反手触发价{reverse_long_trigger:.2f}") - logger.info(f" 开仓价={entry_price:.2f}, 前一根实体={prev_body:.2f}, 实体/5={reverse_offset:.2f}") - return 'long', reverse_long_trigger - else: - logger.debug(f"第一分钟反手信号被过滤：当前价格{first_1m_close:.2f}已远离触发价{reverse_long_trigger:.2f}") - - elif self.start == 1: - # 持有多仓，检测反手开空信号 - # 反手触发价 = 开仓价格 - 前一根实体/5 - reverse_short_trigger = entry_price - reverse_offset - - # 检查第一分钟是否触及反手触发价 - if first_1m_low <= reverse_short_trigger: - # 第一分钟低点触及反手触发价，并且当前价格在触发价附近 - if first_1m_close <= reverse_short_trigger + self.reverse_price_tolerance: - logger.info(f"🔄 第一分钟反手信号检测：持多仓，第一分钟低点{first_1m_low:.2f}<=反手触发价{reverse_short_trigger:.2f}") - logger.info(f" 开仓价={entry_price:.2f}, 前一根实体={prev_body:.2f}, 实体/5={reverse_offset:.2f}") - return 'short', reverse_short_trigger - else: - logger.debug(f"第一分钟反手信号被过滤：当前价格{first_1m_close:.2f}已远离触发价{reverse_short_trigger:.2f}") - - return None, None - - def check_realtime_trigger(self, kline_data, current_position=0): - """ - 实时检测当前3分钟K线是否触发信号 + 检查当前3分钟K线是否触发交易信号（与回测逻辑完全一致） + 若同时触发多空，则用该3分钟内的1分钟K线判断先后顺序。 参数: kline_data: K线数据列表 - current_position: 当前持仓状态 (1=多, -1=空, 0=无) - - 逻辑优化： - - 当已有持仓时，只关心反向信号（有多仓只看空信号，有空仓只看多信号） - - 无仓位时，用1分钟K线判断先触发的方向 - - 【重要】反手信号不仅要求K线高/低点触及触发价，还要求当前价格在触发价附近 - 避免"先涨后跌"或"先跌后涨"的情况下错误开仓 返回：(方向, 触发价格, 有效前一根K线, 当前K线) 或 (None, None, None, None) """ @@ -288,60 +190,44 @@ class BitmartOneFifthStrategy: c_high = float(curr['high']) c_low = float(curr['low']) - c_close = float(curr['close']) # 当前价格（用于检查价格是否仍在触发价附近） - long_triggered = c_high >= long_trigger short_triggered = c_low <= short_trigger + both_triggered = long_triggered and short_triggered direction = None trigger_price = None - # 关键优化：根据当前持仓状态决定关注哪个方向的信号 - if current_position == 1: - # 当前是多仓，只关心空信号（用于平多开空） - if short_triggered: - # 【重要】额外检查：当前价格必须在做空触发价附近或下方 - # 如果价格已经涨回去了（当前价格远高于做空触发价），则不触发 - if c_close <= short_trigger + self.reverse_price_tolerance: + # 如果同时触发多空，用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: - logger.debug(f"空信号被过滤：当前价格{c_close:.2f}已远离做空触发价{short_trigger:.2f}（容差{self.reverse_price_tolerance}）") - - elif current_position == -1: - # 当前是空仓，只关心多信号（用于平空开多） - if long_triggered: - # 【重要】额外检查：当前价格必须在做多触发价附近或上方 - # 如果价格已经跌回去了（当前价格远低于做多触发价），则不触发 - if c_close >= long_trigger - self.reverse_price_tolerance: direction = 'long' trigger_price = long_trigger - else: - logger.debug(f"多信号被过滤：当前价格{c_close:.2f}已远离做多触发价{long_trigger:.2f}（容差{self.reverse_price_tolerance}）") - else: - # 无仓位，按原逻辑判断先触发的方向 - if long_triggered and short_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 - ) - trigger_price = long_trigger if direction == 'long' else short_trigger - if direction is None: - c_open = float(curr['open']) - d_long = abs(long_trigger - c_open) - d_short = abs(short_trigger - c_open) - direction = 'short' if d_short <= d_long else 'long' - trigger_price = long_trigger if direction == 'long' else short_trigger - elif short_triggered: - direction = 'short' - trigger_price = short_trigger - elif long_triggered: - 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 @@ -583,73 +469,12 @@ class BitmartOneFifthStrategy: curr = kline_data[-1] curr_time_str = datetime.datetime.fromtimestamp(curr['id']).strftime('%H:%M:%S') - # 优化：先获取持仓状态，再检测信号（传入持仓状态以便只关注反向信号） + # 获取持仓状态 if not self.get_position_status(): logger.warning("获取仓位信息失败，使用缓存的持仓状态") - # ========== 第一分钟反手信号检测 ========== - # 如果有持仓，先检测基于开仓价格的第一分钟反手信号 - first_min_direction = None - first_min_trigger_price = None - if self.start != 0: - first_min_direction, first_min_trigger_price = self.check_first_minute_reverse_signal(curr, kline_data) - - # 如果检测到第一分钟反手信号，优先执行 - if first_min_direction: - curr_kline_id = curr['id'] - if self.last_trade_kline_id != curr_kline_id: - logger.info(f"{'=' * 50}") - # 安全获取开仓价格和前一根实体 - entry_price_display = self.entry_price if self.entry_price else (float(self.open_avg_price) if self.open_avg_price else 0) - entry_body_display = self.entry_prev_body if self.entry_prev_body else 0 - logger.info(f"🔄 第一分钟反手信号触发！方向: {first_min_direction}, 触发价: {first_min_trigger_price:.2f}") - logger.info(f" 开仓价: {entry_price_display:.2f}, 前一根实体/5: {entry_body_display/5:.2f}") - logger.info(f" 当前持仓: {self.start} (1=多, -1=空)") - - balance = self.get_available_balance() - trade_size = (balance or 0) * self.risk_percent - - if first_min_direction == 'long' and self.start == -1: - logger.info("📈 第一分钟反手：平空仓，反手开多") - self.平仓() - time.sleep(1) - self.开单(marketPriceLongOrder=1, size=trade_size) - self.first_minute_reverse_executed = True - self.last_trade_kline_id = curr_kline_id - # 更新开仓信息 - valid_prev_idx, valid_prev = self.find_valid_prev_bar(kline_data, len(kline_data) - 1, self.min_body_size) - if valid_prev: - self.entry_prev_body = self.get_body_size(valid_prev) - self.entry_price = float(curr['close']) - self.entry_kline_id = curr_kline_id - self.get_position_status() - self._send_position_message(curr) - - elif first_min_direction == 'short' and self.start == 1: - logger.info("📉 第一分钟反手：平多仓，反手开空") - self.平仓() - time.sleep(1) - self.开单(marketPriceLongOrder=-1, size=trade_size) - self.first_minute_reverse_executed = True - self.last_trade_kline_id = curr_kline_id - # 更新开仓信息 - valid_prev_idx, valid_prev = self.find_valid_prev_bar(kline_data, len(kline_data) - 1, self.min_body_size) - if valid_prev: - self.entry_prev_body = self.get_body_size(valid_prev) - self.entry_price = float(curr['close']) - self.entry_kline_id = curr_kline_id - self.get_position_status() - self._send_position_message(curr) - - logger.info(f"{'=' * 50}") - time.sleep(self.check_interval) - continue - - # ========== 原有的五分之一信号检测 ========== - # 传入当前持仓状态，确保有持仓时只关注反向信号 - direction, trigger_price, valid_prev, curr_kline = self.check_realtime_trigger( - kline_data, current_position=self.start - ) + # 检测五分之一策略信号（与回测逻辑完全一致） + direction, trigger_price, valid_prev, curr_kline = self.check_realtime_trigger(kline_data) if direction: curr_kline_id = curr_kline['id'] @@ -663,15 +488,6 @@ class BitmartOneFifthStrategy: prev_type = "阳线" if self.is_bullish(valid_prev) else "阴线" prev_body = self.get_body_size(valid_prev) - # 由于 check_realtime_trigger 已经考虑了持仓状态，这里理论上不会出现同向信号 - # 但保留这个检查作为安全措施 - if (direction == "long" and self.start == 1) or (direction == "short" and self.start == -1): - logger.debug(f"同向信号被过滤: direction={direction}, position={self.start}") - self.last_trigger_kline_id = curr_kline_id - self.last_trigger_direction = direction - time.sleep(self.check_interval) - continue - logger.info(f"{'=' * 50}") logger.info(f"🚨 检测到{direction}信号！触发价格: {trigger_price:.2f}") logger.info( @@ -684,40 +500,44 @@ class BitmartOneFifthStrategy: trade_size = (balance or 0) * self.risk_percent executed = False + # 执行交易逻辑（与回测一致） if direction == "long": if self.start == -1: + # 持空仓遇到做多信号 -> 平空并反手开多 logger.info("📈 平空仓，反手开多") self.平仓() time.sleep(1) self.开单(marketPriceLongOrder=1, size=trade_size) executed = True elif self.start == 0: + # 无仓位遇到做多信号 -> 开多 logger.info("📈 无仓位，开多") self.开单(marketPriceLongOrder=1, size=trade_size) executed = True + elif self.start == 1: + # 持多仓遇到做多信号 -> 不操作 + logger.debug("已持有多仓，忽略做多信号") elif direction == "short": if self.start == 1: + # 持多仓遇到做空信号 -> 平多并反手开空 logger.info("📉 平多仓，反手开空") self.平仓() time.sleep(1) self.开单(marketPriceLongOrder=-1, size=trade_size) executed = True elif self.start == 0: + # 无仓位遇到做空信号 -> 开空 logger.info("📉 无仓位，开空") self.开单(marketPriceLongOrder=-1, size=trade_size) executed = True + elif self.start == -1: + # 持空仓遇到做空信号 -> 不操作 + logger.debug("已持有空仓，忽略做空信号") self.last_trigger_kline_id = curr_kline_id self.last_trigger_direction = direction if executed: self.last_trade_kline_id = curr_kline_id - # 记录开仓信息，用于第一分钟反手信号检测 - self.entry_price = trigger_price # 开仓触发价格 - self.entry_prev_body = self.get_body_size(valid_prev) # 前一根K线实体大小 - self.entry_kline_id = curr_kline_id # 开仓时的K线ID - self.first_minute_reverse_executed = False # 重置第一分钟反手标记 - logger.info(f" 记录开仓信息：开仓价={self.entry_price:.2f}, 前一根实体={self.entry_prev_body:.2f}, K线ID={self.entry_kline_id}") - self.get_position_status() self._send_position_message(curr_kline) last_report_time = time.time()