"""Analysis of the launch-campaign study: the event-level ledger, intent outcomes,
attribution, brain-LTV under stated assumptions, uncertainty and sensitivity.

    .venv/bin/python research/studies/market-perception/press-brain-ltv/analyze.py \\
        [--runs research/studies/market-perception/press-brain-ltv/evidence]

Reads each seed's traces (``evidence/seed-<n>/trace-<arm>.jsonl.gz``) and writes
``results/``: ``summary.json`` (what the page shows), ``events.csv`` (every action),
``exposures.csv`` (every first sighting of a campaign post, with its inferred channel),
``conversions.csv`` (every commercial-intent reader with the credit each post receives)
and ``arms.csv``.

Every number in ``summary.json`` sits under one of three keys, and they are never mixed
without saying so:

- ``measured``: counted from simulator events (follows, posts, likes, reposts, sightings);
- ``derived``: computed from measured events by a stated, deterministic rule (lifts,
  bootstrap intervals, attribution shares, channel inference, the intent classifier);
- ``assumed``: commercial parameters from ``assumptions.json``, which no simulation
  measured, and every dollar figure, which depends on them.
"""

import argparse
import csv
import gzip
import hashlib
import json
import pathlib
import random
import re
import sys

HERE = pathlib.Path(__file__).resolve().parent
ROOT = HERE.parents[3]
sys.path.insert(0, str(ROOT / "scripts/society"))

import factors  # noqa: E402
import ledger  # noqa: E402
import schema  # noqa: E402

PREREG = json.loads((HERE / "preregistration.json").read_text())
REFERENCE = re.compile(PREREG["definitions"]["REFERENCE"], re.I)
INTENT = re.compile(PREREG["definitions"]["INTENT"], re.I)
COMPANY = ("ferrovane", "mara_ferrovane")
# Post hoc (after reading the traces): a reader claiming first-hand use of, or price
# negotiations over, a product launched that evening is confabulating. The pattern was
# checked by hand against every member post that names the product (the page lists them).
CONFAB = re.compile(r"(we(?:'ve| have)? (?:run|ran|piloted|trialed|installed|been running) |we run vx|we(?:'re| are) piloting|"
                    r"we went with|vx-?4s? on (?:our|\d|pump|a |the )|we got ferrovane|ferrovane moved|were quoted|we swap)", re.I)
ARTIFACTS = ("announcement", "article", "founder", "discussion-1", "discussion-2", "critique", "placebo")
# A reader's own top-level post naming the product is campaign content too (word of
# mouth): it is not a reply to anything, so lineage alone would leave it out.
WOM = "word-of-mouth"
FAMILIES = ARTIFACTS + (WOM,)
AMBIENT = tuple(f"a-{i}" for i in range(1, 10))
ARMS = ("control", "announcement", "article", "founder", "article-discussion", "article-critique", "placebo")
COHORTS = ("practitioners", "generalists", "buyers")
BOOT = 2000


def cohort_of(person):
    return person.rsplit("-", 1)[0]


def read_trace(path):
    path = pathlib.Path(path)
    if not path.exists() and path.with_suffix(path.suffix + ".gz").exists():
        path = path.with_suffix(path.suffix + ".gz")
    opener = gzip.open if path.suffix == ".gz" else open
    with opener(path, "rt") as f:
        return [json.loads(line) for line in f if line.strip()]


def initial_social(spec, seed):
    """The x.com state the members start from: posts, and who follows whom."""
    compiled = schema.compile_spec(json.loads(json.dumps(spec)), HERE / "spec/launch-campaign.json", seed)
    state = compiled["definition"]["internet_overlays"]["x-social"]["initial_state"]
    people = [p["id"] for p in compiled["people"]]
    return state, people, compiled


