# Analysis recipes

Ten field-tested patterns, each distilled from real product investigations (e-commerce
checkout, game telemetry at 1.6M events, catalog browsing, navigation-game paths).
Each recipe: when to use → skeleton → how to read → pitfalls. Pick the SMALLEST recipe
that answers the question; combine only when each addition resolves a distinct uncertainty.

Skeletons assume `stream` is an `Eventstream` (see `api-map.md`).

---

## R1 · First picture of a product ("what is going on here?")

```python
d = stream.describe(top_events=None)          # full frequency table, no truncation
f = stream.funnel_data(steps=[...])           # your best guess of the core flow
tg = stream.transition_graph(edge_weight="proba_out")
tg.export_html("artifacts/overview_graph.html", title="...", analysis="...")
```

Read: `step_conversion_rate` locates the worst cliff; the graph shows where traffic
actually goes. Pitfalls: start the funnel at a meaningful entry (not the landing page —
many users enter mid-flow); optional steps (e.g. a review screen only 10% pass) do not
belong in `steps`.

## R2 · Converted vs dropped ("what do winners do differently?")

The single highest-yield move in the toolkit.

```python
lab = stream.add_segment("stage", funnel_events=["basket", "checkout", "purchase"])
lab.transition_graph(diff=("stage", "purchase", "basket"))     # diff = purchase − basket
lab.step_matrix(path_pattern=".*->basket->.*", diff=("stage", "purchase", "basket"))
```

Read: positive cells = over-represented among converters. Pitfalls: `funnel_events` is
closed/ordered — a path that hit `checkout` before ever hitting `basket` is labeled
`basket`, target-only paths get `out_of_funnel`; check `get_segment_levels()` and group
sizes before interpreting.

## R3 · Around an anchor event ("what happens right before/after X?")

```python
(m,) = stream.step_matrix_data(max_steps=6, path_pattern=".*->basket->.*")
m[[0, 1, 2]]        # column 0 = the anchor itself, 1..n = steps after
```

Read: rising `path_end` share right after the anchor = the anchor is where paths die.
Pitfalls: with `path_pattern` the headless return is a tuple of per-anchor blocks —
unpack; with `diff` it is `(blocks, g1_blocks, g2_blocks)`.

## R4 · Micro-journey between two funnel levels

```python
micro = stream.truncate_paths(start_anchor="basket", end_anchor="checkout")
micro.transition_graph(edge_weight="proba_out")
micro.get_metrics([{"metric": "length"}, {"metric": "duration"}]).describe()
```

Read: what successful walkers actually traverse (detours, loops, help pages) and how
long the crossing takes (median steps/minutes → timing for recovery nudges).
Pitfalls: paths lacking either anchor are DROPPED — population = completers only. To
compare with non-completers, split first (`filter_paths` on `has_event`), truncate each
population by its own rules, analyze separately.

## R5 · Behavioral segmentation without target leakage

```python
KEY_FAMILIES = ["listing", "product", "add_to_cart", "search", "promo"]  # NO outcome events!
FEATURES = (
    [{"metric": "length"}, {"metric": "duration"}, {"metric": "active_days"}]
    + [{"metric": "event_count", "metric_args": {"event": e}} for e in KEY_FAMILIES]
)
res = stream.cluster_analysis_data(
    features=FEATURES, method_args={"n_clusters": "3-8"},
    overview_metrics=FEATURES + [
        {"metric": "has_event", "metric_args": {"event": "purchase"}, "agg": "mean"},
    ],   # outcome goes HERE, for validation only
)
labeled = stream.add_clusters(name="behavior", features=FEATURES,
                              **res["best_params"])
```

Read: conversion spread ACROSS clusters (from overview) is the finding; cluster profiles
name the personas. Pitfalls: outcome events in `features` produce silhouette≈0.9
"clusters" that merely restate the funnel — impressive and useless. Inspect the
silhouette curve yourself: a best-K at the range boundary or a >90/10 split means the
clustering is weak regardless of `best_params`; on degenerate data `best_params` may be
absent entirely. Cluster labels are strings (`"cluster_0"`); interpret via top routes and
sizes, and treat clusters as centroids, not rules.

