feat: minis launch, cold-start watch and delivery watch skills

Three TikTok Minis skills for Codex and Claude Code, with the cold-start playbook and five benchmark accounts through 88776-1 (2026-09-30). install.sh links them into ~/.codex/skills and ~/.claude/skills. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-30 15:10:54 +08:00
commit 9f9e70ea2d
18 changed files with 2250 additions and 0 deletions
@@ -0,0 +1,90 @@
+"""Gather everything a lifecycle summary needs after an account is stopped or punished.
+
+    python3 summary.py --account <external_account_id> --date 2026-09-29 --dir ~/minis-watch/88794-9 [--window today|yesterday]
+
+Prints: hourly curve (direct TikTok), per-ad totals, the milestone events, the loss/peak points from
+the timeline, and the same-lifecycle comparison against references/benchmarks.json. The narrative is
+written by the agent from this output using references/summary-template.md.
+"""
+import argparse
+import json
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import swads_mcp as mcp  # noqa: E402
+
+ap = argparse.ArgumentParser()
+ap.add_argument("--account", required=True)
+ap.add_argument("--date", required=True)
+ap.add_argument("--dir", required=True)
+ap.add_argument("--window", default="today", help="ad breakdown window: today or yesterday (the day being summarised)")
+a = ap.parse_args()
+D = os.path.expanduser(a.dir)
+
+h = mcp.call("runtime_hourly_metrics", {"external_account_id": a.account, "date": a.date})
+hours = {}
+for r in h.get("rows") or []:
+    x = hours.setdefault(int(r["hour"]), [0.0, 0.0])
+    x[0] += float(r.get("spend") or 0); x[1] += float(r.get("ad_revenue_d0") or 0)
+print(f"== 每小时(直读 TikTok,{h.get('reporting_timezone')},{h.get('date')})")
+cs = cr = 0.0
+for k in sorted(hours):
+    s, r = hours[k]
+    if s <= 0:
+        continue
+    cs += s; cr += r
+    print(f"{k:02d} 点  消耗 {s:8.2f}  ROAS {r / s:5.2f}  累计 {cs:8.2f}/{cr:8.2f} = {cr / cs:.2f}")
+print(f"合计 消耗 {cs:.2f} 回收 {cr:.2f} ROAS {(cr / cs if cs else 0):.2f} 盈亏 {cr - cs:+.2f}")
+
+q = mcp.call("metrics_semantic_query", {"external_account_id": a.account, "query": {
+    "query_kind": "ad_breakdown", "metrics": ["spend", "impressions", "clicks", "d0_ad_revenue"],
+    "dimensions": ["external_campaign_name", "external_ad_name", "external_ad_id"],
+    "window": a.window, "exclude_zero_spend": True, "order_by_metric": "spend"}})
+cols = [c["name"] for c in q.get("columns") or []]
+print("\n== 每条广告(按消耗)")
+for r in q.get("rows") or []:
+    d = dict(zip(cols, r)); s = d["spend"] or 0; rv = d["d0_ad_revenue"] or 0
+    ctr = d["clicks"] / d["impressions"] * 100 if d.get("impressions") else 0
+    print(f"{d['external_campaign_name']} | {' '.join(d['external_ad_name'].split(' ')[-2:])} | {d['external_ad_id']} | 消耗 {s:.2f} | ROAS {(rv / s if s else 0):.2f} | CTR {ctr:.1f}%")
+
+ev = os.path.join(D, "events.jsonl")
+if os.path.exists(ev):
+    print("\n== 生命周期事件")
+    for line in open(ev):
+        x = json.loads(line)
+        if x["kind"] not in ("广告审核",):
+            print(x["t"], x["kind"], x["text"][:160])
+
+tl = os.path.join(D, "timeline.jsonl")
+if os.path.exists(tl):
+    snaps = [json.loads(l) for l in open(tl) if l.strip() and l.startswith('{"t": "' + a.date)]
+    snaps = [s for s in snaps if s.get("spend")]
+    if snaps:
+        worst = max(snaps, key=lambda s: s["spend"] - s["rev"])
+        best = max(snaps, key=lambda s: s["rev"] / s["spend"] if s["spend"] >= 20 else 0)
+        print(f"\n== 关键点\n亏得最多:{worst['t']}(数据截至 {worst['fresh']})消耗 {worst['spend']} 回收 {worst['rev']} 亏 {worst['spend'] - worst['rev']:.2f}")
+        print(f"累计 ROAS 最高:{best['t']} {best['rev'] / best['spend']:.2f}")
+
+bench = json.load(open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "references", "benchmarks.json")))
+mine = sorted((k, v) for k, v in hours.items() if v[0] > 0)
+ramp = [k for k, v in mine if v[0] >= 20]
+print("\n== 同生命周期对比(以各户第一个消耗 ≥ 20 的整点为起量第 1 小时,列累计 ROAS)")
+rows = [("本户", ramp[0] if ramp else None, dict(mine))]
+for b in bench["accounts"]:
+    hs = {int(k): v for k, v in b["hourly"].items()}
+    rh = [k for k in sorted(hs) if hs[k][0] >= 20]
+    rows.append((f"{b['short']}({b['kind']})", rh[0] if rh else None, hs))
+print("起量第 N 小时 | " + " | ".join(r[0] for r in rows))
+for n in range(0, 10):
+    cells = []
+    for _, r0, hs in rows:
+        if r0 is None:
+            cells.append("-"); continue
+        upto = [v for k, v in hs.items() if k <= r0 + n]
+        s = sum(v[0] for v in upto); r = sum(v[1] for v in upto)
+        cells.append(f"{r / s:.2f}" if s and (r0 + n) in hs else "-")
+    if any(c != "-" for c in cells):
+        print(f"第 {n + 1} 小时 | " + " | ".join(cells))
+for b in bench["accounts"]:
+    print(f"- {b['short']} {b['date']}:{b['outcome']}。{b['lesson']}")
@@ -0,0 +1,121 @@
+"""Minimal Streamable-HTTP client for the swads MCP gateway, so one watch round is one script run.
+
+Credentials are never stored here. Resolution order:
+  1. SWADS_MCP_URL + SWADS_MCP_TOKEN environment variables;
+  2. the `swads` entry under `mcpServers` in ~/.claude.json (project-scoped first, then global),
+     i.e. the same server the operator's Claude Code session already uses;
+  3. the `swads` entry under `mcp_servers` in ~/.codex/config.toml (Codex users).
+"""
+import json
+import os
+import time
+import urllib.error
+import urllib.request
+
+
+def _load_config():
+    url, token = os.environ.get("SWADS_MCP_URL"), os.environ.get("SWADS_MCP_TOKEN")
+    if url and token:
+        return url, {"Authorization": f"Bearer {token}"}
+    path = os.path.expanduser("~/.claude.json")
+    try:
+        cfg = json.load(open(path))
+    except (OSError, ValueError):
+        cfg = {}
+    candidates = []
+    cwd = os.environ.get("SWADS_PROJECT_DIR") or os.getcwd()
+    projects = cfg.get("projects") or {}
+    for proj in [cwd] + [p for p in projects if p != cwd]:
+        server = ((projects.get(proj) or {}).get("mcpServers") or {}).get("swads")
+        if server:
+            candidates.append(server)
+    if (cfg.get("mcpServers") or {}).get("swads"):
+        candidates.append(cfg["mcpServers"]["swads"])
+    for server in candidates:
+        if server.get("url", "").startswith("http") and "127.0.0.1" not in server["url"]:
+            return server["url"], dict(server.get("headers") or {})
+    for server in candidates:
+        if server.get("url"):
+            return server["url"], dict(server.get("headers") or {})
+    try:
+        import tomllib
+        server = (tomllib.load(open(os.path.expanduser("~/.codex/config.toml"), "rb"))
+                  .get("mcp_servers") or {}).get("swads") or {}
+    except (OSError, ValueError, ImportError):
+        server = {}
+    if server.get("url"):
+        return server["url"], dict(server.get("http_headers") or {})
+    raise SystemExit("找不到 swads MCP 配置:设置 SWADS_MCP_URL / SWADS_MCP_TOKEN,或先在 Claude Code / Codex 里配置 swads MCP 服务器")
+
+
+URL, HEADERS = _load_config()
+_sid = None
+_n = 0
+
+
+def _post_once(body):
+    global _sid
+    headers = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream", **HEADERS}
+    if _sid:
+        headers["Mcp-Session-Id"] = _sid
+    req = urllib.request.Request(URL, data=json.dumps(body).encode(), headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=300) as resp:
+        if resp.headers.get("Mcp-Session-Id"):
+            _sid = resp.headers["Mcp-Session-Id"]
+        raw = resp.read().decode()
+        ctype = resp.headers.get("Content-Type", "")
+    if not raw.strip():
+        return None
+    if "event-stream" in ctype:
+        msgs = [json.loads(line[5:]) for line in raw.splitlines() if line.startswith("data:")]
+        return msgs[-1] if msgs else None
+    return json.loads(raw)
+
+
+def _post(body, tries=4):
+    """Retry transport failures (SSL EOF, 5xx, dropped connections); never retry a 4xx."""
+    for i in range(tries):
+        try:
+            return _post_once(body)
+        except urllib.error.HTTPError as e:
+            if e.code < 500 or i == tries - 1:
+                raise
+        except Exception:
+            if i == tries - 1:
+                raise
+        time.sleep(5 * (i + 1))
+
+
+def init():
+    global _sid
+    _sid = None
+    _post({"jsonrpc": "2.0", "id": 0, "method": "initialize",
+           "params": {"protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": {"name": "minis-launch-watch", "version": "1"}}})
+    _post({"jsonrpc": "2.0", "method": "notifications/initialized"})
+
+
+def call(name, args):
+    """Call one gateway tool and return its structured result; raise on a tool error."""
+    global _n
+    _n += 1
+    r = _post({"jsonrpc": "2.0", "id": _n, "method": "tools/call", "params": {"name": name, "arguments": args}})
+    if "error" in r:
+        raise RuntimeError(json.dumps(r["error"], ensure_ascii=False))
+    res = r["result"]
+    sc = res.get("structuredContent")
+    if sc is not None:
+        out = sc.get("result", sc) if isinstance(sc, dict) and list(sc) == ["result"] else sc
+    else:
+        text = "".join(c.get("text", "") for c in res.get("content", []))
+        try:
+            out = json.loads(text)
+        except ValueError:
+            out = text
+    if res.get("isError"):
+        raise RuntimeError(str(out)[:2000])
+    if isinstance(out, dict) and out.get("error_code") and out.get("upstream_status"):
+        raise RuntimeError(json.dumps(out, ensure_ascii=False)[:2000])
+    return out
+
+
+init()
@@ -0,0 +1,353 @@
+"""One watch round for a freshly launched Minis account. Prints the alerts, then one status line.
+
+    python3 watch.py --account <external_account_id> --start "2026-09-29 10:08" --dir ~/minis-watch/88794-9
+
+Everything it concludes is a suggestion; it never writes to the platform. State, the per-round
+timeline and the milestone events live in --dir, so the same directory must be passed every round.
+Thresholds come from references/playbook.md; keep the two in step.
+"""
+import argparse
+import datetime
+import json
+import os
+import re
+import sys
+import time
+from collections import Counter
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import swads_mcp as mcp  # noqa: E402
+
+ap = argparse.ArgumentParser()
+ap.add_argument("--account", required=True)
+ap.add_argument("--start", required=True, help="actual enable time, account timezone, 'YYYY-MM-DD HH:MM'")
+ap.add_argument("--dir", required=True)
+ap.add_argument("--prefix", default="SW-ADS-")
+ap.add_argument("--sync-every", type=int, default=3, help="sync platform status every N rounds")
+a = ap.parse_args()
+
+A = a.account
+D = os.path.expanduser(a.dir)
+os.makedirs(D, exist_ok=True)
+SF, TL, EV = f"{D}/state.json", f"{D}/timeline.jsonl", f"{D}/events.jsonl"
+st = json.load(open(SF)) if os.path.exists(SF) else {}
+rep = st.setdefault("reported", {})
+lc = st.setdefault("lc", {})
+
+
+def f(x):
+    try:
+        return float(x)
+    except (TypeError, ValueError):
+        return 0.0
+
+
+def short(name):
+    s = re.sub(r"^" + re.escape(a.prefix) + r"[^-]+-", "", name)
+    return re.sub(r"(-图文)?-\d{8}$", "", s)
+
+
+def rows(res):
+    cols = [c if isinstance(c, str) else c.get("name") for c in res.get("columns") or []]
+    return [r if isinstance(r, dict) else dict(zip(cols, r)) for r in res.get("rows") or []]
+
+
+# ---- clock: the account's reporting timezone, not the machine's ----
+ctx = mcp.call("campaign_list", {"external_account_id": A, "page_size": 100})
+acct = ctx.get("account_context") or {}
+TZ = datetime.timezone(datetime.timedelta(minutes=int(acct.get("reporting_utc_offset_minutes", 480))))
+now = datetime.datetime.now(TZ)
+START = datetime.datetime.strptime(a.start, "%Y-%m-%d %H:%M").replace(tzinfo=TZ)
+hours_in = (now - START).total_seconds() / 3600
+today = now.strftime("%Y-%m-%d")
+
+
+def event(kind, text, t=None):
+    open(EV, "a").write(json.dumps({"t": t or now.strftime("%Y-%m-%d %H:%M"), "kind": kind, "text": text}, ensure_ascii=False) + "\n")
+
+
+if not lc.get("started"):
+    lc["started"] = True
+    event("开投", f"账户 {A} 开投", t=START.strftime("%Y-%m-%d %H:%M"))
+
+camps = {c["campaign_name"]: c for c in ctx["items"]
+         if c["campaign_name"].startswith(a.prefix) and c.get("operation_status") == "ENABLE"}
+
+# ---- ad labels: 剧名 Ad N(图文/视频), keyed by every remote id the workspace knows ----
+admap, workspaces = {}, {}
+for name, c in camps.items():
+    w = mcp.call("campaign_workspace", {"campaign_id": c["campaign_id"], "external_account_id": A})
+    workspaces[name] = w
+    for ad in w["groups"][0]["ads"]:
+        kind = "视频" if any(v.get("variant_kind") == "video" for v in ad["creative_variants"]) else "图文"
+        label = f"{short(name)} {' '.join(ad['ad_name'].split(' ')[-2:])}({kind})"
+        p = ad.get("payload") or {}
+        for rid in (ad.get("remote_id"), p.get("smart_plus_ad_id"), p.get("external_ad_id"), p.get("ad_id")):
+            if rid:
+                admap[str(rid)] = (label, name)
+
+# ---- today's per-ad numbers (synced; can lag 10-60 min) ----
+q = mcp.call("metrics_semantic_query", {"external_account_id": A, "query": {
+    "query_kind": "ad_breakdown", "metrics": ["spend", "impressions", "clicks", "d0_ad_revenue"],
+    "dimensions": ["external_campaign_name", "external_ad_name", "external_ad_id"],
+    "window": "today", "exclude_zero_spend": True, "order_by_metric": "spend"}})
+fr = [x.get("freshness") for x in q["query_meta"].get("attribution_resolutions", []) if x.get("freshness")]
+fresh_iso = fr[0] if fr else None
+fresh = datetime.datetime.fromisoformat(fresh_iso.replace("Z", "+00:00")).astimezone(TZ).strftime("%H:%M") if fresh_iso else "?"
+ads, agg = [], {}
+for r in rows(q):
+    name = r.get("external_campaign_name")
+    if name not in camps:
+        continue
+    sp, rv = f(r.get("spend")), f(r.get("d0_ad_revenue"))
+    im, cl = f(r.get("impressions")), f(r.get("clicks"))
+    ad = str(r.get("external_ad_id"))
+    label = admap.get(ad, (short(name) + ' ' + ' '.join((r.get('external_ad_name') or ad).split(' ')[-2:]), name))[0]
+    ads.append(dict(camp=name, ad=ad, label=label, sp=sp, rv=rv, ctr=cl / im * 100 if im else 0.0, ro=rv / sp if sp else 0.0))
+    g = agg.setdefault(name, [0.0, 0.0])
+    g[0] += sp
+    g[1] += rv
+tsp, trv = sum(v[0] for v in agg.values()), sum(v[1] for v in agg.values())
+cum = trv / tsp if tsp else 0.0
+
+# ---- direct TikTok hourly numbers: the truth when the synced numbers lag ----
+hourly = mcp.call("runtime_hourly_metrics", {"external_account_id": A})
+hsum = {}
+for r in hourly.get("rows") or []:
+    x = hsum.setdefault(int(r["hour"]), [0.0, 0.0, 0.0])
+    x[0] += f(r.get("spend")); x[1] += f(r.get("ad_revenue_d0")); x[2] += f(r.get("impressions"))
+h_total = sum(v[0] for v in hsum.values())
+
+alerts, H = [], []
+
+
+def once(key, sp, ro):
+    """Report the first time; again only if spend doubled or ROAS fell another 0.1."""
+    p = rep.get(key)
+    if p is None or sp >= 2 * p["sp"] or ro <= p["ro"] - 0.1:
+        rep[key] = {"sp": sp, "ro": ro}
+        return True
+    return False
+
+
+def line(x):
+    return f"{x['label']} [{x['ad']}] 消耗 {x['sp']:.2f} 回收 {x['rv']:.2f} ROAS {x['ro']:.2f} CTR {x['ctr']:.1f}%"
+
+
+# ---- 30-minute window from our own snapshots (same day only) ----
+snaps = []
+if os.path.exists(TL):
+    snaps = [json.loads(l) for l in open(TL) if l.strip()]
+snaps = [s for s in snaps if s["t"][:10] == today]
+older = [s for s in snaps if datetime.datetime.strptime(s["t"], "%Y-%m-%d %H:%M").replace(tzinfo=TZ) <= now - datetime.timedelta(minutes=30)]
+w30, ad_w30 = None, {}
+if older:
+    b = older[-1]
+    bs = sum(v[0] for v in b["camps"].values()); br = sum(v[1] for v in b["camps"].values())
+    if tsp - bs > 0.5:
+        w30 = (tsp - bs, trv - br, (trv - br) / (tsp - bs))
+    prev_ads = {x[1]: x for x in b.get("ads", [])}
+    for x in ads:
+        p = prev_ads.get(x["ad"])
+        ds = x["sp"] - (p[2] if p else 0); dr = x["rv"] - (p[3] if p else 0)
+        if ds > 0.5:
+            ad_w30[x["ad"]] = (ds, dr, dr / ds)
+
+# ---- rule: campaign below 0.8 (look inside, name the losing main ads by ROAS) ----
+by_camp = {}
+for x in ads:
+    by_camp.setdefault(x["camp"], []).append(x)
+for name, (sp, rv) in sorted(agg.items(), key=lambda kv: -kv[1][0]):
+    ro = rv / sp if sp else 0.0
+    cid = camps[name]["external_campaign_id"]
+    if sp >= 1 and ro < 0.8 and once(f"camp08|{cid}", sp, ro):
+        msg = f"计划低效预警  {short(name)}({cid}):今天消耗 {sp:.2f} 回收 {rv:.2f} ROAS {ro:.2f}"
+        small = sp < 5 or hours_in < 3
+        if small:
+            msg += "\n    样本小或仍在前 3 小时:多数会自己回来,先等"
+        for x in sorted(by_camp.get(name, []), key=lambda x: -x["sp"])[:4]:
+            if x["sp"] >= 1:
+                msg += f"\n    {line(x)}"
+        alerts.append(msg)
+
+# ---- rule: losing ad — judged on ROAS, not CTR (88794-9: CTR 20%+ everywhere, still losing) ----
+for x in ads:
+    w = ad_w30.get(x["ad"])
+    if x["sp"] >= 10 and x["ro"] < 0.8 and (w is None or w[2] < 0.8) and once(f"adloss|{x['ad']}", x["sp"], x["ro"]):
+        wtxt = f",最近 30 分钟 {w[0]:.2f}/{w[1]:.2f} = {w[2]:.2f}" if w else ""
+        alerts.append(f"亏损广告  {line(x)}{wtxt} → 建议关这条")
+
+# ---- rule: decay — a main ad that had CTR >= 25% falls below ----
+hi = st.setdefault("ctr_hi", {})
+for x in ads:
+    if x["sp"] >= 1 and x["ctr"] >= 25:
+        hi[x["ad"]] = True
+    elif hi.get(x["ad"]) and x["sp"] >= 1 and x["ctr"] < 25 and f"decay|{x['ad']}" not in rep:
+        rep[f"decay|{x['ad']}"] = {"sp": x["sp"], "ro": x["ro"]}
+        alerts.append(f"衰退观察  {line(x)}(之前 CTR ≥ 25%)")
+
+# ---- rule: earned then losing, two 10-minute windows in a row ----
+peak = st.setdefault("peak", {}); streak = st.setdefault("loss_streak", {})
+prev = st.get("snap")
+for name, (sp, rv) in agg.items():
+    if sp >= 1:
+        peak[name] = max(peak.get(name, 0.0), rv / sp)
+if prev and prev.get("fresh") != fresh_iso and prev.get("day") == today:
+    for name, (sp, rv) in agg.items():
+        psp, prv = prev["camps"].get(name, [0.0, 0.0])
+        ds, dr = sp - psp, rv - prv
+        losing = ds >= 1 and dr < ds
+        streak[name] = streak.get(name, 0) + 1 if losing else 0
+        if losing and peak.get(name, 0) > 1 and streak[name] >= 2 and f"slide|{name}|{now:%H}" not in rep:
+            rep[f"slide|{name}|{now:%H}"] = {"sp": sp, "ro": 0}
+            alerts.append(f"先赚后亏  {short(name)}({camps[name]['external_campaign_id']}):最近 10 分钟 {ds:.2f}/{dr:.2f},连续 {streak[name]} 个窗口亏;今天 {sp:.2f} ROAS {rv / sp:.2f},峰值 {peak[name]:.2f} → 建议关这个计划")
+st["snap"] = {"fresh": fresh_iso, "day": today, "camps": {n: v for n, v in agg.items()}}
+
+# ---- rule: budget 80% / 95% while profitable; say whether adding still makes sense ----
+for name, (sp, rv) in agg.items():
+    c = camps[name]; bud = f(c.get("budget")); ro = rv / sp if sp else 0.0
+    if ro > 1 and bud:
+        for pct in (95, 80):
+            if sp >= bud * pct / 100:
+                key = f"budget{pct}|{c['external_campaign_id']}|{bud}"
+                if key not in rep:
+                    rep[key] = {"sp": sp, "ro": ro}
+                    late = now.hour >= 17
+                    hint = "已过 17 点,手册建议不加" if late else ("ROAS > 1.3,可以加" if ro > 1.3 else "ROAS 不到 1.3,加不加看你")
+                    alerts.append(f"预算快用完  {short(name)}:今天 {sp:.0f}/{bud:.0f}({sp / bud * 100:.0f}%)ROAS {ro:.2f} → {hint}")
+                break
+
+# ---- judgement: first full hours after the account starts spending ----
+ramp = [h for h in sorted(hsum) if hsum[h][0] >= 20 and h < now.hour]
+if ramp and "judge" not in lc:
+    h0 = ramp[0]; r0 = hsum[h0][1] / hsum[h0][0]
+    verdict = "好户" if r0 >= 1.2 else ("观察" if r0 >= 0.9 else "差户信号")
+    lc["judge"] = {"hour": h0, "roas": round(r0, 2), "verdict": verdict}
+    event("判断", f"起量后第一个整点 {h0} 点:消耗 {hsum[h0][0]:.2f} ROAS {r0:.2f} → {verdict}")
+    alerts.append(f"起量判断  {h0} 点这一小时消耗 {hsum[h0][0]:.2f} ROAS {r0:.2f} → {verdict}(好户 ≥ 1.2,差户 < 0.9)")
+if len(ramp) >= 2 and "judge2" not in lc:
+    r1, r2 = (hsum[h][1] / hsum[h][0] for h in ramp[:2])
+    if r1 < 0.9 and r2 < 0.9:
+        lc["judge2"] = True
+        H.append(f"起量后连续两个整点 ROAS < 0.9({ramp[0]} 点 {r1:.2f}、{ramp[1]} 点 {r2:.2f}),按差户处理:只留最近 30 分钟 ROAS ≥ 1 的广告,其余按亏损关")
+
+# ---- platform status: every few rounds, and whenever spend looks stalled ----
+lc["round"] = lc.get("round", 0) + 1
+stalled_sync = lc.get("last_sp") is not None and tsp - lc["last_sp"] < 0.5 and lc["last_sp"] >= 5 and fresh != lc.get("last_fresh")
+h_stalled = lc.get("last_h") is not None and h_total - lc["last_h"] < 0.3 and lc["last_h"] >= 5
+status = {}
+if lc["round"] % a.sync_every == 1 or stalled_sync or h_stalled:
+    try:
+        sid = mcp.call("account_sync", {"external_account_id": A})["sync_run_id"]
+        for _ in range(40):
+            s_ = mcp.call("account_sync_status", {"sync_run_id": sid})
+            if s_["status"].startswith("completed") or s_["status"] in ("failed", "error"):
+                break
+            time.sleep(15)
+        for name, c in camps.items():
+            w = mcp.call("campaign_workspace", {"campaign_id": c["campaign_id"], "external_account_id": A})
+            g = w["groups"][0]
+            status[short(name)] = {"campaign": w.get("platform_status"), "group": g.get("platform_status"),
+                                   "ads": dict(Counter(ad.get("platform_status") for ad in g["ads"]))}
+    except Exception as e:
+        status = {"error": str(e)[:200]}
+    old = lc.get("status", {})
+    for n, v in status.items():
+        if n == "error":
+            continue
+        pv = old.get(n, {})
+        if (pv.get("campaign"), pv.get("group")) != (v["campaign"], v["group"]):
+            event("状态", f"{n}:计划 {pv.get('campaign')}→{v['campaign']},广告组 {pv.get('group')}→{v['group']}")
+        if pv.get("ads") != v["ads"]:
+            event("广告审核", f"{n}:广告 {pv.get('ads')}→{v['ads']}")
+    if status and "error" not in status:
+        lc["status"] = status
+
+# ---- health: when to call the operator to stop the account ----
+if "PUNISH" in json.dumps(status):
+    H.insert(0, "账户被 TikTok 处罚(ADVERTISER_ACCOUNT_PUNISH),继续开只会更糟")
+if status and "error" not in status and camps and all(v.get("campaign") != "CAMPAIGN_STATUS_ENABLE" for v in status.values()):
+    H.append("在投计划在平台上全部不是启用状态(不是本工具停的),先去后台看原因")
+if stalled_sync and h_stalled:
+    H.append(f"全户在投但消耗停住(同步数与 TikTok 小时数都不动,累计 {tsp:.2f})")
+if hours_in >= 4 and tsp >= 50 and cum < 0.90 and w30 and w30[2] < 0.90:
+    H.append(f"开投 {hours_in:.1f} 小时累计 ROAS {cum:.2f} < 0.90,最近 30 分钟 {w30[2]:.2f} 也 < 0.90")
+if w30 and w30[0] >= 30 and w30[2] < 0.80:
+    H.append(f"最近 30 分钟放量 {w30[0]:.2f} USD 但 ROAS 只有 {w30[2]:.2f} < 0.80")
+if (now.hour >= 18 or hours_in >= 8) and w30 and w30[0] >= 3 and w30[2] < 1.0:
+    H.append(f"进入收尾期(18 点后或开投满 8 小时),最近 30 分钟 ROAS {w30[2]:.2f} < 1")
+if tsp >= 50 and cum >= 1.0 and lc.get("was_below_1") and "breakeven" not in lc:
+    lc["breakeven"] = now.strftime("%H:%M")
+    H.append(f"累计 ROAS 回到 1 以上({cum:.2f}),可以考虑收尾锁住")
+if tsp >= 50 and cum < 1.0:
+    lc["was_below_1"] = True
+if H:
+    key = f"health|{now:%Y-%m-%d %H}|{len(H)}"
+    if key not in rep:
+        rep[key] = {"sp": tsp, "ro": cum}
+        msg = "‼ 建议暂停关户:" + ";".join(H) + f"。今天 {tsp:.2f}/{trv:.2f} ROAS {cum:.2f}。要关就告诉我"
+        alerts.insert(0, msg)
+        event("健康度", msg)
+
+# ---- milestones ----
+if tsp > 0 and not lc.get("had_spend"):
+    lc["had_spend"] = True
+    event("起量", f"首次出现消耗:{tsp:.2f} USD ROAS {cum:.2f}(数据截至 {fresh})")
+cp = lc.setdefault("camp", {})
+for name, (sp, rv) in agg.items():
+    k = short(name); ro = rv / sp if sp else 0.0; c = cp.setdefault(k, {})
+    if sp > 0 and not c.get("first"):
+        c["first"] = now.strftime("%H:%M"); event("起量", f"{k} 开始花钱:{sp:.2f} ROAS {ro:.2f}")
+    if sp >= 5:
+        if ro > c.get("peak", 0):
+            c["peak"], c["peak_t"] = ro, now.strftime("%H:%M")
+        if ro < 1 and not c.get("below1"):
+            c["below1"] = now.strftime("%H:%M"); event("变差", f"{k} ROAS 跌破 1:{sp:.2f} / {ro:.2f}(峰值 {c.get('peak', 0):.2f} @ {c.get('peak_t')})")
+        if ro >= 1 and c.get("below1") and not c.get("recovered"):
+            c["recovered"] = now.strftime("%H:%M"); event("回升", f"{k} ROAS 回到 1 以上:{ro:.2f}")
+if stalled_sync and h_stalled and not lc.get("stall_logged"):
+    lc["stall_logged"] = now.strftime("%H:%M"); event("停量", f"全户消耗停住(累计 {tsp:.2f})")
+elif not (stalled_sync and h_stalled):
+    lc.pop("stall_logged", None)
+for al in alerts:
+    if not al.startswith("‼"):
+        event("预警", al.split("\n")[0])
+lc["last_sp"], lc["last_fresh"], lc["last_h"] = tsp, fresh, h_total
+
+open(TL, "a").write(json.dumps({
+    "t": now.strftime("%Y-%m-%d %H:%M"), "fresh": fresh, "spend": round(tsp, 2), "rev": round(trv, 2),
+    "hourly_direct": {str(h): [round(v[0], 2), round(v[1], 2)] for h, v in sorted(hsum.items())},
+    "camps": {short(n): [round(v[0], 2), round(v[1], 2)] for n, v in agg.items()},
+    "ads": [[x["label"], x["ad"], round(x["sp"], 2), round(x["rv"], 2), round(x["ctr"], 1)] for x in ads],
+    "status": status or None}, ensure_ascii=False) + "\n")
+json.dump(st, open(SF, "w"), indent=1, ensure_ascii=False)
+
+# ---- same-lifecycle comparison against the benchmark accounts ----
+bench_line = ""
+try:
+    bench = json.load(open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "references", "benchmarks.json")))
+    if ramp:
+        k = now.hour - ramp[0]  # hours since this account ramped
+        parts = []
+        for b in bench["accounts"]:
+            hs = sorted((int(h), v) for h, v in b["hourly"].items())
+            rh = [h for h, v in hs if v[0] >= 20]
+            if not rh:
+                continue
+            upto = [v for h, v in hs if h <= rh[0] + k]
+            s_ = sum(v[0] for v in upto); r_ = sum(v[1] for v in upto)
+            if s_:
+                parts.append(f"{b['short']} {r_ / s_:.2f}")
+        mine = sum(v[1] for h, v in hsum.items() if h <= now.hour) / max(h_total, 0.01)
+        bench_line = f";同阶段(起量后第 {k + 1} 小时)累计 ROAS:本户 {mine:.2f} vs " + "、".join(parts)
+except (OSError, ValueError, KeyError):
+    pass
+
+print("\n".join(alerts) if alerts else "一轮无异常")
+top = sorted(agg.items(), key=lambda kv: -kv[1][0])[:3]
+stage = "等量期" if not lc.get("had_spend") and hours_in < 2 else ("判断期" if hours_in < 3 else ("放量期" if now.hour < 18 and hours_in < 8 else "收尾期"))
+w30t = f";最近 30 分钟 {w30[0]:.2f}/{w30[1]:.2f} ROAS {w30[2]:.2f}" if w30 else ""
+lag = f";按天数据落后,TikTok 小时数合计 {h_total:.2f}" if h_total - tsp > 2 else ""
+print(f"[{stage} {hours_in:.1f}h|{len(camps)} 个在投计划 今天 {tsp:.2f}/{trv:.2f} ROAS {cum:.2f}{w30t};"
+      + "、".join(f"{short(n)} {v[0]:.2f}/{(v[1] / v[0] if v[0] else 0):.2f}" for n, v in top)
+      + f";数据截至 {fresh},现在 {now:%H:%M}{lag}{bench_line}]")