vep/scripts/p0/run.py

# VEP — Phase 0 Falsification Harness (GST reference tenant)
#
# PURPOSE
# Run the four load-bearing GST hypotheses (H1-H4) against real data and emit a
# KILL / KEEP / REVISE verdict per the gst_p0_falsification.md worksheet.
# This is the gate that, per the spine, should have preceded the P1 build. Since P1
# exists, this harness lets the gate run now: supply data -> get verdicts -> prune or keep.
#
# DESIGN RULES (spine-consistent)
#  - No fabricated numbers. If data is missing, the verdict is INSUFFICIENT, never a guess.
#  - Each hypothesis names its falsify-condition and the data column(s) it needs.
#  - Verdicts are explainable: every KILL cites the observed number that killed it.
#
# USAGE
#   python scripts/p0/run.py --data-dir scripts/p0/sample_data
# Data files expected (see templates in sample_data/README):
#   signups.csv        : email, intent_bucket, converted_to_foundation(bool), date
#   cohort_jan2026.csv : attendee, bought_ageless(bool), upgraded_tier2(bool)
#   subscribers.csv    : sub_id, tier, practice_signal(0-1), continued(bool), purchase_recency_days
#   capped_clients.csv : client_id, revenue_stream, converted_to_scalable(bool)
#
# With no data the harness reports INSUFFICIENT for every hypothesis (honest default).

from __future__ import annotations
import argparse, csv, json, os, sys
from dataclasses import dataclass, field


@dataclass
class Hypothesis:
    id: str
    statement: str
    needs: str
    falsify: str
    verdict: str = "INSUFFICIENT"
    evidence: str = "no data supplied"

    def report(self) -> str:
        return f"[{self.id}] {self.verdict}\n   claim: {self.statement}\n   needs: {self.needs}\n   falsify-if: {self.falsify}\n   evidence: {self.evidence}\n"


HYPS = [
    Hypothesis("H1", "Intent-based routing + Foundation Series landing converts the 190-signup audience at a rate worth building funnel infra for.",
               "signups.csv (intent_bucket, converted_to_foundation)",
               "conversion < threshold needed to justify build (define threshold before run)"),
    Hypothesis("H2", "Behavioral recency (practice signal) predicts subscription continuation better than 90/180-day purchase timers.",
               "subscribers.csv (practice_signal, continued, purchase_recency_days)",
               "purchase_recency correlates with continuation >= practice_signal"),
    Hypothesis("H3", "Movement Lab as qualification hits Ageless Body 25%+ / Tier 2 40%+ conversion.",
               "cohort_jan2026.csv (bought_ageless, upgraded_tier2)",
               "actual conversion materially below targets, or Lab adds no lift vs non-attendees"),
    Hypothesis("H4", "Founder-time-capped revenue (56%) shifts to scalable streams via acquisition instrumentation alone, without practitioner training.",
               "capped_clients.csv (revenue_stream, converted_to_scalable)",
               "capped clients do not convert to scalable streams at the needed rate"),
]


def _load(path: str) -> list[dict]:
    if not os.path.exists(path):
        return []
    with open(path, newline="") as f:
        return list(csv.DictReader(f))


def _rate(rows, col) -> float | None:
    if not rows:
        return None
    n = sum(1 for r in rows if str(r.get(col, "")).strip().lower() in ("1", "true", "yes", "t"))
    return n / len(rows)


def evaluate(h: Hypothesis, data_dir: str) -> None:
    if h.id == "H1":
        rows = _load(os.path.join(data_dir, "signups.csv"))
        if not rows:
            return
        # Require a defined threshold to avoid a fabricated verdict.
        thr = os.environ.get("H1_THRESHOLD")
        rate = _rate(rows, "converted_to_foundation")
        if rate is None:
            h.evidence = "signups.csv present but column 'converted_to_foundation' empty"
            return
        if thr is None:
            h.verdict = "INSUFFICIENT"
            h.evidence = f"conversion={rate:.1%} but H1_THRESHOLD undefined; set it to decide KILL/KEEP"
            return
        thr_f = float(thr)
        h.evidence = f"conversion={rate:.1%} (threshold={thr_f:.1%})"
        h.verdict = "KEEP" if rate >= thr_f else "KILL"
    elif h.id == "H2":
        rows = _load(os.path.join(data_dir, "subscribers.csv"))
        if not rows:
            return
        # Compute correlation of practice_signal vs continued, and purchase_recency vs continued.
        def corr(col):
            xs, ys = [], []
            for r in rows:
                try:
                    xs.append(float(r[col]))
                    ys.append(1.0 if str(r.get("continued", "")).lower() in ("1", "true", "t") else 0.0)
                except (ValueError, TypeError):
                    pass
            if len(xs) < 3:
                return None
            mx, my = sum(xs) / len(xs), sum(ys) / len(ys)
            num = sum((x - mx) * (y - my) for x, y in zip(xs, ys))
            den = (sum((x - mx) ** 2 for x in xs) * sum((y - my) ** 2 for y in ys)) ** 0.5
            return num / den if den else 0.0
        cp, cr = corr("practice_signal"), corr("purchase_recency_days")
        if cp is None or cr is None:
            h.evidence = "subscribers.csv present but insufficient numeric rows"
            return
        h.evidence = f"corr(practice,continued)={cp:.2f}; corr(purchase_recency,continued)={cr:.2f}"
        h.verdict = "KILL" if cr >= cp else "KEEP"
    elif h.id == "H3":
        rows = _load(os.path.join(data_dir, "cohort_jan2026.csv"))
        if not rows:
            return
        a = _rate(rows, "bought_ageless")
        t = _rate(rows, "upgraded_tier2")
        if a is None or t is None:
            h.evidence = "cohort_jan2026.csv present but target columns empty"
            return
        h.evidence = f"Ageless={a:.1%} (target 25%); Tier2={t:.1%} (target 40%)"
        h.verdict = "KEEP" if (a >= 0.25 and t >= 0.40) else "REVISE"
    elif h.id == "H4":
        rows = _load(os.path.join(data_dir, "capped_clients.csv"))
        if not rows:
            return
        r = _rate(rows, "converted_to_scalable")
        if r is None:
            h.evidence = "capped_clients.csv present but column empty"
            return
        h.evidence = f"capped->scalable conversion={r:.1%}"
        h.verdict = "KEEP" if r >= 0.30 else "REVISE"


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--data-dir", default=os.path.join(os.path.dirname(__file__), "sample_data"))
    args = ap.parse_args()
    print("VEP P0 Falsification Harness — GST reference tenant")
    print(f"data-dir: {args.data_dir}\n")
    kills = 0
    for h in HYPS:
        evaluate(h, args.data_dir)
        print(h.report())
        if h.verdict in ("KILL", "REVISE"):
            kills += 1
    print("=" * 60)
    if kills == 0 and all(h.verdict == "INSUFFICIENT" for h in HYPS):
        print("GATE: INSUFFICIENT DATA — supply data files to run the gate.")
        print("This is the honest default: no fabricated verdicts.")
    else:
        print(f"GATE: {kills} hypothesis(es) KILL/REVISE. Per P0 acceptance, >=1 kill/revise = gate MET.")
        print("If KILL/REVISE: prune or revise the build accordingly (the P1 spine is small + reversible).")
    return 0


if __name__ == "__main__":
    sys.exit(main())