Delayed conversion credit for recommendations

"""Delayed conversion credit: which exposure most likely earned this purchase?

Simulates a shop whose recommender shows items to 1,800 users on five surfaces
(home, search, email, detail, push) over 60 days. Every exposure is an event with
features surface:<s>, action:<impression|click|save|cart>, category:<c> and
pos:<top|mid|low>. Purchases arrive hours to weeks later. HeapDB runs as a
sidecar: it learns from purchases without labels (observe), from exposure
windows that pass without a purchase (settle), and from the few purchases that
carry a tracked email coupon code (feedback).

Hidden ground truth (per exposure, noisy-OR; lag is LogNormal, median 3 days):

* email digest, opened (surface:email + action:click): 0.10
* email digest, delivered but not opened (action:impression): 0.008
* saved-for-later reminder (surface:push + action:click): 0.04
* item detail page clicks: 0.012; home impressions: 0.001
* search: 0.0. Search is navigation: before most purchases the user searches
  for the item they already decided to buy, 5 minutes to 3 hours earlier.

So last-click credit goes to search. That is the confounder this example shows.

Run:
    pip install heapdb
    python 0003-delayed-conversion-credit.py
"""

from __future__ import annotations

import math
import random

from heapdb import HeapDB, LogNormalKernel, format_result

M, H, D = 60.0, 3600.0, 86400.0
SEED = 3
N_USERS = 1800
DAYS = 60

SURFACES = ["home", "search", "email", "detail", "push"]
CATEGORIES = ["shoes", "kitchen", "books", "toys", "garden", "audio"]
ACTION_WEIGHT = {"impression": 0.3, "click": 1.0, "save": 1.0, "cart": 1.2}


def truth(surface: str, action: str) -> float:
    """Hidden probability that one exposure causes a purchase."""
    if surface == "email":
        return 0.10 if action == "click" else 0.008
    if surface == "push":
        return 0.04 if action == "click" else 0.0
    if surface == "detail":
        return 0.012
    if surface == "home" and action == "impression":
        return 0.001
    return 0.0


BACKGROUND_PER_USER = 0.25   # purchases with no exposure as cause
NAV_SEARCH_P = 0.75          # share of purchases preceded by a navigational search
COUPON_P = 0.20              # share of email-caused purchases with a tracked coupon


def true_lag(rng: random.Random) -> float:
    return math.exp(rng.gauss(math.log(3 * D), 0.5))


def exposure(rng: random.Random, t: float) -> tuple[float, str, str, str, str]:
    s = rng.choices(SURFACES, weights=[40, 22, 8, 22, 8])[0]
    if s == "email":
        a = "click" if rng.random() < 0.25 else "impression"
    elif s == "push":
        a = "click" if rng.random() < 0.3 else "impression"
    elif s == "search":
        a = rng.choices(["click", "impression"], weights=[6, 4])[0]
    elif s == "detail":
        a = rng.choices(["click", "save", "cart"], weights=[7, 2, 1])[0]
    else:
        a = rng.choices(["impression", "click", "save"], weights=[8, 2, 1])[0]
    pos = rng.choices(["top", "mid", "low"], weights=[3, 4, 3])[0]
    return t, s, a, rng.choice(CATEGORIES), pos


def simulate(rng: random.Random):
    """Per user: exposures (t, surface, action, category, pos) and purchases (t, cause_index)."""
    users = {}
    for u in range(N_USERS):
        exps = []
        for _ in range(rng.randint(10, 40)):
            exps.append(exposure(rng, rng.uniform(0, DAYS * D)))
        for day in range(0, DAYS, 7):          # weekly email digest
            if rng.random() < 0.7:
                t = (day + rng.uniform(0, 1)) * D
                exps.append((t, "email", "click" if rng.random() < 0.3 else "impression",
                             rng.choice(CATEGORIES), "top"))
        purchases = []
        for i, (t, s, a, _c, _p) in enumerate(exps):
            if rng.random() < truth(s, a):
                purchases.append((t + true_lag(rng), i))
        if rng.random() < BACKGROUND_PER_USER:
            purchases.append((rng.uniform(5 * D, DAYS * D), None))
        purchases = [p for p in purchases if p[0] < DAYS * D]
        for tp, _ in purchases:                 # navigation: search for the thing
            if rng.random() < NAV_SEARCH_P:
                exps.append((tp - rng.uniform(5 * M, 3 * H), "search", "click",
                             rng.choice(CATEGORIES), "top"))
        users[f"user-{u:04d}"] = (exps, sorted(purchases))
    return users


def surf(ev) -> str:
    """Surface of an event, or 'none'."""
    if ev is None:
        return "none"
    return next(f for f in ev.features if f.startswith("surface:"))[len("surface:"):]


