"""phx_analyze.py — turn scoring runs + live fetches into publishable stats. Inputs (whichever exist): scores.npz (post_ids, author_ids, probs [n,19]) age_sweep.npz (ages_h, probs [n_ages, n, 19], post_ids) fetched.jsonl (live public data per post id) Output: lab_stats.json — everything the /lab page needs, with honest labeling: only 6 of 19 head indices are named in the release (run_pipeline.py comments); the rest are published as anonymous distributions. """ from __future__ import annotations import json import math import os import sys import numpy as np OUT = sys.argv[1] if len(sys.argv) > 1 else "lab_stats.json" NAMED = {1: "favorite", 4: "reply", 5: "quote", 6: "repost", 11: "dwell", 13: "video_quality_view"} result = {"named_head_indices": {str(k): v for k, v in NAMED.items()}, "checkpoint": "oss-phoenix-artifacts (mini Phoenix, frozen)", "corpus": "sports_corpus.npz (~537K real posts, 6h window, Sports topic)"} def spearman(a, b): ra = np.argsort(np.argsort(a)).astype(np.float64) rb = np.argsort(np.argsort(b)).astype(np.float64) ra -= ra.mean(); rb -= rb.mean() d = math.sqrt((ra @ ra) * (rb @ rb)) return float(ra @ rb / d) if d else 0.0 if os.path.exists("scores.npz"): z = np.load("scores.npz") probs = z["probs"] n, k = probs.shape heads = {} for i in range(k): col = probs[:, i] heads[str(i)] = { "name": NAMED.get(i, f"unnamed_{i}"), "mean": round(float(col.mean()), 5), "median": round(float(np.median(col)), 5), "p90": round(float(np.percentile(col, 90)), 5), "p99": round(float(np.percentile(col, 99)), 5), } result["head_distributions"] = heads result["n_scored"] = int(n) # implied relative rarity: mean P(action) ratios vs favorite fav = probs[:, 1].mean() result["implied_ratios_vs_favorite"] = { NAMED[i]: round(float(probs[:, i].mean() / fav), 4) for i in NAMED if fav > 0} # head correlation (spearman) among named heads corr = {} named_idx = list(NAMED) for a in named_idx: for b in named_idx: if a < b: corr[f"{NAMED[a]}~{NAMED[b]}"] = round(spearman(probs[:, a], probs[:, b]), 3) result["head_rank_correlations"] = corr if os.path.exists("age_sweep.npz"): z = np.load("age_sweep.npz") ages = z["ages_h"]; sw = z["probs"] # [A, n, 19] curves = {} for i, name in NAMED.items(): m = sw[:, :, i].mean(axis=1) base = m[0] if m[0] else 1.0 curves[name] = { "ages_h": [float(a) for a in ages], "mean_prob": [round(float(x), 5) for x in m], "relative_to_first": [round(float(x / base), 4) for x in m], } result["age_curves"] = curves if os.path.exists("fetched.jsonl"): rows = [] with open("fetched.jsonl", encoding="utf-8") as f: for line in f: try: r = json.loads(line) if r.get("status") == "ok": rows.append(r) except Exception: pass result["n_fetched_live"] = len(rows) if rows and os.path.exists("scores.npz"): z = np.load("scores.npz") probs = z["probs"]; pids = z["post_ids"].astype(str) idx = {p: i for i, p in enumerate(pids)} joined = [(r, idx[r["id"]]) for r in rows if r["id"] in idx] result["n_joined"] = len(joined) if len(joined) >= 50: favs = np.array([r["favs"] for r, _ in joined], dtype=np.float64) reps = np.array([r["replies"] for r, _ in joined], dtype=np.float64) pf = np.array([probs[i, 1] for _, i in joined]) pr = np.array([probs[i, 4] for _, i in joined]) result["backtest"] = { "spearman_Pfav_vs_actual_likes": round(spearman(pf, favs), 3), "spearman_Preply_vs_actual_replies": round(spearman(pr, reps), 3), "spearman_Pfav_vs_actual_replies": round(spearman(pf, reps), 3), "likes_by_Pfav_decile": [ round(float(np.median(favs[(pf >= np.quantile(pf, q / 10)) & (pf < np.quantile(pf, (q + 1) / 10) + (1e9 if q == 9 else 0))])), 1) for q in range(10)], } # checker-feature validation on ACTUAL engagement def eng(mask): m = mask.astype(bool) if m.sum() < 20 or (~m).sum() < 20: return None return { "n_with": int(m.sum()), "median_likes_with": float(np.median(favs[m])), "median_likes_without": float(np.median(favs[~m])), "median_replies_with": float(np.median(reps[m])), "median_replies_without": float(np.median(reps[~m])), "mean_Pfav_with": round(float(pf[m].mean()), 5), "mean_Pfav_without": round(float(pf[~m].mean()), 5), } slop = json.load(open(os.environ.get("CHECKS_JSON", "checks.json"), encoding="utf-8")) if os.path.exists( os.environ.get("CHECKS_JSON", "checks.json")) else None texts = [r["text"].lower() for r, _ in joined] feats = { "has_photo": np.array([r["has_photo"] for r, _ in joined]), "has_video": np.array([r["has_video"] for r, _ in joined]), "has_link": np.array([r["n_urls"] > 0 for r, _ in joined]), "hashtags_3plus": np.array([r["n_hashtags"] >= 3 for r, _ in joined]), "is_reply": np.array([bool(r.get("is_reply")) for r, _ in joined]), "verified_author": np.array([r["verified"] for r, _ in joined]), "question_mark": np.array(["?" in t for t in texts]), } if slop: sp = [p.lower() for p in slop["slop_patterns"]] bp = [p.lower() for p in slop["bait_patterns"]] feats["slop_pattern"] = np.array([any(x in t for x in sp) for t in texts]) feats["bait_pattern"] = np.array([any(x in t for x in bp) for t in texts]) result["feature_effects"] = {k: eng(v) for k, v in feats.items()} if os.path.exists("fetched_authors.jsonl") and os.path.exists("scores_authors.npz"): za = np.load("scores_authors.npz") aprobs = za["probs"]; apids = za["post_ids"].astype(str) aauth = za["author_ids"].astype(str) aidx = {p: i for i, p in enumerate(apids)} rows = [] with open("fetched_authors.jsonl", encoding="utf-8") as f: for line in f: try: r = json.loads(line) if r.get("status") == "ok" and r["id"] in aidx: rows.append((r, aidx[r["id"]])) except Exception: pass byauth = {} for r, i in rows: byauth.setdefault(aauth[i], []).append((r, i)) per_author_fav, per_author_reply = [], [] n_authors_used = 0 for a, items in byauth.items(): if len(items) < 6: continue favs = np.array([r["favs"] for r, _ in items], dtype=np.float64) reps = np.array([r["replies"] for r, _ in items], dtype=np.float64) pf = np.array([aprobs[i, 1] for _, i in items]) pr = np.array([aprobs[i, 4] for _, i in items]) if np.ptp(favs) > 0 and np.ptp(pf) > 0: per_author_fav.append(spearman(pf, favs)) if np.ptp(reps) > 0 and np.ptp(pr) > 0: per_author_reply.append(spearman(pr, reps)) n_authors_used += 1 if per_author_fav: pa = np.array(per_author_fav) result["within_author_backtest"] = { "n_authors": n_authors_used, "n_posts": len(rows), "note": "audience size held constant: rank correlation of model P(action) vs actual counts WITHIN each prolific author's posts (>=6 live posts each)", "fav_spearman_median": round(float(np.median(pa)), 3), "fav_spearman_mean": round(float(pa.mean()), 3), "fav_frac_positive": round(float((pa > 0).mean()), 3), "fav_iqr": [round(float(np.percentile(pa, 25)), 3), round(float(np.percentile(pa, 75)), 3)], "reply_spearman_median": round(float(np.median(per_author_reply)), 3) if per_author_reply else None, "reply_frac_positive": round(float((np.array(per_author_reply) > 0).mean()), 3) if per_author_reply else None, } with open(OUT, "w", encoding="utf-8") as f: json.dump(result, f, indent=1) print("[done]", OUT) for k in result: print(" -", k)