#!/usr/bin/env python3 """ CobaltKite Research 001 (CKR-001): analysis script. Rebuilds mentions.csv and summary.csv from runs.csv, and prints every number used in the study's findings. python analysis.py # reads ./runs.csv, writes ./mentions.csv and ./summary.csv python analysis.py --data DIR # read/write in DIR Requires: Python 3.9+, pandas, numpy. Fixed random seed, so output is identical on every run. Definitions - Mentioned: the brand's name appears anywhere in the visible answer text (detected by the alias table below; URLs are stripped first so a link address never counts as a mention). - Listed: the brand appears in the human-coded recommendation list for that run (column coded_list_raw in runs.csv), including "also consider" asides. - List rank: the brand's position in that coded list, starting at 1. - Top 3: list rank of 1, 2 or 3. """ import argparse import re from pathlib import Path import numpy as np import pandas as pd SEED = 20261014 N_BOOT = 10_000 Z = 1.959964 # 95% # ---------------------------------------------------------------- brand tables # Text detection: case-sensitive, whole-word. Only HR / payroll / PEO / EOR vendors # count as brands. Integrations (Slack, Google Workspace, Okta...) and review sites # (G2, Capterra, Reddit...) are deliberately excluded. TEXT_ALIASES = { "Rippling": [r"Rippling"], "Gusto": [r"Gusto"], "BambooHR": [r"BambooHR"], "Justworks": [r"Justworks"], "Deel": [r"Deel"], "HiBob": [r"HiBob"], "TriNet": [r"TriNet"], "MambaHR": [r"MambaHR"], # "remote" is an ordinary word, so Remote (the vendor) is only matched in brand-like forms. "Remote": [r"Remote\.com", r"(?<=Deel or )Remote", r"Remote(?= or Deel)"], "ADP": [r"ADP"], "Workday": [r"Workday"], "UKG": [r"UKG"], "Paychex": [r"Paychex"], "Paylocity": [r"Paylocity"], "Paycom": [r"Paycom"], "Paycor": [r"Paycor"], "Lattice": [r"Lattice"], "QuickBooks": [r"QuickBooks"], "Namely": [r"Namely"], "Insperity": [r"Insperity"], "Zenefits": [r"Zenefits"], "Personio": [r"Personio"], "Culture Amp": [r"Culture Amp"], "Oyster": [r"Oyster HR", r"Oyster(?= \()"], "OnPay": [r"OnPay"], } _TEXT_RX = {b: re.compile("|".join(rf"\b{a}\b" for a in al)) for b, al in TEXT_ALIASES.items()} # Coded-list normalisation: product/plan names collapse to the vendor. LIST_CANON = { "Gusto Plus": "Gusto", "Gusto Plus / Premium": "Gusto", "BambooHR Core": "BambooHR", "Deel HR Core": "Deel", "Deel HR": "Deel", "HiBob (Bob)": "HiBob", "Justworks (PEO)": "Justworks", "Justworks Payroll": "Justworks", "Justworks PEO": "Justworks", } LIST_SPLIT = {"Deel or Remote": ["Deel", "Remote"]} # one list slot naming two vendors: both get that rank MD_LINK = re.compile(r"\[([^\]]*)\]\((?:https?://|www\.)[^)]*\)") URL = re.compile(r"https?://\S+|www\.\S+") def visible_text(t: str, drop_link_labels: bool = False) -> str: """Answer text with URLs removed. Optionally also drop citation-chip labels.""" t = MD_LINK.sub("" if drop_link_labels else r"\1", t) return URL.sub("", t) def detect_mentions(text: str, drop_link_labels: bool = False) -> dict: """{brand: char offset of first mention} for every brand named in the answer.""" vt = visible_text(text, drop_link_labels) out = {} for b, rx in _TEXT_RX.items(): m = rx.search(vt) if m: out[b] = m.start() return out def parse_coded_list(raw: str) -> dict: """{brand: rank} from a coded list like '#1 Rippling | #2 Gusto Plus | ...' (first slot wins).""" ranks = {} for slot in str(raw).split("|"): slot = slot.strip() m = re.match(r"#(\d+)\s+(.*)$", slot) if not m: continue rank, name = int(m.group(1)), m.group(2).strip() names = LIST_SPLIT.get(name, [LIST_CANON.get(name, name)]) for n in names: ranks.setdefault(n, rank) return ranks # ---------------------------------------------------------------- statistics def wilson(k: int, n: int): if n == 0: return (np.nan, np.nan, np.nan) p = k / n den = 1 + Z**2 / n centre = (p + Z**2 / (2 * n)) / den half = Z * np.sqrt(p * (1 - p) / n + Z**2 / (4 * n**2)) / den return p, max(0.0, centre - half), min(1.0, centre + half) def boot_ci(df: pd.DataFrame, stat, strata=("platform", "prompt_id"), n_boot=N_BOOT, seed=SEED): """Percentile CI, resampling runs with replacement within each platform x prompt cell.""" rng = np.random.default_rng(seed) groups = [g.index.to_numpy() for _, g in df.groupby(list(strata))] vals = np.empty(n_boot) for i in range(n_boot): idx = np.concatenate([rng.choice(g, size=len(g), replace=True) for g in groups]) vals[i] = stat(df.loc[idx]) return float(np.percentile(vals, 2.5)), float(np.percentile(vals, 97.5)) # ---------------------------------------------------------------- build tables def build_mentions(runs: pd.DataFrame) -> pd.DataFrame: rows = [] for _, r in runs.iterrows(): txt = detect_mentions(r.answer_text) txt_strict = detect_mentions(r.answer_text, drop_link_labels=True) listed = parse_coded_list(r.coded_list_raw) order = {b: i + 1 for i, (b, _) in enumerate(sorted(txt.items(), key=lambda kv: kv[1]))} for b in sorted(set(txt) | set(listed)): rank = listed.get(b) rows.append({ "run_id": r.run_id, "platform": r.platform, "prompt_id": r.prompt_id, "brand": b, "mentioned": int(b in txt), "mentioned_outside_citation_links": int(b in txt_strict), "listed": int(rank is not None), "position": rank if rank is not None else "", "in_top3": int(rank is not None and rank <= 3), "first_mention_order": order.get(b, ""), }) return pd.DataFrame(rows) def wide(runs, mentions, col): """runs x brands 0/1 matrix for a mentions column.""" m = mentions.pivot_table(index="run_id", columns="brand", values=col, aggfunc="max", fill_value=0) return m.reindex(runs.run_id).fillna(0).astype(int) def main(): ap = argparse.ArgumentParser() ap.add_argument("--data", default=".") a = ap.parse_args() d = Path(a.data) runs = pd.read_csv(d / "runs.csv", keep_default_na=False) mentions = build_mentions(runs) mentions.to_csv(d / "mentions.csv", index=False) allb = sorted(mentions.brand.unique()) fix = lambda X: X.reindex(columns=allb, fill_value=0) M = fix(wide(runs, mentions, "mentioned")) MS = fix(wide(runs, mentions, "mentioned_outside_citation_links")) L = fix(wide(runs, mentions, "listed")) T = fix(wide(runs, mentions, "in_top3")) lm = mentions[mentions.listed == 1].copy() lm["rank1"] = (lm.position.astype(int) == 1).astype(int) first = fix(wide(runs, lm, "rank1")) tm = mentions.copy() tm["text_top3"] = (pd.to_numeric(tm.first_mention_order, errors="coerce") <= 3).astype(int) TT = fix(wide(runs, tm, "text_top3")) base = runs.set_index("run_id")[["platform", "prompt_id", "web_search"]] R = base.join(M.add_prefix("m_")).join(L.add_prefix("l_")).join(T.add_prefix("t3_")) \ .join(first.add_prefix("r1_")).join(MS.add_prefix("ms_")).join(TT.add_prefix("tt_")).reset_index() R = R.fillna(0) brands = list(M.columns) R["n_mentioned"] = R[[f"m_{b}" for b in brands]].sum(axis=1) R["n_listed"] = R[[f"l_{b}" for b in brands]].sum(axis=1) summary = [] def add(fid, metric, group, k, n, method="Wilson 95%"): p, lo, hi = wilson(int(k), int(n)) summary.append({"finding_id": fid, "metric": metric, "group": group, "k": int(k), "n": int(n), "value": round(p, 4), "ci_low": round(lo, 4), "ci_high": round(hi, 4), "method": method}) def add_stat(fid, metric, group, value, lo, hi, method): summary.append({"finding_id": fid, "metric": metric, "group": group, "k": "", "n": "", "value": round(value, 4), "ci_low": round(lo, 4), "ci_high": round(hi, 4), "method": method}) slices = [("All", R)] + [(p, g) for p, g in R.groupby("platform")] + [(p, g) for p, g in R.groupby("prompt_id")] \ + [(f"{a}-{b}", g) for (a, b), g in R.groupby(["platform", "prompt_id"])] # F0: brand-level rates for b in brands: for name, g in slices: add("F0", f"mentioned:{b}", name, g[f"m_{b}"].sum(), len(g)) add("F0", f"top3:{b}", name, g[f"t3_{b}"].sum(), len(g)) add("F0", f"rank1:{b}", name, g[f"r1_{b}"].sum(), len(g)) k_m = R[f"m_{b}"].sum() add("F0", f"top3_given_mentioned:{b}", "All", ((R[f"m_{b}"] == 1) & (R[f"t3_{b}"] == 1)).sum(), k_m) # F1 hero: Deel gap = lambda g: g["m_Deel"].mean() - g["t3_Deel"].mean() lo, hi = boot_ci(R, gap) add_stat("F1", "deel_mention_minus_top3", "All", gap(R), lo, hi, "stratified bootstrap 95%, 10k") add("F1", "deel_listed", "All", R["l_Deel"].sum(), len(R)) add("F1", "deel_mentioned_outside_citation_links", "All", R["ms_Deel"].sum(), len(R)) add("F1", "deel_in_first3_named_in_text", "All", R["tt_Deel"].sum(), len(R)) for col, grp in [("prompt_id", "drop "), ("platform", "drop ")]: for v in sorted(R[col].unique()): g = R[R[col] != v] add("F1-LOO", "deel_mentioned", grp + v, g["m_Deel"].sum(), len(g)) add("F1-LOO", "deel_top3", grp + v, g["t3_Deel"].sum(), len(g)) # F2: Rippling #1 add("F2", "rippling_rank1", "All", R["r1_Rippling"].sum(), len(R)) for col in ["prompt_id", "platform"]: for v in sorted(R[col].unique()): g = R[R[col] != v] add("F2-LOO", "rippling_rank1", "drop " + v, g["r1_Rippling"].sum(), len(g)) # F3: big-4 share of top-3 slots big4 = ["Rippling", "BambooHR", "Gusto", "Justworks"] t3cols = [f"t3_{b}" for b in brands] share = lambda g: g[[f"t3_{b}" for b in big4]].values.sum() / g[t3cols].values.sum() lo, hi = boot_ci(R, share) add_stat("F3", "big4_share_of_top3_slots", "All", share(R), lo, hi, "stratified bootstrap 95%, 10k") all3 = lambda g: (g[[f"t3_{b}" for b in big4]].sum(axis=1) >= 3).mean() add("F3", "runs_top3_all_from_big4", "All", int(round(all3(R) * len(R))), len(R)) # F4: brands per answer, ChatGPT vs Gemini (mentioned brands) for c in ["n_mentioned", "n_listed"]: diff = lambda g, c=c: g[g.platform == "ChatGPT"][c].mean() - g[g.platform == "Gemini"][c].mean() lo, hi = boot_ci(R, diff) add_stat("F4", f"{c}_ChatGPT_minus_Gemini", "All", diff(R), lo, hi, "stratified bootstrap 95%, 10k") for name, g in slices[:3]: add_stat("F4", f"{c}_mean", name, g[c].mean(), g[c].min(), g[c].max(), "mean (min, max)") # F5: which moves mention rates more, the assistant or the wording? # Spread = mean absolute gap in a brand's mention rate between groups, averaged over # brands mentioned at least once. Prompt spread uses the mean of the 3 pairwise gaps so it # is on the same scale as the 2-group platform gap. Raw gaps are inflated by sampling noise # (n=10-30 per group), so each is compared with a permutation null (labels shuffled within # the other factor) and reported as excess over that null. X = R[[f"m_{b}" for b in brands]].to_numpy(float) plat = R.platform.to_numpy() prm = R.prompt_id.to_numpy() def pgap(lbl): a, b = sorted(set(lbl)) return np.abs(X[lbl == a].mean(0) - X[lbl == b].mean(0)).mean() def qgap(lbl): ks = sorted(set(lbl)) mu = [X[lbl == k].mean(0) for k in ks] return np.mean([np.abs(mu[i] - mu[j]).mean() for i in range(3) for j in range(i + 1, 3)]) def shuffle_within(lbl, by, rng): out = lbl.copy() for v in set(by): ix = np.where(by == v)[0] out[ix] = rng.permutation(lbl[ix]) return out rng = np.random.default_rng(SEED) sp, sq = pgap(plat), qgap(prm) null_p = np.array([pgap(shuffle_within(plat, prm, rng)) for _ in range(N_BOOT)]) null_q = np.array([qgap(shuffle_within(prm, plat, rng)) for _ in range(N_BOOT)]) add_stat("F5", "platform_gap_raw", "All", sp, np.nan, np.nan, "mean abs gap across brands") add_stat("F5", "prompt_gap_raw", "All", sq, np.nan, np.nan, "mean pairwise abs gap across brands") add_stat("F5", "platform_gap_excess_over_null", "All", sp - null_p.mean(), sp - np.percentile(null_p, 97.5), sp - np.percentile(null_p, 2.5), "permutation, 10k (range = obs minus null 95% band)") add_stat("F5", "prompt_gap_excess_over_null", "All", sq - null_q.mean(), sq - np.percentile(null_q, 97.5), sq - np.percentile(null_q, 2.5), "permutation, 10k (range = obs minus null 95% band)") add_stat("F5", "platform_gap_perm_p", "All", float((null_p >= sp).mean()), np.nan, np.nan, "one-sided permutation p") add_stat("F5", "prompt_gap_perm_p", "All", float((null_q >= sq).mean()), np.nan, np.nan, "one-sided permutation p") # F8: problem-led wording (P2) and BambooHR at #1 for v, g in R.groupby("prompt_id"): add("F8", "bamboohr_rank1", v, g["r1_BambooHR"].sum(), len(g)) for p, g in R[R.prompt_id == "P2"].groupby("platform"): add("F8", "bamboohr_rank1_P2", p, g["r1_BambooHR"].sum(), len(g)) # F9: HiBob by platform for p, g in R.groupby("platform"): add("F9", "hibob_mentioned", p, g["m_HiBob"].sum(), len(g)) hd = lambda g: g[g.platform == "ChatGPT"]["m_HiBob"].mean() - g[g.platform == "Gemini"]["m_HiBob"].mean() lo, hi = boot_ci(R, hd) add_stat("F9", "hibob_mentioned_ChatGPT_minus_Gemini", "All", hd(R), lo, hi, "stratified bootstrap 95%, 10k") # F10: named but never listed (enterprise suites) for b in ["Workday", "ADP", "UKG"]: add("F10", f"named_never_listed:{b}", "All", R[f"m_{b}"].sum(), len(R)) add("F10", f"listed:{b}", "All", R[f"l_{b}"].sum(), len(R)) ent = R[["m_Workday", "m_ADP", "m_UKG"]].max(axis=1) add("F10", "any_enterprise_suite_named", "All", ent.sum(), len(R)) # F6: Deel by platform (is the 'mentioned not recommended' pattern on both?) for p in ["ChatGPT", "Gemini"]: g = R[R.platform == p] add("F6", "deel_mentioned", p, g["m_Deel"].sum(), len(g)) add("F6", "deel_top3", p, g["t3_Deel"].sum(), len(g)) pdiff = lambda g: g[g.platform == "ChatGPT"]["m_Deel"].mean() - g[g.platform == "Gemini"]["m_Deel"].mean() lo, hi = boot_ci(R, pdiff) add_stat("F6", "deel_mentioned_ChatGPT_minus_Gemini", "All", pdiff(R), lo, hi, "stratified bootstrap 95%, 10k") # F7: web search add("F7", "web_search_used", "ChatGPT", (R[R.platform == "ChatGPT"].web_search == "yes").sum(), 30) add("F7", "web_search_used", "Gemini", (R[R.platform == "Gemini"].web_search == "yes").sum(), 30) S = pd.DataFrame(summary) S.to_csv(d / "summary.csv", index=False) # ---------- console report def line(metric, group="All"): x = S[(S.metric == metric) & (S.group == group)].iloc[0] kn = f"{x.k}/{x.n} = " if x.k != "" else "" return f"{metric:42s} {group:14s} {kn}{x.value:.3f} [{x.ci_low:.3f}, {x.ci_high:.3f}]" print("\n== Brand rates (all 60 runs): mentioned / top-3 / ranked #1") for b in sorted(brands, key=lambda b: -R[f"m_{b}"].sum()): print(f" {b:12s} mentioned {R[f'm_{b}'].sum():2d} listed {R[f'l_{b}'].sum():2d} top3 {R[f't3_{b}'].sum():2d} #1 {R[f'r1_{b}'].sum():2d}") print("\n== F1 Deel"); [print(" ", line(m)) for m in ["mentioned:Deel", "deel_listed", "deel_mentioned_outside_citation_links", "top3:Deel", "deel_in_first3_named_in_text", "deel_mention_minus_top3", "top3_given_mentioned:Deel"]] [print(" ", line(m, g)) for g in sorted(S[S.finding_id == "F1-LOO"].group.unique()) for m in ["deel_mentioned", "deel_top3"]] print("\n== F2 Rippling #1"); print(" ", line("rippling_rank1")); [print(" ", line("rippling_rank1", g)) for g in sorted(S[S.finding_id == "F2-LOO"].group.unique())] print("\n== F3 Big-4"); print(" ", line("big4_share_of_top3_slots")); print(" ", line("runs_top3_all_from_big4")) print("\n== F4 Brands per answer"); [print(" ", line(m, g)) for m in ["n_mentioned_mean", "n_listed_mean"] for g in ["All", "ChatGPT", "Gemini"]] print(" ", line("n_mentioned_ChatGPT_minus_Gemini")); print(" ", line("n_listed_ChatGPT_minus_Gemini")) print("\n== F5 Platform vs prompt spread"); [print(" ", line(m)) for m in ["platform_gap_raw", "prompt_gap_raw", "platform_gap_excess_over_null", "prompt_gap_excess_over_null", "platform_gap_perm_p", "prompt_gap_perm_p"]] print("\n== F8 BambooHR #1 by prompt"); [print(" ", line("bamboohr_rank1", v)) for v in ["P1", "P2", "P3"]]; [print(" ", line("bamboohr_rank1_P2", p)) for p in ["ChatGPT", "Gemini"]] print("\n== F9 HiBob by platform"); [print(" ", line("hibob_mentioned", p)) for p in ["ChatGPT", "Gemini"]]; print(" ", line("hibob_mentioned_ChatGPT_minus_Gemini")) print("\n== F10 Named, never listed"); [print(" ", line(m)) for m in ["named_never_listed:Workday", "named_never_listed:ADP", "named_never_listed:UKG", "any_enterprise_suite_named"]] print("\n== F6 Deel by platform"); [print(" ", line(m, p)) for p in ["ChatGPT", "Gemini"] for m in ["deel_mentioned", "deel_top3"]] print(" ", line("deel_mentioned_ChatGPT_minus_Gemini")) print("\n== F7 Web search"); [print(" ", line("web_search_used", p)) for p in ["ChatGPT", "Gemini"]] print(f"\nWrote {d/'mentions.csv'} ({len(mentions)} rows) and {d/'summary.csv'} ({len(S)} rows)") if __name__ == "__main__": main()