优化bitmart交易

This commit is contained in:
27942
2026-02-03 23:27:14 +08:00
parent fd63961e3c
commit 757b7033fe
2 changed files with 78 additions and 94 deletions

10
1111
View File

@@ -24,8 +24,8 @@ Abc12345678
基于前一根有效 K 线(实体 ≥ 0.1
做多触发价 = 当前 k 线开盘价 + 实体 / 5(收盘价向上 1/5 实体)
做空触发价 = 当前 k 线开盘价 - 实体 / 5(收盘价向下 1/5 实体)
做多触发价 = 当前 k 线开盘价 + 实体 / 3(收盘价向上 1/3 实体)
做空触发价 = 当前 k 线开盘价 - 实体 / 3(收盘价向下 1/3 实体)
若已有持仓,在 3 分钟 K 线的第一分钟可单独检测反手:
持空反手做多:价格涨到 开仓价 + 前一根实体 / 5
@@ -35,8 +35,10 @@ Abc12345678
反手信号,
当前这根线线跌倒了上一根的三分之一,我开了空,但是这根涨到了上一根 k 线的最高价格,上一根 k 线的上阴线涨幅大于 0.01%,反手做多
第二个反手信号,当前这根做了多,上一根 k 线的上阴线涨幅大于 0.01%,当前这个根跌倒了上一根的开盘价格,反手做空
反手信号:
反手做空,当前这根做多,上一根 k 线的上阴线涨幅大于 0.01%当前这个根跌倒了上一根的k线的实体边反手做空
反手做多,当前这根做空,上一根 k 线的下阴线跌幅大于 0.01%当前这个根涨到了上一根的k线的实体边反手做多

View File

@@ -1,12 +1,12 @@
"""
BitMart 三分之一策略 — 5分钟K线交易
策略规则(与 交易/bitmart-三分之一策略交易.py、回测数据-三分之一策略-5分钟精准版 一致)
1. 触发价格基于前一根有效K线实体>=0.1
- 做多触发价 = 盘价 + 实体/3从收盘价往上涨1/3
- 做空触发价 = 盘价 - 实体/3从收盘价往下跌1/3
策略规则:
1. 触发价格基于前一根有效K线实体>=0.1使用当前K线开盘价
- 做多触发价 = 当前K线开盘价 + 实体/3
- 做空触发价 = 当前K线开盘价 - 实体/3
2. 信号触发:
2. 信号触发(无持仓时)
- 当前5分钟K线最高价 >= 做多触发价 → 做多信号
- 当前5分钟K线最低价 <= 做空触发价 → 做空信号
@@ -17,14 +17,11 @@ BitMart 三分之一策略 — 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线涨到上一根开盘价 -> 平空反手开多
5. 反手信号(有持仓时检测
- 反手做:当前持多,上一根K线上线涨幅 > 0.01%当前K线跌到上一根K线的实体下边 -> 平反手开
- 反手做:当前持空,上一根K线下线跌幅 > 0.01%当前K线涨到上一根K线的实体上边 -> 平反手开
(上线 = high - max(open,close),上线涨幅 = (上线/开盘价)*100线同理)
(实体上边 = max(open,close),实体下边 = min(open,close)
"""
import random
import time
@@ -48,7 +45,7 @@ ding_executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="dingtalk")
class BitmartOneThirdStrategy:
"""三分之一策略:触发价 = 盘价 ± 实体/35分钟K线"""
"""三分之一策略:触发价 = 当前K线开盘价 ± 实体/35分钟K线"""
def __init__(self, bit_id):
self.page: ChromiumPage | None = None
@@ -100,18 +97,20 @@ class BitmartOneThirdStrategy:
return i, prev
return None, None
def get_one_third_levels(self, prev):
def get_one_third_levels(self, prev, curr):
"""
做多触发价 = 盘价 + 实体/3
做空触发价 = 盘价 - 实体/3
做多触发价 = 当前K线开盘价 + 实体/3
做空触发价 = 当前K线开盘价 - 实体/3
实体来自前一根有效K线
"""
p_open = float(prev['open'])
p_close = float(prev['close'])
curr_open = float(curr['open'])
body = abs(p_open - p_close)
if body < 0.001:
return None, None
long_trigger = p_close + body / 3
short_trigger = p_close - body / 3
long_trigger = curr_open + body / 3
short_trigger = curr_open - body / 3
return long_trigger, short_trigger
def get_upper_shadow(self, candle):
@@ -138,36 +137,19 @@ class BitmartOneThirdStrategy:
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'])
def get_body_upper_edge(self, candle):
"""实体上边 = max(open, close)"""
return max(float(candle['open']), float(candle['close']))
# 持空反手做多:当前涨到上一根最高价,且上一根上影线涨幅 > 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 get_body_lower_edge(self, candle):
"""实体下边 = min(open, close)"""
return min(float(candle['open']), float(candle['close']))
def check_reverse_by_prev_open(self, kline_data):
def check_reverse_signal(self, kline_data):
"""
反手信号二:持多时上一根上影线涨幅>0.01%且当前K线跌到上一根开盘价 -> 反手做空;
持空时上一根下影线跌幅>0.01%当前K线到上一根开盘价 -> 反手做多。
反手信号(有持仓时检测):
- 反手做空当前持多上一根K线上阴线涨幅 > 0.01%当前K线到上一根K线的实体下边
- 反手做多当前持空上一根K线下阴线跌幅 > 0.01%当前K线涨到上一根K线的实体上边
使用紧邻的上一根K线kline_data[-2])。
返回:(方向 'long'|'short', 上一根K线) 或 (None, None)
"""
@@ -177,22 +159,23 @@ class BitmartOneThirdStrategy:
prev = kline_data[-2]
curr_high = float(curr['high'])
curr_low = float(curr['low'])
prev_open = float(prev['open'])
prev_body_upper = self.get_body_upper_edge(prev) # 实体上边
prev_body_lower = self.get_body_lower_edge(prev) # 实体下边
# 持多反手做空:上一根上线涨幅>0.01%,当前跌到上一根开盘价
# 持多反手做空:上一根上线涨幅>0.01%,当前跌到上一根实体下边
if self.start == 1:
if self.upper_shadow_pct(prev) > self.min_shadow_pct and curr_low <= prev_open:
if self.upper_shadow_pct(prev) > self.min_shadow_pct and curr_low <= prev_body_lower:
return 'short', prev
# 持空反手做多:上一根下线跌幅>0.01%,当前涨到上一根开盘价
# 持空反手做多:上一根下线跌幅>0.01%,当前涨到上一根实体上边
if self.start == -1:
if self.lower_shadow_pct(prev) > self.min_shadow_pct and curr_high >= prev_open:
if self.lower_shadow_pct(prev) > self.min_shadow_pct and curr_high >= prev_body_upper:
return 'long', prev
return None, None
def check_realtime_trigger(self, kline_data):
"""
检查当前5分钟K线是否触发信号。
做多触发价 = 前一根收盘价 + 实体/3做空 = 盘价 - 实体/3。
做多触发价 = 当前K线开盘价 + 实体/3做空 = 当前K线开盘价 - 实体/3。
同根多空都触发时用开盘价与触发价距离判断先后。
返回:(方向, 触发价格, 有效前一根K线, 当前K线) 或 (None, None, None, None)
"""
@@ -209,7 +192,7 @@ class BitmartOneThirdStrategy:
if prev is None:
return None, None, None, None
long_trigger, short_trigger = self.get_one_third_levels(prev)
long_trigger, short_trigger = self.get_one_third_levels(prev, curr)
if long_trigger is None:
return None, None, None, None
@@ -457,12 +440,13 @@ class BitmartOneThirdStrategy:
try:
kline_data = self.get_klines()
if not kline_data or len(kline_data) < 3:
if not kline_data or len(kline_data) < 2:
logger.warning("K线数据不足等待重试...")
time.sleep(self.check_interval)
continue
curr = kline_data[-1]
prev = kline_data[-2]
curr_time_str = datetime.datetime.fromtimestamp(curr['id']).strftime('%H:%M:%S')
if not self.get_position_status():
@@ -470,42 +454,40 @@ class BitmartOneThirdStrategy:
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 '最低'}")
# 有持仓时优先检测反手信号
if self.start != 0:
rev_dir, rev_prev = self.check_reverse_signal(kline_data)
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)
prev_body_upper = self.get_body_upper_edge(rev_prev)
prev_body_lower = self.get_body_lower_edge(rev_prev)
logger.info(f"{'='*50}")
if rev_dir == 'long':
logger.info(f"🔄 反手信号({rev_dir})当前K线涨到上一根实体上边={prev_body_upper:.2f}")
else:
logger.info(f"🔄 反手信号({rev_dir})当前K线跌到上一根实体下边={prev_body_lower:.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(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
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)
@@ -531,7 +513,7 @@ class BitmartOneThirdStrategy:
logger.info(f"{'='*50}")
logger.info(f"🚨 检测到{direction}信号 触发价={trigger_price:.2f}")
logger.info(f" 有效前一根[{prev_time}] {prev_type} 实体={prev_body:.2f}")
logger.info(f" 当前5分钟K线 H={curr_kline['high']:.2f} L={curr_kline['low']:.2f} C={curr_kline['close']:.2f}")
logger.info(f" 当前5分钟K线 O={curr_kline['open']:.2f} H={curr_kline['high']:.2f} L={curr_kline['low']:.2f} C={curr_kline['close']:.2f}")
logger.info(f" 当前持仓: {self.start} (1=多 -1=空 0=无)")
balance = self.get_available_balance()