## R6 · Loops and repeated behavior ("where do users churn in place?")

Self-loops live on the transition-matrix diagonal; A→B→A patterns via `matches_pattern`:

```python
tm = stream.transition_graph_data(edge_weight="proba_out")
selfloops = {e: tm.loc[e, e] for e in tm.index}
pogo = stream.get_metrics([{"metric": "matches_pattern",
                            "metric_args": {"pattern": "listing->product->listing"}}])
```

Pitfalls (both are result-killers): (1) count loops BEFORE `collapse_events(loops=)`
— collapsing erases them; (2) naive "conversion of users with loop vs without" is
confounded by exposure — longer paths have more loops AND more chances to convert;
stratify by path-length quantiles before comparing.

## R7 · Time-to-outcome and intervention windows

```python
tb = stream.get_metrics([{"metric": "time_between",
                          "metric_args": {"start_event": "basket",
                                          "end_event": "checkout"}}]).dropna()
# completers' timing; for a survival view add censored paths (reached basket, no checkout)
```

Read: median/quantiles of completion → after which delay organic completion is <5% —
that is when a nudge fires. Pitfalls: `time_between` = first A to first B globally in the
path, not "first B after A" — for strict semantics compute from `to_dataframe()`; account
for right-censoring near the end of the log window.

## R8 · Compare segments on many metrics at once

```python
stream.segment_overview(segment_col="device", metrics=[
    {"metric": "length", "agg": "median"},
    {"metric": "has_event", "metric_args": {"event": "purchase"}, "agg": "mean"},
])
```

Read: scan for the metric with the largest cross-segment gap, then drill with R2/R3.
Pitfalls: report `n` per segment value alongside every share — a 100% on 5 paths is
noise; flag small cells explicitly (reviewers will ask).

## R9 · What-if on a Markov chain (advanced; custom math on top)

Estimate the value of removing/rerouting a friction step (e.g., guest checkout):

```python
counts = sub.transition_graph_data(edge_weight="count").astype(float).fillna(0)
# build absorbing chain on counts; edit edges (reroute basket->login mass to basket->checkout);
# recompute absorption; bootstrap paths for CIs
```

Non-negotiables learned in the field: (1) build the chain on the RELEVANT SUB-POPULATION
(paths containing the anchor, `truncate_paths` to first anchor→outcome) — a chain over
the full log averages transition rows over unrelated users and badly distorts absorption;
(2) validation gate — base-chain absorption must reproduce the observed conversion before
any scenario is trusted; (3) plug-in absorption equals the training-set rate by
construction, so the model's value is in scenario DELTAS, not level forecasts;
(4) rerouted users converting like organic ones is an upper-bound assumption — present a
sensitivity grid (25/50/100% of the empirical rate).

## R10 · Package results for stakeholders

```python
w = lab.transition_graph(diff=("stage", "purchase", "basket"))
w.export_html("artifacts/conv_vs_drop.html", title="...",
              analysis="Findings in markdown; [basket] and [checkout] become clickable.")
meta = {"recipe": processed.recipe(), "version": retentioneering.__version__,
        "filters": "...", "params": {...}}
```

Rules that survived stakeholder review: write the `analysis=` text AFTER conclusions are
final (a stale caption contradicting the report is a credibility killer); caption numbers
must come from the headless twin, not from memory; ship `recipe()` + versions in
`run-metadata.json` so any artifact is regenerable from raw data.

---

## Choosing quickly

| Question shape | Recipe |
|---|---|
| "What's going on / where do we lose people?" | R1 → R2 |
| "What happens around event X?" | R3 |
| "What do successful users do between A and B?" | R4 |
| "What kinds of users do we have?" | R5 |
| "Why do users go in circles?" | R6 |
| "When should we intervene?" | R7 |
| "Which segment behaves differently?" | R8 |
| "What is fixing X worth?" | R9 |
| "How do I hand this to a PM?" | R10 |

# Retentioneering 5.0 — verified API map

