import os import time import uuid import datetime from dataclasses import dataclass from tqdm import tqdm from loguru import logger from bitmart.api_contract import APIContract from bitmart.lib.cloud_exceptions import APIException from 交易.tools import send_dingtalk_message @dataclass class StrategyConfig: # ============================= # 1m | ETH 永续 | 控止损≤5/日 # ============================= # ===== 合约 ===== contract_symbol: str = "ETHUSDT" open_type: str = "cross" leverage: str = "30" # 50 -> 30：显著降低1m噪声导致的连环止损与回撤波动 # ===== K线与指标 ===== step_min: int = 1 lookback_min: int = 240 ema_len: int = 36 # 30 -> 36：均值更稳，信号更挑剔 atr_len: int = 14 # ===== 动态阈值基础（自适应行情）===== entry_dev_floor: float = 0.0012 # 0.10% -> 0.12%：过滤小噪声进场 tp_floor: float = 0.0006 # 0.05% -> 0.06%：更接近“净盈利” sl_floor: float = 0.0018 # 0.15% -> 0.18%：ETH 1m插针多，底线略放宽 # 更挑剔、更少止损（进场更苛刻；止损不过度随波动放大） entry_k: float = 1.45 # 1.20 -> 1.45：减少进场频率 tp_k: float = 0.65 # 0.60 -> 0.65：略抬止盈 sl_k: float = 1.05 # 1.20 -> 1.05：配合sl_floor，避免高波动下止损无限变大 # ===== 时间/冷却 ===== max_hold_sec: int = 75 # 90/120 -> 75：1m回归不恋战 cooldown_sec_after_exit: int = 20 # 10 -> 20：减少“刚出又进”连环单 # ===== 下单/仓位 ===== risk_percent: float = 0.004 # 0.005 -> 0.004：再压一点波动，更贴合止损≤5/日 min_size: int = 1 max_size: int = 5000 # ===== 日内风控 ===== daily_loss_limit: float = 0.02 # -2% 停机 daily_profit_cap: float = 0.01 # +1% 封顶停机 # ===== 危险模式过滤（1m ETH 更敏感）===== atr_ratio_kill: float = 0.0038 # 0.0045 -> 0.0038：更早暂停开仓 big_body_kill: float = 0.010 # 0.012 -> 0.010：更敏感 # ===== 轮询节奏 ===== klines_refresh_sec: int = 10 tick_refresh_sec: int = 1 status_notify_sec: int = 60 # ========================================================= # ✅ 止损后同向入场加门槛（但不禁止同向重入） # ========================================================= reentry_penalty_mult: float = 1.55 # 同向入场门槛×1.55：大幅降低连环止损概率 reentry_penalty_max_sec: int = 180 # 罚时最长持续 reset_band_k: float = 0.45 # dev回到更靠近均值才解除罚则 reset_band_floor: float = 0.0006 # 最小复位带宽（0.06%） # ========================================================= # ✅ 自动阈值：ATR/Price 分位数基准（更稳，不被短时噪声带跑） # ========================================================= vol_baseline_window: int = 120 vol_baseline_quantile: float = 0.65 vol_scale_min: float = 0.80 vol_scale_max: float = 1.60 # ========================================================= # ✅ 升级：止损后同方向 SL 放宽幅度与“止损时 vol_scale”联动 # ========================================================= 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 # mult = 1 + alpha*(vol_scale_at_sl - 1) class BitmartFuturesMeanReversionBot: def __init__(self, cfg: StrategyConfig): self.cfg = cfg # ✅ 只从环境变量读（请务必更换曾经硬编码泄露过的 key） self.api_key = os.getenv("BITMART_API_KEY", "").strip() self.secret_key = os.getenv("BITMART_SECRET_KEY", "").strip() self.memo = os.getenv("BITMART_MEMO", "合约交易").strip() if not self.api_key or not self.secret_key: raise RuntimeError("请先设置环境变量 BITMART_API_KEY / BITMART_SECRET_KEY / BITMART_MEMO(可选)") self.contractAPI = APIContract(self.api_key, self.secret_key, self.memo, timeout=(5, 15)) # 持仓状态: -1 空, 0 无, 1 多 self.pos = 0 self.entry_price = None self.entry_ts = None self.last_exit_ts = 0 # 日内权益基准 self.day_start_equity = None self.trading_enabled = True self.day_tag = datetime.date.today() # 缓存 self._klines_cache = None self._klines_cache_ts = 0 self._last_status_notify_ts = 0 # ✅ 止损后“同向入场加门槛”状态 self.last_sl_dir = 0 # 1=多止损，-1=空止损，0=无 self.last_sl_ts = 0.0 # ✅ 止损后“同方向 SL 联动放宽”状态 self.post_sl_dir = 0 self.post_sl_ts = 0.0 self.post_sl_vol_scale = 1.0 # 记录止损时的 vol_scale self.pbar = tqdm(total=60, desc="运行中(秒)", ncols=90) # ----------------- 通用工具 ----------------- def ding(self, msg, error=False): prefix = "❌bitmart：" if error else "🔔bitmart：" if error: for _ in range(3): 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}") self.ding(f"设置杠杆失败: {resp}", error=True) return False except Exception as e: logger.error(f"设置杠杆异常: {e}") self.ding(f"设置杠杆异常: {e}", error=True) return False # ----------------- 行情/指标 ----------------- def get_klines_cached(self): 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): 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}") self.ding(f"获取K线异常: {e}", error=True) return None def get_last_price(self, fallback_close: float) -> float: """ 优先取更实时的最新价；若SDK不支持/字段不同，回退到K线close。 """ 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, n: int) -> float: k = 2 / (n + 1) e = values[0] for v in values[1:]: e = v * k + e * (1 - k) return e @staticmethod def atr(klines, n: int) -> float: 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, 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: 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: return True return False def atr_ratio_baseline(self, klines) -> float: """ 自动阈值基准：最近 window 根的 atr_ratio 分布的 quantile 作为“典型波动” """ window = self.cfg.vol_baseline_window if len(klines) < (window + self.cfg.atr_len + 5): return 0.0 ratios = [] for i in range(-window, 0): sub = klines[:i] if i != 0 else klines a = self.atr(sub, self.cfg.atr_len) p = sub[-1]["close"] if p > 0 and a > 0: ratios.append(a / p) if not ratios: return 0.0 ratios.sort() q = max(0.0, min(1.0, self.cfg.vol_baseline_quantile)) idx = int(q * (len(ratios) - 1)) return ratios[idx] def dynamic_thresholds(self, atr_ratio: float, base_ratio: float): """ 动态阈值：atr_ratio * vol_scale，并带 floor """ if base_ratio <= 0: vol_scale = 1.0 else: raw = atr_ratio / base_ratio vol_scale = max(self.cfg.vol_scale_min, min(self.cfg.vol_scale_max, raw)) entry_dev = max(self.cfg.entry_dev_floor, self.cfg.entry_k * vol_scale * atr_ratio) tp = max(self.cfg.tp_floor, self.cfg.tp_k * vol_scale * atr_ratio) sl = max(self.cfg.sl_floor, self.cfg.sl_k * vol_scale * atr_ratio) return entry_dev, tp, sl, vol_scale # ----------------- 账户/仓位 ----------------- 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}") self.ding(f"持仓查询异常: {e}", error=True) 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.ding(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: """ 保守仓位估算：按 1张≈0.001ETH（沿用你原假设） """ bal = self.get_assets_available() if bal < 10: return 0 margin = bal * self.cfg.risk_percent 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) return size def place_market_order(self, side: int, size: int) -> bool: """ side: 1 开多 2 平空 3 平多 4 开空 """ if size <= 0: return False client_order_id = f"mr_{int(time.time())}_{uuid.uuid4().hex[:8]}" try: resp = self.contractAPI.post_submit_order( contract_symbol=self.cfg.contract_symbol, client_order_id=client_order_id, side=side, mode=1, type="market", leverage=self.cfg.leverage, open_type=self.cfg.open_type, size=size )[0] logger.info(f"order_resp: {resp}") if resp.get("code") == 1000: return True self.ding(f"下单失败: {resp}", error=True) return False except APIException as e: logger.error(f"API下单异常: {e}") self.ding(f"API下单异常: {e}", error=True) return False except Exception as e: logger.error(f"下单未知异常: {e}") self.ding(f"下单未知异常: {e}", error=True) return False def close_position_all(self): if self.pos == 1: ok = self.place_market_order(3, 999999) if ok: self.pos = 0 elif self.pos == -1: ok = self.place_market_order(2, 999999) if ok: self.pos = 0 # ----------------- 止损后机制（核心优化） ----------------- def _reentry_penalty_active(self, dev: float, entry_dev: float) -> bool: """ 止损后同向入场加门槛： - 只要 dev 还没有回到中性区，就对“上次止损方向”的同向入场门槛提高 - dev 回到 abs(dev) <= reset_band 后自动解除 - 超过 max_sec 自动解除（避免一直卡住） """ 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 放宽倍数与“止损时 vol_scale”联动： mult = 1 + alpha*(vol_scale_at_sl - 1) 并做上下限裁剪 + 有效期控制 """ 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 dev = (price - ema_value) / ema_value if ema_value else 0.0 size = self.calculate_size(price) if size <= 0: 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"enter_check: price={price:.2f}, ema={ema_value:.2f}, dev={dev * 100:.3f}% " f"(entry_dev={entry_dev * 100:.3f}%, long_th={long_th * 100:.3f}%, short_th={short_th * 100:.3f}%) " f"size={size}, penalty={penalty_active}, last_sl_dir={self.last_sl_dir}" ) if dev <= long_th: if self.place_market_order(1, size): # 开多 self.pos = 1 self.entry_price = price self.entry_ts = time.time() self.ding(f"✅开多：dev={dev * 100:.3f}% size={size} entry={price:.2f}") elif dev >= short_th: if self.place_market_order(4, size): # 开空 self.pos = -1 self.entry_price = price self.entry_ts = time.time() self.ding(f"✅开空：dev={dev * 100:.3f}% size={size} entry={price:.2f}") def maybe_exit(self, price: 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 if self.pos == 1: pnl = (price - self.entry_price) / self.entry_price else: pnl = (self.entry_price - price) / self.entry_price # ✅ 同方向止损后：在有效期内放宽 SL（与止损时 vol_scale 联动） 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 if pnl >= tp: self.close_position_all() self.ding(f"🎯止盈：pnl={pnl * 100:.3f}% price={price:.2f} tp={tp * 100:.3f}%") self.entry_price, self.entry_ts = None, None self.last_exit_ts = time.time() elif pnl <= -effective_sl: # 记录止损方向 sl_dir = self.pos # 1=多止损，-1=空止损 self.close_position_all() self.ding( f"🛑止损：pnl={pnl * 100:.3f}% price={price:.2f} " f"sl={sl * 100:.3f}% effective_sl={effective_sl * 100:.3f}%(×{sl_mult:.2f})", error=True ) # ✅ 开启：同向入场加门槛 self.last_sl_dir = sl_dir self.last_sl_ts = time.time() # ✅ 开启：同向 SL 联动放宽（记录止损时 vol_scale） self.post_sl_dir = sl_dir self.post_sl_ts = time.time() self.post_sl_vol_scale = float(vol_scale) self.entry_price, self.entry_ts = None, None self.last_exit_ts = time.time() elif hold >= self.cfg.max_hold_sec: self.close_position_all() self.ding(f"⏱超时：hold={int(hold)}s pnl={pnl * 100:.3f}% price={price:.2f}") self.entry_price, self.entry_ts = None, None self.last_exit_ts = time.time() 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): 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 "无") penalty_active = self._reentry_penalty_active(dev, entry_dev) sl_mult = 1.0 if self.pos != 0 and self.post_sl_dir == self.pos: sl_mult = self._post_sl_dynamic_mult() msg = ( f"【BitMart {self.cfg.contract_symbol}｜1m均值回归(自动阈值+止损智能)】\n" f"方向：{direction_str}\n" f"现价：{price:.2f}\n" f"EMA{self.cfg.ema_len}：{ema_value:.2f}\n" f"dev：{dev * 100:.3f}%（entry_dev={entry_dev * 100:.3f}%）\n" f"ATR比：{atr_ratio * 100:.3f}% 基准：{base_ratio * 100:.3f}% vol_scale={vol_scale:.2f}\n" f"tp/sl：{tp * 100:.3f}% / {sl * 100:.3f}%（postSL×{sl_mult:.2f}, sl@scale={self.post_sl_vol_scale:.2f}）\n" f"止损同向加门槛：{'ON' if penalty_active else 'OFF'}（last_sl_dir={self.last_sl_dir}）\n" f"可用余额：{bal:.2f} USDT 杠杆：{self.cfg.leverage}x\n" f"超时：{self.cfg.max_hold_sec}s 冷却：{self.cfg.cooldown_sec_after_exit}s" ) self.ding(msg) def action(self): if not self.set_leverage(): self.ding("杠杆设置失败，停止运行", error=True) return while True: now_dt = datetime.datetime.now() self.pbar.n = now_dt.second self.pbar.refresh() klines = self.get_klines_cached() if not klines or len(klines) < (self.cfg.ema_len + 5): time.sleep(1) continue 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 # 自动阈值 a = self.atr(klines, self.cfg.atr_len) atr_ratio = (a / price) if price > 0 else 0.0 base_ratio = self.atr_ratio_baseline(klines) entry_dev, tp, sl, vol_scale = self.dynamic_thresholds(atr_ratio, base_ratio) # 日内风控 self.risk_kill_switch() # 刷新仓位 if not self.get_position_status(): time.sleep(1) continue # 停机：平仓+不再开仓 if not self.trading_enabled: if self.pos != 0: self.close_position_all() time.sleep(5) continue # 危险市场：不新开仓（允许已有仓按 tp/sl/超时 退出） if self.is_danger_market(klines, price): logger.warning("危险模式：高波动/大实体K，暂停开仓") self.maybe_exit(price, tp, sl, vol_scale) time.sleep(self.cfg.tick_refresh_sec) continue # 先出场再入场 self.maybe_exit(price, tp, sl, vol_scale) self.maybe_enter(price, ema_value, entry_dev) # 状态通知（限频） bal = self.get_assets_available() self.notify_status_throttled( price, ema_value, dev, bal, atr_ratio, base_ratio, vol_scale, entry_dev, tp, sl ) 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) bot.action() # 9208.96