Designing Event Data for Product Analytics

In plain terms

An event is a note the software writes when something happens: “this person opened the checkout page at 9:03.” These records support questions such as how many observed visitors bought a product and where a checkout sequence stopped. Their definitions determine what can be counted. Explaining why a metric changed also needs context and, for causal claims, an appropriate study design.

The application records events on devices with different clocks and software versions. A changed schema leaves historical records in the old shape and deployed clients still producing it. Preserve source versions and provide tested migrations or versioned views instead of expecting every query to handle every historical format. A tracking plan agreed by producers and consumers makes those transitions deliberate.

The modeling article distinguished current state from events that record what happened. Event history is usually appended; corrections, retention expiry, and deletion still need explicit handling. Here we design names, identifiers, timestamps, versions, and ownership for product analytics. The lab isolates several common failures; it does not simulate every production concern.

What one event has to carry

An event answers five questions: what happened, to whom, when, with what details, and how the platform can trust the answer. Product and engineering teams need to agree on both the business meaning and the evidence the collection system can provide. The usual design puts the answers into a fixed event envelope that every event shares and a properties object that differs per event.

{
  "event_id": "8b1f0c2e-6d5a-4e1b-9a3f-2c7d1e0f4a55",
  "event_name": "checkout_started",
  "event_version": 2,
  "event_time": "2026-03-11T09:03:17.412Z",
  "sent_time": "2026-03-11T09:03:19.020Z",
  "received_time": "2026-03-11T09:03:19.377Z",
  "source": "app",
  "app_version": "4.12.0",
  "anonymous_id": "A7f3e...",
  "user_id": "U10482",
  "session_id": "S9c2b...",
  "properties": {"cart_id": "CART17", "amount_cents": 6300, "currency": "USD", "item_count": 2}
}

A common envelope lets collection and processing share machinery, while properties describe the business event. The envelope is still a claim made by a producer: a supplied source or user_id does not prove authenticity. The collector should stamp received_time, authenticate server producers, and obtain trusted account context from the appropriate identity boundary. Session and identity rules remain part of the product contract.

Names and the tracking plan

One useful naming convention is an object followed by a past-tense verb: product_viewed, cart_item_added, checkout_started, order_placed, subscription_paused. The object comes first so that names sort into families, the verb says what happened to it, and the past tense says it already did. Names that describe the user interface (blue_button_clicked, screen3_shown) break the first time the interface changes, and names that describe intent (conversion, engagement) mean different things to different teams. Within a declared version, each event name has a defined meaning, trigger, and owner.

The tracking plan is the document that holds those meanings. It is a table with one row per event and, under each, one row per property. The columns that matter:

event            checkout_started                         version 2   owner: checkout team   status: active
trigger          the checkout screen opens with a non-empty cart; once per cart, not per tap
source           client (app, web)
properties       cart_id        string    required   stable cart identifier
                 amount_cents   integer   required   USD cart total in cents, before shipping and tax
                 currency       string    required   USD in this example; use an explicit currency contract
                 item_count     integer   required   distinct products in the cart
consumers        checkout funnel dashboard; abandoned-cart email job; experimentation platform
history          v1 (2024-06) sent amount as a decimal number of dollars; supported during migration; retirement pending

The trigger defines the unit being counted. Repeated taps inflate an event count, but not a count of distinct cart IDs. To enforce “once per cart,” retain the creation identity across screen reopenings and retries, or enforce a scoped business key such as (cart_id, event_name) at the counting boundary. Reusing an event_id only for network retries does not prevent a second screen opening from creating another ID. If a cart can begin several checkout attempts, define checkout_id and count attempts instead. The example uses USD cents; a multi-currency contract should use amount_minor plus currency and its minor-unit scale, rather than assume every currency has cents. State whether totals include shipping, tax, discounts, and refunds.

The plan is reviewed like code, by the producing team and at least one consuming team, before the first event is emitted. The engineering-habits article’s review rules apply: the reviewer asks what question this event answers, what the trigger is, and what happens to existing dashboards if it changes. A tracking plan nobody reviews decays into a list of names that nobody can define.