Every signature below was executed against retentioneering 5.0 (branch `v5-migration`,
July 2026). If your installed version differs, verify before use:

```python
import retentioneering; print(retentioneering.__version__)
```

Authoritative per-method reference: docstrings in the installed package and
https://retentioneering.com/docs. This map exists so you do not have to guess names,
argument conventions, or return shapes.

## 1. Eventstream — the hub object

```python
from retentioneering import Eventstream

stream = Eventstream(df, schema={
    "path_cols":     ["user_id"],            # path identity; nesting allowed: ["user_id","session_id"]
    "event_col":     "event",
    "timestamp_col": "timestamp",            # REQUIRED; parseable datetimes (tz-aware OK)
    "segment_cols":  ["device", "country"],  # categorical labels usable in every diff/overview
    "custom_cols":   ["price"],              # carried along; visible to sql= modes
})
```

- Defaults match columns named `user_id` / `event` / `timestamp` — `Eventstream(df)` just works then.
- Unknown schema keys raise `SchemaConfigError` listing valid keys (trust this error).
- Rows with null path values are rejected loudly. Numeric path ids stay numeric
  (they are NOT cast to str) — join freely with your source frame.
- `stream.df` is a READ-ONLY property (assignment raises). For a materialized copy use
  `stream.to_dataframe(exclude_start_end=True)`.
- Event names may not contain `->` (raises `SchemaConfigError` — it is the path-pattern delimiter).
- Every processor returns a NEW Eventstream (immutable chaining).

First calls on any new dataset:

```python
d = stream.describe(top_events=20)      # dict: schema, shape, date_range, event_frequency,
                                        #       path_stats, segments
d["event_frequency"].attrs              # {'truncated': True, 'n_total_events': N} when cut;
                                        # pass top_events=None to disable truncation
stream.get_event_counts()               # {event: count}
stream.get_segment_levels()             # {segment_col: [values...]}
```

## 2. Data processors (verb-first, chainable)

| Processor | Use for | Notes |
|---|---|---|
| `filter_events(keep=/drop=/func=/sql=)` | row-level filtering | exactly one mode; `sql=` is full DuckDB over alias `eventstream` (CTEs OK) |
| `filter_paths(condition)` | keep whole paths by metric condition tree | leaves `{"op","metric","value","metric_args"}`; branches `and/or/not`; top-level list = AND; raises `EmptyEventstreamError` when nothing survives |
| `add_segment(name, rules=/func=/sql=/funnel_events=/time_range=/metric_bins=)` | add a segment column | `funnel_events` labels each path by the furthest step reached IN ORDER (closed funnel); non-reachers get `"out_of_funnel"`. `metric_bins={"metric":..., "edges"|"quantiles":..., "segment_levels":[...]}` bins paths by a per-path metric — cut points are INTERIOR, so N points give N+1 bins |
| `add_clusters(name, features, method_args={"n_clusters": k}, ...)` | materialize clusters as a segment column | labels are strings `"cluster_0"...`; deterministic (seeded) |
| `add_start_end_events()` | explicit `path_start`/`path_end` rows | idempotent; widgets add them implicitly |
| `rename_events(mapping)` | collapse taxonomies (e.g. `"PLP: *"` families by explicit dict) | unknown keys raise |
| `collapse_events(loops=True | event_groups= | group_col= | bounds=, name=)` | merge a run of events into one; `name` labels it (literal, `{"col": …}`, or cases) | count loops BEFORE collapsing if loops are your subject |
| `split_sessions(timeout="30m" | separator= | bounds={"start_event":…,"end_event":…})` | derive sessions | duration strings need units; column params use `session_col` naming |
| `truncate_paths(start_anchor, end_anchor)` | window each path between two anchors | each anchor is an event name, a spec `{pattern, at, occurrence, offset}`, or a LIST of either (narrowest window wins). A list is how you get both "whichever comes first" and a keep-whole fallback: `end_anchor=["purchase", "path_end"]`. A bare name still DROPS paths missing the anchor |
| `drop_events / drop_segment / edit_events / rename_segment_levels / sample_paths / to_daily_states / urls_to_events / add_events` | as named | `sample_paths(frac=, random_state=)` for stable subsamples |

