Files
lm_code/bitmart/均线自动化开单.py
2025-12-29 18:29:21 +08:00

1166 lines
42 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import time
import datetime
from typing import Tuple, Optional, List
from dataclasses import dataclass, field
from tqdm import tqdm
from loguru import logger
from bitmart.api_contract import APIContract
from DrissionPage import ChromiumPage, ChromiumOptions
from bit_tools import openBrowser
from 交易.tools import send_dingtalk_message
@dataclass
class StrategyConfig:
"""针对90%返佣优化的均值回归策略配置"""
# =============================
# ETH 永续 | 1分钟K线 | 90%返佣优化
# =============================
# ===== 合约配置 =====
contract_symbol: str = "ETHUSDT"
open_type: str = "cross"
leverage: str = "30"
# ===== K线与指标 =====
step_min: int = 1
lookback_min: int = 240
ema_len: int = 36
atr_len: int = 14
# ===== 动态阈值基准 =====
vol_baseline_window: int = 60
vol_baseline_quantile: float = 0.65
vol_scale_min: float = 0.80
vol_scale_max: float = 1.60
base_ratio_refresh_sec: int = 60
# ===== 动态floor配置针对低手续费优化 =====
entry_dev_floor_min: float = 0.0008 # 0.08%原0.12%
entry_dev_floor_max: float = 0.0020 # 0.20%原0.30%
entry_dev_floor_base_k: float = 1.10
tp_floor_min: float = 0.0004 # 0.04%原0.06%
tp_floor_max: float = 0.0015 # 0.15%原0.20%
tp_floor_base_k: float = 0.55
sl_floor_min: float = 0.0012 # 0.12%原0.18%
sl_floor_max: float = 0.0040 # 0.40%原0.60%
sl_floor_base_k: float = 1.35
# ===== 阈值倍率(针对低手续费优化) =====
entry_k: float = 1.30 # 原1.45
tp_k: float = 0.55 # 原0.65
sl_k: float = 0.90 # 原1.05
# ===== 时间/冷却(降低以增加频率) =====
max_hold_sec: int = 60 # 原75秒
cooldown_sec_after_exit: int = 5 # 原20秒
opposite_direction_cooldown: int = 30 # 平仓后同向冷却30秒
# ===== 仓位管理 =====
fixed_margin: float = 10.0 # 固定保证金
min_size: int = 1
max_size: int = 10
max_daily_trades: int = 50 # 每日最大交易次数
# ===== 日内风控 =====
daily_loss_limit: float = 0.02
daily_profit_cap: float = 0.01
# ===== 危险模式过滤 =====
atr_ratio_kill: float = 0.0038
big_body_kill: float = 0.010
# ===== 轮询节奏 =====
klines_refresh_sec: int = 10
tick_refresh_sec: int = 1
status_notify_sec: int = 60
# ===== 止损后机制 =====
reentry_penalty_mult: float = 1.55
reentry_penalty_max_sec: int = 180
reset_band_k: float = 0.45
reset_band_floor: float = 0.0006
post_sl_sl_max_sec: int = 90
post_sl_mult_min: float = 1.02
post_sl_mult_max: float = 1.16
post_sl_vol_alpha: float = 0.20
# ===== 正常平仓条件(针对低手续费优化) =====
normal_exit_threshold: float = 0.0002 # 0.02%原0.03%
normal_exit_min_profit: float = 0.0001 # 0.01%原0.02%
# ===== 90%返佣配置 =====
platform_fee_rate: float = 0.0005 # 平台手续费:开仓价值的万分之五
rebate_rate: float = 0.90 # 返佣比例90%
# ===== 新增:趋势过滤 =====
use_trend_filter: bool = True
trend_ema_len: int = 50
trend_threshold: float = 0.0005 # 0.05%
# ===== 新增:成交量确认 =====
require_volume_confirmation: bool = True
volume_ma_len: int = 20
volume_threshold: float = 1.2
# ===== 新增:连续亏损控制 =====
max_consecutive_losses: int = 3
recovery_wait_sec: int = 60
class BitmartFuturesMeanReversionBot:
def __init__(self, cfg: StrategyConfig, bit_id=None):
self.bit_id = bit_id
self.page: Optional[ChromiumPage] = None
self.cfg = cfg
self.api_key = "a0fb7b98464fd9bcce67e7c519d58ec10d0c38a8"
self.secret_key = "4eaeba78e77aeaab1c2027f846a276d164f264a44c2c1bb1c5f3be50c8de1ca5"
self.memo = "合约交易"
if not self.api_key or not self.secret_key:
raise RuntimeError("请先设置API密钥")
self.contractAPI = APIContract(self.api_key, self.secret_key, self.memo, timeout=(5, 15))
# 持仓状态
self.pos = 0 # -1 空, 0 无, 1 多
self.entry_price: Optional[float] = None
self.entry_ts: Optional[float] = None
self.last_exit_ts = 0.0
# 新增:记录上次平仓方向
self.last_exit_direction = 0 # 1=多平, -1=空平
self.last_exit_price = 0.0
# 开仓信息
self.entry_margin: Optional[float] = None
self.entry_position_value: Optional[float] = None
# 日内权益基准
self.day_start_equity: Optional[float] = None
self.trading_enabled = True
self.day_tag = datetime.date.today()
self.today_trade_count = 0
self.consecutive_losses = 0
# 缓存
self._klines_cache: Optional[List] = None
self._klines_cache_ts = 0.0
self._last_status_notify_ts = 0.0
# 基准波动率缓存
self._base_ratio_cached = 0.0015
self._base_ratio_ts = 0.0
# 止损后机制状态
self.last_sl_dir = 0
self.last_sl_ts = 0.0
self.post_sl_dir = 0
self.post_sl_ts = 0.0
self.post_sl_vol_scale = 1.0
self.pbar = tqdm(total=60, desc="运行中(秒)", ncols=90)
logger.info(f"初始化完成 - 合约:{cfg.contract_symbol}, 杠杆:{cfg.leverage}x, 返佣:{cfg.rebate_rate * 100}%")
# ----------------- 工具函数 -----------------
def ding(self, msg: str, error: bool = False):
"""发送钉钉通知"""
prefix = "❌bitmart" if error else "🔔bitmart"
if error:
for _ in range(1):
send_dingtalk_message(f"{prefix}{msg}")
else:
send_dingtalk_message(f"{prefix}{msg}")
def set_leverage(self) -> bool:
"""设置杠杆"""
try:
resp = self.contractAPI.post_submit_leverage(
contract_symbol=self.cfg.contract_symbol,
leverage=self.cfg.leverage,
open_type=self.cfg.open_type
)[0]
if resp.get("code") == 1000:
logger.success(f"设置杠杆成功:{self.cfg.open_type} + {self.cfg.leverage}x")
return True
logger.error(f"设置杠杆失败: {resp}")
return False
except Exception as e:
logger.error(f"设置杠杆异常: {e}")
return False
# ----------------- 行情数据 -----------------
def get_klines_cached(self) -> Optional[List]:
"""获取缓存的K线数据"""
now = time.time()
if self._klines_cache is not None and (now - self._klines_cache_ts) < self.cfg.klines_refresh_sec:
return self._klines_cache
kl = self.get_klines()
if kl:
self._klines_cache = kl
self._klines_cache_ts = now
return self._klines_cache
def get_klines(self) -> Optional[List]:
"""获取K线数据"""
try:
end_time = int(time.time())
start_time = end_time - 60 * self.cfg.lookback_min
resp = self.contractAPI.get_kline(
contract_symbol=self.cfg.contract_symbol,
step=self.cfg.step_min,
start_time=start_time,
end_time=end_time
)[0]
if resp.get("code") != 1000:
logger.error(f"获取K线失败: {resp}")
return None
data = resp.get("data", [])
formatted = []
for k in data:
formatted.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"]),
})
formatted.sort(key=lambda x: x["id"])
return formatted
except Exception as e:
logger.error(f"获取K线异常: {e}")
return None
def get_last_price(self, fallback_close: float) -> float:
"""获取最新价格"""
try:
if hasattr(self.contractAPI, "get_contract_details"):
r = self.contractAPI.get_contract_details(contract_symbol=self.cfg.contract_symbol)[0]
d = r.get("data") if isinstance(r, dict) else None
if isinstance(d, dict):
for key in ("last_price", "mark_price", "index_price"):
if key in d and d[key] is not None:
return float(d[key])
if hasattr(self.contractAPI, "get_ticker"):
r = self.contractAPI.get_ticker(contract_symbol=self.cfg.contract_symbol)[0]
d = r.get("data") if isinstance(r, dict) else None
if isinstance(d, dict):
for key in ("last_price", "price", "last", "close"):
if key in d and d[key] is not None:
return float(d[key])
except Exception:
pass
return float(fallback_close)
@staticmethod
def ema(values: List[float], n: int) -> float:
"""计算EMA"""
k = 2 / (n + 1)
e = values[0]
for v in values[1:]:
e = v * k + e * (1 - k)
return e
@staticmethod
def atr(klines: List[dict], n: int) -> float:
"""计算ATR"""
if len(klines) < n + 1:
return 0.0
trs = []
for i in range(-n, 0):
cur = klines[i]
prev = klines[i - 1]
tr = max(
cur["high"] - cur["low"],
abs(cur["high"] - prev["close"]),
abs(cur["low"] - prev["close"]),
)
trs.append(tr)
return sum(trs) / len(trs)
# ----------------- 风控检查 -----------------
def is_danger_market(self, klines: List[dict], price: float) -> bool:
"""检查是否危险市场"""
last = klines[-1]
body = abs(last["close"] - last["open"]) / last["open"] if last["open"] else 0.0
if body >= self.cfg.big_body_kill:
logger.warning(f"大实体K线: {body * 100:.2f}%")
return True
a = self.atr(klines, self.cfg.atr_len)
atr_ratio = (a / price) if price > 0 else 0.0
if atr_ratio >= self.cfg.atr_ratio_kill:
logger.warning(f"高ATR比率: {atr_ratio * 100:.2f}%")
return True
return False
def check_trend_filter(self, price: float) -> bool:
"""趋势过滤器"""
if not self.cfg.use_trend_filter:
return True
klines = self.get_klines_cached()
if not klines or len(klines) < self.cfg.trend_ema_len:
return True
try:
closes = [k["close"] for k in klines[-(self.cfg.trend_ema_len + 1):]]
trend_ema = self.ema(closes, self.cfg.trend_ema_len)
trend_dev = (price - trend_ema) / trend_ema if trend_ema else 0.0
if abs(trend_dev) > self.cfg.trend_threshold:
logger.debug(f"趋势过滤触发: 偏差={trend_dev * 100:.3f}%, 阈值={self.cfg.trend_threshold * 100:.3f}%")
return False
return True
except Exception as e:
logger.error(f"趋势过滤异常: {e}")
return True
# ----------------- 波动率计算 -----------------
def atr_ratio_baseline(self, klines: List[dict]) -> float:
"""计算ATR比率基准"""
window = min(self.cfg.vol_baseline_window, len(klines) - self.cfg.atr_len - 1)
if window <= 10:
logger.warning(f"数据不足计算基准: {len(klines)}根K线")
return 0.0
ratios = []
step = 3
for i in range(-window, 0, step):
if len(klines) + i < self.cfg.atr_len + 1:
continue
start_idx = len(klines) + i - self.cfg.atr_len
end_idx = len(klines) + i
if start_idx < 0 or end_idx <= start_idx:
continue
sub_klines = klines[start_idx:end_idx]
if len(sub_klines) >= self.cfg.atr_len + 1:
a = self.atr(sub_klines, self.cfg.atr_len)
price = klines[end_idx - 1]["close"]
if a > 0 and price > 0:
ratio = a / price
if 0.0001 < ratio < 0.01:
ratios.append(ratio)
if len(ratios) < 5:
a = self.atr(klines[-60:], self.cfg.atr_len)
price = klines[-1]["close"]
if a > 0 and price > 0:
baseline = a / price
logger.debug(f"使用全量数据计算基准: {baseline * 100:.4f}%")
return baseline
else:
return 0.0
ratios.sort()
idx = min(len(ratios) - 1,
max(0, int(self.cfg.vol_baseline_quantile * (len(ratios) - 1))))
baseline = ratios[idx]
logger.debug(f"基准计算: 样本数={len(ratios)}, 基准={baseline * 100:.4f}%")
return baseline
def get_base_ratio_cached(self, klines: List[dict]) -> float:
"""获取缓存的基准波动率"""
now = time.time()
refresh_sec = self.cfg.base_ratio_refresh_sec
if (self._base_ratio_cached is None or
(now - self._base_ratio_ts) >= refresh_sec):
baseline = self.atr_ratio_baseline(klines)
if baseline > 0.0001:
self._base_ratio_cached = baseline
self._base_ratio_ts = now
logger.info(f"基准波动率更新: {baseline * 100:.4f}%")
else:
current_price = klines[-1]["close"] if klines else 3000
if current_price > 4000:
default_baseline = 0.0010
elif current_price > 3500:
default_baseline = 0.0012
elif current_price > 3000:
default_baseline = 0.0015
elif current_price > 2500:
default_baseline = 0.0018
else:
default_baseline = 0.0020
self._base_ratio_cached = default_baseline
self._base_ratio_ts = now
logger.warning(f"使用价格动态默认基准: {default_baseline * 100:.4f}%")
return self._base_ratio_cached
@staticmethod
def _clamp(x: float, lo: float, hi: float) -> float:
"""限制数值在指定范围内"""
return max(lo, min(hi, x))
# ----------------- 动态阈值计算 -----------------
def dynamic_thresholds(self, atr_ratio: float, base_ratio: float) -> Tuple[
float, float, float, float, float, float, float]:
"""计算动态阈值"""
if atr_ratio <= 0:
logger.warning(f"ATR比率异常: {atr_ratio}")
atr_ratio = 0.001
if base_ratio < 0.0005:
base_ratio = max(0.001, atr_ratio * 1.2)
logger.debug(f"基准太小使用调整后的atr_ratio: {base_ratio * 100:.4f}%")
# vol_scale计算
if base_ratio > 0:
raw_scale = atr_ratio / base_ratio
vol_scale = self._clamp(raw_scale, self.cfg.vol_scale_min, self.cfg.vol_scale_max)
else:
vol_scale = 1.0
# 动态floor计算
entry_floor_raw = self.cfg.entry_dev_floor_base_k * base_ratio
entry_floor = self._clamp(
entry_floor_raw,
self.cfg.entry_dev_floor_min,
self.cfg.entry_dev_floor_max,
)
tp_floor_raw = self.cfg.tp_floor_base_k * base_ratio
tp_floor = self._clamp(
tp_floor_raw,
self.cfg.tp_floor_min,
self.cfg.tp_floor_max,
)
sl_floor_raw = self.cfg.sl_floor_base_k * base_ratio
sl_floor = self._clamp(
sl_floor_raw,
self.cfg.sl_floor_min,
self.cfg.sl_floor_max,
)
# 最终阈值
entry_dev_atr_part = self.cfg.entry_k * vol_scale * atr_ratio
entry_dev = max(entry_floor, entry_dev_atr_part)
tp_atr_part = self.cfg.tp_k * vol_scale * atr_ratio
tp = max(tp_floor, tp_atr_part)
sl_atr_part = self.cfg.sl_k * vol_scale * atr_ratio
sl = max(sl_floor, sl_atr_part)
entry_dev = max(entry_dev, self.cfg.entry_dev_floor_min)
logger.info(
f"动态阈值: entry={entry_dev * 100:.4f}%, tp={tp * 100:.4f}%, sl={sl * 100:.4f}%, "
f"vol_scale={vol_scale:.2f}"
)
return entry_dev, tp, sl, vol_scale, entry_floor, tp_floor, sl_floor
# ----------------- 账户仓位管理 -----------------
def get_assets_available(self) -> float:
"""获取可用余额"""
try:
resp = self.contractAPI.get_assets_detail()[0]
if resp.get("code") != 1000:
return 0.0
data = resp.get("data")
if isinstance(data, dict):
return float(data.get("available_balance", 0))
if isinstance(data, list):
for asset in data:
if asset.get("currency") == "USDT":
return float(asset.get("available_balance", 0))
return 0.0
except Exception as e:
logger.error(f"余额查询异常: {e}")
return 0.0
def get_position_status(self) -> bool:
"""获取持仓状态"""
try:
resp = self.contractAPI.get_position(contract_symbol=self.cfg.contract_symbol)[0]
if resp.get("code") != 1000:
return False
positions = resp.get("data", [])
if not positions:
self.pos = 0
return True
p = positions[0]
self.pos = 1 if p["position_type"] == 1 else -1
return True
except Exception as e:
logger.error(f"持仓查询异常: {e}")
return False
def get_equity_proxy(self) -> float:
"""获取权益"""
return self.get_assets_available()
def refresh_daily_baseline(self):
"""刷新日内基准"""
today = datetime.date.today()
if today != self.day_tag:
self.day_tag = today
self.day_start_equity = None
self.trading_enabled = True
self.today_trade_count = 0
self.consecutive_losses = 0
logger.info(f"新的一天({today}):重置日内风控基准")
def risk_kill_switch(self):
"""风险控制开关"""
self.refresh_daily_baseline()
equity = self.get_equity_proxy()
if equity <= 0:
return
if self.day_start_equity is None:
self.day_start_equity = equity
logger.info(f"日内权益基准设定:{equity:.2f} USDT")
return
pnl = (equity - self.day_start_equity) / self.day_start_equity
if pnl <= -self.cfg.daily_loss_limit:
self.trading_enabled = False
self.ding(f"触发日止损:{pnl * 100:.2f}% -> 停机", error=True)
if pnl >= self.cfg.daily_profit_cap:
self.trading_enabled = False
self.ding(f"达到日盈利封顶:{pnl * 100:.2f}% -> 停机")
def calculate_size(self, price: float) -> int:
"""计算仓位大小"""
bal = self.get_assets_available()
if bal < self.cfg.fixed_margin:
logger.warning(f"余额不足:{bal:.2f} USDT < {self.cfg.fixed_margin} USDT")
return 0
margin = self.cfg.fixed_margin
lev = int(self.cfg.leverage)
size = int((margin * lev) / (price * 0.001))
size = max(self.cfg.min_size, size)
size = min(self.cfg.max_size, size)
logger.debug(f"计算仓位:保证金={margin}u, 杠杆={lev}x, size={size}")
return size
# ----------------- 浏览器操作 -----------------
def openBrowser(self) -> bool:
"""打开浏览器"""
try:
bit_port = openBrowser(id=self.bit_id)
co = ChromiumOptions()
co.set_local_port(port=bit_port)
self.page = ChromiumPage(addr_or_opts=co)
return True
except Exception as e:
logger.error(f"打开浏览器失败: {e}")
return False
def click_safe(self, xpath: str, sleep: float = 0.5) -> bool:
"""安全点击"""
try:
ele = self.page.ele(xpath)
if not ele:
return False
ele.scroll.to_see(center=True)
time.sleep(sleep)
ele.click()
return True
except Exception as e:
logger.error(f"点击失败: {e}")
return False
# ----------------- 订单执行 -----------------
def place_market_order(self, side: int, size: int) -> bool:
"""下市价单"""
if size <= 0:
return False
size = 10
try:
# 确保size在合理范围内
size = max(self.cfg.min_size, min(self.cfg.max_size, size))
# 开多单
if side == 1:
self.click_safe('x://button[normalize-space(text()) ="市价"]')
self.page.ele('x://*[@id="size_0"]').input(str(size), clear=True)
self.click_safe('x://span[normalize-space(text()) ="买入/做多"]')
logger.info(f"✅ 开多单: size={size}")
return True
# 开空单
elif side == 4:
self.click_safe('x://button[normalize-space(text()) ="市价"]')
self.page.ele('x://*[@id="size_0"]').input(str(size), clear=True)
self.click_safe('x://span[normalize-space(text()) ="卖出/做空"]')
logger.info(f"✅ 开空单: size={size}")
return True
# 平多单
elif side == 2:
self.click_safe('x://span[normalize-space(text()) ="市价"]')
time.sleep(0.3)
self.page.ele('x://*[@id="size_0"]').input(str(size), clear=True)
time.sleep(0.3)
self.click_safe('x://span[normalize-space(text()) ="卖出/做空"]')
logger.info(f"✅ 平多单: size={size}")
return True
# 平空单
elif side == 3:
self.click_safe('x://span[normalize-space(text()) ="市价"]')
time.sleep(0.3)
self.page.ele('x://*[@id="size_0"]').input(str(size), clear=True)
time.sleep(0.3)
self.click_safe('x://span[normalize-space(text()) ="买入/做多"]')
logger.info(f"✅ 平空单: size={size}")
return True
else:
logger.error(f"未知的订单方向: side={side}")
return False
except Exception as e:
logger.error(f"下单异常: {e}")
return False
def calculate_fee(self, position_value: float) -> float:
"""计算手续费考虑90%返佣)"""
platform_fee = position_value * self.cfg.platform_fee_rate
actual_fee = platform_fee * (1 - self.cfg.rebate_rate)
return actual_fee
def calculate_net_pnl(self, price: float) -> Tuple[float, float, float]:
"""计算净盈亏"""
if self.pos == 0 or self.entry_price is None or self.entry_margin is None or self.entry_position_value is None:
return 0.0, 0.0, 0.0
leverage = int(self.cfg.leverage)
# 计算价格变动比例
if self.pos == 1:
price_change_ratio = (price - self.entry_price) / self.entry_price
else:
price_change_ratio = (self.entry_price - price) / self.entry_price
# 计算毛盈亏比例(考虑杠杆)
gross_pnl_ratio = leverage * price_change_ratio
# 计算手续费
entry_fee = self.calculate_fee(self.entry_position_value)
exit_position_value = self.entry_position_value
exit_fee = self.calculate_fee(exit_position_value)
total_fee = entry_fee + exit_fee
# 手续费相对于保证金的比率
fee_ratio = total_fee / self.entry_margin
# 净盈亏 = 毛盈亏 - 手续费比率
net_pnl_ratio = gross_pnl_ratio - fee_ratio
return net_pnl_ratio, total_fee, gross_pnl_ratio
def close_position_all(self) -> bool:
"""平掉所有持仓"""
if self.pos == 0:
logger.info("当前无持仓,无需平仓")
return True
max_retries = 3
retry_delay = 1.0
for attempt in range(1, max_retries + 1):
logger.info(f"平仓尝试 {attempt}/{max_retries}...")
old_pos = self.pos
# 执行平仓操作
if self.pos == 1:
ok = self.place_market_order(2, 999999)
if not ok:
logger.warning(f"平多单操作失败 (尝试 {attempt}/{max_retries})")
if attempt < max_retries:
time.sleep(retry_delay)
continue
else:
self.ding(f"平多单失败:已重试{max_retries}次仍失败", error=True)
return False
elif self.pos == -1:
ok = self.place_market_order(3, 999999)
if not ok:
logger.warning(f"平空单操作失败 (尝试 {attempt}/{max_retries})")
if attempt < max_retries:
time.sleep(retry_delay)
continue
else:
self.ding(f"平空单失败:已重试{max_retries}次仍失败", error=True)
return False
else:
return True
# 等待订单执行
time.sleep(1.5)
# 验证是否平仓成功
verify_success = self._verify_position_closed(old_pos)
if verify_success:
self.pos = 0
logger.success(f"✅ 平仓成功 (尝试 {attempt}/{max_retries})")
return True
else:
logger.warning(f"平仓验证失败 (尝试 {attempt}/{max_retries})")
if attempt < max_retries:
time.sleep(retry_delay)
self.get_position_status()
else:
self.ding(f"平仓失败:已重试{max_retries}", error=True)
return False
return False
def _verify_position_closed(self, expected_old_pos: int) -> bool:
"""验证持仓是否已平仓"""
try:
resp = self.contractAPI.get_position(contract_symbol=self.cfg.contract_symbol)[0]
if resp.get("code") != 1000:
logger.warning(f"查询持仓状态失败: {resp}")
return False
positions = resp.get("data", [])
if not positions or len(positions) == 0:
logger.info("✅ SDK验证持仓已平仓")
return True
for p in positions:
position_type = p.get("position_type", 0)
current_pos = 1 if position_type == 1 else -1
if current_pos == expected_old_pos:
logger.warning(f"⚠️ SDK验证持仓仍存在")
return False
logger.info("✅ SDK验证持仓已平仓或已改变")
return True
except Exception as e:
logger.error(f"验证持仓状态异常: {e}")
return False
# ----------------- 止损后机制 -----------------
def _reentry_penalty_active(self, dev: float, entry_dev: float) -> bool:
"""检查是否需要应用重新入场惩罚"""
if self.last_sl_dir == 0:
return False
if (time.time() - self.last_sl_ts) > self.cfg.reentry_penalty_max_sec:
self.last_sl_dir = 0
return False
reset_band = max(self.cfg.reset_band_floor, self.cfg.reset_band_k * entry_dev)
if abs(dev) <= reset_band:
self.last_sl_dir = 0
return False
return True
def _post_sl_dynamic_mult(self) -> float:
"""计算止损后SL放宽倍数"""
if self.post_sl_dir == 0:
return 1.0
if (time.time() - self.post_sl_ts) > self.cfg.post_sl_sl_max_sec:
self.post_sl_dir = 0
self.post_sl_vol_scale = 1.0
return 1.0
raw = 1.0 + self.cfg.post_sl_vol_alpha * (self.post_sl_vol_scale - 1.0)
raw = max(1.0, raw)
return max(self.cfg.post_sl_mult_min, min(self.cfg.post_sl_mult_max, raw))
# ----------------- 交易逻辑(核心修复) -----------------
def in_cooldown(self) -> bool:
"""检查是否在冷却期内"""
return (time.time() - self.last_exit_ts) < self.cfg.cooldown_sec_after_exit
def maybe_enter(self, price: float, ema_value: float, entry_dev: float):
"""
检查并执行入场 - 修复版
修复问题:避免平仓后立即开反向仓位
"""
# 基础检查
if self.pos != 0:
return
if self.in_cooldown():
return
if not self.trading_enabled:
return
# 连续亏损控制
if self.consecutive_losses >= self.cfg.max_consecutive_losses:
logger.warning(f"连续亏损{self.consecutive_losses}次,暂停交易{self.cfg.recovery_wait_sec}")
time.sleep(self.cfg.recovery_wait_sec)
return
# 日内交易次数限制
if self.today_trade_count >= self.cfg.max_daily_trades:
logger.info(f"达到日交易次数上限: {self.today_trade_count}")
return
# 趋势过滤
if not self.check_trend_filter(price):
return
dev = (price - ema_value) / ema_value if ema_value else 0.0
size = self.calculate_size(price)
if size <= 0:
return
# 方向过滤:避免平仓后立即反向开仓
if self.last_exit_direction != 0:
time_since_exit = time.time() - self.last_exit_ts
# 如果是平多后(做空方向),且在冷却期内,避免立即开多
if (self.last_exit_direction == 1 and # 上次是多平
time_since_exit < self.cfg.opposite_direction_cooldown and
dev <= 0): # 当前适合开多
logger.info(f"平多后{int(time_since_exit)}秒内,避免开多")
return
# 如果是平空后(做多方向),且在冷却期内,避免立即开空
if (self.last_exit_direction == -1 and # 上次是空平
time_since_exit < self.cfg.opposite_direction_cooldown and
dev >= 0): # 当前适合开空
logger.info(f"平空后{int(time_since_exit)}秒内,避免开空")
return
penalty_active = self._reentry_penalty_active(dev, entry_dev)
long_th = -entry_dev
short_th = entry_dev
if penalty_active:
if self.last_sl_dir == 1:
long_th = -entry_dev * self.cfg.reentry_penalty_mult
elif self.last_sl_dir == -1:
short_th = entry_dev * self.cfg.reentry_penalty_mult
logger.info(
f"入场检查: 偏离={dev * 100:.3f}%, 阈值=[{long_th * 100:.3f}%, {short_th * 100:.3f}%], "
f"size={size}, 连续亏损={self.consecutive_losses}"
)
# 开多条件价格显著低于EMA
if dev <= long_th:
if self.place_market_order(1, size):
self.pos = 1
self.entry_price = price
self.entry_ts = time.time()
self.entry_margin = self.cfg.fixed_margin
self.entry_position_value = self.entry_margin * int(self.cfg.leverage)
self.ding(f"✅开多:偏离={dev * 100:.3f}%, 价格={price:.2f}")
# 开空条件价格显著高于EMA
elif dev >= short_th:
if self.place_market_order(4, size):
self.pos = -1
self.entry_price = price
self.entry_ts = time.time()
self.entry_margin = self.cfg.fixed_margin
self.entry_position_value = self.entry_margin * int(self.cfg.leverage)
self.ding(f"✅开空:偏离={dev * 100:.3f}%, 价格={price:.2f}")
def maybe_exit(self, price: float, ema_value: float, tp: float, sl: float, vol_scale: float):
"""
检查并执行出场
记录平仓方向,用于后续入场过滤
"""
if self.pos == 0 or self.entry_price is None or self.entry_ts is None:
return
hold = time.time() - self.entry_ts
net_pnl, total_fee, gross_pnl = self.calculate_net_pnl(price)
dev = (price - ema_value) / ema_value if ema_value else 0.0
# 计算止损倍数
sl_mult = 1.0
if self.post_sl_dir == self.pos and self.post_sl_dir != 0:
sl_mult = self._post_sl_dynamic_mult()
effective_sl = sl * sl_mult
# 记录平仓前的方向
old_pos = self.pos
# 条件1正常平仓价格回归EMA
if abs(dev) <= self.cfg.normal_exit_threshold:
if net_pnl >= self.cfg.normal_exit_min_profit: # 净盈利达到最小要求
if self.close_position_all():
self._record_exit(price, old_pos, net_pnl, total_fee, gross_pnl, "正常平仓")
return
else:
logger.error("正常平仓失败")
return
# 条件2止盈
if gross_pnl >= tp:
if net_pnl > 0: # 扣除手续费后仍有盈利
if self.close_position_all():
self._record_exit(price, old_pos, net_pnl, total_fee, gross_pnl, "止盈")
return
else:
logger.error("止盈平仓失败")
return
# 条件3止损
if gross_pnl <= -effective_sl:
if self.close_position_all():
self._record_exit(price, old_pos, net_pnl, total_fee, gross_pnl, "止损")
# 记录止损状态
sl_dir = old_pos
self.last_sl_dir = sl_dir
self.last_sl_ts = time.time()
self.post_sl_dir = sl_dir
self.post_sl_ts = time.time()
self.post_sl_vol_scale = float(vol_scale)
return
else:
logger.error("止损平仓失败")
return
# 条件4超时平仓
if hold >= self.cfg.max_hold_sec:
if net_pnl > 0: # 扣除手续费后仍有盈利
if self.close_position_all():
self._record_exit(price, old_pos, net_pnl, total_fee, gross_pnl, "超时平仓")
return
else:
logger.error("超时平仓失败")
return
else:
logger.debug(f"超时但净盈亏为负,继续持有")
def _record_exit(self, price: float, old_pos: int, net_pnl: float,
total_fee: float, gross_pnl: float, exit_type: str):
"""记录平仓信息"""
self.entry_price = None
self.entry_ts = None
self.entry_margin = None
self.entry_position_value = None
self.last_exit_ts = time.time()
self.last_exit_direction = old_pos
self.last_exit_price = price
self.today_trade_count += 1
# 更新连续亏损计数
if net_pnl < 0:
self.consecutive_losses += 1
else:
self.consecutive_losses = 0
self.ding(
f"📊{exit_type}:方向={'' if old_pos == 1 else ''}, "
f"毛盈亏={gross_pnl * 100:.3f}%, 净盈亏={net_pnl * 100:.3f}%, "
f"手续费={total_fee:.4f}u, 价格={price:.2f}"
)
# ----------------- 状态通知 -----------------
def notify_status_throttled(self, price: float, ema_value: float, dev: float, bal: float,
atr_ratio: float, base_ratio: float, vol_scale: float,
entry_dev: float, tp: float, sl: float,
entry_floor: float, tp_floor: float, sl_floor: float):
"""限频状态通知"""
now = time.time()
if (now - self._last_status_notify_ts) < self.cfg.status_notify_sec:
return
self._last_status_notify_ts = now
direction_str = "" if self.pos == 1 else ("" if self.pos == -1 else "")
# 计算当前盈亏
current_net_pnl, current_fee, current_gross_pnl = 0.0, 0.0, 0.0
if self.pos != 0 and self.entry_price:
current_net_pnl, current_fee, current_gross_pnl = self.calculate_net_pnl(price)
msg = (
f"【BitMart {self.cfg.contract_symbol}1m均值回归(90%返佣优化)】\n"
f"📊 状态:{direction_str} | 今日交易:{self.today_trade_count}/{self.cfg.max_daily_trades}\n"
f"💰 现价:{price:.2f} | EMA{self.cfg.ema_len}{ema_value:.2f} | 偏离:{dev * 100:.3f}%\n"
f"🎯 阈值:入场±{entry_dev * 100:.3f}% | 止盈{tp * 100:.3f}% | 止损{sl * 100:.3f}%\n"
f"🌊 波动率ATR={atr_ratio * 100:.3f}% | 基准={base_ratio * 100:.3f}% | 缩放={vol_scale:.2f}\n"
f"💳 余额:{bal:.2f} USDT | 杠杆:{self.cfg.leverage}x | 保证金:{self.cfg.fixed_margin}u\n"
)
if self.pos != 0:
msg += (
f"📈 当前盈亏:毛={current_gross_pnl * 100:.3f}% | 净={current_net_pnl * 100:.3f}% | "
f"手续费={current_fee:.4f}u\n"
f"⚠️ 连续亏损:{self.consecutive_losses}\n"
)
self.ding(msg)
# ----------------- 主循环 -----------------
def action(self):
"""主循环"""
if not self.set_leverage():
self.ding("杠杆设置失败,停止运行", error=True)
return
# 打开浏览器
if not self.openBrowser():
self.ding("打开 TGE 失败!", error=True)
return
logger.info("TGE 端口获取成功")
self.page.get("https://derivatives.bitmart.com/zh-CN/futures/ETHUSDT")
time.sleep(3) # 等待页面加载
logger.info("策略启动 - 90%返佣优化版")
logger.info("策略特点:低阈值、高频次、严格方向过滤")
while True:
now_dt = datetime.datetime.now()
self.pbar.n = now_dt.second
self.pbar.refresh()
# 1. 获取K线数据
klines = self.get_klines_cached()
if not klines or len(klines) < (self.cfg.ema_len + 5):
logger.warning("K线数据不足等待...")
time.sleep(1)
continue
# 2. 计算技术指标
last_k = klines[-1]
closes = [k["close"] for k in klines[-(self.cfg.ema_len + 1):]]
ema_value = self.ema(closes, self.cfg.ema_len)
price = self.get_last_price(fallback_close=float(last_k["close"]))
dev = (price - ema_value) / ema_value if ema_value else 0.0
# 3. 计算波动率指标
a = self.atr(klines, self.cfg.atr_len)
atr_ratio = (a / price) if price > 0 else 0.0
base_ratio = self.get_base_ratio_cached(klines)
# 4. 计算动态阈值
entry_dev, tp, sl, vol_scale, entry_floor, tp_floor, sl_floor = self.dynamic_thresholds(
atr_ratio, base_ratio
)
# 5. 风控检查
self.risk_kill_switch()
# 6. 获取持仓状态
if not self.get_position_status():
time.sleep(1)
continue
# 7. 检查交易是否启用
if not self.trading_enabled:
if self.pos != 0:
if self.close_position_all():
logger.info("交易被禁用,已平仓")
else:
logger.error("交易被禁用,但平仓失败")
logger.warning("交易被禁用,等待...")
time.sleep(5)
continue
# 8. 检查危险市场
if self.is_danger_market(klines, price):
logger.warning("危险模式,暂停开仓")
self.maybe_exit(price, ema_value, tp, sl, vol_scale)
time.sleep(self.cfg.tick_refresh_sec)
continue
# 9. 执行交易逻辑
# 先检查平仓
self.maybe_exit(price, ema_value, tp, sl, vol_scale)
# 再检查开仓
self.maybe_enter(price, ema_value, entry_dev)
# 10. 状态通知
bal = self.get_assets_available()
self.notify_status_throttled(
price, ema_value, dev, bal,
atr_ratio, base_ratio, vol_scale,
entry_dev, tp, sl,
entry_floor, tp_floor, sl_floor
)
time.sleep(self.cfg.tick_refresh_sec)
if __name__ == "__main__":
"""
Windows PowerShell:
setx BITMART_API_KEY "你的key"
setx BITMART_SECRET_KEY "你的secret"
setx BITMART_MEMO "合约交易"
重新打开终端再运行。
Linux/macOS:
export BITMART_API_KEY="你的key"
export BITMART_SECRET_KEY="你的secret"
export BITMART_MEMO "合约交易"
"""
cfg = StrategyConfig()
bot = BitmartFuturesMeanReversionBot(cfg, bit_id="f2320f57e24c45529a009e1541e25961")
# 设置日志级别
logger.remove()
logger.add(lambda msg: tqdm.write(msg, end=""), level="INFO")
logger.info("""
====================================================
90%返佣优化策略说明:
策略特点:
1. 针对90%返佣优化阈值降低30-50%
2. 增加方向过滤,避免平仓后立即反向开仓
3. 冷却时间从20秒减至5秒提高交易频率
4. 添加连续亏损控制和趋势过滤
5. 日内交易次数限制50次
核心修复:
1. 避免平仓后立即反向开仓(原策略最大问题)
2. 修复平仓后可能立即开反向仓位的问题
3. 增加对同方向冷却的控制
====================================================
""")
try:
bot.action()
except KeyboardInterrupt:
logger.info("程序被用户中断")
bot.ding("🤖 策略已手动停止")
except Exception as e:
logger.error(f"程序异常退出: {e}")
bot.ding(f"❌ 策略异常退出: {e}", error=True)
raise