ddrwode/lm_code
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

代码结构优化

Browse Source
This commit is contained in:
ddrwode
2026-02-02 17:18:31 +08:00
parent 71c025b627
commit 488e8de86a
1 changed files with 62 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files

304
交易/bitmart-五分之一策略交易.py
View File

@@ -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()