## 3. Metrics registry — one config format everywhere

Used by `get_metrics(list)`, `filter_paths(condition)`, `cluster_analysis*(features=/overview_metrics=)`,
`segment_overview(metrics=)`.

```python
stream.get_metrics([
    {"metric": "length"},
    {"metric": "duration"},                                            # seconds
    {"metric": "event_count",   "metric_args": {"event": "basket"}},   # SINGLE event → key 'event'
    {"metric": "has_event",     "metric_args": {"event": "purchase"}},
    {"metric": "has_any_event", "metric_args": {"events": ["a","b"]}}, # LIST → key 'events' (OR)
    {"metric": "has_all_events","metric_args": {"events": ["a","b"]}}, # LIST → key 'events' (AND)
    {"metric": "time_between",  "metric_args": {"start_event": "basket",
                                                "end_event": "purchase"}},
    {"metric": "matches_pattern","metric_args": {"pattern": "basket->.*->purchase"}},
])   # -> DataFrame indexed by path id
```

Argument-key convention (this exact split is version-verified):
**single event → `event` (string); multiple events → `events` (list) and only on
`has_any_event` / `has_all_events`.** Wrong keys raise `InvalidMetricConfigError`
naming the requirement.

Other metrics: `active_days`, `first_event_time`, `in_segment` (modes `any/all/event_share`),
`in_segment_bulk` (same check, one column per segment level: omit `segment_levels` for every
level of `segment_name`, omit both for every level of every segment column).
`matches_pattern` is token-wise (no substring false-positives) and order-deterministic.
One position may be a class of events — `[a|b]` any of, `[^a]` anything but, `.` any
event — which replaces merging events with `rename_events` just to ask one question.
`[^a]` and `.` never match `path_start`/`path_end`; a sentinel takes part only when named.
Quantified with `*` a class becomes a restricted gap: `a->[^x]*->b` is "reached b from a
without passing through x". A restricted gap needs an anchor on both sides — name
`path_start`/`path_end` if that is what you mean.
`time_between` measures first occurrence of A to first occurrence of B **globally in the
path** (not "first B after A") — verify fit for your question.

## 4. Tools: widgets + headless twins

Every widget has a headless `<name>_data(...)` twin returning plain data with the same
data parameters. Common widget params: `diff=`, `path_col=`, `height=`, `sidebar_open=`,
`state_file=`.

| Widget | Headless returns | Key params |
|---|---|---|
| `transition_graph` | events×events DataFrame | `edge_weight` ∈ `proba_out, proba_in, count, unique_paths, share_of_total, avg_per_path, time_median, time_q95` |
| `step_matrix` / `step_sankey` | no `path_pattern`: DataFrame; with `path_pattern`: tuple of per-anchor blocks; with pattern+diff: `(blocks, g1_blocks, g2_blocks)` | `max_steps`, `path_pattern=".*->X->.*"` — the anchor event sits at column **0**. `transition_graph` takes `path_pattern` too, but there it only SELECTS paths (no step axis to centre, nothing is cut) |
| `funnel` | dict; each step has `unique_paths`, `conversion_rate` (share of ALL paths) **and `step_conversion_rate`** (step-to-step) | `steps=[...]` — closed/ordered semantics: a path counts at step N only after passing all previous |
| `segment_overview` | DataFrame metrics×segment values | `metrics=[...]` with `agg` |
| `cluster_analysis` | dict: `overview_df`, `silhouette` (`{"params": [...], "silhouette": [...], "best_index", "selected_index"}`), `cluster_labels`, `best_params` | `method_args={"n_clusters": …}` accepts int, list, or range string `"3-8"`; features = metric configs. `select={"n_clusters": 5}` interprets that grid point instead of the top score, keeping the whole grid — the top silhouette is a hint, not a verdict, and near-ties are common. `best_params` always describes the interpreted point and is shaped as `add_clusters` kwargs (`method`/`method_args`/`scaler`), so `add_clusters(name, features, **best_params)` reproduces it |