The tracking plan is an event-specific data contract; its schema describes structure, while trigger and unit definitions supply meaning.

Who: identifiers at three scopes

The same person shows up under several identifiers, and each answers a different question.

A visitor may browse under an anonymous_id, then log in and order under a user_id. Identity stitching links these observations when a trusted login associates the identifiers. A global anonymous-to-account map is safe only under a one-account-per-anonymous-ID assumption. Shared devices, account switching, and logout require reset or time-bounded link rules; ambiguous links should not silently select the last account. Exercise 4 deliberately uses one device per account and leaves conflicting links unresolved.

Keep permitted source identifiers with their original scope and record the version of the stitching policy used for a metric. More links do not necessarily mean better identity resolution. Counts by browser installation, account, or stitched profile are different quantities; none automatically counts distinct humans. SDK reset behavior also varies: Segment documents that reset clears both userId and anonymousId.

When: three timestamps and what to trust

event_time records the producer’s estimate of occurrence time. sent_time records dispatch on that same clock; received_time records collector arrival. Monitor server clocks too. Arrival time is useful for ingestion progress, but an offline queue can make it much later than the action. None of these timestamps alone proves the true occurrence time.

With synchronized clocks, received_time − event_time estimates delivery lag. With a client offset, it mixes lag and clock error. Under a constant client offset, sent_time − event_time estimates time before dispatch, because the offset cancels. A derived estimate, received_time − (sent_time − event_time), removes that offset but still includes transit delay. Snowplow documents this kind of derived timestamp. Clock jumps and mismatched retry timestamps break the assumptions. Preserve originals, flag the chosen estimate, and treat fallback thresholds as policy rather than proof that a clock is wrong. A measured lag distribution helps choose lookback and correction policy; it cannot prove a day is complete.

Store timestamps in UTC with an explicit offset, and if a question needs the user’s local time (was this order placed in the evening?), carry the user’s time zone as a property and convert at query time. A local timestamp without a known zone cannot be mapped unambiguously to an instant.

Where: client events and server events

A client event observes screens and interactions. Collection can lose it through blocked tracking, queue limits, or failure before durable storage. A server event records a backend fact, such as order creation or payment capture. Server collection can fail too; reconcile important events with the operational source instead of assuming their completeness.

Use the order service as the authority for order_placed and the client for product_viewed or checkout_started. To couple a database change to an event, use a transactional outbox: write an outbox row in the same transaction and relay it afterward; a database commit and a separate broker send are not automatically atomic. The relay may retry, so consumers still deduplicate. AWS explains the transactional outbox and duplicate delivery. Client payment attempts and server payment outcomes can both be useful when separately named and linked, rather than summed as the same fact.

Server events may carry an authenticated account ID, an anonymous context passed with the request, and a checkout or order correlation ID. Guest checkout need not have a user_id. Define which fields are available and trusted for each source. The lab simplifies this to logged-in server orders with no anonymous_id and a shared session_id.

Changing an event without breaking its readers

Events change with the product. The table evaluates changes against existing readers and the declared event meaning; aliases, adapters, and coordinated releases can support a migration, but must be explicit.

Give incompatible changes a new event_version and define the support and retirement period for older producers. Normalize versions into a documented canonical representation only when their meanings can be mapped; a changed trigger may need a separate metric or event. Preserve the source version for audit and replay. Schema compatibility depends on format, reader behavior, and compatibility direction; it does not automatically detect a change in business meaning. Confluent documents these compatibility directions and format-dependent rules. Unregistered versions remain quarantined until reviewed.

Validation, duplicates, and order

Three things the collector and the first transformation have to do to every event, in this order.

Validate against the plan. Check known event names and versions, envelope fields, required properties, types, units, and allowed values. Retain rejected records with reasons in controlled quarantine, subject to the same access and retention rules as raw events. Monitor rejection rates by producer and app version and agree an alert response time with owners. A validator catches only the rules it implements; passing it does not establish source completeness or business correctness.

