#!/usr/bin/env python3
"""
W4 — Event time, processing time, and the answer that is quietly wrong.  (~45 min)

Miniature of P07. Count events per hourly window three ways -- by processing time,
by event time with watermark delay d, and by a batch oracle -- on a stream where
some clients sync late. Then plot the completeness/latency frontier that delay buys.
"""
import random, math
from collections import defaultdict

rng = random.Random(7)
HOUR = 3600
N = 60000

# A realistic mix: most events arrive within seconds; an "offline sync" cohort
# arrives minutes to hours later. This bimodality is the whole problem.
events = []
for _ in range(N):
    et = rng.uniform(0, 24 * HOUR)                     # true event time, over one day
    if rng.random() < 0.12:                            # 12% offline-sync cohort
        lag = rng.uniform(20 * 60, 5 * HOUR)
    else:
        lag = rng.expovariate(1 / 3.0)                 # seconds
    events.append((et, et + lag))
events.sort(key=lambda e: e[1])                        # stream order = arrival order

def window(t): return int(t // HOUR)

# --- the oracle: what a batch job over the complete day would say
truth = defaultdict(int)
for et, _ in events: truth[window(et)] += 1

# --- processing time: bucket by when we SAW it
proc = defaultdict(int)
for _, at in events: proc[window(at)] += 1

def event_time(delay):
    """Bucket by event time; a window closes when watermark = max_seen_et - delay
    passes its end. Events arriving after that are late and dropped."""
    counted, dropped, closed_at = defaultdict(int), 0, {}
    max_et = 0.0
    for et, at in events:
        max_et = max(max_et, et)
        wm = max_et - delay
        w = window(et)
        if (w + 1) * HOUR <= wm:                       # window already closed
            dropped += 1
        else:
            counted[w] += 1
        for ww in list(range(24)):
            if ww not in closed_at and (ww + 1) * HOUR <= wm:
                closed_at[ww] = at                     # processing time of closure
    return counted, dropped, closed_at

print("Same 60,000 events. Three ways of counting them per hour.\n")
err = lambda d: sum(abs(d.get(w, 0) - truth[w]) for w in truth)
print(f"  oracle (batch over the whole day)     total abs error       0")
print(f"  bucketed by PROCESSING time           total abs error {err(proc):>7,}"
      f"   ({err(proc)/N*100:.1f}% of events misattributed)")
c0, d0, _ = event_time(0)
print(f"  bucketed by EVENT time, delay=0       total abs error {err(c0)+d0:>7,}"
      f"   ({d0:,} dropped as late)")

print("\nThe processing-time error is not noise -- it is a systematic bias that")
print("lands entirely on the offline-sync cohort, i.e. exactly one segment of users.\n")

print("The frontier that watermark delay buys:")
print(f"  {'delay':>10} {'completeness':>13} {'dropped':>9} {'mean close lag':>16}")
print("  " + "-" * 52)
for delay in (0, 60, 300, 900, 3600, 5*3600, 6*3600):
    counted, dropped, closed = event_time(delay)
    completeness = 1 - dropped / N
    # how long after a window ENDS do we emit it, in event-time terms
    lag = delay
    lbl = f"{delay//60} min" if delay < 3600 else f"{delay/3600:.0f} h"
    print(f"  {lbl:>10} {completeness*100:>12.2f}% {dropped:>9,} {lag/60:>13.0f} min")

print("\nRead the shape, and note it is NOT the concave curve you may expect.")
print("Completeness is FLAT from 0 to 15 minutes (88.92% -> 89.43%, +0.5pp) and then")
print("climbs to 100% only as the delay reaches 5 hours. Fifteen minutes of extra")
print("latency buys essentially nothing.")
print("\nThe shape is set entirely by the LATENESS DISTRIBUTION, not by any property")
print("of windowing. This stream is bimodal: 88% of events arrive within seconds and")
print("are already counted at delay 0, while a 12% offline-sync cohort is spread")
print("uniformly over 20 min to 5 h. There is nothing in between, so there is no knee.")
print("\nThe operational consequence is uncomfortable: you must either accept ~89%")
print("completeness or wait five hours. Tuning the delay to 5 or 15 minutes -- the")
print("usual instinct -- lands in the dead zone and buys latency for nothing.")
print("\nSo: measure YOUR lateness distribution before choosing a delay. A watermark")
print("is a formal admission that you are trading completeness for the ability to")
print("emit at all, and the exchange rate is a property of your data, not your code.")