def analyze_arm(records, initial, people, arm_ops):
    """Measured events and derived per-person outcomes of one arm of one seed."""
    events = ledger.actions(records)
    sightings = ledger.exposures(records)
    posts = {pid: {"author": p.get("author"), "reply_to": p.get("reply_to"), "quoted": p.get("quoted"),
                   "text": p.get("text", "")} for pid, p in initial["posts"].items()}
    for op in arm_ops:
        p = op["post"]
        posts[p["id"]] = {"author": op["as"], "reply_to": p.get("reply_to"), "quoted": p.get("quoted"),
                          "text": p["text"], "round": op.get("at", 0)}
    for e in events:
        if e["kind"] in ("post", "reply", "quote"):
            posts[e["post"]] = {"author": member_handles.get(e["person"], e["person"]), "reply_to": e["target"] if e["kind"] == "reply" else None,
                                "quoted": e["target"] if e["kind"] == "quote" else None, "text": e["text"],
                                "round": e["round"]}
    family = ledger.lineage(posts, roots=ARTIFACTS)
    wom_roots = {pid for pid, f in family.items() if f == pid and pid not in ARTIFACTS and pid not in initial["posts"]
                 and REFERENCE.search(posts[pid].get("text") or "")}
    family = {pid: (WOM if f in wom_roots else f) for pid, f in family.items()}
    campaign = {p for p, f in family.items() if f in FAMILIES}
    for e in events:
        e["family"] = family.get(e.get("target") or "", None) if e["kind"] != "follow" else None
        if e["kind"] in ("post", "reply", "quote"):
            e["family"] = family.get(e["post"])

    # Who follows whom, as it changes (for the channel a sighting came through).
    follows = {p: set(initial["follows"].get(p, [])) for p in people}
    follow_log = sorted((e for e in events if e["kind"] in ("follow", "unfollow")), key=ledger.order_key)
    repost_log = sorted((e for e in events if e["kind"] == "repost"), key=ledger.order_key)

    def follows_at(person, key):
        current = set(follows.get(person, set()))
        for e in follow_log:
            if e["person"] != person or ledger.order_key(e) > key:
                continue
            (current.add if e["kind"] == "follow" else current.discard)(e["target"])
        return current

    # Intent readers: a held follow begun in the run, or an intent post. Converted at the
    # earliest qualifying event.
    held = {}
    for e in follow_log:
        if e["target"] in COMPANY:
            if e["kind"] == "follow":
                held.setdefault((e["person"], e["target"]), e)
            else:
                held.pop((e["person"], e["target"]), None)
    intent_posts = [e for e in events if e["kind"] in ("post", "reply", "quote")
                    and REFERENCE.search(e["text"] or "") and INTENT.search(e["text"] or "")]
    conversions = {}
    for e in list(held.values()) + intent_posts:
        why = "follow" if e["kind"] == "follow" else "intent post"
        prior = conversions.get(e["person"])
        if prior is None or ledger.order_key(e) < (prior["round"], prior["_index"]):
            conversions[e["person"]] = {"person": e["person"], "round": e["round"], "_index": e["_index"],
                                        "via": why, "event": e}
    # Exploratory, defined after the registration (see the page): the registered
    # classifier needs the product named, but a reply under a campaign post asking about
    # price or a pilot rarely names it. Counted separately, never in the primary outcome.
    broad = {e["person"] for e in events if e["kind"] in ("post", "reply", "quote")
             and INTENT.search(e["text"] or "") and (REFERENCE.search(e["text"] or "")
                                                    or family.get(e["post"]) in FAMILIES)} | set(conversions)
    grounded = {e["person"] for e in held.values()} | {e["person"] for e in intent_posts if not CONFAB.search(e["text"] or "")}
    for c in conversions.values():
        c["grounded"] = c["person"] in grounded
    for c in conversions.values():
        c["via_all"] = sorted({("follow" if e["kind"] == "follow" else "intent post")
                               for e in list(held.values()) + intent_posts if e["person"] == c["person"]})

    # Sightings of campaign posts, first per (person, post), with the channel inferred.
    first = {}
    for s in sightings:
        if s["post"] in campaign and (s["person"], s["post"]) not in first:
            first[(s["person"], s["post"])] = s
    exposures = []
    for (person, post), s in sorted(first.items(), key=lambda kv: ledger.order_key(kv[1])):
        key = ledger.order_key(s)
        author = posts[post]["author"]
        followed = follows_at(person, key)
        reposters = [e["person"] for e in repost_log if e["target"] == post and ledger.order_key(e) <= key
                     and e["person"] != person]
        via = None
        if author in followed or author == member_handles.get(person):
            channel = "follows the author"
        else:
            via = next((r for r in reposters if member_handles.get(r) in followed), None)
            if via:
                channel = "repost by someone followed"
            elif posts[post].get("reply_to") or posts[post].get("quoted"):
                channel = "thread or quote"
            else:
                channel = "search, profile or explore"
        exposures.append({"person": person, "post": post, "family": family[post], "round": s["round"],
                          "_index": s["_index"], "channel": channel, "via": via, "author": author})

    impressions = sum(1 for s in sightings if s["post"] in campaign)
    return {"events": events, "sightings": sightings, "family": family, "campaign": campaign,
            "conversions": conversions, "broad": broad, "exposures": exposures, "impressions": impressions, "posts": posts}