Deduplicate on a stable event key. Retries can deliver an event more than once; delivery guarantees depend on the SDK and collector. Reuse the creation-time event_id, namespaced by producer or tenant if it is not globally unique. Exclude transport timestamps when comparing logical payloads. If the same key carries different business data, quarantine or fail the conflict rather than silently keep one. Retain deduplication state for the supported retry and replay horizon. Also enforce a business key such as order_id: two event IDs can describe the same order.

Define ordering separately from arrival. A constant offset cancels when comparing events on one device, but clocks can jump or produce tied timestamps. A session can also contain both client and server events, which do not share a clock. Use producer sequence numbers with an explicit scope, correlation IDs, and a documented time-estimation policy where needed. A deterministic tie-break is not evidence of causal order. Exercise 6 compares both orders with simulator truth and shows why a session ID alone is insufficient.

Verify instrumentation after release. Exercise page reopenings, retries, logout, account switching, and supported app versions. Trace one known checkout through collection, quarantine, deduplication, and the final metric. Compare server order IDs with committed orders at a shared cutoff. Watch valid-event volume and missing steps by app version as well as rejection rates: a producer that stops sending creates no invalid records. Record bot, test-account, sampling, and collection-eligibility rules because they change the measured population; a rate over observed sessions is not automatically a rate over all visitors.

Event deduplication is one part of making retries safe. Idempotency concerns the repeated operation’s effect, so deduplication state and the output write must be coordinated.

What must not be in an event

Collect only fields needed for the stated purpose. Avoid names, email addresses, and unrestricted free text in this analytics contract. A user_id or anonymous_id can still identify or link to a person; pseudonymous does not mean anonymous. ICO distinguishes pseudonymisation from anonymisation. Classify identifiers as well as properties, restrict access, and define retention and deletion propagation across raw data, derived tables, and vendors. Append-only processing does not make deletion impossible or remove the need to implement it.

The checkout funnel, designed

The subscription commerce company from the architecture article, its web shop and app, and the events that the checkout funnel dashboard needs:

These six events are a catalog, not six mandatory sequential steps: an already logged-in visitor skips user_logged_in. Define a funnel such as product_viewed → cart_item_added → checkout_started → order_placed, its counting unit (session, account, or checkout), conversion window, and rules for skipped and repeated steps. An ordered funnel requires later steps after entry within that window; merely appearing in both weekly sets is a different metric. Amplitude explains ordered steps and conversion windows. Carry cart_id or checkout_id when sessions can contain multiple attempts. Order totals are booked order value, not automatically recognized revenue: payments, cancellations, refunds, tax, and currency rules require reconciliation.

Separate the entry-date range from the observation horizon. A view at 23:50 with a 30-minute conversion window needs observations through 00:20 the following day, plus the agreed delivery allowance. Do not mark it as a completed failure at midnight. Report unfinished cohorts separately or wait until their windows have elapsed; excluding only unfinished non-converters would bias the rate upward. Late deliveries can still require revision. The simulator generates complete sessions even when they extend beyond the nominal week; it does not model a live reporting cutoff.

Anti-patterns

• One generic event. track with a type property and forty optional fields. Validation requires an explicit schema for each type, and every query starts with a CASE on type.
• Names from the interface. home_banner_v2_clicked. The next redesign leaves a year of history under a name nobody maps to the new one.
• Money as a decimal, units in prose. amount: 24.0 in one version and amount: 2400 in the next, with the unit change noted in a chat message.
• Local time without an offset. “09:03:17” from a device in one time zone and a server in another, compared as if they were on one clock.
• Overwriting the anonymous id at login. The pre-login history is orphaned, and the funnel starts at the login screen.
• Personal data in properties. An email address in user_logged_in, copied to every downstream table before anyone noticed.
• Changing meaning in place. checkout_started quietly redefined from “once per cart” to “once per tap”; the funnel improves overnight and nobody knows why.
• No owner. An event that a departed developer added, that three dashboards depend on, and that nobody can say the trigger of.

