# -*- coding: utf-8 -*- """ analyse_campagne.py — MOD Focus marketing avancé / Expérimentation (2026-2027) De l'export brut à la phrase de résultat. Fonctionne dans Google Colab (uploader les deux CSV) ou en local (pandas + scipy). Adapté du script analyses_expe.py du cours Prix M1. Entrées : events.csv — l'export de collect.php (ts, session_id, tag, flow, type, step, value) ads_manager.csv — l'export Meta Ads Manager par annonce (une annonce = une condition), séparateur « ; », colonnes condition / Impressions / Clics sur un lien / Montant dépensé Sorties, dans l'ordre où le rapport les attend : 1. N par condition (impressions, clics pub, visiteurs, clics CTA, questionnaires complets) 2. Manipulation check : M, SD par condition, t de Welch, p, d de Cohen 3. Test principal adapté à la VD : - VD binaire (a cliqué / n'a pas cliqué) -> chi² d'indépendance + h de Cohen - VD continue (Likert) -> t de Welch + d de Cohen 4. La phrase de résultat normée, prête à coller dans le rapport Règle du cours : un résultat non significatif est un résultat. On l'écrit avec ses chiffres. """ import math import numpy as np import pandas as pd from scipy import stats EVENTS = "exemple_events.csv" # remplacer par votre export ADS = "exemple_ads_manager.csv" # remplacer par votre export Ads Manager MC_STEPS = ["mc1", "mc2"] # les items du manipulation check (moyenne = score MC) ATTENTION_STEP = "attention" # l'attention check ATTENTION_OK = {"A": "acte", "B": "abonnement"} # la bonne réponse attendue PAR condition CONTINUOUS_DV = "intention" # une VD déclarative en complément (facultatif) # --------------------------------------------------------------------------- # 1. De l'export brut (un événement par ligne) au tableau propre (une ligne par visiteur) # --------------------------------------------------------------------------- def charger_events(chemin): ev = pd.read_csv(chemin) ev["ts"] = pd.to_datetime(ev["ts"], utc=True).dt.tz_convert("Europe/Paris") ev = ev.sort_values("ts") # 1 ligne = 1 session_id ; pour chaque question, on garde la DERNIÈRE valeur cliquée (règle de view.php) reponses = (ev[ev["type"] == "answer"] .groupby(["session_id", "step"])["value"].last() .unstack("step")) sessions = ev.groupby("session_id").agg(flow=("flow", "first"), tag=("tag", "first"), debut=("ts", "min"), fin=("ts", "max")) sessions["a_clique"] = ev[ev["type"] == "click"].groupby("session_id").size().reindex(sessions.index).fillna(0).gt(0) sessions["questionnaire_fini"] = ev[ev["type"] == "end"].groupby("session_id").size().reindex(sessions.index).fillna(0).gt(0) df = sessions.join(reponses) for col in MC_STEPS + [CONTINUOUS_DV]: if col in df.columns: df[col] = pd.to_numeric(df[col], errors="coerce") df["MC"] = df[[c for c in MC_STEPS if c in df.columns]].mean(axis=1) if ATTENTION_STEP in df.columns: df["attention_ok"] = df.apply(lambda r: r[ATTENTION_STEP] == ATTENTION_OK.get(r["flow"]), axis=1) return df def cohen_d(x, y): x, y = np.asarray(x, float), np.asarray(y, float) nx, ny = len(x), len(y) s = math.sqrt(((nx - 1) * x.var(ddof=1) + (ny - 1) * y.var(ddof=1)) / (nx + ny - 2)) return (x.mean() - y.mean()) / s if s > 0 else float("nan") def cohen_h(p1, p2): return 2 * math.asin(math.sqrt(p1)) - 2 * math.asin(math.sqrt(p2)) def welch(df, col, groupe="flow"): a = df.loc[df[groupe] == "A", col].dropna() b = df.loc[df[groupe] == "B", col].dropna() t, p = stats.ttest_ind(a, b, equal_var=False) # degrés de liberté de Welch–Satterthwaite (ce que R affiche dans t.test) va, vb = a.var(ddof=1) / len(a), b.var(ddof=1) / len(b) ddl = (va + vb) ** 2 / (va ** 2 / (len(a) - 1) + vb ** 2 / (len(b) - 1)) return dict(nA=len(a), nB=len(b), MA=a.mean(), SDA=a.std(ddof=1), MB=b.mean(), SDB=b.std(ddof=1), t=t, ddl=ddl, p=p, d=cohen_d(a, b)) def chi2_clics(df, col="a_clique", groupe="flow"): tab = pd.crosstab(df[groupe], df[col]).reindex(index=["A", "B"], columns=[False, True], fill_value=0) chi2, p, ddl, _ = stats.chi2_contingency(tab, correction=False) pA = tab.loc["A", True] / tab.loc["A"].sum() pB = tab.loc["B", True] / tab.loc["B"].sum() return dict(tab=tab, pA=pA, pB=pB, chi2=chi2, ddl=ddl, p=p, h=cohen_h(pA, pB)) def fmt_p(p): return "p < .001" if p < 0.001 else "p = %.3f" % p if __name__ == "__main__": df = charger_events(EVENTS) ads = pd.read_csv(ADS, sep=";", encoding="utf-8-sig") # ---------------- 1. N par condition print("=" * 72) print("1. N PAR CONDITION".center(72)) print("=" * 72) n = pd.DataFrame({ "impressions (Meta)": ads.set_index("condition")["Impressions"], "clics pub (Meta)": ads.set_index("condition")["Clics sur un lien"], "visiteurs landing (site)": df.groupby("flow").size(), "clics bouton (VD)": df.groupby("flow")["a_clique"].sum(), "questionnaires complets": df.groupby("flow")["questionnaire_fini"].sum(), }) print(n.to_string()) if "attention_ok" in df: q = df[df["questionnaire_fini"]] print("\nAttention check réussi : %.0f %% des questionnaires complets (%d/%d)" % (100 * q["attention_ok"].mean(), q["attention_ok"].sum(), len(q))) print("\nÉcart Meta / site : %d clics pub vs %d arrivées enregistrées — l'écart (clics accidentels," " bloqueurs, rebonds avant chargement) se commente dans les limites." % (ads["Clics sur un lien"].sum(), len(df))) # ---------------- 2. Manipulation check print("\n" + "=" * 72) print("2. MANIPULATION CHECK (moyenne des items %s)" % ", ".join(MC_STEPS)) print("=" * 72) q = df[df["questionnaire_fini"]] mc = welch(q, "MC") print("A : M = %.2f, SD = %.2f (n = %d) | B : M = %.2f, SD = %.2f (n = %d)" % (mc["MA"], mc["SDA"], mc["nA"], mc["MB"], mc["SDB"], mc["nB"])) print("t(%.1f) = %.2f, %s, d = %.2f" % (mc["ddl"], mc["t"], fmt_p(mc["p"]), mc["d"])) if mc["p"] < 0.05: print("-> La manipulation est perçue : on peut interpréter le test principal.") else: print("-> La manipulation N'EST PAS perçue : le test principal reste calculable, mais toute différence" " (ou absence de différence) ne peut pas être attribuée à votre variable. À écrire tel quel.") # ---------------- 3a. Test principal : VD binaire (le clic) print("\n" + "=" * 72) print("3a. TEST PRINCIPAL — VD comportementale : a cliqué sur le bouton (chi²)".center(72)) print("=" * 72) c = chi2_clics(df) print(c["tab"].rename(columns={False: "pas de clic", True: "clic"}).to_string()) print("\nTaux de clic : A = %.1f %% vs B = %.1f %%, χ²(%d) = %.2f, %s, h = %.2f" % (100 * c["pA"], 100 * c["pB"], c["ddl"], c["chi2"], fmt_p(c["p"]), c["h"])) # ---------------- 3b. Test complémentaire : VD continue (Likert) if CONTINUOUS_DV in df.columns: print("\n" + "=" * 72) print("3b. COMPLÉMENT — VD déclarative « %s » (t de Welch)" % CONTINUOUS_DV) print("=" * 72) w = welch(q, CONTINUOUS_DV) print("A : M = %.2f, SD = %.2f (n = %d) | B : M = %.2f, SD = %.2f (n = %d)" % (w["MA"], w["SDA"], w["nA"], w["MB"], w["SDB"], w["nB"])) print("t(%.1f) = %.2f, %s, d = %.2f" % (w["ddl"], w["t"], fmt_p(w["p"]), w["d"])) # ---------------- 4. La phrase normée print("\n" + "=" * 72) print("4. PHRASES À COLLER DANS LE RAPPORT (à relire, pas à recopier sans comprendre)".center(72)) print("=" * 72) print("MC : « Les répondants de la condition B (M = %.2f, SD = %.2f) ont perçu davantage [le construit manipulé]" " que ceux de la condition A (M = %.2f, SD = %.2f ; t(%.1f) = %.2f, %s, d = %.2f) : la manipulation est %s. »" % (mc["MB"], mc["SDB"], mc["MA"], mc["SDA"], mc["ddl"], mc["t"], fmt_p(mc["p"]), mc["d"], "réussie" if mc["p"] < 0.05 else "NON vérifiée")) verdict = "H1 est acceptée" if c["p"] < 0.05 else "H1 est rejetée (résultat non significatif, pas un échec)" print("H1 : « Le taux de clic est de %.1f %% en condition A (n = %d) contre %.1f %% en condition B (n = %d)," " χ²(%d) = %.2f, %s, h = %.2f. %s. »" % (100 * c["pA"], c["tab"].loc["A"].sum(), 100 * c["pB"], c["tab"].loc["B"].sum(), c["ddl"], c["chi2"], fmt_p(c["p"]), c["h"], verdict)) print("\nEnsuite, et seulement ensuite : la décision pour le client, et une limite.")