"""自动化小额实盘执行器。读信号总线，直接调 Bitget v2 REST 下单。 ## 为什么不用 Hummingbot 的 PositionExecutor 它的连接器只暴露 LIMIT / LIMIT_MAKER / MARKET，没有触发单，于是 `control_stop_loss()` 只能在本地盯价、触发时才发市价单——**进程一死仓位就是 裸的**。而交易所本身支持 `place-order` 带 `presetStopLossPrice`，下单时就把 止损挂到服务端。绕过连接器不是图省事，是为了消掉一整类故障。 另外 `TripleBarrierConfig` 只有单级止盈，装不下 3 ATR 减半 + 8 ATR 目标； 自己写反而更短。 ## 出场结构为什么能拆成两个半仓 回测结构是 2 ATR 止损 / 3 ATR 减半 / 8 ATR 目标 / 48 根超时，且**剩余半仓的 止损保持在入场价的 2 ATR、不移动**。已核实 `research/lib/exit_model.py:151`—— `runner_stops` 的 `ret` 是 `(entry - low[j]) / a`，从入场价算，且 `RUNNER_STOP == SL == 2.0`。两半共用同一个不动的止损，所以： 半仓 A 市价入场 + 服务端止损 2 ATR · maker 止盈 3 ATR 半仓 B 市价入场 + 服务端止损 2 ATR · maker 止盈 8 ATR 止损先到则两半都在 -2 ATR 出场；3 ATR 先到则 A 出场、B 继续且止损仍在 2 ATR。 与回测逐情形一致。若哪天把 RUNNER_STOP 改成不等于 SL（比如移到成本），这个 分解就**不再成立**，`assert_decomposable()` 会在启动时挡住。 ## 三条出场腿各自挂在哪 止损 交易所侧（presetStopLossPrice，随入场单一起到）→ 进程死了仍在 止盈 交易所侧（post_only reduce-only 限价） → 进程死了仍在 超时 **本进程**，48 分钟到点市价平 所以进程死掉只会让持仓超过 48 根，不会变成裸仓——退化是良性的。 ## 硬约束才是这个文件的重点 一笔止损只亏约 1 USDT，所以"亏损可控"对单笔成立。但三类故障的代价**不随仓位 缩小**，必须显式封住： 失控下单 循环里的 bug 反复开仓，单笔小但笔数无界 → MAX_OPEN / MAX_DAY 亏损累积 策略真的不行，但没人盯着 → MAX_DAY_LOSS 裸仓 进程在"已入场、止损未挂"之间死掉 → 服务端止损 + 重启对账 ## 为什么单独一个 live/ 子树、不放在 research/ 下 生产与研究共处一个目录/进程/机器有四条具体代价，其中第一条已经咬过一次： 1. `live_state.json` 原先落在 `research/out/`，而那里 `shadow_hb.py` 会在 CSV 表头变化时自动 rename 归档、研究脚本会写、人也会手工清数据。那个文件 装的是 MAX_DAY_LOSS 累计与在场仓位，**闸的状态被清掉不报错，只是静默 失效**。所以生产状态改到独立目录（LIVE_HOME）。 2. 采集器十币清空 300~560ms，直接叠在信号到达执行器的延迟上。 3. 研究侧的探针 OOM 过一次（14.9GB、负载 12），当时若有仓位在场，执行器会 被一起杀掉，只剩交易所侧止损兜着。 4. 依赖面：本文件只需标准库 + aiohttp。原先为读两个常量 import 研究侧的 step43，把 numpy/pandas/pyarrow 全拖进生产进程。 因此本目录**不 import research/ 下的任何东西**（`exit_params.py` 是生产自己 持有的契约，研究侧反过来读它）。 python live/live_exec.py --dry-run # 只打印不下单 """ from __future__ import annotations import argparse import asyncio import json import os import signal import sys import time from decimal import Decimal from pathlib import Path HERE = Path(__file__).resolve() # 只插自己所在目录。**不要**把 research/ 加进来——见文件头第 4 条 sys.path.insert(0, str(HERE.parent)) import signal_bus # noqa: E402 from bitget_rest import Bitget # noqa: E402 import tg # noqa: E402 from exit_params import MAXB, RUNNER, RUNNER_STOP, SCALE_AT, SL # noqa: E402 # 生产状态的根目录。默认放 ~/chan-live，**不落在仓库里**：仓库会被 git # checkout/clean 动，而这里存的是日亏损累计与在场仓位，丢了等于闸失忆 LIVE_HOME = Path(os.environ.get("LIVE_HOME", Path.home() / "chan-live")) SYMS = os.environ.get( "SYMS", "BTC,ETH,SOL,BNB,XRP,DOGE,ADA,AVAX,LINK,LTC").split(",") NOTIONAL = float(os.environ.get("LIVE_NOTIONAL", "500")) LEVERAGE = int(os.environ.get("LIVE_LEVERAGE", "10")) # ── 硬约束 ──────────────────────────────────────────────────────────── # 并发仓位数。1 笔约占 50 USDT 保证金，3 笔 150 USDT。信号速率 5.3 笔/天、 # 持仓 48 分钟，期望并发只有 0.18 笔，所以 3 已经很宽——超了说明有 bug MAX_OPEN = int(os.environ.get("LIVE_MAX_OPEN", "3")) # 日开仓上限。实测 5.3 笔/天，给 3 倍余量。这一条专门封"失控下单" MAX_DAY = int(os.environ.get("LIVE_MAX_DAY", "15")) # 日亏损上限（USDT）。一笔止损约 1 USDT，15 笔全亏 15 USDT MAX_DAY_LOSS = float(os.environ.get("LIVE_MAX_DAY_LOSS", "20")) # 信号超过这么久就不做了。参考成交价是次根开盘价，过期后跑的不是回测那个价 STALE_S = float(os.environ.get("LIVE_STALE_S", "20")) # 盯交易所侧出场的轮询间隔。10s 足够：出场后要做的只是记账与放开 MAX_OPEN # 名额，不涉及下单时效。太密会白耗 API 配额 WATCH_S = float(os.environ.get("LIVE_WATCH_S", "10")) STATE = Path(os.environ.get("LIVE_STATE", LIVE_HOME / "state" / "live_state.json")) TRADES = Path(os.environ.get("LIVE_TRADES", LIVE_HOME / "state" / "live_trades.jsonl")) # 出场结构从 exit_params 读，本文件不再抄一份字面量。抄一份的问题不是难看， # 是改了回测参数后这边不会跟上，而且不报错 SL_ATR, SCALE_ATR, RUNNER_ATR = SL, SCALE_AT, RUNNER def assert_decomposable() -> None: """两个半仓的分解依赖 RUNNER_STOP == SL，不成立就必须停机。 若有人把剩余半仓的止损改成移到成本（RUNNER_STOP=0）或任何 != SL 的值， 这个分解就变成"两半共用同一止损"的错误近似，实盘跑的是另一个收益结构， 而且不会报错。所以在启动时硬挡。 """ if float(RUNNER_STOP) != float(SL): raise SystemExit( f"⛔ RUNNER_STOP({RUNNER_STOP}) != SL({SL})，两个半仓的分解不再\n" f" 成立。live_exec 的出场结构会与回测不一致且不报错。\n" f" 要改成单执行器 + 手工两级止盈，或把这两个值改回一致。") class Guard: """硬约束与当日计数。状态落盘，重启后不清零。 不落盘的话，进程反复重启就等于反复重置日上限——"失控下单"这一类恰好常常 伴随反复重启，那时上限必须还记得。 """ def __init__(self, path: Path = STATE): self.path = path self.day = time.strftime("%Y-%m-%d") self.n_day = 0 self.pnl_day = 0.0 self.done: set = set() self.pending_roll: tuple | None = None self._load() def _load(self) -> None: try: d = json.loads(self.path.read_text()) except Exception: return # 跨日则计数归零，但已处理过的信号键要保留，否则会重开旧仓 if d.get("day") == self.day: self.n_day = int(d.get("n_day", 0)) self.pnl_day = float(d.get("pnl_day", 0.0)) self.done = set(d.get("done", [])) def save(self) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) tmp = self.path.with_suffix(".tmp") # 原子替换：直接覆写时若在写一半崩溃，状态文件会变成半个 JSON， # 重启后读不出来 → 日计数归零 → 上限失效 tmp.write_text(json.dumps({ "day": self.day, "n_day": self.n_day, "pnl_day": self.pnl_day, # 只留最近的，否则文件无界增长 "done": sorted(self.done)[-5000:]})) tmp.replace(self.path) def roll(self) -> None: today = time.strftime("%Y-%m-%d") if today != self.day: print(f" [guard] 跨日 {self.day} → {today}，" f"当日 {self.n_day} 笔 / PnL {self.pnl_day:+.2f} USDT", flush=True) # roll() 是同步的，推送要 await，所以只留个待发件，由心跳取走 self.pending_roll = (self.day, self.n_day, self.pnl_day) self.day, self.n_day, self.pnl_day = today, 0, 0.0 self.save() def blocks(self, key: str, n_open: int) -> str | None: """返回拒绝原因，None 表示放行。""" self.roll() if key in self.done: return "已处理过（幂等）" if n_open >= MAX_OPEN: return f"并发仓位已达上限 {MAX_OPEN}" if self.n_day >= MAX_DAY: return f"当日开仓已达上限 {MAX_DAY}" if self.pnl_day <= -MAX_DAY_LOSS: return (f"当日亏损 {self.pnl_day:.2f} 已达上限 " f"-{MAX_DAY_LOSS}，停止开新仓") return None def took(self, key: str) -> None: self.done.add(key) self.n_day += 1 self.save() def realized(self, pnl: float) -> None: self.pnl_day += pnl self.save() def legs() -> list[dict]: """两个半仓的止盈位，用 ATR 倍数表达。 止损两半相同（SL_ATR），所以不写在这里——它在 open_position 里算一次。 """ return [{"tag": "scale", "atr": SCALE_ATR}, {"tag": "runner", "atr": RUNNER_ATR}] def oid_of(key: str, tag: str) -> str: """把信号键变成交易所能接受的 clientOid。 信号键形如 `SOL:1787904388411:+1`，里面的 `:` 和 `+` 未必被交易所接受， 带过去会直接拒单——而拒单发生在入场腿上，等于这笔信号静默漏掉。只留 字母数字和下划线。 clientOid 是**交易所级幂等**：重发同一个 oid 会被拒。这比本地去重可靠， 因为「已发出但没收到回复」这种情况本地判不了，重试就会开两次仓。 """ # 方向必须显式编码：直接把非字母数字换成下划线，会让 `+1` 和 `-1` 都变成 # `_1`，同一根上的多空信号得到相同 oid，第二笔被交易所当重复拒掉 k = key.replace(":+1", ":L").replace(":-1", ":S") safe = "".join(c if c.isalnum() else "_" for c in f"{k}_{tag}") return safe[:60] def log_trade(rec: dict, path: Path = TRADES) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("a") as f: f.write(json.dumps(rec) + "\n") f.flush() os.fsync(f.fileno()) class Exec: def __init__(self, dry: bool, bus: Path): self.dry = dry self.bus = bus self.guard = Guard() self.api: Bitget | None = None self.rules: dict = {} self.execs: dict = {} # key → 该笔的腿与超时时刻 self.offset = 0 # 已读到总线的哪一行 self.n_seen = self.n_took = self.n_skip = 0 # ── 启动 ────────────────────────────────────────────────────── async def start(self) -> None: assert_decomposable() # 只从"现在"往后做。历史信号的参考成交价早已过期，补做等于随机入场 self.offset = sum(1 for _ in signal_bus.read_all(self.bus)) print(f" 总线已有 {self.offset} 条历史信号，全部跳过（参考价已过期）", flush=True) self.api = Bitget(dry=self.dry) self.rules = await self.api.contracts() print(f" 合约规则 {len(self.rules)} 个", flush=True) # 启动推送兼作"通道通不通"的自检：配错了这里就收不到，而不是等到 # 几小时后第一个真信号来时才发现 await tg.started(NOTIONAL, LEVERAGE, len(SYMS), self.dry) print(f" Telegram {'已启用' if tg.ENABLED else '未配置（不推送）'}", flush=True) if self.dry: print(" ⚠ 空跑模式：不下真单", flush=True) # 但仍要发一次**带签名**的请求。上面的 contracts() 是公开端点， # 不验签，光靠它空跑会"通过"却根本没测到密钥与 IP 白名单—— # 那种假保证比不测更糟：等到第一个真信号来时才暴露，而信号那时 # 正在过期，没有从容排查的余地 if self.api.key and self.api.secret and self.api.passphrase: try: acc = await self.api.account() or {} print(" ✓ 密钥与 IP 白名单通", flush=True) # 打全几个余额字段，不只看 available：我们用逐仓，真正 # 决定能不能开的是 isolatedMaxAvailable。只看一个字段， # 取错了就会把"有钱"误报成"没钱"，或者反过来 fields = ("accountEquity", "usdtEquity", "available", "isolatedMaxAvailable", "crossedMaxAvailable", "maxTransferOut", "locked", "unrealizedPL") shown = {k: acc.get(k) for k in fields if k in acc} print(f" 余额 {shown}", flush=True) # 逐仓下取 isolatedMaxAvailable，缺了才退回 available av = float(acc.get("isolatedMaxAvailable") or acc.get("available") or 0) need = NOTIONAL / LEVERAGE * MAX_OPEN print(f" 可开保证金 {av:.2f} USDT · {MAX_OPEN} 笔并发" f"需约 {need:.0f}（名义 {NOTIONAL:.0f} / " f"{LEVERAGE}x）", flush=True) if av < need: per = NOTIONAL / LEVERAGE fit = int(av // per) if per > 0 else 0 print(f" ⚠ 保证金只够 {fit} 笔，而 MAX_OPEN=" f"{MAX_OPEN}。", flush=True) # 这里不只是"少做几笔"。两条腿是分别下单的，第一条 # 成了、第二条因保证金不足失败，就留下一个半仓—— # 收益结构从「50% 在 3 ATR + 50% 在 8 ATR」变成只剩 # 一条腿，而且是静默的。让闸按真实余额拦，比让交易所 # 拒单干净 print(f" 要么充钱到 {need:.0f}+ USDT，要么把 " f"LIVE_MAX_OPEN 降到 {max(fit, 1)}。不改的话" f"超出的信号会下单失败，且可能只成一条腿、" f"静默变成半仓（收益结构就不是设计的那个了）", flush=True) if fit == 0: print(f" 现在连一笔都开不了（每笔需 " f"{per:.0f} USDT）", flush=True) except Exception as e: # noqa: BLE001 # 退出前要关会话。SystemExit 会绕过 run() 里的收尾， # 漏了就在日志尾部留一串 "Unclosed client session"， # 把真正的报错顶出视野 await self.api.close() raise SystemExit( f"⛔ 带签名的请求失败：{type(e).__name__}: {e}\n" f" 40018 = 出口 IP 不在该 key 的白名单里；\n" f" 40037 = key 不存在（填错或已删）；\n" f" 40001/40009 = secret/passphrase 不对；\n" f" 40099 之类 = 权限没开够（要读+交易）。") else: print(" ⚠ 没填密钥，本次**未**验证密钥与 IP 白名单", flush=True) return if not (self.api.key and self.api.secret and self.api.passphrase): await self.api.close() raise SystemExit("⛔ 缺 BITGET_API_KEY / BITGET_API_SECRET / " "BITGET_PASSPHRASE（末项无 API_）。先跑 --dry-run。") await self.setup_symbols() await self.reconcile() async def setup_symbols(self) -> None: """逐仓 + 杠杆。每次启动都设一遍，不假设交易所侧的状态。 杠杆若被人在 App 里改过，仓位大小就不是我们算的那个。设成幂等操作比 读回来核对简单，且失败会直接暴露。 """ bad: list[str] = [] for s in SYMS: sym = f"{s}USDT" for fn, arg in ((self.api.set_margin_mode, "isolated"), (self.api.set_leverage, LEVERAGE)): try: await fn(sym, arg) except Exception as e: print(f" ⚠ {sym} 设置失败 {type(e).__name__}: {e}", flush=True) bad.append(f"{sym}/{fn.__name__}") if not bad: print(f" 已设 {len(SYMS)} 个币为逐仓 {LEVERAGE}x", flush=True) return # 不能无条件报"已设好"。杠杆设失败是有经济后果的：仓位大小由名义额 # 算、与杠杆无关，但保证金要求会变。若交易所侧实际是 1x，每笔要 100 # USDT 保证金，第 2、3 笔必然失败，且可能只成一条腿变成半仓 print(f" ⛔ {len(bad)} 项设置失败，交易所侧的逐仓/杠杆**不是** " f"{LEVERAGE}x：{bad[:6]}{' …' if len(bad) > 6 else ''}", flush=True) print(" 后果不是不能交易，而是保证金要求与我们算的不一致——" "可能只成一条腿、静默变成半仓。先在 App 里核一遍再跑", flush=True) await tg.error(f"{len(bad)} 项逐仓/杠杆设置失败", f"交易所侧不是 {LEVERAGE}x：{bad[:10]}") async def reconcile(self) -> None: """启动时把交易所的实际持仓对上。 崩溃重启后交易所可能还有仓位。它们的服务端止损仍在（presetStopLossPrice 挂在交易所侧，不随进程消失），但**超时腿丢了**，会一直持有到止损或止盈。 选择平掉而非接管：接管要重建入场价、ATR、剩余半仓状态和已过根数，任一项 猜错就让出场结构变成另一个东西且不报错；平掉的代价只是一笔小额亏损， 且行为确定。 """ try: pos = await self.api.positions() except Exception as e: print(f" ⚠ 对账读持仓失败 {type(e).__name__}: {e}", flush=True) return if not pos: print(" 对账：交易所无持仓，干净启动", flush=True) return print(f" ⚠ 对账：发现 {len(pos)} 个遗留持仓，撤挂单后市价平掉", flush=True) await tg.error(f"对账：发现 {len(pos)} 个遗留持仓，撤挂单后市价平掉", "、".join(f"{p['symbol']} {p['holdSide']} {p['total']}" for p in pos)) for p in pos: sym, hs, sz = p["symbol"], p["holdSide"], p["total"] print(f" {sym} {hs} {sz} @ {p.get('openPriceAvg')}", flush=True) try: await self.api.cancel_all(sym) await self.api.close_market( sym, hs, sz, f"recon:{int(time.time() * 1000)}") log_trade({"ev": "reconcile_flatten", "symbol": sym, "hold_side": hs, "size": sz}) except Exception as e: print(f" ⛔ 平仓失败 {type(e).__name__}: {e}，" f"需人工介入", flush=True) # ── 主循环 ──────────────────────────────────────────────────── async def poll(self) -> None: while True: try: await self.step() except Exception as e: import traceback print(f" ⛔ 主循环异常 {type(e).__name__}: {e}", flush=True) traceback.print_exc() await asyncio.sleep(0.2) async def step(self) -> None: recs = list(signal_bus.read_all(self.bus)) if len(recs) <= self.offset: return new, self.offset = recs[self.offset:], len(recs) for r in new: self.n_seen += 1 await self.on_signal(r) # 一条记录要能下单，这些字段一个都不能缺 NEEDED = ("key", "sym", "emit_ms", "direction", "entry_px", "atr_pct") async def on_signal(self, r: dict) -> None: # 先校验再取值。缺了这一步，总线上一条字段不全的记录会抛 KeyError， # 把 poll 任务打死，进而**整个执行器停止交易**——一行坏数据换全面停摆， # 代价完全不对等。搬运侧已挡半行，但挡不住字段级的不全 miss = [k for k in self.NEEDED if r.get(k) is None] if miss: self.n_skip += 1 print(f" ⊘ 丢弃畸形记录，缺字段 {miss}：{str(r)[:200]}", flush=True) log_trade({"ev": "malformed", "missing": miss, "rec": str(r)[:500]}) await tg.error("总线上有畸形记录，已丢弃", f"缺字段 {miss}\n{str(r)[:300]}") return age = time.time() - r["emit_ms"] / 1000.0 n_open = sum(1 for v in self.execs.values() if v) why = self.guard.blocks(r["key"], n_open) if why is None and age > STALE_S: why = f"信号已过期 {age:.1f}s > {STALE_S:.0f}s" if why: self.n_skip += 1 print(f" ⊘ {r['key']} 跳过：{why}", flush=True) log_trade({"ev": "skip", "key": r["key"], "why": why, "age_s": round(age, 2)}) # 只有被硬约束挡住才推。过期跳过是常态（搬运重连会重放旧信号）， # 推了会把真事淹掉 if "过期" not in why and "已做过" not in why: await tg.blocked(r["key"], why) return self.guard.took(r["key"]) self.n_took += 1 lg = legs() side = "LONG" if r["direction"] > 0 else "SHORT" print(f" ▶ {r['key']} {side} 名义 {NOTIONAL:.0f} {LEVERAGE}x " f"· 延后 {age:.1f}s · ATR {r['atr_pct'] * 1e4:.1f}bp", flush=True) for x in lg: print(f" {x['tag']:<7}止盈 {x['atr']:.0f} ATR = " f"{x['atr'] * r['atr_pct'] * 1e4:.1f}bp · 止损 " f"{SL_ATR * r['atr_pct'] * 1e4:.1f}bp · 超时 {MAXB}min", flush=True) log_trade({"ev": "entry", "key": r["key"], "side": side, "entry_px": r["entry_px"], "atr_pct": r["atr_pct"], "notional": NOTIONAL, "leverage": LEVERAGE, "age_s": round(age, 2), "legs": lg, "dry": self.dry}) # 空跑也要走完 open_position：数量取整、价位对齐 tick、请求体构造都在 # 那里，跳过等于什么都没验。不下真单由 Bitget(dry=True) 负责 await self.open_position(r, lg) def qty_of(self, sym: str, entry: float) -> tuple[str, str]: """入场量与半仓量，都对齐步长。 入场量取到**步长的偶数倍**，半仓才是精确一半。不这么做 SOL 的半仓会是 全仓的 43%（步长 0.1 币 ≈ 10.7 USDT），而回测假设 50/50。 """ r = self.rules.get(f"{sym}USDT") if not r: raise RuntimeError(f"{sym} 没有合约规则") step = Decimal(str(r["sizeMultiplier"])) px = Decimal(str(entry)) grid = step * 2 n = max(Decimal("1"), (Decimal(str(NOTIONAL)) / px / grid).quantize(Decimal("1"))) qty = n * grid return str(qty), str(qty / 2) def snap(self, sym: str, px: float) -> str: r = self.rules[f"{sym}USDT"] tick = Decimal(str(r["priceEndStep"])) * ( Decimal(10) ** -int(r["pricePlace"])) q = (Decimal(str(px)) / tick).quantize(Decimal("1")) * tick return str(q) async def open_position(self, r: dict, lg: list[dict]) -> None: """两笔「市价入场 + 服务端止损」，再各挂一个 maker 止盈。 止损随入场单一起到交易所（presetStopLossPrice），所以不存在"已入场、 止损未挂"的裸仓窗口——那是本地盯价方案最危险的一段。 止盈单独挂 post_only 限价：成本模型里止盈按 maker 计且不吃滑点，用 preset（触发后市价）会让这部分变成 taker，预算就不成立了。 """ sym, d = r["sym"], r["direction"] pair = f"{sym}USDT" entry = r["entry_px"] a = entry * r["atr_pct"] _, half = self.qty_of(sym, entry) side = "buy" if d > 0 else "sell" close_side = "sell" if d > 0 else "buy" hold = "long" if d > 0 else "short" stop_px = self.snap(sym, entry - d * SL_ATR * a) opened = [] for x in lg: oid = oid_of(r["key"], x["tag"]) try: await self.api.entry_with_stop(pair, side, half, stop_px, oid) except Exception as e: print(f" ⛔ {x['tag']} 入场失败 {e}", flush=True) log_trade({"ev": "entry_fail", "key": r["key"], "tag": x["tag"], "err": str(e)}) continue tp_px = self.snap(sym, entry + d * x["atr"] * a) try: await self.api.tp_limit(pair, close_side, half, tp_px, oid + "-tp") except Exception as e: # 入场成了但止盈没挂上：仓位仍有服务端止损，不是裸仓。 # 超时腿会兜住它，所以只告警不强平 print(f" ⚠ {x['tag']} 止盈挂单失败 {e}" f"（仓位有服务端止损，超时腿会兜）", flush=True) opened.append({"tag": x["tag"], "oid": oid, "size": half, "tp_px": tp_px, "stop_px": stop_px}) print(f" {x['tag']:<7}{half} 币 · 止损 {stop_px} · " f"止盈 {tp_px}", flush=True) log_trade({"ev": "opened", "key": r["key"], "legs": opened, "stop_px": stop_px}) if opened: self.execs[r["key"]] = { "sym": sym, "pair": pair, "hold": hold, "deadline": time.time() + MAXB * 60, "legs": opened, # watch() 靠这个时间戳去 history-position 里认领对应的平仓记录 "opened_ms": int(time.time() * 1000), "side": "LONG" if d > 0 else "SHORT"} await tg.opened(sym, self.execs[r["key"]]["side"], entry, r["atr_pct"], NOTIONAL, LEVERAGE, stop_px, opened, time.time() - r["emit_ms"] / 1000.0) async def watch(self) -> None: """盯交易所侧的出场，把已实现盈亏记回闸。 为什么必须有这个循环：止损与止盈都挂在交易所侧，成交时本进程收不到 任何通知。缺了它有两个后果，都是静默的： 1. `Guard.realized()` 没人调用 → `pnl_day` 恒为 0 → **MAX_DAY_LOSS 这道闸完全不生效**。三条硬约束里最重要的一条。 2. `self.execs` 的条目要挂到 48 分钟截止才清 → `MAX_OPEN` 把已经 出场的仓位继续算在场 → 新信号被白挡掉。方向保守但不是本意。 盈亏取交易所的 `netProfit`（= pnl + 资金费 + 开平手续费），不自己按 标记价估——估会漏掉费用，而且方向总是偏乐观。 """ while True: await asyncio.sleep(WATCH_S) if self.dry or not self.execs: continue try: live = {p["symbol"] for p in await self.api.positions()} except Exception as e: # noqa: BLE001 print(f" ⚠ 盯仓读持仓失败 {type(e).__name__}: {e}", flush=True) continue gone = [k for k, st in self.execs.items() if st["pair"] not in live] if not gone: continue # 两个半仓在同一 symbol 上会被交易所净成一个仓位，所以一个 key # 对应一条历史记录。按最早的入场时间取一次历史，够覆盖全部 since = min(self.execs[k]["opened_ms"] for k in gone) - 60_000 try: hist = await self.api.history_positions(start_ms=since) except Exception as e: # noqa: BLE001 print(f" ⚠ 读历史持仓失败 {type(e).__name__}: {e}，" f"本轮不结算（下轮重试，不会漏）", flush=True) continue for key in gone: st = self.execs[key] rec = self._match_hist(hist, st) if rec is None: # 常见于刚平掉、历史还没落库。留着下轮再试；真丢了也有 # 48 分钟截止那条路兜住 execs 的清理 print(f" … {key} 已出场但历史未就绪，下轮再结算", flush=True) continue pnl = float(rec.get("netProfit") or 0.0) self.guard.realized(pnl) self.guard.save() self.execs.pop(key, None) print(f" ◀ {key} 交易所侧出场 · 已实现 {pnl:+.2f} USDT " f"· 当日累计 {self.guard.pnl_day:+.2f}", flush=True) log_trade({"ev": "closed", "key": key, "net_profit": pnl, "open_px": rec.get("openAvgPrice"), "close_px": rec.get("closeAvgPrice")}) await tg.closed(st["sym"], st["side"], pnl, float(rec.get("openAvgPrice") or 0), float(rec.get("closeAvgPrice") or 0), self.guard.pnl_day, self.guard.n_day) @staticmethod def _match_hist(hist: list, st: dict) -> dict | None: """在历史持仓里认领属于这一笔的记录。 按 symbol + holdSide 匹配，并要求收盘时间不早于入场时间（减 60s 容差， 两边时钟与落库都有抖动）。同一 symbol 有多条时取最近的一条。 """ best, best_t = None, -1.0 for r in hist: if r.get("symbol") != st["pair"] or r.get("holdSide") != st["hold"]: continue t = float(r.get("utime") or r.get("uTime") or 0) if t < st["opened_ms"] - 60_000: continue if t > best_t: best, best_t = r, t return best async def sweep(self) -> None: """超时腿：48 分钟到点市价平。 这是唯一必须靠本进程存活的出场腿。止损与止盈都在交易所侧，所以进程 死掉只会让持仓超过 48 根，不会变成裸仓——退化是良性的。 """ while True: await asyncio.sleep(5) now = time.time() for key, st in list(self.execs.items()): if now < st["deadline"]: continue try: pos = [p for p in await self.api.positions() if p["symbol"] == st["pair"]] if not pos: print(f" ◀ {key} 超时前已全部出场", flush=True) log_trade({"ev": "timeout_noop", "key": key}) else: for p in pos: await self.api.cancel_all(st["pair"]) await self.api.close_market( st["pair"], p["holdSide"], p["total"], oid_of(key, "timeout")) print(f" ◀ {key} 超时市价平 {pos[0]['total']} 币", flush=True) log_trade({"ev": "timeout_close", "key": key, "size": pos[0]["total"]}) except Exception as e: print(f" ⛔ {key} 超时平仓失败 {type(e).__name__}: {e}", flush=True) continue self.execs.pop(key, None) async def heartbeat(self) -> None: while True: await asyncio.sleep(300) # 跨日结算是 roll() 里同步留下的，在这里发出去 self.guard.roll() if self.guard.pending_roll: await tg.day_rolled(*self.guard.pending_roll) self.guard.pending_roll = None n_open = sum(1 for v in self.execs.values() if v) print(f" [心跳] 见信号 {self.n_seen} · 已做 {self.n_took} · " f"跳过 {self.n_skip} · 在场 {n_open}/{MAX_OPEN} · " f"当日 {self.guard.n_day}/{MAX_DAY} 笔 · " f"当日 PnL {self.guard.pnl_day:+.2f}/-{MAX_DAY_LOSS}", flush=True) async def shutdown(self) -> None: """收到停机信号：撤挂单 + 平掉在场仓位，然后退出。 为什么停机要平仓，而崩溃不需要：崩溃后 systemd/docker 会在几秒内重启， `reconcile` 接着就把遗留仓位清掉，空窗期有交易所侧止损兜着。而**主动 停机后没人重启**，仓位会一直挂到止损或止盈——48 分钟超时腿丢了，跑的 就不是回测那个出场结构了。 直接复用 reconcile：它做的正是"撤挂单 + 市价平掉一切"。 """ print("\n 收到停机信号，撤挂单并平掉在场仓位", flush=True) await tg.stopping(len(self.execs)) try: if self.dry: print(" 空跑模式，无仓位可平", flush=True) else: await asyncio.wait_for(self.reconcile(), timeout=60.0) except asyncio.TimeoutError: print(" ⛔ 平仓超过 60s 未完成。仓位仍有交易所侧止损，" "但超时腿已丢，去交易所确认", flush=True) except Exception as e: # noqa: BLE001 print(f" ⛔ 停机平仓失败 {type(e).__name__}: {e}，需人工介入", flush=True) finally: # 不关会话会在日志里留 "Unclosed client session"，且反复重启 # （Restart=always）会漏 socket try: await self.api.close() except Exception: # noqa: BLE001 pass async def run(self) -> None: await self.start() stop = asyncio.Event() loop = asyncio.get_running_loop() # 必须显式挂 SIGTERM：docker stop 与 systemd stop 默认发的都是它， # 而 Python 对 SIGTERM 不抛 KeyboardInterrupt，不挂就是直接消失、 # 没有任何清理。SIGINT 一并挂上，省得依赖 KillSignal= 那种绕法 for sig in (signal.SIGTERM, signal.SIGINT): loop.add_signal_handler(sig, stop.set) work = [asyncio.create_task(c) for c in (self.poll(), self.watch(), self.sweep(), self.heartbeat())] done, _ = await asyncio.wait( [*work, asyncio.create_task(stop.wait())], return_when=asyncio.FIRST_COMPLETED) for t in work: t.cancel() await asyncio.gather(*work, return_exceptions=True) # 任一主循环自己退出（异常）也走这里：仓位不能留给没人管的进程 for t in done: if t in work and (exc := t.exception()) is not None: print(f" ⛔ 主循环异常退出 {type(exc).__name__}: {exc}", flush=True) await tg.error("主循环异常退出，正在平仓并退出", f"{type(exc).__name__}: {exc}") await self.shutdown() def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--dry-run", action="store_true", help="不连交易所、不下单，只验总线与约束逻辑") ap.add_argument("--bus", default=str(signal_bus.BUS)) a = ap.parse_args() print(f"实盘执行器 · 名义 {NOTIONAL:.0f} USDT · {LEVERAGE}x · " f"并发≤{MAX_OPEN} · 日开仓≤{MAX_DAY} · 日亏损≤{MAX_DAY_LOSS}") asyncio.run(Exec(a.dry_run, Path(a.bus)).run()) if __name__ == "__main__": main()