Lab

The standard-library lab generates 400 devices and 1,200 sessions over a week. It implements five event names, omitting payment_submitted and optional list_name from the six-event catalog. Each session has one cart and at most one order. Each anonymous ID belongs to exactly one account, and its clock offset remains fixed. About 5% of sessions buffer client analytics for one to three days while backend ordering remains available; 3% of distinct events receive one duplicate. Timestamps are timezone-aware UTC values. sent_time models dispatch one second before first receipt, while collector redeliveries retain the original sent_time. true_times is a separate simulator oracle, not a collected field. Run setup once. Exercises isolate failures on the generated records; they are not a fully connected production validation pipeline.

import random, math
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone

rng = random.Random(0)
start = datetime(2026, 3, 9, tzinfo=timezone.utc)
PRICES = {"P1": 2400, "P2": 3900, "P3": 1500, "P4": 5900}
events, next_id = [], [1]
# Simulator oracle, never sent to the collector.
true_times = {}

def emit(name, version, source, when, device, session, user, props, delay=None):
    """Append one event; client events carry the device's clock, server events carry the true time."""
    clock_offset = device["skew"] if source in ("app", "web") else timedelta(0)
    if delay is None:
        delay = timedelta(seconds=rng.randint(1, 30))
    true_times[f"E{next_id[0]}"] = when
    events.append({"event_id": f"E{next_id[0]}", "event_name": name, "event_version": version, "source": source,
                   "event_time": when + clock_offset, "received_time": when + delay,
                   "sent_time": when + delay - timedelta(seconds=1) + clock_offset,
                   "app_version": str(device["app_version"]) if source != "server" else None,
                   "anonymous_id": device["anonymous_id"] if source != "server" else None,
                   "user_id": user, "session_id": session, "properties": props})
    next_id[0] += 1

devices = []
for n in range(1, 401):
    roll = rng.random()
    if roll < 0.90:
        skew = timedelta(seconds=rng.randint(-2, 2))
    elif roll < 0.95:
        skew = timedelta(minutes=rng.choice([-5, 5, 12]))
    elif roll < 0.98:
        skew = timedelta(hours=rng.choice([-8, 9]))
    else:
        skew = timedelta(days=-rng.randint(365, 2200))
    devices.append({"anonymous_id": f"A{n}", "user_id": f"U{n}", "skew": skew,
                    "source": rng.choice(["app", "app", "web"]), "app_version": rng.choice([1, 2, 2, 2, 2, 2, 2, 3])})

for session_no in range(1, 1201):
    device = rng.choice(devices)
    session, source, version = f"S{session_no}", device["source"], device["app_version"]
    logged_in = rng.random() < 0.4
    user = device["user_id"] if logged_in else None
    when = start + timedelta(days=rng.randrange(7), seconds=rng.randrange(86400))
    offline = timedelta(days=rng.randint(1, 3)) if rng.random() < 0.05 else None

    def arrival():
        return None if offline is None else offline + timedelta(seconds=rng.randint(1, 30))

    cart = []
    for _ in range(rng.randint(1, 4)):
        product = rng.choice(list(PRICES))
        props = {} if (source == "web" and version == 1 and rng.random() < 0.05) else {"product_id": product}
        emit("product_viewed", 1, source, when, device, session, user, props, arrival())
        when += timedelta(seconds=rng.randint(5, 90))
        if rng.random() < 0.5:
            cart.append(product)
            quantity = "1" if (source == "web" and version == 1) else 1
            emit("cart_item_added", 1, source, when, device, session, user, {"product_id": product, "quantity": quantity}, arrival())
            when += timedelta(seconds=rng.randint(5, 60))
    if cart and rng.random() < 0.6:
        total = sum(PRICES[p] for p in cart)
        props = {"amount": total / 100, "currency": "USD"} if version == 1 else {"amount_cents": total, "currency": "USD"}
        props["item_count"] = len(set(cart))
        props["cart_id"] = f"CART{session_no}"
        if version == 3:
            props["coupon"] = None
        emit("checkout_started", version, source, when, device, session, user, props, arrival())
        when += timedelta(seconds=rng.randint(20, 120))
        if not logged_in:
            user = device["user_id"]
            emit("user_logged_in", 1, source, when, device, session, user, {"method": "password"}, arrival())
            when += timedelta(seconds=rng.randint(5, 30))
        if rng.random() < 0.8:
            emit("order_placed", 1, "server", when, device, session, user,
                 {"order_id": f"O{session_no}", "amount_cents": total, "currency": "USD", "items": len(cart)})

