Exposure attribution for creative work

"""Exposure attribution for creative work: which send most likely leaked this track?

Simulates one year of pre-release promotion at a music label: 200 tracks, each
sent before release to a promo pool (DJs), press, playlist curators and sync
agents, plus public teaser snippets on social media. Every send is an exposure
event of the track. Features encode only risk-bearing PRESENCE: download:on,
format:wav (full quality), expiry:none, watermark:off and audience:large. A
watermarked 14-day stream to a few people carries no risk feature. Recipient
group, file version (asset: full, edit, stems, snippet) and the watermark flag
are attrs, not features. Weight is the log-scaled audience size, 1 + 0.5 * log10(recipients).

A monitoring service finds unauthorized copies on the platforms it covers. It
also tells which file version (asset) the copy matches. HeapDB ranks the
earlier sends of that track; a bridge outside the core keeps only candidates
of the matched asset.

Hidden ground truth (per send, noisy-OR, then scaled by the send's weight):

* Seeded risky settings: download:on 0.06, format:wav 0.05, expiry:none 0.04;
  watermark:off 0.02. Any send also has a base 0.01 (screen or line recording).
* Confounder: audience:large has no effect of its own (audience size is already
  in the weight), but most large sends are promo-pool sends of full WAVs,
  downloadable, with no expiry.
* Near-miss: every format:wav send is also download:on, so feedback cannot
  fully separate the two.
* Teaser snippets (type post) are public on purpose and never leak a full copy;
  a rule scores them 0.
* Noise: 10% of monitored tracks also leak from an untracked source (a studio,
  a laptop); 25% of leaks land on platforms the watcher does not cover; 15% of
  tracks are not monitored at all.

Learning: a per-recipient watermark (canary) survives in 55% of leaks from
watermarked sends and confirms the source (feedback). Only watermarked sends
can be confirmed this way. Every found copy is observe()d. Sends whose window
passed with no copy are settle()d, only for monitored tracks, with
w_neg = 0.75 (the watcher's platform coverage).

Run:
    pip install heapdb
    python 0005-creative-exposure-attribution.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 = 5
N_TRACKS = 200
T0 = 1767571200.0            # 2026-01-05T00:00:00Z; simulation times are seconds after T0

TRUTH = {"download:on": 0.06, "format:wav": 0.05, "expiry:none": 0.04, "watermark:off": 0.02}
BASE = 0.01                  # any send: a screen or line recording
SEEDED = ("download:on", "expiry:none", "format:wav")
NO_EFFECT = "audience:large"
UNTRACKED = 0.10             # P(leak from an untracked source) per monitored track
COVERAGE = 0.75              # share of leak destinations the watcher covers
MONITORED = 0.85             # share of tracks enrolled in monitoring
CANARY_SURVIVES = 0.55       # P(per-recipient watermark survives in a leaked copy)


def weight(recipients: int) -> float:
    return 1.0 + 0.5 * math.log10(recipients)


def send(t, group, asset, fmt, wm, expiry, n):
    """One exposure: (time, type, presence features, weight, attrs)."""
    feats = set()
    if fmt != "stream_only":
        feats.add("download:on")
    if fmt == "wav":
        feats.add("format:wav")
    if expiry == "none":
        feats.add("expiry:none")
    if not wm:
        feats.add("watermark:off")
    if 50 < n <= 5000:
        feats.add("audience:large")
    return (t, "post" if group == "social" else "send", feats, weight(n),
            {"group": group, "asset": asset, "watermarked": wm})


def campaign(rng: random.Random, release: float) -> list:
    """Pre-release sends of one track, 25-45 days before release."""
    out = []
    def when():
        return release - rng.uniform(25, 45) * D
    for _ in range(rng.choice([1, 1, 2])):          # promo pool: hundreds of DJs
        n = rng.randint(200, 800)
        r = rng.random()
        if r < 0.65:
            out.append(send(when(), "promo_pool", rng.choice(["full", "edit"]), "wav", True, "none", n))
        elif r < 0.85:
            out.append(send(when(), "promo_pool", "full", "stream_only", True, "14d", n))
        else:
            out.append(send(when(), "promo_pool", "full", "mp3", True, "14d", n))
    for _ in range(rng.randint(2, 5)):              # press
        fmt = "mp3" if rng.random() < 0.5 else "stream_only"
        out.append(send(when(), "press", "full", fmt, rng.random() < 0.7,
                        rng.choice(["none", "14d"]), rng.randint(3, 80)))
    for _ in range(rng.randint(3, 8)):              # curators, one by one
        fmt = "mp3" if rng.random() < 0.25 else "stream_only"
        out.append(send(when(), "curator", rng.choice(["full", "edit"]), fmt, True,
                        "14d" if rng.random() < 0.6 else "none", 1))
    for _ in range(rng.randint(0, 2)):              # sync agents: WAV stems, small audience
        out.append(send(when(), "sync", rng.choice(["full", "stems"]), "wav", rng.random() < 0.5,
                        rng.choice(["none", "14d"]), rng.randint(2, 6)))
    for _ in range(rng.randint(1, 2)):              # public teaser snippets
        out.append(send(when(), "social", "snippet", "stream_only", False, "none", 50000))
    return sorted(out, key=lambda s: s[0])


def p_leak(feats, w) -> float:
    miss = 1.0 - BASE
    for f in feats:
        miss *= 1.0 - TRUTH.get(f, 0.0)
    return min(0.95, w * (1.0 - miss))


def leak_lag(rng: random.Random) -> float:
    """Send -> copy found by the watcher: the leak itself plus detection delay."""
    return math.exp(rng.gauss(math.log(4 * D), 0.6)) + rng.expovariate(1 / (12 * H))


def main() -> None:
    rng = random.Random(SEED)
    db = HeapDB(bucket_seconds=2 * D)
    db.register_effect("leak_prerelease", LogNormalKernel(median=4 * D, sigma=0.6, tail_sigmas=3.0),
                       default_prior=(1, 20), leak=0.02, w_neg=COVERAGE)
    # A public teaser snippet cannot be the source of a full-length copy.
    db.add_rule("leak_prerelease", "public snippet", lambda e: e.attrs.get("asset") == "snippet", 0.0)

    tracks, copies = {}, []          # copy: (t_found, track, true send event or None, asset, canary)
    n_leaks = n_unseen = 0
    for i in range(N_TRACKS):
        track = f"track-{i:03d}"
        release = rng.uniform(50, 360) * D
        monitored = rng.random() < MONITORED
        evs = [db.insert(track, T0 + t, typ, feats, weight=w, attrs=at)
               for t, typ, feats, w, at in campaign(rng, release)]
        tracks[track] = (monitored, evs)
        leaks = [(ev.time - T0 + leak_lag(rng), ev, ev.attrs["asset"]) for ev in evs
                 if ev.type == "send" and rng.random() < p_leak(ev.features, ev.weight)]
        if monitored and rng.random() < UNTRACKED:
            leaks.append((release - rng.uniform(1, 30) * D, None, rng.choice(["full", "edit"])))
        for t, ev, asset in leaks:
            n_leaks += 1
            if not monitored or t >= release or rng.random() > COVERAGE:
                n_unseen += 1
                continue
            canary = ev is not None and ev.attrs["watermarked"] and rng.random() < CANARY_SURVIVES
            copies.append((T0 + t, track, ev, asset, canary))
    copies.sort(key=lambda c: c[0])

    for t, track, ev, _asset, canary in copies:
        db.observe(track, "leak_prerelease", t)
        if canary:
            db.feedback(track, "leak_prerelease", t, confirmed_event_id=ev.id)
    horizon = T0 + 400 * D
    rejected = sum(db.settle(horizon, "leak_prerelease", entity=tr).events_rejected
                   for tr, (mon, _) in tracks.items() if mon)

    def candidates(track, t, asset, k=50):
        """top_causes, then the bridge's variant filter: same asset as the copy."""
        return [c for c in db.top_causes(track, "leak_prerelease", t, k=k) if c.event.attrs["asset"] == asset]

    n_send = sum(1 for _, (_, evs) in tracks.items() for e in evs if e.type == "send")
    n_post = sum(len(evs) for _, (_, evs) in tracks.items()) - n_send
    n_canary = sum(c[4] for c in copies)
    print("SCENARIO")
    print(f"  {N_TRACKS} tracks ({sum(m for m, _ in tracks.values())} monitored), {n_send:,} sends "
          f"and {n_post} teaser posts before release")
    print(f"  {n_leaks} leaks simulated, {n_unseen} never found (unmonitored, uncovered or post-release);")
    print(f"  {len(copies)} copies found: {n_canary} with a surviving canary (feedback), "
          f"{len(copies) - n_canary} observed only")
    print(f"  settle: {rejected:,} sends of monitored tracks passed their window with no copy")

    # Held out from feedback: copies with a known true send but no surviving canary.
    held = [c for c in copies if c[2] is not None and not c[4]]
    hit = last_hit = big_hit = 0
    pick = None
    for t, track, ev, asset, _ in held:
        cands = candidates(track, t, asset)
        prior = [e for e in tracks[track][1] if e.type == "send" and e.time < t and e.attrs["asset"] == asset]
        last = max(prior, key=lambda e: e.time)
        big = max(prior, key=lambda e: (e.weight, e.time))
        top = cands[0].event if cands else None
        hit += top is not None and top.id == ev.id
        last_hit += last.id == ev.id
        big_hit += big.id == ev.id
        raw = db.top_causes(track, "leak_prerelease", t, k=3)
        if (pick is None and top is not None and top.id == ev.id and raw.top is not None
                and raw.top.event.id != ev.id and len(raw) == 3):
            pick = (t, track, ev, asset)

    print()
    t, track, ev, asset = pick
    print(f"ONE LEAK, EXPLAINED ({track}: copy found, matches the {asset} file, canary stripped)")
    print(format_result(db.top_causes(track, "leak_prerelease", t, k=3, explain=True)))
    kept = candidates(track, t, asset)[:3]
    print(f"  variant filter keeps {', '.join('#%d' % c.event.id for c in kept)}: most likely exposure "
          f"#{kept[0].event.id}; true source #{ev.id}")

    print()
    print("ACROSS ALL COPIES (link_report, global view, all risk features)")
    feats = sorted({f for _, (_, evs) in tracks.items() for e in evs if e.type == "send"
                    for f in e.features})
    rep = db.link_report("leak_prerelease", min_support=20, features=feats)
    print("\n".join(str(rep).splitlines()[:10]))

    print()
    print("BEFORE THE NEXT RELEASE (two planned promo-pool sends of 600 DJs, both watermarked)")
    plans = {
        "A: full WAV, download, no expiry": {"format:wav", "download:on", "expiry:none",
                                             "audience:large"},
        "B: stream only, 14-day expiry": {"audience:large"},
    }
    risk, true_p = {}, {}
    for name, fs in plans.items():
        risk[name] = db.feature_risk("leak_prerelease", fs, weight=weight(600), type="send")
        true_p[name] = p_leak(fs, weight(600))
        print(f"  {name:<34} relative risk score {risk[name]:.3f}   true P(leak) {true_p[name]:.3f}")
    (a, b), (ta, tb) = risk.values(), true_p.values()
    print(f"  A / B: HeapDB {a / b:.1f}x, simulated truth {ta / tb:.1f}x "
          f"(scores rank sends; they are not probabilities)")

    order = [r.feature for r in rep]
    ranks = {f: order.index(f) + 1 for f in SEEDED + (NO_EFFECT,)}
    print()
    print("CHECK")
    print("  seeded risky settings rank " + ", ".join(f"{f} #{ranks[f]}" for f in SEEDED)
          + f" of {len(order)}; no-effect {NO_EFFECT} #{ranks[NO_EFFECT]}")
    print(f"  {len(held)} held-out copies (true send known, canary stripped): HeapDB top-1 = true send "
          f"{hit}/{len(held)} ({hit / len(held):.0%}); last send {last_hit}/{len(held)} "
          f"({last_hit / len(held):.0%}); largest audience {big_hit}/{len(held)} "
          f"({big_hit / len(held):.0%})")
    # The two separable seeded settings must hold #1 and #2, above the no-effect
    # confounder. format:wav is not asserted: it always rides with download:on.
    top_two = {ranks["download:on"], ranks["expiry:none"]} == {1, 2}
    if (not top_two or ranks[NO_EFFECT] <= 2 or hit <= max(last_hit, big_hit)
            or a <= 2 * b):
        raise SystemExit(1)


if __name__ == "__main__":
    main()