Common point of compromise

"""Common point of compromise: where was the card data stolen?

Simulates one card issuer's portfolio: 1,500 cards making legitimate purchases
for 38 days at a grocery chain, a gas station chain, a big online marketplace
and about 120 smaller merchants. Fraud on a card shows up days to weeks after
its data was stolen. HeapDB runs as a sidecar to the fraud case system: every
new fraud card is observe()d, clean cards are settle()d, and a few cards whose
source an investigator confirmed get feedback(). On "today" (day 38) we ask
which merchant is the common point, then score the cards exposed there that
have not shown fraud yet, and check them against the rest of the simulation.

Hidden ground truth (the seeded cause):

* merchant:trailhead-outfitters (e-commerce, mcc:5941) is breached from day 14
  to day 28 (June 15-29). Each purchase there in that window is stolen with
  probability 0.6. First fraud follows after a log-normal lag (median 10 days,
  sigma 0.6): data is sold and used in batches.
* Distractors: almost every card shops at merchant:freshmart (grocery),
  merchant:fuelstop (gas) and merchant:megamart (online marketplace). They are
  innocent but appear in nearly every fraud card's history.
* Near-miss: merchant:peak-supply sells to the same outdoor customers, uses the
  same payment processor and is often visited on the same day. It is innocent.
* Noise: 2% of cards get fraud for unrelated reasons (lost, stolen, phishing)
  at random times.

Run:
    pip install heapdb
    python 0002-common-point-of-compromise.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 = 11
N_CARDS = 1500
NOW = 38                  # the investigation runs on day 38
HORIZON = 100             # the simulation keeps running (truth only) to day 100
T0 = 1780272000.0         # 2026-06-01T00:00:00Z; simulated times are seconds after T0

BREACHED = "merchant:trailhead-outfitters"
NEAR_MISS = "merchant:peak-supply"
POPULAR = ["merchant:freshmart", "merchant:fuelstop", "merchant:megamart"]
BREACH = (14 * D, 28 * D)
STEAL = 0.6               # P(card data stolen | purchase at the breached merchant in the window)
BACKGROUND = 0.02         # P(fraud for unrelated reasons) per card over the 100 days
N_CONFIRMED = 5           # fraud cards whose source an investigator confirms
CHECKPOINTS = (10, 25, 35, 45)
FLAG = 0.02               # risk threshold for proactive action on an exposed card

# merchant -> (mcc, channel, visits per day for a typical card)
MERCHANTS = {
    "merchant:freshmart": ("mcc:5411", "card-present", 0.25),
    "merchant:fuelstop": ("mcc:5541", "card-present", 0.15),
    "merchant:megamart": ("mcc:5399", "ecommerce", 0.10),
    BREACHED: ("mcc:5941", "ecommerce", 0.0),
    NEAR_MISS: ("mcc:5941", "ecommerce", 0.0),
}
MCCS = ["mcc:5812", "mcc:5814", "mcc:5651", "mcc:5732", "mcc:5942", "mcc:7832", "mcc:5912",
        "mcc:4121"]
TAIL = [f"merchant:m{i:03d}" for i in range(120)]
for i, m in enumerate(TAIL):
    MERCHANTS[m] = (MCCS[i % len(MCCS)], "ecommerce" if i % 3 == 0 else "card-present", 0.0)


def fraud_lag(rng: random.Random) -> float:
    return math.exp(rng.gauss(math.log(10 * D), 0.6))


def simulate(rng: random.Random):
    """Return (transactions to HORIZON, first-fraud time per card, theft time per card)."""
    txns, first_fraud, stolen = [], {}, {}
    for c in range(N_CARDS):
        card = f"card-{c:04d}"
        outdoor = rng.random() < 0.12          # shops at outdoor stores
        favorites = rng.sample(TAIL, 8)
        fraud = math.inf
        for day in range(HORIZON):
            visits = [m for m in POPULAR if rng.random() < MERCHANTS[m][2]]
            if rng.random() < 0.35:
                visits.append(rng.choice(favorites))
            if rng.random() < (0.06 if outdoor else 0.002):
                visits.append(BREACHED)
                if rng.random() < 0.6:          # same-day co-visit: the near-miss
                    visits.append(NEAR_MISS)
            elif rng.random() < (0.05 if outdoor else 0.002):
                visits.append(NEAR_MISS)
            for m in visits:
                t = day * D + rng.uniform(7, 22) * H
                mcc, channel, _ = MERCHANTS[m]
                txns.append((t, card, {m, mcc, f"channel:{channel}"}))
                if m == BREACHED and BREACH[0] <= t < BREACH[1] and rng.random() < STEAL:
                    stolen.setdefault(card, t)
                    fraud = min(fraud, t + fraud_lag(rng))
        if rng.random() < BACKGROUND:
            fraud = min(fraud, rng.uniform(0, HORIZON) * D)
        if fraud < HORIZON * D:
            first_fraud[card] = fraud
    txns.sort(key=lambda x: x[0])
    return txns, first_fraud, stolen


def make_db() -> HeapDB:
    db = HeapDB(bucket_seconds=2 * D)
    # Compromise-to-fraud lag: LogNormal, median 10 days, peak near 7 days, long tail
    # (data is resold and used weeks later). Truncated at median * exp(3 sigma), about
    # 60 days. One first fraud per card (repeatable=False). global_pseudocount=1e6 lets
    # every card borrow the whole portfolio's evidence: the CPP case-control view.
    db.register_effect("first_fraud", LogNormalKernel(median=10 * D, sigma=0.6, tail_sigmas=3.0),
                       leak=0.3, repeatable=False, cause_types={"txn"},
                       global_pseudocount=1e6)
    return db


def merchant_report(db: HeapDB, min_support: float = 20):
    return db.link_report("first_fraud", features=[m for m in MERCHANTS], min_support=min_support)


def main() -> None:
    rng = random.Random(SEED)
    txns, first_fraud, stolen = simulate(rng)
    db = make_db()
    now = NOW * D
    stream = [(t, 0, (card, feats)) for t, card, feats in txns if t < now]
    stream += [(t, 1, card) for card, t in first_fraud.items() if t < now]
    stream.sort(key=lambda x: (x[0], x[1]))

    by_card: dict[str, list] = {}
    cases, confirmed, snapshots, day = [], [], [], 0
    for t, kind, item in stream:
        while t >= (day + 1) * D:              # nightly: settle windows that fully passed
            day += 1
            db.settle(T0 + day * D)
        if kind == 0:
            card, feats = item
            by_card.setdefault(card, []).append(db.insert(card, T0 + t, "txn", feats))
            continue
        card = item
        db.observe(card, "first_fraud", T0 + t)
        cases.append(card)
        # From case 30 on, investigators confirm the source of a few cards (the merchant
        # discloses the breach, or forensics ties the card to it). The popular merchants
        # those cards used in the window are cleared at the same time.
        visit = [e for e in by_card.get(card, []) if BREACHED in e.features
                 and BREACH[0] <= e.time - T0 < BREACH[1] and e.time < T0 + t]
        if visit and len(cases) >= 30 and len(confirmed) < N_CONFIRMED:
            cleared = [e.id for e in by_card[card] if e.features & set(POPULAR)
                       and 0 < T0 + t - e.time < 30 * D]
            db.feedback(card, "first_fraud", T0 + t, confirmed_event_id=visit[-1].id,
                        rejected_ids=cleared)
            confirmed.append(card)
        if len(cases) in CHECKPOINTS:
            db.settle(T0 + t, partial=True)
            feats = [r.feature for r in merchant_report(db, min_support=3)]
            snapshots.append((len(cases), t, feats.index(BREACHED) + 1, feats.index(NEAR_MISS) + 1,
                              len(feats)))
    db.settle(T0 + now, partial=True)
    rep = merchant_report(db)
    feats = [r.feature for r in rep]
    snapshots.append((len(cases), now, feats.index(BREACHED) + 1, feats.index(NEAR_MISS) + 1,
                      len(feats)))

    print("SCENARIO")
    print(f"  {N_CARDS} cards, {NOW} days, {len(stream) - len(cases):,} legitimate transactions")
    print(f"  {len(cases)} cards with fraud so far, {len(confirmed)} with a source confirmed "
          f"by investigators")
    share = []
    for m in [BREACHED] + POPULAR:
        n = sum(any(m in e.features for e in by_card.get(c, []) if e.time < T0 + first_fraud[c])
                for c in cases)
        share.append(f"{m.split(':')[1]} {n}")
    print(f"  fraud cards that used each merchant before their fraud:\n    {', '.join(share)}")

    print()
    print("STANDING QUERY: SUSPECT MERCHANTS AS FRAUD CARDS ARRIVE")
    for n, t, r_b, r_n, tot in snapshots:
        print(f"  {n:>3} cards, day {t / D:4.1f}: trailhead-outfitters #{r_b:<3} "
              f"peak-supply #{r_n:<3} of {tot}")

    # One card whose data was stolen at the breach (truth), with no confirmed source, a
    # visit to the near-miss merchant in its window, and the theft visit ranked first.
    for pick in sorted(c for c in cases if c in stolen and c not in confirmed):
        res = db.top_causes(pick, "first_fraud", T0 + first_fraud[pick], k=3, explain=True)
        if (len(res.causes) == 3 and any(NEAR_MISS in c.event.features for c in res)
                and abs(res.top.event.time - T0 - stolen[pick]) < 1):
            break
    print()
    print(f"ONE CARD, EXPLAINED ({pick}: data stolen day {stolen[pick] / D:.1f}, "
          f"first fraud day {first_fraud[pick] / D:.1f})")
    print(format_result(res))

    print()
    print("ACROSS ALL FRAUD CARDS (merchant features, global links)")
    lines = str(rep).splitlines()
    print("\n".join(lines[:9]))
    for f in POPULAR:
        print(f"  ... {f} #{feats.index(f) + 1}, lift {rep.row(f).lift:.2f}")

    # Exposure window: the middle 80% of the suspect's visits that are the top cause of a
    # fraud card (the tails are timing noise and unrelated fraud).
    top = feats[0]
    tops = []
    for c in cases:
        best = db.top_causes(c, "first_fraud", T0 + first_fraud[c], k=1).top
        if best is not None and top in best.event.features:
            tops.append(best.event.time - T0)
    tops.sort()
    lo, hi = tops[len(tops) // 10], tops[-1 - len(tops) // 10]
    kern = db.effect("first_fraud").kernel
    exposed, flagged = [], []
    for card, evs in by_card.items():
        if card in first_fraud and first_fraud[card] < now:
            continue
        visits = [e for e in evs if top in e.features and lo <= e.time - T0 <= hi]
        if not visits:
            continue
        exposed.append(card)
        # Risk still ahead, per visit: learned link (feature_risk) x share of the lag kernel
        # not yet elapsed; visits combine by noisy-OR.
        miss = 1.0
        for e in visits:
            miss *= 1.0 - (db.feature_risk("first_fraud", e.features, entity=card, type="txn")
                           * (1.0 - kern.mass_fraction(T0 + now - e.time)))
        if 1.0 - miss >= FLAG:
            flagged.append(card)
    later = [c for c in first_fraud if first_fraud[c] >= now]
    rest = [c for c in by_card if c not in exposed and not (c in first_fraud
                                                             and first_fraud[c] < now)]
    print()
    print(f"PROACTIVE: CARDS EXPOSED AT {top.split(':')[1]} (day {lo / D:.1f}-{hi / D:.1f}), "
          "NO FRAUD YET")
    print(f"  {len(exposed)} exposed cards; {len(flagged)} flagged at risk >= {FLAG} "
          "(feature_risk x lag mass still ahead)")
    groups = [("flagged", flagged), ("exposed, not flagged", [c for c in exposed if c not in flagged]),
              ("all other clean cards", rest)]
    print(f"  fraud later (simulated to day {HORIZON}):")
    for name, g in groups:
        n = sum(c in later for c in g)
        print(f"    {name:<24} {n:>3}/{len(g):<5} ({n / len(g):.0%})")

    r_b, r_n = feats.index(BREACHED) + 1, feats.index(NEAR_MISS) + 1
    print()
    print("CHECK")
    print(f"  seeded breach trailhead-outfitters ranks #{r_b} of {len(feats)} merchants "
          f"(lift {rep.row(BREACHED).lift:.2f})")
    print(f"  near-miss peak-supply #{r_n} (lift {rep.row(NEAR_MISS).lift:.2f}); popular merchants "
          + ", ".join(f"#{feats.index(f) + 1}" for f in POPULAR))
    print(f"  estimated exposure window day {lo / D:.1f}-{hi / D:.1f} vs seeded "
          f"day {BREACH[0] / D:.0f}-{BREACH[1] / D:.0f}")
    if r_b != 1 or r_n == 1 or top != BREACHED:
        raise SystemExit(1)


if __name__ == "__main__":
    main()