for e in rng.sample(events, len(events) * 3 // 100):
    events.append(dict(e, received_time=e["received_time"] + timedelta(seconds=rng.randint(1, 60))))
events.sort(key=lambda e: e["received_time"])
print(len(events), "records,", len({e["event_id"] for e in events}), "distinct events")
print(sorted(Counter(e["event_name"] for e in events).items()))
# 5972 records, 5799 distinct events
# [('cart_item_added', 1565), ('checkout_started', 561), ('order_placed', 458), ('product_viewed', 3046), ('user_logged_in', 342)]

Nearly six thousand records for fewer than six thousand events, which is the duplicate rate showing before anything is analysed.

1. Write the tracking plan and validate against it. Encode the five simulated event names and their six registered name/version pairs’ required properties and types as a Python structure, check every record’s envelope and properties, and count the outcomes by reason.

PLAN = {
    ("product_viewed", 1): {"product_id": str},
    ("cart_item_added", 1): {"product_id": str, "quantity": int},
    ("checkout_started", 1): {"amount": (int, float), "currency": str, "item_count": int, "cart_id": str},
    ("checkout_started", 2): {"amount_cents": int, "currency": str, "item_count": int, "cart_id": str},
    ("user_logged_in", 1): {"method": str},
    ("order_placed", 1): {"order_id": str, "amount_cents": int, "currency": str, "items": int},
}
ENVELOPE = ["event_id", "event_name", "source", "session_id"]
ALLOWED_ENVELOPE = set(ENVELOPE) | {
    "event_version", "event_time", "sent_time", "received_time",
    "app_version", "anonymous_id", "user_id", "properties",
}

def validate(e):
    if not isinstance(e, dict):
        return "event is not an object"
    if set(e) - ALLOWED_ENVELOPE:
        return "unregistered envelope field"
    for field in ENVELOPE:
        if not isinstance(e.get(field), str) or not e[field].strip():
            return f"missing or invalid {field}"
    if type(e.get("event_version")) is not int or e["event_version"] < 1:
        return "invalid event_version"
    if e["source"] not in {"app", "web", "server"}:
        return "unknown source"
    for field in ("event_time", "sent_time", "received_time"):
        value = e.get(field)
        if not isinstance(value, datetime) or value.utcoffset() is None:
            return f"invalid {field}: timezone-aware datetime required"
    for field in ("anonymous_id", "user_id"):
        value = e.get(field)
        if value is not None and (not isinstance(value, str) or not value.strip()):
            return f"invalid {field}"
    if not (e.get("anonymous_id") or e.get("user_id")):
        return "no identifier"
    if e["source"] != "server" and (not isinstance(e.get("app_version"), str) or not e["app_version"]):
        return "missing app_version"
    spec = PLAN.get((e["event_name"], e["event_version"]))
    if spec is None:
        return f"unknown event {e['event_name']} v{e['event_version']}"
    expected = "server" if e["event_name"] == "order_placed" else "client"
    if (e["source"] == "server") != (expected == "server"):
        return "unexpected source for event"
    if e["event_name"] in {"user_logged_in", "order_placed"} and not e.get("user_id"):
        return "missing account for this lab event"
    props = e.get("properties")
    if not isinstance(props, dict):
        return "properties is not an object"
    if set(props) - set(spec):
        return "unregistered property"
    for prop, kind in spec.items():
        if prop not in props:
            return f"{e['event_name']} without {prop}"
        kinds = kind if isinstance(kind, tuple) else (kind,)
        if type(props[prop]) not in kinds:
            names = "/".join(k.__name__ for k in kinds)
            return f"{e['event_name']}.{prop} is not {names}"
        value = props[prop]
        if isinstance(value, str) and not value:
            return f"empty {prop}"
        if prop in {"amount", "amount_cents", "quantity", "items", "item_count"}:
            if (type(value) is float and not math.isfinite(value)) or value < 0:
                return f"invalid {prop}"
            if prop in {"quantity", "items", "item_count"} and value == 0:
                return f"invalid {prop}"
        if prop == "currency" and value != "USD":
            return "unsupported currency in this USD-only lab"
    return None

verdicts = Counter(validate(e) or "valid" for e in events)
print(sorted(verdicts.items(), key=lambda kv: -kv[1]))
quarantine = [(e, validate(e)) for e in events if validate(e) is not None]
valid = [e for e in events if validate(e) is None]
# [('valid', 5871), ('cart_item_added.quantity is not int', 51), ('unknown event checkout_started v3', 43), ('product_viewed without product_id', 7)]

2. Normalise versions. Version 1 of checkout_started sent amount as a decimal number of dollars; versions 2 and 3 send amount_cents. First sum “whatever amount field exists” across all versions, as a query written against the raw table would. Exclude unregistered v3 records from both compared totals, then normalize v1 and v2 to the canonical cents shape and sum again.

from decimal import Decimal

all_checkouts = [e for e in events if e["event_name"] == "checkout_started"]
print(Counter(e["event_version"] for e in all_checkouts))
checkouts = [e for e in all_checkouts if e["event_version"] in (1, 2)]
print("unregistered checkout records excluded:", len(all_checkouts) - len(checkouts))
naive = sum(e["properties"].get("amount_cents", e["properties"].get("amount")) for e in checkouts)
print("naive mixed-unit total:", naive)

def normalize(e):
    """Normalize registered checkout versions; retain the source version."""
    if e["event_name"] != "checkout_started":
        return e
    if type(e["event_version"]) is not int or e["event_version"] not in (1, 2):
        raise ValueError("unregistered checkout version")
    props = dict(e["properties"])
    if e["event_version"] == 1:
        cents = Decimal(str(props.pop("amount"))) * 100
        if not cents.is_finite() or cents < 0 or cents != cents.to_integral_value():
            raise ValueError("amount is not a non-negative whole number of cents")
        props["amount_cents"] = int(cents)
    elif type(props.get("amount_cents")) is not int or props["amount_cents"] < 0:
        raise ValueError("invalid amount_cents")
    return dict(e, canonical_version=2, properties=props)

normalized = [normalize(e) for e in checkouts]
total = sum(e["properties"]["amount_cents"] for e in normalized)
print("normalized checkout value:", total, "USD cents")
print("mixed-unit shortfall: {:.1%}".format(1 - naive / total))
# Counter({2: 427, 1: 91, 3: 43})
# unregistered checkout records excluded: 43
# naive mixed-unit total: 2453445.0
# normalized checkout value: 2952900 USD cents
# mixed-unit shortfall: 16.9%

3. Deduplicate. Count events delivered more than once and how far apart the copies arrive. Compute order count and order value in USD cents from order_placed with and without deduplication on event_id.

by_id = defaultdict(list)
for e in events:
    by_id[e["event_id"]].append(e)
dupes = {k: v for k, v in by_id.items() if len(v) > 1}
gaps = [(v[-1]["received_time"] - v[0]["received_time"]).total_seconds() for v in dupes.values()]
print(len(dupes), "events delivered twice; copies arrive", int(min(gaps, default=0)), "to", int(max(gaps, default=0)), "seconds apart")

orders = [e for e in events if e["event_name"] == "order_placed"]
print("orders counted from records:", len(orders), "order value (USD cents)", sum(e["properties"]["amount_cents"] for e in orders))
def deduplicate(records):
    seen, result = {}, []
    for e in sorted(records, key=lambda e: e["received_time"]):
        # Event IDs are globally unique in this simulator.
        key = e["event_id"]
        payload = {k: v for k, v in e.items() if k not in {"received_time", "sent_time"}}
        if key in seen:
            if seen[key] != payload:
                raise ValueError(f"conflicting payload for {key}")
            continue
        seen[key] = payload
        result.append(e)
    return result

deduped = deduplicate(events)
orders = [e for e in deduped if e["event_name"] == "order_placed"]
print("orders after dedupe by event_id:", len(orders), "order value (USD cents)", sum(e["properties"]["amount_cents"] for e in orders))
print("dedupe window needed for this stream:", int(max(gaps, default=0)), "seconds; retries in production can be hours apart")
# 173 events delivered twice; copies arrive 1 to 60 seconds apart
# orders counted from records: 458 order value (USD cents) 2643200
# orders after dedupe by event_id: 440 order value (USD cents) 2530700
# dedupe window needed for this stream: 60 seconds; retries in production can be hours apart

4. Stitch identities. Build a link table from every event that carries both an anonymous_id and a user_id. Compute weekly viewer/orderer set overlap twice: once using the available identifiers and once resolving unambiguous links. This does not test event order or a conversion window.

def build_links(records):
    candidates = defaultdict(set)
    for e in records:
        if e.get("anonymous_id") and e.get("user_id"):
            candidates[e["anonymous_id"]].add(e["user_id"])
    # Ambiguous device/account links are left unresolved, never last-writer-wins.
    return {a: next(iter(users)) for a, users in candidates.items() if len(users) == 1}

links = build_links(events)
print(len(links), "anonymous ids linked to a user")

def person(e, stitch):
    if e["user_id"]:
        return ("account", e["user_id"])
    a = e["anonymous_id"]
    return ("account", links[a]) if stitch and a in links else ("anonymous", a)

for stitch in (False, True):
    viewed = {person(e, stitch) for e in events if e["event_name"] == "product_viewed"}
    ordered = {person(e, stitch) for e in events if e["event_name"] == "order_placed"}
    print(f"stitch={stitch}: {len(viewed)} identities viewed, {len(ordered)} ordered, "
          f"{len(viewed & ordered)} did both, weekly overlap {len(viewed & ordered) / len(viewed):.1%}")
# 346 anonymous ids linked to a user
# stitch=False: 618 identities viewed, 273 ordered, 215 did both, weekly overlap 34.8%
# stitch=True: 372 identities viewed, 273 ordered, 273 did both, weekly overlap 73.4%

5. Measure the clocks. For client events, compute event_time - received_time and bucket it. Count events dated in the future and events dated years in the past. Write a rule that decides which timestamp to trust, apply it, and compare the range of calendar days before and after. Finally count records whose estimated lag exceeds a day and compare a sent-time-based estimate with simulator truth.

client = [e for e in events if e["source"] != "server"]
skews = [(e["event_time"] - e["received_time"]).total_seconds() for e in client]
buckets = Counter("within 1 minute" if abs(s) < 60 else "within 1 hour" if abs(s) < 3600
                  else "within 1 day" if abs(s) < 86400 else "more than a day" for s in skews)
print(sorted(buckets.items(), key=lambda kv: -kv[1]))
print("events dated in the future:", sum(s > 60 for s in skews), " dated years in the past:", sum(s < -86400 * 300 for s in skews))

def best_time(e):
    """Illustrative fallback policy, not proof of a broken clock."""
    skew = (e["event_time"] - e["received_time"]).total_seconds()
    if skew > 60 or skew < -86400 * 30:
        return e["received_time"], "arrival_fallback"
    return e["event_time"], "event_time"

print("date range by client clock:", min(e["event_time"] for e in client).date(), "to", max(e["event_time"] for e in client).date())
print("date range after the rule:  ", min(best_time(e)[0] for e in client).date(), "to", max(best_time(e)[0] for e in client).date())
print("events whose calendar day differs between client clock and arrival:",
      sum(e["event_time"].date() != e["received_time"].date() for e in client))
late = [e for e in client if e["received_time"] - best_time(e)[0] > timedelta(days=1)]
print(len(late), "records have estimated lag over a day, maximum whole days:",
      max((e["received_time"] - best_time(e)[0] for e in late), default=timedelta(0)).days, "days")
print("fallback records:", sum(best_time(e)[1] == "arrival_fallback" for e in client))
def derived_time(e):
    """Cancel a constant client offset; transit delay remains."""
    return e["received_time"] - (e["sent_time"] - e["event_time"])
print("maximum derived-time error in simulator:",
      max(abs((derived_time(e) - true_times[e["event_id"]]).total_seconds()) for e in client), "seconds")
# [('within 1 minute', 4687), ('within 1 hour', 301), ('more than a day', 296), ('within 1 day', 230)]
# events dated in the future: 342  dated years in the past: 99
# date range by client clock: 2020-03-23 to 2026-03-16
# date range after the rule:   2026-03-09 to 2026-03-15
# events whose calendar day differs between client clock and arrival: 403
# 197 records have estimated lag over a day, maximum whole days: 3 days
# fallback records: 441
# maximum derived-time error in simulator: 61.0 seconds

6. Order within sessions. Group events by session and compare the arrival order with the event-time order. Compare each order with simulator truth, repeat for client-only events, then calculate a two-step session funnel with an explicit time policy and a 30-minute window.

sessions = defaultdict(list)
seen = set()
for e in events:
    if e["event_id"] not in seen:
        seen.add(e["event_id"])
        sessions[e["session_id"]].append(e)

def order_of(evs, key):
    return [e["event_id"] for e in sorted(evs, key=key)]

def wrong_order(evs, key):
    return order_of(evs, key) != order_of(evs, lambda e: true_times[e["event_id"]])

print("sessions with arrival order different from simulator truth:",
      sum(wrong_order(evs, lambda e: e["received_time"]) for evs in sessions.values()))
print("sessions with mixed-clock event order different from truth:",
      sum(wrong_order(evs, lambda e: e["event_time"]) for evs in sessions.values()))
client_sessions = [[e for e in evs if e["source"] != "server"] for evs in sessions.values()]
print("client-only sessions with event order different from truth:",
      sum(wrong_order(evs, lambda e: e["event_time"]) for evs in client_sessions))

def analysis_time(e):
    """Use an explicit estimated-time policy for the session funnel."""
    if e["source"] == "server":
        return e["event_time"]
    return e["received_time"] - (e["sent_time"] - e["event_time"])

def converts(evs, window=timedelta(minutes=30)):
    if not isinstance(window, timedelta) or window < timedelta(0):
        raise ValueError("window must be a non-negative timedelta")
    views = [analysis_time(e) for e in evs if e["event_name"] == "product_viewed"]
    orders = [analysis_time(e) for e in evs if e["event_name"] == "order_placed"]
    return any(t <= u <= t + window for t in views for u in orders)

eligible = [evs for evs in sessions.values() if any(e["event_name"] == "product_viewed" for e in evs)]
converted = sum(converts(evs) for evs in eligible)
print("view-to-order sessions within 30 minutes:", converted, "/", len(eligible),
      "= {:.1%}".format(converted / len(eligible) if eligible else float("nan")))
# sessions with arrival order different from simulator truth: 183
# sessions with mixed-clock event order different from truth: 29
# client-only sessions with event order different from truth: 0
# view-to-order sessions within 30 minutes: 440 / 1200 = 36.7%

7. Write the tracking plan. Turn the checkout funnel table into the full document: for each of the six events the trigger sentence, source, owner, version history, consumers, and every property with type, unit, requiredness, classification, and an example. Put it in the project’s docs/ next to the data dictionary from the modeling article.

Keep the event definitions, validator, and identity and deduplication policies together with their tests. Before using the prototype for a published metric, connect the acceptance stages and define cohort maturity, reconciliation, and correction rules.