def main() -> None:
    rng = random.Random(SEED)
    users = simulate(rng)

    db = HeapDB(bucket_seconds=12 * H)
    db.register_event_type("exposure", weight=lambda a: ACTION_WEIGHT[a["action"]])
    db.register_effect("purchase", LogNormalKernel(median=3 * D, sigma=0.6, tail_sigmas=3.0),
                       default_prior=(1, 9), leak=0.05)

    ids: dict[str, list] = {}      # user -> event per exposure index (insertion order)
    n_ev = 0
    for user, (exps, _) in users.items():
        order = sorted(range(len(exps)), key=lambda i: exps[i][0])
        evs = [None] * len(exps)
        for i in order:
            t, s, a, c, p = exps[i]
            evs[i] = db.insert(user, t, "exposure",
                               {f"surface:{s}", f"action:{a}", f"category:{c}", f"pos:{p}"},
                               attrs={"action": a})
            n_ev += 1
        ids[user] = evs

    all_purchases = sorted((tp, user, cause) for user, (_, ps) in users.items() for tp, cause in ps)
    for tp, user, _ in all_purchases:
        db.observe(user, "purchase", tp)
    settled = db.settle(DAYS * D + 30 * D)

    # Hard links: a tracked email coupon code ties some purchases to one email.
    n_fb = 0
    for tp, user, cause in all_purchases:
        if cause is not None and users[user][0][cause][1] == "email" and rng.random() < COUPON_P:
            db.feedback(user, "purchase", tp, confirmed_event_id=ids[user][cause].id)
            n_fb += 1

    seeded = [(tp, u, c) for tp, u, c in all_purchases if c is not None]
    print("SCENARIO")
    print(f"  {N_USERS:,} users, {n_ev:,} exposures over {DAYS} days, {len(all_purchases):,} purchases")
    print(f"  ({len(seeded):,} caused by an exposure, the rest background); "
          f"{n_fb} confirmed by coupon code")
    print(f"  settle: {settled.events_rejected:,} exposures passed their window with no purchase")

    # Last-click vs HeapDB top-1 on purchases whose true cause is known.
    last_ok = top_ok = 0
    last_by: dict[str, int] = {}
    top_by: dict[str, int] = {}
    true_by: dict[str, int] = {}
    pick = None
    for tp, user, cause in seeded:
        evs = ids[user]
        true_ev = evs[cause]
        clicks = [e for e in evs if e.time < tp and e.attrs["action"] != "impression"]
        last = max(clicks, key=lambda e: e.time) if clicks else None
        res = db.top_causes(user, "purchase", tp, k=3)
        n_cand = sum(c.score >= 0.01 for c in res)
        top = res.top.event if res.top else None
        last_by[surf(last)] = last_by.get(surf(last), 0) + 1
        top_by[surf(top)] = top_by.get(surf(top), 0) + 1
        true_by[surf(true_ev)] = true_by.get(surf(true_ev), 0) + 1
        last_ok += last is not None and last.id == true_ev.id
        top_ok += top is not None and top.id == true_ev.id
        if (pick is None and n_cand == 3 and top is not None and top.id == true_ev.id
                and surf(last) == "search" and surf(true_ev) == "email"):
            pick = (user, tp, true_ev)

    print()
    print("ONE CASE, EXPLAINED")
    user, tp, true_ev = pick
    print(f"  {user} bought at day {tp / D:.1f}; true cause: #{true_ev.id} "
          f"(email, {(tp - true_ev.time) / D:.1f} days earlier); last click was a search")
    print(format_result(db.top_causes(user, "purchase", tp, k=3, explain=True)))

    print()
    print("ACROSS ALL CASES")
    print(f"  credited surface, {len(seeded):,} seeded purchases:  true   last-click   HeapDB top-1")
    for s in ["email", "push", "detail", "home", "search"]:
        print(f"    {s:<10} {true_by.get(s, 0):>34} {last_by.get(s, 0):>12} {top_by.get(s, 0):>14}")
    rep = db.link_report("purchase", min_support=20,
                         features=[f"surface:{s}" for s in SURFACES]
                         + [f"action:{a}" for a in ACTION_WEIGHT])
    print(rep)

    print()
    print("CHECK")
    surf_rank = [r.feature for r in rep if r.feature.startswith("surface:")]
    last_share, top_share = last_ok / len(seeded), top_ok / len(seeded)
    print(f"  true causal exposure = last click in {last_share:.1%} of seeded purchases; "
          f"= HeapDB top-1 in {top_share:.1%}")
    print(f"  surface:email ranks #{surf_rank.index('surface:email') + 1} of "
          f"{len(surf_rank)} surfaces; surface:search ranks "
          f"#{surf_rank.index('surface:search') + 1} (last-click credits it most)")
    if (surf_rank[0] != "surface:email" or top_share <= last_share
            or max(last_by, key=last_by.get) != "search"):
        raise SystemExit(1)


if __name__ == "__main__":
    main()