diff --git a/live/bitget_rest.py b/live/bitget_rest.py index 80ee06f..ca86f2c 100644 --- a/live/bitget_rest.py +++ b/live/bitget_rest.py @@ -99,6 +99,26 @@ class Bitget: "marginCoin": MARGIN_COIN}) return [p for p in (d or []) if float(p.get("total") or 0) != 0] + async def history_positions(self, start_ms: int | None = None, + limit: int = 100) -> list: + """已平仓位，用来取**已实现盈亏**。 + + 为什么必须问交易所而不是自己算：止损与止盈都挂在交易所侧成交，本进程 + 看不到成交价；而且要算准还得含手续费与资金费。这个端点的 `netProfit` + 已经是 `pnl + totalFunding + openFee + closeFee`，正是日亏损上限该用 + 的数。自己按标记价估会把费用漏掉，方向还总是偏乐观。 + + 返回形状按文档是 `data.list`，但也见过直接给数组的写法，两种都收。 + 时间字段文档写 `ctime/utime`，官方 TS 类型写 `cTime/uTime`，同样都读。 + """ + p: dict = {"productType": PRODUCT, "limit": str(limit)} + if start_ms: + p["startTime"] = str(int(start_ms)) + d = await self._req("GET", "/api/v2/mix/position/history-position", p) + if isinstance(d, dict): + return list(d.get("list") or []) + return list(d or []) + async def fee_rate(self, symbol: str) -> dict: """账户在该合约上的**实际**费率档。 diff --git a/live/deploy/README.md b/live/deploy/README.md index 5e67596..62c31f7 100644 --- a/live/deploy/README.md +++ b/live/deploy/README.md @@ -111,6 +111,11 @@ journalctl -u chan-live-ship -f # 搬运日志 重连次数。管道死了但进程还活着是这里最危险的状态——`ServerAliveInterval=15` 负责让它变成一次可见的断开。 +**Telegram** 在 `live.env` 里填 `TG_TOKEN` / `TG_CHAT` 就开。启动时会推一条 +"执行器启动"，兼作通道自检——配错了当场就知道，而不是等几小时后第一个真信号 +来时才发现。推开仓、平仓（带已实现盈亏）、被硬约束挡住、报错、对账平仓、跨日 +结算；不推信号过期跳过（常态）和心跳。约 30 条/天上限。 + ## 停机与回滚 ```bash @@ -153,3 +158,9 @@ cd /opt/chan && sudo git reset --hard && sudo systemctl restart chan-live- staleness 闸挡掉。这是对的——参考成交价是次根开盘价，过了就不是回测那个价。 - **重启时会平掉交易所上已有的仓位**（`reconcile`）。接管要重建入场价、ATR、 剩余半仓状态和已过根数，任一项猜错就跑成另一个收益结构，所以选择平掉。 +- **日亏损上限依赖 `watch()` 循环**。止损与止盈在交易所侧成交，本进程收不到 + 通知，所以有个 10s 轮询去 `history-position` 取 `netProfit`（含手续费与资金 + 费）记回闸。这个循环停了，`pnl_day` 会恒为 0，`MAX_DAY_LOSS` 静默失效。 + 心跳里会打 `当日 PnL x/-20`，值一直是 0.00 而又确实有平仓，就是它出了问题。 +- **平仓后 `MAX_OPEN` 名额由 `watch()` 释放**，不是立刻。最坏延迟 10s。若 + 历史记录还没落库，会等下一轮，日志里打"历史未就绪，下轮再结算"。 diff --git a/live/deploy/live.env.example b/live/deploy/live.env.example index 800ea42..ab8edaf 100644 --- a/live/deploy/live.env.example +++ b/live/deploy/live.env.example @@ -48,3 +48,21 @@ LIVE_STALE_S=20 # 交易的币池。要与采集机一致，否则会收到不做的币的信号（会被忽略但徒增噪声） SYMS=BTC,ETH,SOL,BNB,XRP,DOGE,ADA,AVAX,LINK,LTC + +# 盯交易所侧出场的轮询间隔（秒）。**别关掉这个循环**：止损与止盈在交易所侧 +# 成交，本进程收不到通知，缺了它 pnl_day 恒为 0，MAX_DAY_LOSS 就是死的 +LIVE_WATCH_S=10 + +# ── Telegram 通知 ───────────────────────────────────────────────── +# 拿 token：Telegram 里找 @BotFather → /newbot +# 拿 chat id：给 bot 随便发一句，再开 +# https://api.telegram.org/bot/getUpdates 看 result[0].message.chat.id +# +# 留空则完全不推（不报错）。推的内容：开仓、平仓（带已实现盈亏）、被硬约束 +# 挡住、报错、对账平仓、跨日结算、启动与停机。 +# **不推**信号过期跳过（常态，搬运重连会重放旧信号）和心跳（日志里有）。 +# 量级约 30 条/天上限。 +TG_TOKEN= +TG_CHAT= +# 前缀，用来和采集机推的手工信号区分开——两边可以共用同一个 bot 和对话 +TG_TAG=实盘 diff --git a/live/deploy/status.sh b/live/deploy/status.sh index 597b9da..17e2a7b 100755 --- a/live/deploy/status.sh +++ b/live/deploy/status.sh @@ -54,7 +54,10 @@ today = time.strftime("%Y-%m-%d") day = d.get("day", "?") stale = "" if day == today else f" ⚠ 是 {day} 的，跨日后首次开仓时才归零" print(f" 当日 {day}{stale}") -print(f" 已开 {d.get('n_day', 0)} 笔 · 盈亏 {d.get('pnl_day', 0.0):+.2f} USDT") +pnl, n = d.get("pnl_day", 0.0), d.get("n_day", 0) +# pnl 恒为 0 而又确实开过仓，说明 watch() 没在记账 → MAX_DAY_LOSS 是死的 +warn = " ⚠ 开过仓但盈亏仍为 0，查 watch() 是否在跑" if n and pnl == 0 else "" +print(f" 已开 {n} 笔 · 盈亏 {pnl:+.2f} USDT{warn}") print(f" 已处理信号键 {len(d.get('done', []))} 个（幂等去重用，留最近 5000）") PY else diff --git a/live/live_exec.py b/live/live_exec.py index bc1e488..fc97a70 100644 --- a/live/live_exec.py +++ b/live/live_exec.py @@ -78,6 +78,7 @@ 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 @@ -99,6 +100,9 @@ MAX_DAY = int(os.environ.get("LIVE_MAX_DAY", "15")) 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")) @@ -135,6 +139,7 @@ class Guard: 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: @@ -165,6 +170,8 @@ class Guard: 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() @@ -248,6 +255,9 @@ class Exec: 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) if self.dry: print(" ⚠ 空跑模式：不下真单", flush=True) return @@ -294,6 +304,9 @@ class Exec: 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')}", @@ -339,6 +352,10 @@ class Exec: 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"]) @@ -428,12 +445,95 @@ class Exec: 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} - log_trade({"ev": "opened", "key": r["key"], "legs": opened, - "stop_px": stop_px}) + "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 分钟到点市价平。 @@ -472,6 +572,11 @@ class Exec: 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} · " @@ -490,6 +595,7 @@ class Exec: 直接复用 reconcile：它做的正是"撤挂单 + 市价平掉一切"。 """ print("\n 收到停机信号，撤挂单并平掉在场仓位", flush=True) + await tg.stopping(len(self.execs)) try: if self.dry: print(" 空跑模式，无仓位可平", flush=True) @@ -519,7 +625,8 @@ class Exec: 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.sweep(), self.heartbeat())] + 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) @@ -531,6 +638,8 @@ class Exec: 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() diff --git a/live/tg.py b/live/tg.py new file mode 100644 index 0000000..cf7b405 --- /dev/null +++ b/live/tg.py @@ -0,0 +1,112 @@ +"""生产侧 Telegram 通知。只用标准库 + aiohttp。 + +## 推什么、不推什么 + +推送的价值在于**你会因此做点什么**。按这个标准筛： + + 推 开仓 能立刻眼看一遍方向/价位对不对 + 推 平仓 带已实现盈亏（含手续费与资金费），这是唯一的真账 + 推 闸拦截 仅 MAX_OPEN / MAX_DAY / MAX_DAY_LOSS —— 说明有 bug 或策略在流血 + 推 报错、对账平仓、跨日结算 + 不推 信号过期跳过 这是常态（重连重放会带上旧信号），推了就淹掉真事 + 不推 心跳 日志里有，推了每天 288 条 + +量级：信号 6.8 个/天、日开仓上限 15，所以最多约 30 条/天。 + +## 失败一律只打日志 + +推送挂了不能连坐交易。所有异常在这里吞掉——上层不该因为 Telegram 抽风而 +影响下单或平仓。 +""" +from __future__ import annotations + +import os +import time + +TOKEN = os.environ.get("TG_TOKEN", "") +CHAT = os.environ.get("TG_CHAT", "") +ENABLED = bool(TOKEN and CHAT) +# 生产机上给这条推送打个前缀，免得和采集机推的手工信号混在一个对话里分不清 +TAG = os.environ.get("TG_TAG", "实盘") + + +async def send(text: str) -> None: + """推一条。任何失败都只打日志——推送挂了不能连坐交易。""" + if not ENABLED: + return + try: + import aiohttp + url = f"https://api.telegram.org/bot{TOKEN}/sendMessage" + async with aiohttp.ClientSession() as s: + async with s.post(url, + json={"chat_id": CHAT, "text": text}, + timeout=aiohttp.ClientTimeout(total=10)) as r: + if r.status != 200: + body = (await r.text())[:200] + print(f" [TG] 推送失败 HTTP {r.status} {body}", flush=True) + except Exception as e: # noqa: BLE001 + print(f" [TG] 推送异常 {type(e).__name__}: {e}", flush=True) + + +def _fmt(px: float) -> str: + """按量级选小数位。跨币种从 79000 到 0.087，固定位数会难读或丢精度。""" + a = abs(px) + if a >= 1000: + return f"{px:.1f}" + if a >= 10: + return f"{px:.3f}" + if a >= 1: + return f"{px:.4f}" + return f"{px:.6f}" + + +async def opened(sym: str, side: str, entry: float, atr_pct: float, + notional: float, leverage: int, stop_px: str, + legs: list[dict], age_s: float) -> None: + arrow = "多" if side == "LONG" else "空" + lines = [f"[{TAG}] 开仓 {sym} {arrow}", + f"参考价 {_fmt(entry)} · ATR {atr_pct * 1e4:.1f}bp", + f"名义 {notional:.0f} USDT · {leverage}x · " + f"保证金 {notional / leverage:.0f} USDT", + f"止损 {stop_px}（两腿共用，不移动）"] + for lg in legs: + lines.append(f" {lg['tag']:<6} {lg['size']} 币 · 止盈 {lg['tp_px']}") + lines.append(f"距参考价成立 {age_s:.1f}s") + await send("\n".join(lines)) + + +async def closed(sym: str, side: str, pnl: float, open_px: float, + close_px: float, day_pnl: float, day_n: int, + reason: str = "") -> None: + arrow = "多" if side == "LONG" else "空" + mark = "盈" if pnl > 0 else ("亏" if pnl < 0 else "平") + tail = f" · {reason}" if reason else "" + await send(f"[{TAG}] 平仓 {sym} {arrow} {mark} {pnl:+.2f} USDT{tail}\n" + f"开 {_fmt(open_px)} → 平 {_fmt(close_px)}\n" + f"当日 {day_n} 笔 · 累计 {day_pnl:+.2f} USDT") + + +async def blocked(key: str, why: str) -> None: + """只在被硬约束挡住时推。过期跳过属于常态，不走这里。""" + await send(f"[{TAG}] ⛔ 信号被挡 {key}\n{why}") + + +async def error(what: str, detail: str) -> None: + await send(f"[{TAG}] ⛔ {what}\n{detail[:500]}") + + +async def started(notional: float, leverage: int, syms: int, + dry: bool) -> None: + mode = "空跑（不下真单）" if dry else "真跑" + await send(f"[{TAG}] 执行器启动 · {mode}\n" + f"名义 {notional:.0f} USDT · {leverage}x · {syms} 币") + + +async def day_rolled(day: str, n: int, pnl: float) -> None: + await send(f"[{TAG}] {day} 结算 · {n} 笔 · {pnl:+.2f} USDT") + + +async def stopping(n_open: int) -> None: + await send(f"[{TAG}] 收到停机信号，平掉在场 {n_open} 笔后退出。" + f"\n注意：停机后不再有超时平仓与新开仓。" + f"发生时间 {time.strftime('%H:%M:%S')}") diff --git a/research/live/tg_notify.py b/research/live/tg_notify.py index de85f58..aa57ee8 100644 --- a/research/live/tg_notify.py +++ b/research/live/tg_notify.py @@ -185,20 +185,18 @@ def build(sym: str, direction: int, entry: float, atr_pct: float, async def send(text: str) -> None: - """推一条。任何失败都只打日志——推送挂了不能连坐采集。""" - if not ENABLED: - return - try: - import aiohttp - url = f"https://api.telegram.org/bot{TOKEN}/sendMessage" - async with aiohttp.ClientSession() as s: - async with s.post(url, json={"chat_id": CHAT, "text": text}, - timeout=aiohttp.ClientTimeout(total=10)) as r: - if r.status != 200: - print(f" [TG] 推送失败 HTTP {r.status} " - f"{(await r.text())[:200]}", flush=True) - except Exception as e: - print(f" [TG] 推送异常 {type(e).__name__}: {e}", flush=True) + """推一条。传输层复用生产侧的 `live/tg.py`，这里不再维护第二份。 + + 方向与 `signal_bus` 一致：**生产持有实现，研究侧反过来 import**。反过来 + 写成生产 import 研究侧，就等于把研究侧的依赖树绑到实盘进程上。 + """ + from pathlib import Path + import sys + p = str(Path(__file__).resolve().parents[2] / "live") + if p not in sys.path: + sys.path.insert(0, p) + import tg + await tg.send(text) async def push_signal(sym: str, direction: int, entry: float, atr_pct: float,