member_handles = {}


def per_person_outcomes(result, people):
    """0/1 and count outcomes per person for the bootstrap."""
    events = result["events"]
    out = {}
    for p in people:
        out[p] = {
            "intent": int(p in result["conversions"]),
            "intent_broad": int(p in result["broad"]),
            "intent_grounded": int(p in result["conversions"] and result["conversions"][p]["grounded"]),
            "confabulated_posts": sum(1 for e in events if e["person"] == p and e["kind"] in ("post", "reply", "quote")
                                      and CONFAB.search(e["text"] or "") and REFERENCE.search(e["text"] or "")),
            "follow_company": int(any(e["person"] == p and e["kind"] == "follow" and e["target"] == "ferrovane"
                                      for e in events)),
            "reached": int(any(x["person"] == p for x in result["exposures"])),
            "ambient_likes": sum(1 for e in events if e["person"] == p and e["kind"] == "like"
                                 and e["target"] in AMBIENT) - sum(1 for e in events if e["person"] == p
                                                                   and e["kind"] == "unlike" and e["target"] in AMBIENT),
            "northstar_follow": int(any(e["person"] == p and e["kind"] == "follow" and e["target"] == "NorthstarHQ"
                                        for e in events)),
            "impressions": sum(1 for x in result["sightings"] if x["person"] == p and x["post"] in result["campaign"]),
            "via_repost": int(any(x["person"] == p and x["channel"] == "repost by someone followed" for x in result["exposures"])),
            "campaign_engagements": sum(1 for e in events if e["person"] == p and e.get("family") in FAMILIES
                                        and e["kind"] in ("like", "repost", "reply", "quote")),
        }
    return out


def bootstrap(table, seeds, people, outcome, arms, rng, n=BOOT):
    """Paired percentile intervals of arm minus control, mean per panel member: resample
    seeds, and members within each seed, with one draw shared by every arm."""
    draws = {a: [] for a in arms}
    for _ in range(n):
        chosen = [rng.choice(seeds) for _ in seeds]
        sample = [(s, rng.choice(people)) for s in chosen for _ in people]
        base = sum(table[s]["control"][p][outcome] for s, p in sample) / len(sample)
        for a in arms:
            draws[a].append(sum(table[s][a][p][outcome] for s, p in sample) / len(sample) - base)
    out = {}
    for a, values in draws.items():
        values.sort()
        out[a] = {"lo90": values[int(0.05 * n)], "hi90": values[int(0.95 * n) - 1],
                  "share_above_0": sum(v > 0 for v in values) / n}
    return out, draws


def clv(a, product):
    arr = product["rollout_sensors"] * product["software_price_per_sensor_month"] * 12
    return sum(arr * product["software_gross_margin"] * (1 - a["annual_churn"]) ** (t - 1) / (1 + a["discount_rate"]) ** t
               for t in range(1, a["horizon_years"] + 1))


def value_per_intent(a, product):
    """Expected discounted contribution margin of one commercial-intent reader (assumed)."""
    pilot = product["pilot_kit_price"] * product["pilot_kit_hardware_margin"]
    rollout = clv(a, product) - a["onboarding_cost_per_rollout"]
    return a["intent_to_conversation"] * (-a["sales_cost_per_conversation"] + a["conversation_to_pilot"] * (
        pilot + a["pilot_to_rollout"] * rollout))


