#!/usr/bin/env python3 """Study #7 analysis (PREREG v3). This file IS the definition of every measure and gate. Hashed in step0/LOCK.log before any trend is computed. Any change = logged amendment first. python3 -I analyze_trend.py sample # completeness + STOP check, draw the seeded precision samples python3 -I analyze_trend.py gates # needs work/class_*.json (classified samples) + step0/d1_labs.json All dates UTC. Seed 20261012. """ import glob, json, os, random, re, sys from collections import defaultdict HERE = os.path.dirname(os.path.abspath(__file__)) RAW, WORK, STEP0 = os.path.join(HERE, "raw"), os.path.join(HERE, "work"), os.path.join(HERE, "step0") SEED = 20261012 B = 10000 # ---- copied verbatim from study #6 social-media-mcp-census/collect.py (PLATFORMS_A3, GENERIC_A3, X_FALSE_STEMS; X rule = its L197) ---- PLATFORMS_A3 = { "x": r"twitter|\btweets?\b|\bx\.com\b|\bx api\b|\bx-mcp\b|\bxmcp\b|\bmcp-x\b|\bx \(formerly twitter\)|\bx/twitter\b|\bx \(twitter\)|\bxurl\b|\bx search\b|\bsearch x\b|\bx posts?\b", "linkedin": r"linkedin", "facebook": r"facebook|\bfb\b|\bfb page|\bmeta graph", "instagram": r"instagram|\big\b|\binsta\b|\big graph", "threads": r"threads\.net|threads api|meta threads|threads by meta|\bthreads\b", "tiktok": r"tiktok|tik tok", "youtube": r"youtube|\byt\b", "pinterest": r"pinterest", "bluesky": r"bluesky|\bbsky\b|atproto|at protocol", "reddit": r"reddit|subreddit", } GENERIC_A3 = re.compile(r"social[ -_]?media|social network|\bsocials\b|\bsocial[ -](posts?|content|listening|channels?|scheduler|scheduling|accounts?|profiles?|apis?|inbox|analytics|data|presence|marketing|management|feeds?|networks?|platforms?|publishing|posting|intelligence|monitoring|signals?|engagement|automation)\b|cross-?post", re.I) X_FALSE_STEMS = re.compile(r"x64dbg|xcode|xero|xmind|xmpp|xlsx|xml|xray|xterm|xbox", re.I) MCP_TEXT = re.compile(r"mcp", re.I) # D's text rule; N must also meet it (N subset of D) P1 = ["2025-10", "2025-11", "2025-12", "2026-01", "2026-02", "2026-03"] P2 = ["2026-04", "2026-05", "2026-06", "2026-07", "2026-08", "2026-09"] QUARTERS = {"2025Q1": ["2025-01", "2025-02", "2025-03"], "2025Q2": ["2025-04", "2025-05", "2025-06"], "2025Q3": ["2025-07", "2025-08", "2025-09"], "2025Q4": P1[:3], "2026Q1": P1[3:], "2026Q2": P2[:3], "2026Q3": P2[3:]} D1_SET = ["social media mcp", "social media mcp server", "mcp social media", "twitter mcp", "x mcp server", "facebook mcp", "instagram mcp", "instagram mcp server", "threads mcp", "tiktok mcp", "pinterest mcp", "bluesky mcp", "reddit mcp", "post to instagram with claude", "post to linkedin with claude", "claude post to twitter", "chatgpt post to instagram", "chatgpt post to linkedin", "ai agent social media posting", "automate social media with ai agent", "social media api for ai agents"] def text_of(it): return " ".join([it.get("name") or "", it.get("description") or "", " ".join(it.get("topics") or [])]) def classify_match(it): t = text_of(it) if not MCP_TEXT.search(t): return None plats = sorted(k for k, rx in PLATFORMS_A3.items() if re.search(rx, t, re.I)) # study #6 collect.py L197: false X stem in the NAME, no twitter/tweet anywhere -> drop the x match only if "x" in plats and X_FALSE_STEMS.search(it.get("name") or "") and not re.search(r"twitter|tweet", t, re.I): plats = [p for p in plats if p != "x"] generic = bool(GENERIC_A3.search(t)) if not plats and not generic: return None stratum = plats[0] if len(plats) == 1 else ("multi" if plats else "generic") only = plats[0] if (len(plats) == 1 and not generic) else ("generic" if (not plats and generic) else None) return {"plats": plats, "generic": generic, "stratum": stratum, "only": only} def tol(total): return max(2, 0.005 * total) def load_cells(step): """Returns (repos by id, incomplete-cell totals by month). A recheck file overrides its cell.""" repos, inc = {}, defaultdict(int) for fn in sorted(glob.glob(os.path.join(RAW, step, "*.json"))): c = json.load(open(fn)) rc = os.path.join(RAW, step, "recheck", os.path.basename(fn)) if os.path.exists(rc): c = json.load(open(rc)) month = os.path.basename(fn)[:7] total = c.get("total_count") if c.get("error") or total is None: inc[month] += 10 ** 9 # unknown size: treated as blocking until rechecked continue if c["n_ids"] < total - tol(total): inc[month] += total - c["n_ids"] # A4 (2026-10-13): count the missing repos, not the whole cell total for it in c.get("items", []): repos[it["id"]] = it return repos, inc def keep_n(repos): out = {} for rid, it in repos.items(): m = classify_match(it) if m: out[rid] = dict(it, **m, month=it["created_at"][:7]) return out def keep_d(repos): return {rid: dict(it, month=it["created_at"][:7]) for rid, it in repos.items() if MCP_TEXT.search(text_of(it))} def reps(repos, drop=frozenset()): """Owner cap: one representative per (owner, month) = earliest-created matching repo (tie: lowest id).""" best = {} for rid, it in repos.items(): if rid in drop: continue k = (it.get("owner") or it["full_name"].split("/")[0], it["month"]) if k not in best or (it["created_at"], rid) < (best[k]["created_at"], best[k]["id"]): best[k] = it return best def capped(repos, drop=frozenset()): c = defaultdict(int) for (_, month) in reps(repos, drop): c[month] += 1 return c def ssum(c, months): return sum(c.get(m, 0) for m in months) def stop_check(inc, raw_by_month, periods): bad = [] for name, months in periods.items(): it = sum(inc.get(m, 0) for m in months) rw = sum(raw_by_month.get(m, 0) for m in months) if it > 0.01 * max(rw, 1): bad.append((name, it, rw)) return bad def raw_by_month(repos): c = defaultdict(int) for it in repos.values(): c[it["month"]] += 1 return c def alloc(strata_counts, n): total = sum(strata_counts.values()) if total <= n: return dict(strata_counts) q = {k: v * n / total for k, v in strata_counts.items()} a = {k: int(v) for k, v in q.items()} for k in sorted(q, key=lambda k: (-(q[k] - a[k]), k))[: n - sum(a.values())]: a[k] += 1 return a def slim(it): return {k: it.get(k) for k in ("id", "full_name", "description", "topics", "created_at", "stratum")} def step_sample(): nr, ninc = load_cells("n") dr, dinc = load_cells("d") N, D = keep_n(nr), keep_d(dr) stopN = stop_check(ninc, raw_by_month(N), QUARTERS) stopD = stop_check(dinc, raw_by_month(D), {"P1": P1, "P2": P2}) json.dump({"stopN": stopN, "stopD": stopD}, open(os.path.join(WORK, "stop_check.json"), "w"), indent=1) if stopN or stopD: print("STOP RULE TRIGGERED (run recheck first, then re-run):", stopN, stopD) return rnd = random.Random(SEED) rn, rd = reps(N), reps(D) for per, months in (("P1", P1), ("P2", P2)): pool = sorted((it for (o, m), it in rn.items() if m in months), key=lambda x: x["id"]) by = defaultdict(list) for it in pool: by[it["stratum"]].append(it) a = alloc({k: len(v) for k, v in by.items()}, 150) samp = [] for k in sorted(a): samp += rnd.sample(by[k], a[k]) json.dump([slim(x) for x in samp], open(os.path.join(WORK, f"sample_N_{per}.json"), "w"), indent=1) dpool = sorted((it for (o, m), it in rd.items() if m in months), key=lambda x: x["id"]) dsamp = rnd.sample(dpool, min(100, len(dpool))) json.dump([slim(x) for x in dsamp], open(os.path.join(WORK, f"sample_D_{per}.json"), "w"), indent=1) print(per, "N sample", len(samp), "D sample", len(dsamp)) def load_class(name, field): s = json.load(open(os.path.join(WORK, f"sample_{name}.json"))) c = json.load(open(os.path.join(WORK, f"class_{name}.json"))) return [(x, bool(c[x["full_name"]][field])) for x in s] def q(xs, p): xs = sorted(xs) return xs[max(0, min(len(xs) - 1, int(p * len(xs))))] def boot(n1, n2, d1, d2, seed=SEED): """Joint bootstrap. Returns (lower 2.5% of R_p, lower 2.5% of R_p/R_d, point R_p, point R_d).""" rnd = random.Random(seed) mean = lambda v: sum(v) / len(v) if v else 0.0 rp, rr = [], [] for _ in range(B): a = mean([rnd.choice(n1) for _ in n1]); b = mean([rnd.choice(n2) for _ in n2]) c = mean([rnd.choice(d1) for _ in d1]); d = mean([rnd.choice(d2) for _ in d2]) r_p = b / a if a else 0.0 r_d = d / c if c else float("inf") rp.append(r_p); rr.append(r_p / r_d if r_d not in (0.0, float("inf")) else 0.0) pt_p = mean(n2) / mean(n1) if mean(n1) else 0.0 pt_d = mean(d2) / mean(d1) if mean(d1) else 0.0 return q(rp, 0.025), q(rr, 0.025), pt_p, pt_d def core_gates(Nc, Dc, lb_p, lb_ratio, NcS=None): """Nc: capped N (G0-G2). NcS: capped N restricted to repos present in collected D (share numerator, G4).""" NcS = Nc if NcS is None else NcS g = {} n1, n2 = ssum(Nc, P1), ssum(Nc, P2) g["G0"] = {"pass": n1 >= 30, "capN_P1": n1} g["G1"] = {"raw_ratio": n2 / n1 if n1 else None, "lb_Rp": lb_p, "adj": (n2 / n1) * lb_p if n1 else 0, "pass": bool(n1) and (n2 / n1) * lb_p >= 1.25} qs = {k: ssum(Nc, v) for k, v in QUARTERS.items()} peak = max(qs[k] for k in ["2025Q1", "2025Q2", "2025Q3", "2025Q4", "2026Q1", "2026Q2"]) g["G2"] = {"quarters": qs, "pass": qs["2026Q3"] >= qs["2026Q2"] and qs["2026Q3"] >= 0.8 * peak} S = lambda months: ssum(NcS, months) / ssum(Dc, months) if ssum(Dc, months) else 0.0 s1, s2, sq2, sq3 = S(P1), S(P2), S(QUARTERS["2026Q2"]), S(QUARTERS["2026Q3"]) g["G4"] = {"S_P1": s1, "S_P2": s2, "raw_ratio": s2 / s1 if s1 else None, "lb_ratio": lb_ratio, "adj": (s2 / s1) * lb_ratio if s1 else 0, "S_Q2": sq2, "S_Q3": sq3, "pass": bool(s1) and (s2 / s1) * lb_ratio >= 1.10 and sq3 >= 0.9 * sq2} return g def d1_gate(): fn = os.path.join(STEP0, "d1_labs.json") if not os.path.exists(fn): return {"status": "missing"} rows = json.load(open(fn)) kws = {} for r in rows: k = (r.get("keyword") or "").lower() ms = r.get("monthly_searches") or [] if k in D1_SET and len(ms) == 12 and all(m.get("search_volume") is not None for m in ms): ms = sorted(ms, key=lambda m: (m["year"], m["month"])) kws[k] = [m["search_volume"] for m in ms] if not kws: return {"status": "insufficient", "kept": 0} largest = max(kws, key=lambda k: (sum(kws[k]), k)) rest = {k: v for k, v in kws.items() if k != largest} prior = sum(sum(v[:6]) for v in rest.values()); latest = sum(sum(v[6:]) for v in rest.values()) out = {"kept": sorted(kws), "largest_removed": largest, "prior6": prior, "latest6": latest} if len(rest) < 4 or prior < 300: out["status"] = "insufficient"; return out r = latest / prior out.update(ratio=r, status="pass" if r >= 1.10 else ("block" if r < 0.80 else "neutral")) return out def d2_gate(): rx = lambda t: MCP_TEXT.search(t) and (any(re.search(v, t, re.I) for v in PLATFORMS_A3.values()) or GENERIC_A3.search(t)) tot = {} for fn in glob.glob(os.path.join(RAW, "npm", "meta", "*.json")): m = json.load(open(fn)) created = ((m.get("time") or {}).get("created") or "") t = " ".join([m.get("name") or "", m.get("description") or "", " ".join(m.get("keywords") or []) if isinstance(m.get("keywords"), list) else ""]) if not m.get("name") or not created or created >= "2025-10-01" or not rx(t): continue dl = os.path.join(RAW, "npm", "dl", os.path.basename(fn)) if not os.path.exists(dl): continue days = ((json.load(open(dl)).get("resp") or {}).get("downloads") or []) a = sum(x["downloads"] for x in days if "2025-10-01" <= x["day"] <= "2026-03-31") b = sum(x["downloads"] for x in days if "2026-04-01" <= x["day"] <= "2026-09-30") tot[m["name"]] = (a, b) if not tot: return {"status": "empty"} largest = max(tot, key=lambda k: (sum(tot[k]), k)) a = sum(v[0] for k, v in tot.items() if k != largest); b = sum(v[1] for k, v in tot.items() if k != largest) r = b / a if a else 0.0 return {"cohort": len(tot), "largest_removed": largest, "P1": a, "P2": b, "ratio": r, "status": "pass" if r >= 1.25 else "fail"} def step_gates(): nr, _ = load_cells("n"); dr, _ = load_cells("d") N, D = keep_n(nr), keep_d(dr) Nc, Dc = capped(N), capped(D) NS = {rid: it for rid, it in N.items() if rid in D} # A1: share numerator only counts repos present in collected D NcS = capped(NS) sN1, sN2 = load_class("N_P1", "server"), load_class("N_P2", "server") sD1, sD2 = load_class("D_P1", "mcp_code"), load_class("D_P2", "mcp_code") v = lambda s: [int(b) for _, b in s] lb_p, lb_ratio, pt_p, pt_d = boot(v(sN1), v(sN2), v(sD1), v(sD2)) G = core_gates(Nc, Dc, lb_p, lb_ratio, NcS) G["N_not_in_D"] = {m: sum(1 for it in N.values() if it["month"] == m and it["id"] not in D) for m in P1 + P2} G["precision"] = {"p1": sum(v(sN1)) / len(sN1), "p2": sum(v(sN2)) / len(sN2), "d1": sum(v(sD1)) / len(sD1), "d2": sum(v(sD2)) / len(sD2), "R_p": pt_p, "R_d": pt_d} d1, d2 = d1_gate(), d2_gate() G["G3"] = {"D1": d1, "D2": d2, "pass": (d1.get("status") == "pass" or d2.get("status") == "pass") and d1.get("status") != "block"} # G5a: largest stratum = the "only-match" stratum whose removal lowers capped N in P2 the most cands = list(PLATFORMS_A3) + ["generic"] drops = {k: frozenset(rid for rid, it in N.items() if it["only"] == k) for k in cands} base2 = ssum(Nc, P2) largest = max(cands, key=lambda k: (base2 - ssum(capped(N, drops[k]), P2), k)) Nc_a = capped(N, drops[largest]) keep = lambda s: [int(b) for x, b in s if x["id"] not in drops[largest]] lb_p_a, lb_ratio_a, _, _ = boot(keep(sN1), keep(sN2), v(sD1), v(sD2)) ga = core_gates(Nc_a, Dc, lb_p_a, lb_ratio_a, capped(NS, drops[largest])) # G5b: drop the largest P2 month mN = max(P2, key=lambda m: (Nc.get(m, 0), m)) g1b = (ssum(Nc, [m for m in P2 if m != mN]) / 5) / (ssum(Nc, P1) / 6) * lb_p if ssum(Nc, P1) else 0 Sm = lambda m: NcS.get(m, 0) / Dc[m] if Dc.get(m) else 0.0 mS = max(P2, key=lambda m: (Sm(m), m)) rest = [m for m in P2 if m != mS] s1 = ssum(NcS, P1) / ssum(Dc, P1) if ssum(Dc, P1) else 0 s2b = ssum(NcS, rest) / ssum(Dc, rest) if ssum(Dc, rest) else 0 g4b = (s2b / s1) * lb_ratio if s1 else 0 G["G5"] = {"largest_stratum": largest, "a_G1": ga["G1"]["pass"], "a_G4": ga["G4"]["pass"], "a_detail": {"G1": ga["G1"], "G4": ga["G4"]}, "b_drop_month_G1": mN, "b_G1_adj": g1b, "b_drop_month_G4": mS, "b_G4_adj": g4b, "pass": ga["G1"]["pass"] and ga["G4"]["pass"] and g1b >= 1.25 and g4b >= 1.10} pub = lambda s, cl: [int(bool(cl[x["full_name"]].get("publishes"))) for x, b in s if b] c1 = json.load(open(os.path.join(WORK, "class_N_P1.json"))); c2 = json.load(open(os.path.join(WORK, "class_N_P2.json"))) a, b = pub(sN1, c1), pub(sN2, c2) rnd = random.Random(SEED); diffs = [] for _ in range(B): if a and b: diffs.append(sum(rnd.choice(b) for _ in b) / len(b) - sum(rnd.choice(a) for _ in a) / len(a)) G["publish_share"] = {"P1": sum(a) / len(a) if a else None, "P2": sum(b) / len(b) if b else None, "n1": len(a), "n2": len(b), "diff_ci95": [q(diffs, 0.025), q(diffs, 0.975)] if diffs else None} G["monthly"] = {"capN": dict(sorted(Nc.items())), "capN_inD": dict(sorted(NcS.items())), "capD": dict(sorted(Dc.items())), "rawN": dict(sorted(raw_by_month(N).items())), "rawD": dict(sorted(raw_by_month(D).items()))} G["PASS"] = all(G[k]["pass"] for k in ("G0", "G1", "G2", "G3", "G4", "G5")) json.dump(G, open(os.path.join(WORK, "gates.json"), "w"), indent=1) for k in ("G0", "G1", "G2", "G3", "G4", "G5"): print(k, "PASS" if G[k]["pass"] else "FAIL") print("OVERALL", "PASS" if G["PASS"] else "SKIP") if __name__ == "__main__": {"sample": step_sample, "gates": step_gates}[sys.argv[1]]()