"""One watch round for a freshly launched Minis account. Prints the alerts, then one status line. python3 watch.py --account --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()}} # ---- lifecycle clock: stages follow this account's own ramp, not the wall clock (operator 2026-09-30) ---- ramp = [h for h in sorted(hsum) if hsum[h][0] >= 20 and h < now.hour] since_ramp = (now.hour + now.minute / 60 - ramp[0]) if ramp else None # hours since the first >= 20 USD hour began # ---- 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 = since_ramp is not None and since_ramp >= 5 hint = "起量已满 5 小时,手册建议不加" 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 ---- 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 ((since_ramp is not None and since_ramp >= 6) or hours_in >= 8) and w30 and w30[0] >= 3 and w30[2] < 1.0: H.append(f"进入收尾期(起量后满 6 小时或开投满 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 ramp else ("判断期" if since_ramp < 3 else ("放量期" if since_ramp < 6 and hours_in < 8 else "收尾期"))) if 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}]")