def cost_of(arm, costs):
    total = 0.0
    for item in costs["arms"][arm]:
        c = costs["items"][item]
        total += c["staff_hours"] * costs["loaded_hourly_rate"] + c["founder_hours"] * costs["founder_hourly_rate"] + c["cash"]
    return total


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--runs", default=str(HERE / "evidence"))
    parser.add_argument("--out", default=str(HERE / "results"))
    args = parser.parse_args(argv)
    runs, out = pathlib.Path(args.runs), pathlib.Path(args.out)
    out.mkdir(parents=True, exist_ok=True)
    spec = json.loads((HERE / "spec/launch-campaign.json").read_text())
    assumptions = json.loads((HERE / "assumptions.json").read_text())
    seeds = PREREG["design"]["seeds"]
    ops = {a["id"]: a.get("ops", []) for a in spec["interventions"]["arms"]}
    table, results, run_meta, people = {}, {}, {}, None
    event_rows, exposure_rows, conversion_rows = [], [], []
    for seed in seeds:
        folder = runs / f"seed-{seed}"
        if not folder.exists():
            print(f"missing {folder}", file=sys.stderr)
            return 2
        initial, people, compiled = initial_social(spec, seed)
        member_handles.clear()
        member_handles.update({p["id"]: p["accounts"]["x-social"] for p in compiled["people"]})
        summary = json.loads((folder / "summary.json").read_text())
        run_meta[seed] = {"journal": summary.get("journal"), "elapsed_s": summary.get("elapsed_s"),
                          "crn": {a: c["holds"] for a, c in summary["crn"].items()},
                          "decisions": {a: d["decisions"] for a, d in summary["arms"].items()},
                          "activations": {a: d["activations"] for a, d in summary["arms"].items()},
                          "final_hash": {a: d["state_hashes"][-1][:16] for a, d in summary["arms"].items()},
                          "graph_edges": {k: len(v) for k, v in compiled["graphs"].items()}}
        # The runner's common-random-numbers check compares gestures by control id. A post
        # a reader wrote is numbered in creation order, so the same on-screen post can carry
        # another id in another arm: classify each flagged pair.
        explained = []
        for arm, c in summary["crn"].items():
            if not c["violations"]:
                continue
            ctl = {(r["round"], r["person"], r["step"]): r for r in read_trace(folder / "trace-control.jsonl") if r.get("type") == "decision"}
            other = {(r["round"], r["person"], r["step"]): r for r in read_trace(folder / f"trace-{arm}.jsonl") if r.get("type") == "decision"}
            for k in c["violations"]:
                x, y = ctl[tuple(k)], other[tuple(k)]
                norm = lambda g: re.sub(r"p-\d+", "p-N", json.dumps(g, sort_keys=True))  # noqa: E731
                explained.append({"arm": arm, "key": k, "same_reply": x["decision"] == y["decision"],
                                  "same_gesture_modulo_post_id": norm(x["gesture"]) == norm(y["gesture"])})
        run_meta[seed]["crn_flags"] = explained
        table[seed], results[seed] = {}, {}
        for arm in ARMS:
            records = read_trace(folder / f"trace-{arm}.jsonl")
            # Replies the policy could not read (empty: the model spent its completion
            # budget reasoning) become "wait" and end that activation. Counted per arm,
            # because an uneven spread would be a confounder.
            decs = [x for x in records if x.get("type") == "decision"]
            run_meta[seed].setdefault("unreadable", {})[arm] = {
                "decisions": len(decs), "unreadable": sum(1 for x in decs if x["decision"].get("why") == "unreadable reply")}
            r = analyze_arm(records, initial, people, ops[arm])
            results[seed][arm] = r
            table[seed][arm] = per_person_outcomes(r, people)
            for e in r["events"]:
                event_rows.append({"seed": seed, "arm": arm, "round": e["round"], "person": e["person"],
                                   "cohort": cohort_of(e["person"]), "kind": e["kind"], "target": e.get("target") or "",
                                   "post": e.get("post", ""), "family": e.get("family") or "",
                                   "text": (e.get("text") or "").replace("\n", " ")})
            for x in r["exposures"]:
                exposure_rows.append({"seed": seed, "arm": arm, "round": x["round"], "person": x["person"],
                                      "cohort": cohort_of(x["person"]), "post": x["post"], "family": x["family"],
                                      "author": x["author"], "channel": x["channel"], "via": x["via"] or ""})

    rng = random.Random(20261028)
    outcomes = ("intent", "intent_grounded", "confabulated_posts", "intent_broad", "follow_company", "reached", "ambient_likes", "northstar_follow", "campaign_engagements")
    n_units = len(seeds) * len(people)
    measured = {"arms": {}, "per_seed": {}, "cohorts": {}, "trajectories": {}}
    derived = {"lift": {}, "attribution": {}, "nested": {}, "channels": {}}
    for arm in ARMS:
        totals = {o: sum(table[s][arm][p][o] for s in seeds for p in people) for o in outcomes}
        measured["arms"][arm] = dict(totals, impressions=sum(results[s][arm]["impressions"] for s in seeds),
                                     units=n_units)
        measured["per_seed"][arm] = {str(s): {o: sum(table[s][arm][p][o] for p in people) for o in outcomes} for s in seeds}
        measured["cohorts"][arm] = {c: {"intent": sum(table[s][arm][p]["intent"] for s in seeds for p in people if cohort_of(p) == c),
                                        "reached": sum(table[s][arm][p]["reached"] for s in seeds for p in people if cohort_of(p) == c),
                                        "units": sum(1 for s in seeds for p in people if cohort_of(p) == c)} for c in COHORTS}
        # Unique reach and impressions by round, mean over seeds.
        traj = []
        for rnd in range(spec["clock"]["rounds"]):
            reach = sum(len({x["person"] for x in results[s][arm]["exposures"] if x["round"] <= rnd}) for s in seeds)
            imp = sum(sum(1 for x in results[s][arm]["sightings"] if x["post"] in results[s][arm]["campaign"] and x["round"] == rnd) for s in seeds)
            reposts = sum(sum(1 for e in results[s][arm]["events"] if e["kind"] == "repost" and e.get("family") in FAMILIES and e["round"] <= rnd) for s in seeds)
            conv = sum(sum(1 for c in results[s][arm]["conversions"].values() if c["round"] <= rnd) for s in seeds)
            traj.append({"round": rnd, "reach": reach / len(seeds), "impressions": imp / len(seeds),
                         "reposts": reposts / len(seeds), "intent": conv / len(seeds)})
        measured["trajectories"][arm] = traj

    intervals = {}
    for o in outcomes:
        intervals[o], draws = bootstrap(table, seeds, people, o, ARMS[1:], rng)
        if o == "intent":
            intent_draws = draws
    for arm in ARMS[1:]:
        lift = {}
        for o in outcomes:
            diff = (measured["arms"][arm][o] - measured["arms"]["control"][o]) / n_units
            lift[o] = {"per_member": diff, **intervals[o][arm]}
        lift["intent"]["per_seed"] = [measured["per_seed"][arm][str(s)]["intent"] - measured["per_seed"]["control"][str(s)]["intent"] for s in seeds]
        derived["lift"][arm] = lift
    # Marginal effect of the follow-up artifacts, given the article.
    v = {frozenset(): measured["arms"]["control"]["intent"], frozenset(["article"]): measured["arms"]["article"]["intent"],
         frozenset(["article", "discussion"]): measured["arms"]["article-discussion"]["intent"],
         frozenset(["article", "critique"]): measured["arms"]["article-critique"]["intent"]}
    derived["nested"] = {
        "discussion_given_article": ledger.increments(v, [[], ["article"], ["article", "discussion"]]),
        "critique_given_article": ledger.increments(v, [[], ["article"], ["article", "critique"]]),
        "note": "Counts of intent readers summed over seeds. Shapley values are not identified: no arm runs the discussion or the critique without the article."}
    draws_nested = {}
    for follow_arm in ("article-discussion", "article-critique"):
        d = sorted(x - y for x, y in zip(intent_draws[follow_arm], intent_draws["article"]))
        draws_nested[follow_arm] = {"per_member": (measured["arms"][follow_arm]["intent"] - measured["arms"]["article"]["intent"]) / n_units,
                                    "lo90": d[int(0.05 * BOOT)], "hi90": d[int(0.95 * BOOT) - 1]}
    derived["nested"]["intervals_vs_article"] = draws_nested

    # Attribution: each arm's conversions over the campaign posts their person saw first.
    for arm in ARMS[1:]:
        per_method = {}
        for method in ledger.METHODS:
            by_artifact, by_kind, unexposed, total = {}, {}, 0.0, 0
            for s in seeds:
                r = results[s][arm]
                convs = list(r["conversions"].values())
                total += len(convs)
                res = ledger.attribute(convs, r["sightings"], r["family"], method,
                                       keep=lambda f: f in FAMILIES)
                unexposed += res["unexposed"]
                for k, c in res["by_artifact"].items():
                    by_artifact[k] = by_artifact.get(k, 0) + c
                for post, c in res["by_post"].items():
                    kind = "artifact" if post in ARTIFACTS else ("reader's own post naming the product" if r["family"].get(post) == WOM else "reader's reply or quote")
                    by_kind[kind] = by_kind.get(kind, 0) + c
                if method == "linear":
                    for row in res["rows"]:
                        conversion_rows.append({"seed": s, "arm": arm, "person": row["person"], "cohort": cohort_of(row["person"]),
                                                "round": row["round"], "via": row["via"],
                                                "credits": json.dumps({k: round(v, 4) for k, v in row["credits"].items()})})
            lift = measured["arms"][arm]["intent"] - measured["arms"]["control"]["intent"]
            anchored, info = ledger.anchor(by_artifact, lift)
            per_method[method] = {"conversions": total, "credited": sum(by_artifact.values()), "unexposed": unexposed,
                                  "by_artifact": by_artifact, "by_post_kind": by_kind,
                                  "anchored_to_lift": anchored, "anchor": info}
        derived["attribution"][arm] = per_method
        # Channel split of linear credit: how each credited post first reached the reader.
        chan = {}
        for s in seeds:
            r = results[s][arm]
            first = {(x["person"], x["post"]): x["channel"] for x in r["exposures"]}
            res = ledger.attribute(list(r["conversions"].values()), r["sightings"], r["family"], "linear",
                                   keep=lambda f: f in FAMILIES)
            for row in res["rows"]:
                for post, c in row["credits"].items():
                    ch = first.get((row["person"], post), "unknown")
                    chan[ch] = chan.get(ch, 0) + c
        derived["channels"][arm] = chan
    # Exposure channels of all first sightings (not only converters).
    derived["exposure_channels"] = {arm: {} for arm in ARMS}
    for row in exposure_rows:
        d = derived["exposure_channels"][row["arm"]]
        d[row["channel"]] = d.get(row["channel"], 0) + 1

    # Economics: assumed parameters times derived lifts.
    product = assumptions["product"]
    economics = {"value_per_intent_reader": {}, "clv_per_rollout": {}, "arms": {}, "sensitivity": {}}
    for name, a in assumptions["sets"].items():
        economics["value_per_intent_reader"][name] = value_per_intent(a, product)
        economics["clv_per_rollout"][name] = clv(a, product)
    for arm in ARMS[1:]:
        cost = cost_of(arm, assumptions["costs"])
        lift = derived["lift"][arm]["intent"]
        rows = {}
        for name, a in assumptions["sets"].items():
            k = a["readers_represented_per_panel_member"] * len(people)
            val = economics["value_per_intent_reader"][name]
            ltv = lift["per_member"] * len(people) * a["readers_represented_per_panel_member"] * val
            lo = lift["lo90"] * len(people) * a["readers_represented_per_panel_member"] * val
            hi = lift["hi90"] * len(people) * a["readers_represented_per_panel_member"] * val
            rows[name] = {"audience_readers": k, "brain_ltv": ltv, "brain_ltv_lo90": lo, "brain_ltv_hi90": hi,
                          "per_1000_readers": lift["per_member"] * 1000 * val,
                          "cost": cost, "net": ltv - cost, "net_lo90": lo - cost, "net_hi90": hi - cost,
                          "roi": (ltv - cost) / cost if cost else None}
        economics["arms"][arm] = rows
    # One-at-a-time sensitivity of the article's base brain-LTV.
    base = assumptions["sets"]["base"]
    lift_article = derived["lift"]["article"]["intent"]
    def ltv_with(a, lift_per_member):
        return lift_per_member * len(people) * a["readers_represented_per_panel_member"] * value_per_intent(a, product)
    center = ltv_with(base, lift_article["per_member"])
    tornado = [{"parameter": "intent lift (measured, 90% interval)", "low": ltv_with(base, lift_article["lo90"]),
                "high": ltv_with(base, lift_article["hi90"])}]
    for key in base:
        lo = dict(base, **{key: assumptions["sets"]["conservative"][key]})
        hi = dict(base, **{key: assumptions["sets"]["optimistic"][key]})
        tornado.append({"parameter": key, "low": ltv_with(lo, lift_article["per_member"]),
                        "high": ltv_with(hi, lift_article["per_member"])})
    tornado.sort(key=lambda t: -abs(t["high"] - t["low"]))
    economics["sensitivity"] = {"arm": "article", "set": "base", "center": center, "rows": tornado}

    # Example trajectories from the traces: what intent readers saw and did.
    examples = []
    for s in seeds:
        for arm in ARMS[1:]:
            r = results[s][arm]
            for c in r["conversions"].values():
                seen = [x for x in r["exposures"] if x["person"] == c["person"] and ledger.order_key(x) <= (c["round"], c["_index"])]
                acts = [e for e in r["events"] if e["person"] == c["person"] and (e.get("family") in FAMILIES or e.get("target") in COMPANY)]
                examples.append({"seed": s, "arm": arm, "person": c["person"], "cohort": cohort_of(c["person"]),
                                 "converted_round": c["round"], "via": c["via_all"],
                                 "saw": [{"round": x["round"], "post": x["post"], "channel": x["channel"]} for x in seen],
                                 "did": [{"round": e["round"], "kind": e["kind"], "target": e.get("target"),
                                          "text": e.get("text", "")} for e in acts]})
    intent_texts = [{"seed": s, "arm": arm, "person": e["person"], "kind": e["kind"], "text": e["text"],
                     "classified_intent": bool(REFERENCE.search(e["text"] or "") and INTENT.search(e["text"] or "")),
                     "confabulated": bool(REFERENCE.search(e["text"] or "") and CONFAB.search(e["text"] or ""))}
                    for s in seeds for arm in ARMS for e in results[s][arm]["events"]
                    if e["kind"] in ("post", "reply", "quote") and (REFERENCE.search(e["text"] or "") or e.get("family") in FAMILIES)]

    # A cascade for the figure: one seed's article-critique and article arms, every
    # member's first sighting, its channel, and what they did.
    cascades = {}
    for s in seeds:
        cascades[str(s)] = {}
        for arm in ARMS[1:]:
            r = results[s][arm]
            nodes = {}
            for x in r["exposures"]:
                n = nodes.setdefault(x["person"], {"person": x["person"], "cohort": cohort_of(x["person"]),
                                                   "first_round": x["round"], "channel": x["channel"], "via": x["via"],
                                                   "post": x["post"], "actions": []})
            for e in r["events"]:
                if e["person"] in nodes and (e.get("family") in FAMILIES or e.get("target") in COMPANY):
                    nodes[e["person"]]["actions"].append({"round": e["round"], "kind": e["kind"], "target": e.get("target")})
            for p, c in r["conversions"].items():
                nodes.setdefault(p, {"person": p, "cohort": cohort_of(p), "first_round": None, "channel": "not exposed",
                                     "via": None, "post": None, "actions": []})["converted"] = c["round"]
            cascades[str(s)][arm] = {"nodes": list(nodes.values()),
                                     "member_posts": [{"post": pid, "author": p["author"], "family": r["family"].get(pid),
                                                       "round": p.get("round"), "text": p["text"]}
                                                      for pid, p in r["posts"].items()
                                                      if r["family"].get(pid) in FAMILIES and pid not in ARTIFACTS]}

    # Factor ablation (factors.ablation): does knowing what a reader saw and did with
    # campaign posts predict who converts, beyond their cohort? Held out by seed,
    # bootstrapped by reader. Rows: every reader-run in a publishing arm.
    rows = []
    for s in seeds:
        for arm in ARMS[1:]:
            for p in people:
                o = table[s][arm][p]
                rows.append({"seed": s, "person": p, "intent": o["intent"],
                             "cohort_practitioner": int(cohort_of(p) == "practitioners"),
                             "cohort_buyer": int(cohort_of(p) == "buyers"),
                             "artifact_exposure": o["reached"], "artifact_impressions": o["impressions"],
                             "exposure_via_repost": o["via_repost"], "artifact_engagement": o["campaign_engagements"]})
    baseline = ["cohort_practitioner", "cohort_buyer"]
    candidates = ["artifact_exposure", "artifact_impressions", "exposure_via_repost", "artifact_engagement"]
    if sum(r["intent"] for r in rows) >= 2:
        abl = factors.ablation(rows, "intent", baseline, baseline + candidates, "seed", "person", reps=1000, seed=1)
        inc = factors.incremental_value(rows, "intent", baseline, candidates, "seed", "person", reps=500, seed=1)
        derived["factor_ablation"] = {"outcome": "commercial_intent_reader", "rows": len(rows),
                                      "positives": sum(r["intent"] for r in rows), "baseline": baseline,
                                      "expanded": baseline + candidates, "held_out_by": "seed", "cluster": "reader",
                                      "result": {k: v for k, v in abl.items()}, "incremental": inc}

    summary = {
        "study": "press-brain-ltv",
        "generated_by": "research/studies/market-perception/press-brain-ltv/analyze.py",
        "preregistration_sha256": hashlib.sha256((HERE / "preregistration.json").read_bytes()).hexdigest(),
        "spec_sha256": hashlib.sha256((HERE / "spec/launch-campaign.json").read_bytes()).hexdigest(),
        "assumptions_sha256": hashlib.sha256((HERE / "assumptions.json").read_bytes()).hexdigest(),
        "design": {"seeds": seeds, "arms": list(ARMS), "panel": len(people), "units_per_arm": n_units,
                   "cohort_sizes": {c: sum(1 for p in people if cohort_of(p) == c) for c in COHORTS},
                   "rounds": spec["clock"]["rounds"], "bootstrap_resamples": BOOT},
        "runs": run_meta,
        "measured": measured,
        "derived": derived,
        "assumed": {"parameters": assumptions, "economics": economics},
        "examples": examples,
        "member_posts_about_campaign": intent_texts,
        "cascades": cascades,
    }
    (out / "summary.json").write_text(json.dumps(summary, indent=1, sort_keys=True) + "\n")
    for name, rows in (("events.csv", event_rows), ("exposures.csv", exposure_rows), ("conversions.csv", conversion_rows)):
        with open(out / name, "w", newline="") as f:
            if rows:
                w = csv.DictWriter(f, fieldnames=list(rows[0]))
                w.writeheader()
                w.writerows(rows)
    with open(out / "arms.csv", "w", newline="") as f:
        w = csv.writer(f)
        w.writerow(["arm", "units", "intent_readers", "intent_lift_per_member", "lift_lo90", "lift_hi90", "reached",
                    "impressions", "follow_company", "campaign_engagements", "ambient_likes", "cost_usd",
                    "brain_ltv_base_usd", "net_base_usd", "roi_base"])
        for arm in ARMS:
            m = measured["arms"][arm]
            l = derived["lift"].get(arm, {}).get("intent", {})
            e = economics["arms"].get(arm, {}).get("base", {})
            w.writerow([arm, m["units"], m["intent"], round(l.get("per_member", 0), 4), round(l.get("lo90", 0), 4),
                        round(l.get("hi90", 0), 4), m["reached"], m["impressions"], m["follow_company"],
                        m["campaign_engagements"], m["ambient_likes"], e.get("cost", 0), round(e.get("brain_ltv", 0)),
                        round(e.get("net", 0)), None if e.get("roi") is None else round(e["roi"], 3)])
    print(json.dumps({a: {"intent": measured["arms"][a]["intent"], "reached": measured["arms"][a]["reached"],
                          "lift": derived["lift"].get(a, {}).get("intent")} for a in ARMS}, indent=1, default=str))
    return 0


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