From 881cc5e77d9742adfabf210f6b39d0caa0c62ca6 Mon Sep 17 00:00:00 2001 From: jack Date: Fri, 28 Aug 2026 18:58:47 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E4=B8=8A=E3=80=8C=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E5=B8=81=E4=B8=80=E4=B8=AA=E4=BB=93=E4=BD=8D=E3=80=81=E8=B4=A6?= =?UTF-8?q?=E6=88=B7=E4=B8=93=E7=94=A8=E3=80=8D=E8=BF=99=E4=B8=A4=E4=B8=AA?= =?UTF-8?q?=E6=B2=A1=E8=90=BD=E5=9C=A8=E4=BB=A3=E7=A0=81=E9=87=8C=E7=9A=84?= =?UTF-8?q?=E5=89=8D=E6=8F=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 三处都源于同一个隐含假设从未被写成代码。 ① 同币并发会让 pnl_day 翻倍。blocks() 只查幂等键与三条计数闸，没有按币的 占用检查。同币开两笔时交易所净成一个仓位，于是 watch() 按 pair 判出场会让 两个 key 同时进 gone，_match_hist 给它们返回同一条历史记录，realized() 被 调两次。而 pnl_day 正是 MAX_DAY_LOSS 读的数：亏损翻倍提前停机，盈利翻倍让 闸变迟钝。sweep() 也会在第一笔截止时平掉合并后的整个仓位。 合并后的行为（一个止损、两个不同价位的止盈、超时一锅端）不是任何一版回测 建模的东西，所以在 on_signal 里跳过第二个信号，是最接近安全的近似。期望并发 0.18 笔，损失极小。另外给 _match_hist 加 positionId 独占认领，让这个不变量 在记账处本地成立，而不是依赖两百行外的检查——花钱的路径值得两道。 ② positions() 返回账户全部仓位，三个调用点都没过滤。账户上任何第三方仓位 都会被 reconcile 在重启时市价平掉，而 watch() 会因该 symbol 一直在场而永不 结算，MAX_OPEN 名额泄漏、pnl_day 不再更新。加 PAIRS 过滤与 my_positions()。 残留局限记在注释里：同币上的第三方仓位仍分不出来，账户仍应专用。 ③ oid_of 把非字母数字换成下划线，理由是 : 和 + 未必被接受，但调用方又拼了 "-tp" 把 - 加回去，自相矛盾。真被拒时止盈单会全部挂不上，而那条路径只告警 不停机，收益结构静默退化成「只有止损 + 超时」，且空跑验不到（dry 返回假 成功）。改用 _tp，并把字符集约束写进 oid_of 的文档。 实测：同币第二个信号被挡、别币放行、独占认领不重复计账、PAIRS 排除 PEPEUSDT、全部 clientOid 只含字母数字下划线。 Co-authored-by: Cursor --- live/live_exec.py | 63 +++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 55 insertions(+), 8 deletions(-) diff --git a/live/live_exec.py b/live/live_exec.py index 053fd16..2315324 100644 --- a/live/live_exec.py +++ b/live/live_exec.py @@ -87,6 +87,12 @@ 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(",") +# 只认这些交易对。交易所的 all-position 返回**账户全部**仓位，不过滤的话 +# 账户上任何第三方仓位（手工单、另一个策略、试单忘了平）都会被 reconcile +# 在下次重启时市价平掉；watch() 还会因为该 symbol 一直在场而永不结算对应的 +# key，MAX_OPEN 名额泄漏、pnl_day 不再更新。把「账户只归执行器」这个前提 +# 从口头约定变成代码里的过滤 +PAIRS = frozenset(f"{s}USDT" for s in SYMS) NOTIONAL = float(os.environ.get("LIVE_NOTIONAL", "500")) LEVERAGE = int(os.environ.get("LIVE_LEVERAGE", "10")) @@ -217,6 +223,11 @@ def oid_of(key: str, tag: str) -> str: clientOid 是**交易所级幂等**：重发同一个 oid 会被拒。这比本地去重可靠， 因为「已发出但没收到回复」这种情况本地判不了，重试就会开两次仓。 + + 调用方拼后缀时也要守这个字符集（用 `_tp` 而不是 `-tp`）。曾经拼过 `-`， + 和这里的理由自相矛盾；真被拒的话止盈单挂不上，而那条路径只告警不停机， + 收益结构会静默退化成「只有止损 + 超时」，空跑还验不到（dry 直接返回 + 假成功）。 """ # 方向必须显式编码：直接把非字母数字换成下划线，会让 `+1` 和 `-1` 都变成 # `_1`，同一根上的多空信号得到相同 oid，第二笔被交易所当重复拒掉 @@ -368,7 +379,7 @@ class Exec: 且行为确定。 """ try: - pos = await self.api.positions() + pos = await self.my_positions() except Exception as e: print(f" ⚠ 对账读持仓失败 {type(e).__name__}: {e}", flush=True) return @@ -417,6 +428,15 @@ class Exec: # 一条记录要能下单，这些字段一个都不能缺 NEEDED = ("key", "sym", "emit_ms", "direction", "entry_px", "atr_pct") + async def my_positions(self) -> list: + """只看 SYMS 里那些币的仓位。 + + **不要**直接用 api.positions()：它返回账户全部仓位，理由见 PAIRS。 + 残留的局限要知道：同一个币上的第三方仓位（比如你手工开了 ADA）仍然 + 分不出来，那需要按 clientOid 逐单认领，成本不划算。所以账户仍应专用。 + """ + return [p for p in await self.api.positions() if p["symbol"] in PAIRS] + async def on_signal(self, r: dict) -> None: # 先校验再取值。缺了这一步，总线上一条字段不全的记录会抛 KeyError， # 把 poll 任务打死，进而**整个执行器停止交易**——一行坏数据换全面停摆， @@ -432,6 +452,21 @@ class Exec: 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) + # 同币占用检查。整套记账隐含「一个币最多一个仓位」这个前提，但它原先 + # 只存在于口头上。同币开两笔时交易所会**净成一个仓位**，于是： + # · watch() 按 pair 判出场，两个 key 同时进 gone；_match_hist 给它们 + # 返回同一条历史记录，realized() 被调两次 → pnl_day 翻倍。而这正是 + # MAX_DAY_LOSS 读的数：亏损翻倍会提前停机，盈利翻倍会让闸变迟钝 + # · sweep() 在第一笔截止时平掉合并后的整个仓位，把第二笔才持有十分钟 + # 的部分一并平掉 + # 合并后的行为（一个止损、两个不同价位的止盈、超时一锅端）不是任何一版 + # 回测建模的东西，所以跳过第二个信号是最接近安全的近似。期望并发 0.18 + # 笔，这一跳损失极小 + if why is None: + dup = [k for k, v in self.execs.items() if v["sym"] == r["sym"]] + if dup: + why = (f"{r['sym']} 已有在场仓位 {dup[0]}——同币开两笔会在" + f"交易所侧净成一个仓位，把盈亏记账和超时腿都搞错") if why is None and age > STALE_S: why = f"信号已过期 {age:.1f}s > {STALE_S:.0f}s" if why: @@ -521,7 +556,7 @@ class Exec: tp_px = self.snap(sym, entry + d * x["atr"] * a) try: await self.api.tp_limit(pair, close_side, half, tp_px, - oid + "-tp") + oid + "_tp") except Exception as e: # 入场成了但止盈没挂上：仓位仍有服务端止损，不是裸仓。 # 超时腿会兜住它，所以只告警不强平 @@ -564,7 +599,7 @@ class Exec: if self.dry or not self.execs: continue try: - live = {p["symbol"] for p in await self.api.positions()} + live = {p["symbol"] for p in await self.my_positions()} except Exception as e: # noqa: BLE001 print(f" ⚠ 盯仓读持仓失败 {type(e).__name__}: {e}", flush=True) continue @@ -581,9 +616,13 @@ class Exec: print(f" ⚠ 读历史持仓失败 {type(e).__name__}: {e}，" f"本轮不结算（下轮重试，不会漏）", flush=True) continue + # 一条历史记录只能被一个 key 认领。同币并发已在 on_signal 拦住， + # 但记账是花钱的路径：让这个不变量在本地成立，而不是依赖两百行外 + # 的另一处检查。重复认领会让 realized() 被调两次、pnl_day 翻倍 + claimed: set = set() for key in gone: st = self.execs[key] - rec = self._match_hist(hist, st) + rec = self._match_hist(hist, st, claimed) if rec is None: # 常见于刚平掉、历史还没落库。留着下轮再试；真丢了也有 # 48 分钟截止那条路兜住 execs 的清理 @@ -605,21 +644,29 @@ class Exec: self.guard.pnl_day, self.guard.n_day) @staticmethod - def _match_hist(hist: list, st: dict) -> dict | None: + def _match_hist(hist: list, st: dict, + claimed: set | None = None) -> dict | None: """在历史持仓里认领属于这一笔的记录。 按 symbol + holdSide 匹配，并要求收盘时间不早于入场时间（减 60s 容差， 两边时钟与落库都有抖动）。同一 symbol 有多条时取最近的一条。 + + `claimed` 装已被别的 key 认走的 positionId，防止两个 key 认到同一条。 """ - best, best_t = None, -1.0 + best, best_t, best_id = None, -1.0, None for r in hist: if r.get("symbol") != st["pair"] or r.get("holdSide") != st["hold"]: continue + pid = r.get("positionId") + if claimed is not None and pid is not None and pid in claimed: + 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 + best, best_t, best_id = r, t, pid + if best is not None and claimed is not None and best_id is not None: + claimed.add(best_id) return best async def sweep(self) -> None: @@ -635,7 +682,7 @@ class Exec: if now < st["deadline"]: continue try: - pos = [p for p in await self.api.positions() + pos = [p for p in await self.my_positions() if p["symbol"] == st["pair"]] if not pos: print(f" ◀ {key} 超时前已全部出场", flush=True)