**Headless-only, no widget:** `get_conversion_rate(start_anchor, end_anchor, within=None, path_col=None)`
→ DataFrame, one row per (start, end) pair: `paths_with_start` (the denominator), `converted`,
`conversion_rate`, `base_rate` (share of ALL paths where the target occurs at all) and `lift`
(= rate / base_rate; **< 1 means the start event makes the outcome LESS likely**). Report the
denominator and the lift, never the rate alone. Both sides take event names or `truncate_paths`
anchor specs (`path_start` / `path_end` are ordinary names — `end_anchor="path_end", within=1`
is an exit rate); a LIST on either side FANS OUT into separate questions, one row per
combination, unlike `truncate_paths` where a list describes one bound. `within` is an int
(events) or a duration string (`"30m"`), measured from the start anchor, far edge inclusive;
`None` = to the end of the path. The unit of observation is the PATH, not the occurrence — a
path where the start happened three times counts once, so per-visit questions ("of N visits,
how many were entrances") need a different tool. Prefer `path_col="session_id"` when the
question is about a visit rather than a person.

**diff semantics:** `diff=(segment_col, v1, v2)` (also `(path_ids1, path_ids2)`; `v2` may be
`"<REST>"`). Returned diff block = **value1 − value2**. Segment values must match exactly
(check `get_segment_levels()`).

## 5. Deliverables: standalone interactive HTML

```python
w = stream.transition_graph(diff=("device", "mobile", "desktop"))
w.export_html("artifacts/graph.html",
              title="Mobile vs desktop",
              analysis="Markdown text; [event_name] references become clickable node links.")
```

- `export_html` is a method of the **widget**, not the stream.
- Files are fully self-contained (~1.2 MB, no CDN) — safe to attach or email.
- A failed recompute raises `WidgetExportError` (no silent empty exports).
- Widgets construct headlessly in plain scripts — Jupyter is not required for export.

## 6. Reproducibility: lineage recipes

Processor chains are recorded and replayable:

```python
processed = stream.filter_events(drop={"event": ["noise"]}).rename_events({"a": "A"})
rec = processed.recipe()                     # JSON-serializable list of ops
replayed = Eventstream.from_recipe(raw_df, rec)   # re-applies the same chain
```

Persist `rec` in run metadata so any artifact can be regenerated from raw data.

## 7. MCP server (agent-driven analysis)

```python
import retentioneering.mcp as mcp
mcp.serve()                    # data-agnostic: agent loads an eventstream on demand
mcp.serve(stream, port=8765)   # or pre-loaded; raises OSError if port is taken
```

Report-building tools (`add_transition_graph` / `add_step_matrix` /
`add_segment_overview`) each register a tab; `get_conversion_rate` is the exception —
it answers a pair question in numbers and registers nothing, so the agent must quote its
figures in backticks (`check_analysis` requires an anchor link for every number that
came from a tab). `playbook("conversion_rate")` carries the procedure.

## 8. Environment notes

- Requires Python ≥ 3.10. Backend is an embedded engine (DuckDB family) — millions of rows
  run in seconds on a laptop; no network needed.
- Interactive widgets from a *source checkout* need the JS bundle built (`make build`,
  Node required). pip-installed wheels ship the bundle.
- Telemetry: the library reports anonymous usage (method names, never data);
  see the docs "Tracking" page for scope and opt-out.

# Gotchas and validation

Two lists with different natures. **API gotchas**: current library behaviors that
surprise; work around them as described. **Analysis integrity**: methodological traps
that produced wrong-but-plausible conclusions in real investigations until a reviewer
caught them; validate against each before shipping numbers.

## A. API gotchas (version-verified)

| # | Behavior | Handle it |
|---|---|---|
| G1 | a bare `truncate_paths(start_anchor="A", end_anchor="B")` DROPS paths missing either anchor | pass a LIST to add a fallback: `end_anchor=["B", "path_end"]` cuts at B where there is one and keeps the rest whole. Comparing two groups, go further and bound BOTH by the same budget — `end_anchor=["B", {"pattern": "A", "offset": 10}]` — or the diff measures window length, not behaviour |
| G2 | No order-only mode: `timestamp_col` is mandatory | for logs with click order but no clock: synthesize `base_date + order * 1s`; then durations/`time_median` are MEANINGLESS — never report them, only step counts |
| G3 | `time_between` = first A to first B **globally**, not first-B-after-A | off-by-few on paths where B precedes A; compute strict semantics via `to_dataframe()` when it matters |
| G4 | `funnel` is closed/ordered only | for presence-based ("did all of these happen, any order") use `has_all_events` via `get_metrics` |
| G5 | Metric arg keys: single event → `event`, lists → `event_groups` (only on `has_any_event`/`has_all_events`) | copy from api-map, not from memory; errors are loud and name the requirement |
| G6 | Cluster result may LACK `best_params` when silhouette is undefined for every candidate (degenerate/duplicate feature rows) | guard `res.get("best_params")`; treat as "no cluster structure" |
| G7 | Cluster labels are strings `"cluster_0"...`; diff values must match segment levels exactly | check `get_segment_levels()` before writing `diff=(...)` |
| G8 | Transition matrix may be integer-typed with NaN gaps | before matrix math: `.astype(float).fillna(0)` |
| G9 | Graph widgets with >~100 nodes open as a dense hairball | teach the edge-weight threshold in the sidebar; or pre-aggregate events into families (`rename_events`) |
| G10 | `stream.df` is read-only | mutate via processors; materialize with `to_dataframe()` |

Fixed in 5.0 — do NOT carry these legacy workarounds forward: identifier quoting for
SQL-reserved column names; nondeterministic `matches_pattern` ordering; `n_clusters`
range strings; silent `describe()` truncation (now flagged in `.attrs`, disable with
`top_events=None`); silent empty widget exports (now `WidgetExportError`); substring
pattern matching; anchor off-by-one in step matrix; `diff` sign (now v1−v2).

## B. Analysis integrity checklist

Run before presenting conclusions. Each item traces to a real wrong-conclusion incident.

1. **Denominators.** Every percentage names its base. `funnel_data.conversion_rate` is
   share of ALL paths; step-to-step is `step_conversion_rate`. *(Incident: "2% conversion"
   nearly shipped where the true step rate was 32%.)*
2. **Population.** Who exactly is in the analysis? Watch entry filters (`truncate_paths`
   drops non-completers, G1) and "engaged users only" thresholds. *(Incident: a ≥10-min
   activity cutoff silently excluded 43% — the weakest users, the ones the decision was
   about; on the full cohort the verdict got harsher, not softer.)*
3. **Survivorship.** Late-stage aggregates describe survivors. Report "of all who
   started / of those who reached" side by side. Ratings/feedback exist only for
   finishers.
4. **Exposure confound.** Longer paths contain more of everything (loops, feature
   touches) AND convert more. Any "users who did X convert better" claim needs
   stratification by path length or a matched design.
5. **Target leakage.** No outcome events in clustering features (R5) or in
   sequence/pattern features computed over the full path — cut paths at the first
   outcome before mining predictive patterns.
6. **Structural vs behavioral lift.** If the flow FORCES step A before outcome B, "users
   who did A convert ×100" is architecture, not behavior. Say which one you measured.
7. **Small cells.** Shares without `n` are noise bait; flag cells with n below ~30 and
   add intervals (Wilson for proportions) when a decision hangs on them.
8. **Right/left censoring.** Log windows cut both ends: "never returned" near the window
   edge and "first session" of users active before the window are both suspect.
9. **Correlation discipline.** Observational path data supports "associated with", not
   "drives", unless there is an experiment. End reports with hypotheses + suggested A/B,
   not causal claims.
10. **Numbers ↔ visuals.** Every caption number comes from the headless twin of the
    widget shown; write `analysis=` texts after conclusions are frozen.

## C. Self-checks worth automating

```python
# conservation: filters explain themselves
before = stream.describe()["shape"]; after = filtered.describe()["shape"]
# funnel sanity: step counts are non-increasing
# diff sanity: diff block equals g1 - g2 on a sample cell
# determinism: rerun the pipeline; assert identical key outputs
# reproducibility: Eventstream.from_recipe(raw_df, processed.recipe()) equals processed
```

#!/usr/bin/env python
"""Profile an event log and suggest a Retentioneering Eventstream schema.

Usage:
    python inspect_event_log.py PATH [--sep SEP] [--user-col C] [--event-col C]
                                [--ts-col C] [--sample-rows N] [--out DIR]

Reads CSV/TSV/Parquet, infers user/event/timestamp columns when not given,
runs data-quality checks, writes artifacts/data-profile.json, and prints a
human report plus a ready-to-paste Eventstream(...) snippet.

Only requires pandas. If retentioneering is importable, its version is recorded.
"""

from __future__ import annotations

import argparse
import json
import os
import sys

import pandas as pd

USER_HINTS = (
    "user_id",
    "user",
    "client_id",
    "customer_id",
    "uid",
    "session_id",
    "hashed",
    "player",
    "visitor",
)
EVENT_HINTS = ("event_name", "event", "action", "event_type_name", "page", "screen")
TS_HINTS = ("timestamp", "event_time", "time", "datetime", "ts", "date")
SEGMENT_HINTS = (
    "device",
    "platform",
    "country",
    "source",
    "utm",
    "channel",
    "plan",
    "os",
    "browser",
    "campaign",
    "medium",
    "group_name",
    "variant",
)


def sniff_sep(path: str) -> str:
    with open(path, "r", encoding="utf-8", errors="replace") as fh:
        head = fh.readline()
    for sep in ("\t", ";", ",", "|"):
        if head.count(sep) >= 2:
            return sep
    return ","


def pick(columns, hints, taken):
    lowered = {c.lower(): c for c in columns}
    for h in hints:
        for lc, orig in lowered.items():
            if h == lc and orig not in taken:
                return orig
    for h in hints:
        for lc, orig in lowered.items():
            if h in lc and orig not in taken:
                return orig
    return None


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("path")
    ap.add_argument("--sep", default=None)
    ap.add_argument("--user-col", default=None)
    ap.add_argument("--event-col", default=None)
    ap.add_argument("--ts-col", default=None)
    ap.add_argument(
        "--sample-rows",
        type=int,
        default=None,
        help="read only the first N rows (for very large files)",
    )
    ap.add_argument("--out", default="artifacts")
    a = ap.parse_args()

    if a.path.endswith((".parquet", ".pq")):
        df = pd.read_parquet(a.path)
        sep = None
    else:
        sep = a.sep or sniff_sep(a.path)
        df = pd.read_csv(a.path, sep=sep, nrows=a.sample_rows, low_memory=False)

    cols = list(df.columns)
    taken: set = set()
    user_col = a.user_col or pick(cols, USER_HINTS, taken)
    taken.add(user_col)
    event_col = a.event_col or pick(cols, EVENT_HINTS, taken)
    taken.add(event_col)
    ts_col = a.ts_col or pick(cols, TS_HINTS, taken)
    taken.add(ts_col)
    segment_cols = [
        c
        for c in cols
        if c not in taken
        and any(h in c.lower() for h in SEGMENT_HINTS)
        and df[c].nunique(dropna=True) <= max(50, len(df) // 100)
    ]

    problems: list[str] = []
    if not user_col or not event_col:
        problems.append(
            "could not infer user/event columns — pass --user-col/--event-col"
        )

    profile: dict = {
        "path": os.path.abspath(a.path),
        "sep": sep,
        "rows_read": int(len(df)),
        "sampled": a.sample_rows is not None,
        "columns": cols,
        "inferred": {
            "user_col": user_col,
            "event_col": event_col,
            "ts_col": ts_col,
            "segment_candidates": segment_cols,
        },
    }

    if user_col and event_col:
        nulls_user = int(df[user_col].isna().sum())
        nulls_event = int(df[event_col].isna().sum())
        profile["n_paths"] = int(df[user_col].nunique(dropna=True))
        profile["n_event_types"] = int(df[event_col].nunique(dropna=True))
        profile["nulls"] = {"user": nulls_user, "event": nulls_event}
        if nulls_user:
            problems.append(
                f"{nulls_user} rows with null {user_col} — Eventstream will "
                "reject them; decide drop/repair explicitly"
            )
        top = df[event_col].value_counts()
        profile["top_events"] = {str(k): int(v) for k, v in top.head(25).items()}
        share = float(top.iloc[0]) / max(len(df), 1)
        if share > 0.30:
            problems.append(
                f"event '{top.index[0]}' is {share:.0%} of all rows — likely "
                "noise (heartbeat/dialogue); consider drop_events before path tools"
            )
        plen = df.groupby(user_col, dropna=True).size()
        profile["path_length"] = {
            q: float(plen.quantile(x))
            for q, x in [("p50", 0.5), ("p90", 0.9), ("p99", 0.99)]
        } | {"max": int(plen.max()), "n_len1": int((plen == 1).sum())}
        if plen.max() > 50 * max(plen.median(), 1):
            problems.append(
                f"max path length {plen.max()} vs median {plen.median():.0f} "
                "— check bots/shared accounts"
            )

    ts_ok = False
    if ts_col:
        parsed = pd.to_datetime(df[ts_col], errors="coerce", utc=True, format="mixed")
        bad = int(parsed.isna().sum())
        ts_ok = bad == 0
        profile["timestamp"] = {
            "unparseable": bad,
            "min": str(parsed.min()),
            "max": str(parsed.max()),
        }
        if bad:
            problems.append(f"{bad} unparseable timestamps in {ts_col}")
        if user_col:
            ties = int(df.assign(_p=parsed).duplicated([user_col, "_p"]).sum())
            profile["timestamp"]["within_path_ties"] = ties
            if ties:
                problems.append(
                    f"{ties} same-timestamp ties within paths — ensure a "
                    "stable secondary order (row order is preserved)"
                )
        dup = int(df.duplicated([c for c in (user_col, event_col, ts_col) if c]).sum())
        profile["exact_duplicates"] = dup
        if dup:
            problems.append(
                f"{dup} exact duplicate (user,event,ts) rows — dedupe or justify"
            )
    else:
        problems.append(
            "no timestamp column found — if only an order column exists, "
            "synthesize base_date + order*1s and NEVER report durations (gotcha G2)"
        )

    try:
        import retentioneering

        profile["retentioneering_version"] = retentioneering.__version__
    except Exception:
        profile["retentioneering_version"] = None
        problems.append("retentioneering not importable in this environment")

    profile["problems"] = problems

    os.makedirs(a.out, exist_ok=True)
    out_path = os.path.join(a.out, "data-profile.json")
    with open(out_path, "w", encoding="utf-8") as fh:
        json.dump(profile, fh, indent=2, ensure_ascii=False, default=str)

    print(
        f"rows={profile['rows_read']:,}  paths={profile.get('n_paths', '?'):,}  "
        f"event_types={profile.get('n_event_types', '?')}"
    )
    if ts_col and "timestamp" in profile:
        print(f"period: {profile['timestamp']['min']} .. {profile['timestamp']['max']}")
    print(
        f"inferred: user={user_col}  event={event_col}  ts={ts_col}  "
        f"segments={segment_cols}"
    )
    for p in problems:
        print(f"  ⚠ {p}")
    print(f"\nprofile written: {out_path}\n")
    print("suggested schema:\n")
    seg = ", ".join(f'"{c}"' for c in segment_cols)
    print(f"""from retentioneering import Eventstream
stream = Eventstream(df, schema={{
    "path_cols": ["{user_col}"],
    "event_col": "{event_col}",
    "timestamp_col": "{ts_col}",
    "segment_cols": [{seg}],
}})
stream.describe()""")
    return 1 if (not user_col or not event_col or not ts_ok) else 0


if __name__ == "__main__":
    sys.exit(main())