investigate-metric

Box plot metrics playbook

For TrendsQuery insights with trendsFilter.display = "BoxPlot". The metric is the distribution of a numeric math_property per bucket (min, p25, median, mean, p75, max), not a count. Box plots silently drop breakdownFilter (it's in NON_BREAKDOWN_DISPLAY_TYPES), so segmentation and tail drilldown route through HogQL.

1. Which statistic moved

Read boxplot_data per bucket and classify before hypothesizing:

• Median moved — behavioral change in the middle of the population.
• IQR widened — a new slow / fast segment appeared.
• IQR narrowed — a tail was removed (feature gate, rate limit, dependency outage).
• Outliers changed — toggle trendsFilter.excludeBoxPlotOutliers. If the shift disappears, it's outlier-driven; if it persists, the bulk moved.
• Mean / median diverge — skew changed; tails are pulling the mean.

Tail-driven and bulk shifts have different next steps.

2. Zoom

Rerun at interval: "hour" on the anomaly window. A one-hour IQR compression usually points to a deploy or incident; a sustained shift suggests a broader population or tracking change.

3. Segment

Two options instead of breakdownFilter:

Parallel series — one EventsNode per segment value, filtered via properties:

posthog:query-trends
{
  "kind": "TrendsQuery",
  "dateRange": { "date_from": "-30d" },
  "interval": "day",
  "trendsFilter": { "display": "BoxPlot" },
  "series": [
    {
      "kind": "EventsNode", "event": "checkout_completed",
      "math": "p75", "math_property": "order_value",
      "properties": [{ "type": "event", "key": "plan", "value": "free", "operator": "exact" }]
    },
    {
      "kind": "EventsNode", "event": "checkout_completed",
      "math": "p75", "math_property": "order_value",
      "properties": [{ "type": "event", "key": "plan", "value": "paid", "operator": "exact" }]
    }
  ]
}

HogQL — see shared-patterns.md (./shared-patterns.md#hogql-quantile-template). Better when the segment has many values. Watch for noisy quartiles on small n.

Once a candidate segment looks elevated, don't conclude yet. A segment can look guilty for two very different reasons:

• The segment's measurement changed at the anomaly start (real cause).
• The segment was always elevated, and its share of total volume grew at the anomaly start (cohort composition change — same numbers, different fix).

Two cheap checks before concluding:

• Pre-anomaly baseline. Run the same per-segment quantiles on a window before the anomaly. If the elevated segment was already elevated, it's composition not measurement.
• Cross-tab. If two dimensions both look elevated (e.g. host + lib_version), GROUP BY both in HogQL to see whether one is causal or they're correlated.

4. Tail actor drilldown

posthog:query-trends-actors can't select by percentile. Use HogQL with a quantile filter:

SELECT distinct_id, max(toFloat(properties.order_value)) AS max_value
FROM events
WHERE event = 'checkout_completed'
  AND timestamp >= '2026-04-15 00:00:00'
  AND timestamp <  '2026-04-16 00:00:00'
  AND toFloat(properties.order_value) > (
    SELECT quantile(0.9)(toFloat(properties.order_value))
    FROM events
    WHERE event = 'checkout_completed'
      AND timestamp >= '2026-04-15 00:00:00'
      AND timestamp <  '2026-04-16 00:00:00'
  )
GROUP BY distinct_id
ORDER BY max_value DESC
LIMIT 50

Mirror with < quantile(0.1)(...) for the lower tail. Feed IDs into posthog:query-session-recordings-list for UI-shaped shifts.

5. Errors / logs

Did a failure mode truncate the distribution? Timeouts killing the slow tail, a validation error blocking expensive orders. Confirm timing, plausible mechanism, and user overlap.

6. Cohort composition

Run posthog:query-lifecycle on the same event. A distribution shift often reflects a mix change — an influx of new free-tier users can pull p75 order value down while no existing user changed behavior.

Common causes

Hypothesis taxonomy with confirming queries. Rank by evidence count when writing findings.

Release / deploy / flag rollout

A version, flag, or experiment shipped near the anomaly start.

• Breakdown on app_version / $lib_version — shift concentrated in one version is strong evidence.
• For a flag: breakdown on $feature/<flag_key> separates exposed from control.

Suggest pausing or reverting; offer posthog:annotation-create if no annotation exists.

Marketing / traffic-source shift

A campaign started or ended, or source mix changed.

• Breakdown on utm_source, utm_medium, utm_campaign, $referring_domain.
• A composition change can leave overall count stable while conversion moves — break the conversion metric down by source.

Tracking regression

The measurement changed, not the metric.

• Total events vs. unique users — stable users + falling events = fire condition changed.
• An adjacent stable event while the target fell = target's tracking changed.
• Breakdown on $lib_version — concentrated drop = SDK regression.
• posthog:read-data-schema (kind: "events") for recently renamed / deprecated events.

Cohort / lifecycle shift

Same product, different mix of users. New-user influx pulls engagement metrics down.

• posthog:query-lifecycle — change in new / returning / resurrecting / dormant mix.
• posthog:query-retention comparing affected-period cohorts to prior.

Split the metric per lifecycle status in findings.

Seasonality / day-of-week artifact

Weekend dip, holiday trough, end-of-quarter spike. The compare_to_prior_periods.py (../scripts/compare_to_prior_periods.py) script catches this directly.

Platform / device / browser-specific

JS error on a Safari release, mobile crash on a specific OS, regional CDN issue.

• Breakdown on $browser, $browser_version, $os, $device_type, $geoip_country_code.
• Cross-check with posthog:query-error-tracking-issues-list.

Rate limit / upstream outage

A consumer hit a quota or an upstream dependency degraded.

• posthog:query-trends on API error events.
• posthog:query-logs for an error surge in the window.

Upstream data integration

For warehouse-backed metrics: schema change, pipeline failure, view altered. The investigation tools can't confirm this directly — flag as a candidate when no product-side cause fits and recommend the user check pipeline health.

Funnel metrics playbook

For "conversion fell", "drop-off increased at step X".

0. Rule out an incomplete latest bucket

If (now − latest_bucket_start) < funnel_window, users in the bucket haven't had time to convert. Tells:

• Drop is uniform across breakdowns (real regressions usually aren't).
• First-step volume is stable or growing.
• Prior bucket sits on the historical baseline.

If incomplete: report it as the cause; suggest setting dateRange.date_to one funnel window in the past, and offer to annotate.

1. Which step regressed

FunnelsQuery doesn't support compareFilter — run the funnel twice with date ranges of equal length and compare.

posthog:query-funnel
{
  "kind": "FunnelsQuery",
  "dateRange": { "date_from": "-7d" },
  "series": [
    { "kind": "EventsNode", "event": "signed up" },
    { "kind": "EventsNode", "event": "completed onboarding" },
    { "kind": "EventsNode", "event": "first purchase" }
  ],
  "funnelsFilter": { "funnelWindowInterval": 7, "funnelWindowIntervalUnit": "day" }
}

Then rerun with "dateRange": { "date_from": "-14d", "date_to": "-7d" }.

2. Entries or completions?

Run posthog:query-trends on the events at step N-1 and step N. Steady entries with falling completions = problem at that step. Falling entries = problem upstream.

3. Who dropped off

posthog:query-trends-actors only accepts a trends source. Run trends on step N-1 completions and step N completions for the same window, drill into actors of each, and diff — users present in the first but not the second are the drop-offs.

4. Errors / logs

Filter posthog:query-error-tracking-issues-list and posthog:query-logs to the surface where step N lives — a 500 on the submit endpoint can plausibly cause failures; a console warning elsewhere usually can't.

5. What they do instead

posthog:query-paths with endPoint set to the failing step. Paths that don't reach that endpoint show where users bail.

Lifecycle metrics playbook

For "new users fell", "returning users crashed", "resurrecting stopped coming back".

1. Identify which status moved

Run posthog:query-lifecycle with the user's metric. Read which of new / returning / resurrecting / dormant changed.

2. Segment

AssistantLifecycleQuery doesn't support breakdownFilter. To compare segments, run posthog:query-lifecycle once per segment with properties filters on the series, or focus on a single status with lifecycleFilter.toggledLifecycles: ["new"] / ["returning"].

posthog:query-lifecycle
{
  "kind": "LifecycleQuery",
  "dateRange": { "date_from": "-30d" },
  "interval": "day",
  "series": [
    {
      "kind": "EventsNode",
      "event": "$pageview",
      "properties": [{ "type": "event", "key": "$geoip_country_code", "value": "US", "operator": "exact" }]
    }
  ]
}

3. Diagnose by status

• New-user drop — find the project's first-session event via read-data-schema ($session_start, $pageview, or a signup event), then run posthog:query-paths from there to see where onboarding loses people.
• Returning-user drop — posthog:query-trends on the cohort's key engagement events. Use interval zoom + actor drilldown if a specific day stands out.
• Resurrecting drop — usually external. Re-check annotations and re-engagement campaign / email events.

Paths metrics playbook

For "the path from X to Y changed", "a different dominant path emerged". Paths are shape metrics — the change is usually a shift in navigation between events, not a single scalar moving.

1. What changed

PathsQuery doesn't support compareFilter. Run posthog:query-paths twice with date ranges of equal length and compare which edges gained or lost volume.

posthog:query-paths
{
  "kind": "PathsQuery",
  "dateRange": { "date_from": "-7d" },
  "pathsFilter": {
    "includeEventTypes": ["$pageview"],
    "startPoint": "/home",
    "endPoint": "/checkout",
    "edgeLimit": 50
  }
}

Then rerun with "dateRange": { "date_from": "-14d", "date_to": "-7d" }.

2. Endpoint volume

A "drop in paths A → B" is often a drop in A or B alone. Run posthog:query-trends on each endpoint event separately. If either moved, drop into the trend playbook on that event instead.

3. Wrong tool?

If the user's actual question is conversion rate, paths is the wrong tool — build a funnel and run the funnel-playbook (./funnel-playbook.md). Paths is for shape, not conversion.

4. Segment

AssistantPathsQuery doesn't support breakdownFilter. Filter via top-level properties and rerun per segment. A path shape that differs sharply across browsers / countries / plans is evidence the change is segment-specific.

5. Recordings + errors

Pull session recordings for users on the new dominant edge — usually surfaces the UI change faster than more queries. Cross-check query-error-tracking-issues-list for errors on the page where users diverge.

Retention metrics playbook

For "week-1 retention regressed", "March cohort isn't coming back".

0. Config sanity check

• If totalIntervals exceeds the date range, the tail cells are always zero and look like a drop. Match totalIntervals to the range.
• The most recent cohort hasn't had its full retention window yet — note as partial rather than a regression.

1. Isolate the cohort

targetEntity / returningEntity use {type: "events", name: "<event>"} and nest in retentionFilter. Compare affected cohort(s) to prior baselines side by side.

posthog:query-retention
{
  "kind": "RetentionQuery",
  "dateRange": { "date_from": "-90d" },
  "retentionFilter": {
    "targetEntity": { "type": "events", "name": "$pageview" },
    "returningEntity": { "type": "events", "name": "$pageview" },
    "totalIntervals": 8,
    "period": "Week",
    "retentionType": "retention_first_time"
  }
}

2. Event vs. users

Is the drop in the activity event itself or in the cohort doing it? Create or reuse a cohort for the affected period (posthog:cohorts-create / -list), then run posthog:query-trends on the activity event filtered to that cohort.

3. Split the dropout

Run posthog:query-lifecycle scoped to the affected cohort to separate "never retained" (new users who didn't return) from "lost later" (returning users who churned).

Shared patterns

Recipes used across playbooks.

Contents

• When to reach for HogQL
• HogQL insights
• HogQL quantile template
• Property discovery
• Related-metrics sweep
• Breakdown dimensions
• Interpreting breakdown results
• Interval zoom
• Actor drilldown
• Session recordings
• Error / logs cross-check

When to reach for HogQL

Use the typed query tools by default. Use posthog:execute-sql only when the question can't be expressed structurally:

• Ratios across different events.
• Joins with data-warehouse tables.
• Custom aggregations like quantile, arrayJoin, regex extraction.

HogQL insights

For saved insights with query.kind === "HogQLQuery", the playbook routing is shape-based rather than kind-based. Read the SQL and pick the closest playbook:

• count(...) GROUP BY toStartOfDay(...) or similar count-over-time → trend playbook.
• Multi-step windowFunnel or sequential filtering → funnel playbook.
• Cohort-keyed return aggregates → retention playbook.

Run the insight's SQL through posthog:execute-sql to get the data, then follow the chosen playbook's steps using the typed tools where they fit. Drop back to execute-sql for the breakdown / drilldown variants when the original SQL has shape the typed schema can't express.

HogQL quantile template

For per-bucket distribution stats with a segment dimension (box plots can't take breakdownFilter).

SELECT
    toStartOfDay(timestamp) AS day,
    properties.plan         AS segment,
    quantiles(0.25, 0.5, 0.75, 0.9)(toFloat(properties.order_value)) AS q,
    count()                 AS n
FROM events
WHERE event = 'checkout_completed'
  AND timestamp >= now() - INTERVAL 30 DAY
GROUP BY day, segment

• Cast toFloat(properties.X) — properties are loose-typed.
• Report n alongside quantiles; a p75 swing on small cells is usually noise.

Property discovery

Use posthog:read-data-schema to discover events and properties before filtering or breaking down. The query.kind field selects:

• events — list events for fuzzy-matching by name.
• event_properties with event_name — properties on a specific event.
• entity_properties with entity: "person" — person properties.
• event_property_values with event_name + property_name — sample values.

Related-metrics sweep

Before going deep on one metric, run the same anomaly-window query on 2–3 adjacent metrics — upstream funnel steps, sibling events, total event volume, parent-event counts. If they all moved together, the cause is broader than the specific metric (ingestion gap, cohort shift, tracking regression). If only the target metric moved, the investigation is correctly scoped.

Use when the metric sits inside a larger pipeline (a funnel step, a retention activity event, a derived rate) and you want to rule out an upstream cause cheaply.

posthog:query-trends
{
  "kind": "TrendsQuery",
  "dateRange": { "date_from": "-30d" },
  "interval": "day",
  "series": [
    { "kind": "EventsNode", "event": "$pageview", "math": "total" },
    { "kind": "EventsNode", "event": "user signed up", "math": "total" },
    { "kind": "EventsNode", "event": "first team event ingested", "math": "total" }
  ]
}

A single multi-series query is cheaper than three breakdowns and the visual alignment (or lack of it) answers the question immediately.

Breakdown dimensions

A few candidate dimensions, roughly in order of signal:

• $feature/<flag_key> — highest signal post-release.
• $browser, $os, $device_type, $geoip_country_code — for platform / regional issues.
• app_version, $lib_version — for SDK regressions.
• is_identified, $is_first_session, plan / tier — for user-state issues.
• Custom event properties from read-data-schema — usually most diagnostic.

Interpreting breakdown results

Rank by absolute delta, not %. Pipe through breakdown_attribution.py (./../scripts/breakdown_attribution.py) — it ranks by contribution and detects offsetting cases (aggregate flat, segments moved oppositely).

If no breakdown isolates the delta the cause is system-wide (deploy, tracking, infra), not segment-specific. Try a compound breakdown of up to 3 properties for interaction effects (e.g. one browser × one country).

If event volume per interval < ~100, percentages are unreliable — report absolutes too.

When a segment looks guilty, check its pre-anomaly baseline before concluding: the segment may have always behaved that way, with its share of volume just growing at the anomaly start (cohort composition change, not a real regression).

Interval zoom

When a daily point looks anomalous, rerun the query at interval: "hour" scoped to that day. A one-hour cliff is an incident; a sustained shift is something broader. Use interval: "minute" for tight incident windows.

Actor drilldown

posthog:query-trends-actors only accepts a trends source. The selector fields are day, series, and (if breakdown) breakdown:

posthog:query-trends-actors
{
  "kind": "InsightActorsQuery",
  "source": {
    "kind": "TrendsQuery",
    "dateRange": { "date_from": "2026-03-10", "date_to": "2026-03-10" },
    "series": [{ "kind": "EventsNode", "event": "$pageview", "math": "dau" }],
    "breakdownFilter": { "breakdowns": [{ "property": "plan", "type": "event" }] }
  },
  "day": "2026-03-10",
  "breakdown": "free"
}

Session recordings

For UI-shaped drops, pull recordings matching the affected segment via posthog:query-session-recordings-list. Watching three or four is often faster than running more queries. Fetch individual ones with posthog:session-recording-get.

Error / logs cross-check

Run posthog:query-error-tracking-issues-list and posthog:query-logs for the anomaly window. A correlated error is a candidate, not a conclusion — confirm three things:

1. Timing — error volume aligns with the metric movement.
2. Plausible mechanism — the error actually affects the metric's surface (a 500 on a submit endpoint can; a console warning usually can't).
3. User overlap — affected users overlap with users hitting the error.

If any check fails, note the error as coincidental and move on.

Stickiness metrics playbook

For "DAU/MAU dropped", "sessions per week fell", "engagement decayed".

1. Segment

AssistantStickinessQuery doesn't support breakdownFilter. Run posthog:query-stickiness once per segment with property filters on the series, and compare side-by-side.

posthog:query-stickiness
{
  "kind": "StickinessQuery",
  "dateRange": { "date_from": "-30d" },
  "interval": "day",
  "series": [
    {
      "kind": "EventsNode",
      "event": "$pageview",
      "properties": [{ "type": "person", "key": "plan", "value": "pro", "operator": "exact" }]
    }
  ]
}

2. Drill into the low-stickiness segment

Run posthog:query-trends on a key engagement event filtered to that segment, then posthog:query-trends-actors on the trend. Pull recordings for a handful.

3. What sticky users do that non-sticky users don't

With a sticky and a non-sticky cohort, run posthog:query-trends on candidate core events scoped to each (filter via properties cohort filter). Events where the two diverge are the ones driving stickiness.

Trend metrics playbook

For "DAU dropped", "revenue spiked", "clicks fell after the release".

1. Zoom in

Rerun posthog:query-trends at interval: "hour" scoped to the suspicious day(s). A one-hour cliff is an incident or deploy; a full-day shift is a broader cause (campaign, cohort change, tracking regression).

2. Break down

Try several breakdowns. Use read-data-schema to find candidate properties. Pipe results through breakdown_attribution.py (../scripts/breakdown_attribution.py) — it ranks by absolute delta and flags offsetting moves. If no segment isolates the delta, the cause is system-wide.

posthog:query-trends
{
  "kind": "TrendsQuery",
  "dateRange": { "date_from": "-30d" },
  "interval": "day",
  "series": [{ "kind": "EventsNode", "event": "$pageview", "math": "dau" }],
  "breakdownFilter": {
    "breakdowns": [{ "property": "plan", "type": "event" }],
    "breakdown_limit": 10
  }
}

3. Identify affected users

Run posthog:query-trends-actors on the anomalous bucket (or breakdown value that moved). For UI-shaped drops, pull session recordings for the same segment.

4. Errors / logs

posthog:query-error-tracking-issues-list and posthog:query-logs for the window. Confirm timing, plausible mechanism, and user overlap before treating as the cause.

5. Cohort composition

Run posthog:query-lifecycle on the same event. If the drop is concentrated in one status (new users didn't arrive, dormant users didn't resurrect), the user mix changed rather than per-user behavior.

#!/usr/bin/env python3
"""Rank breakdown segments by absolute contribution to a metric delta.

Reads the JSON output of a `posthog:query-trends` call with `breakdownFilter`
on stdin or as a file path. The payload's `results` array contains one entry
per breakdown value, each with a `data` and `days` series.

Implements the "interpreting breakdown results" guidance from
shared-patterns.md: a 50% swing on a series that's 1% of volume only explains
0.5% of the aggregate delta. Sort by absolute contribution, not by percentage.

Usage:
    python3 scripts/breakdown_attribution.py < breakdown_result.json
    python3 scripts/breakdown_attribution.py breakdown_result.json

Optional env:
    WINDOW=N            How many trailing intervals (days/hours/etc., depending
                        on the input's interval) to treat as the anomaly window.
                        Default 7. The preceding N intervals are the baseline.
    TOP=N               Show only the top N segments (default 10).
"""

from __future__ import annotations

import json
import os
import sys


def load_input() -> dict:
    if len(sys.argv) > 1:
        with open(sys.argv[1]) as f:
            raw = f.read()
    else:
        raw = sys.stdin.read()
    parsed = json.loads(raw)
    if isinstance(parsed, list) and parsed and parsed[0].get("type") == "text":
        parsed = json.loads(parsed[0]["text"])
    return parsed


def fmt(v: float, signed: bool = False) -> str:
    if signed:
        prefix = "+" if v >= 0 else "-"
    else:
        prefix = "" if v >= 0 else "-"
    a = abs(v)
    if a >= 1_000_000:
        return f"{prefix}{a / 1_000_000:.2f}M"
    if a >= 1_000:
        return f"{prefix}{a / 1_000:.1f}K"
    return f"{prefix}{a:,.0f}"


def fmt_pct(v: float) -> str:
    # `v == v` is False only when v is NaN.
    return f"{v:+.1f}%" if v == v else "n/a"


def main() -> int:
    window = int(os.environ.get("WINDOW", "7"))
    top = int(os.environ.get("TOP", "10"))

    payload = load_input()
    results = payload.get("results") or payload.get("result") or []
    if not results:
        raise SystemExit("No results in payload — is this a breakdown trends response?")

    rows = []
    total_anomaly = 0.0
    total_baseline = 0.0

    for series in results:
        data = series.get("data") or []
        if len(data) < 2 * window:
            print(
                f"warn: series '{series.get('breakdown_value', series.get('label'))}' "
                f"has {len(data)} points but window*2={2 * window} — "
                "skipping (extend dateRange).",
                file=sys.stderr,
            )
            continue
        baseline = sum(data[-2 * window : -window])
        current = sum(data[-window:])
        delta = current - baseline
        total_anomaly += current
        total_baseline += baseline

        seg = series.get("breakdown_value")
        if seg is None or seg == "":
            seg = series.get("label", "(none)")
        if isinstance(seg, list):
            seg = " / ".join(str(x) for x in seg)
        rows.append({
            "segment": str(seg),
            "baseline": baseline,
            "current": current,
            "delta": delta,
            "pct": (delta / baseline * 100) if baseline else float("nan"),
        })

    if not rows:
        raise SystemExit(
            "No usable series — every breakdown had fewer than 2 windows of data. "
            "Run with a wider dateRange."
        )

    rows.sort(key=lambda r: abs(r["delta"]), reverse=True)
    total_delta = total_anomaly - total_baseline

    print(f"# Breakdown attribution — last {window} intervals vs preceding {window} intervals")
    print()
    print(f"Aggregate: {fmt(total_baseline)} → {fmt(total_anomaly)}  ({fmt(total_delta, signed=True)}, "
          f"{fmt_pct((total_delta / total_baseline * 100) if total_baseline else float('nan'))})")
    print()
    print("Segments ranked by **absolute** delta contribution:")
    print()
    print("| Segment | Baseline | Current | Δ | Δ% | Share of total Δ |")
    print("| --- | ---: | ---: | ---: | ---: | ---: |")

    for r in rows[:top]:
        share = (r["delta"] / total_delta * 100) if total_delta else float("nan")
        print(
            f"| {r['segment']} "
            f"| {fmt(r['baseline'])} "
            f"| {fmt(r['current'])} "
            f"| {fmt(r['delta'], signed=True)} "
            f"| {fmt_pct(r['pct'])} "
            f"| {fmt_pct(share)} |"
        )

    print()
    # If the aggregate barely moved but segments did, segments are offsetting.
    # That's a different diagnostic than "one segment absorbs the delta".
    aggregate_pct = (total_delta / total_baseline * 100) if total_baseline else 0
    largest_segment_move = max(abs(r["delta"]) for r in rows)
    aggregate_is_quiet = abs(aggregate_pct) < 5 and largest_segment_move > abs(total_delta) * 2

    if aggregate_is_quiet:
        print(
            "**Aggregate barely moved but individual segments did — segments are "
            "offsetting each other. Investigate the largest movers separately rather "
            "than as a 'share of total delta'.**"
        )
        return 0

    top_row = rows[0]
    top_share = (top_row["delta"] / total_delta * 100) if total_delta else float("nan")
    if total_delta and abs(top_share) >= 50:
        print(
            f"**Top segment '{top_row['segment']}' absorbs "
            f"{abs(top_share):.0f}% of the aggregate delta — strong segment signal.**"
        )
    elif total_delta and sum(abs(r["delta"]) for r in rows[:3]) / abs(total_delta) >= 0.7:
        print(
            "**Top 3 segments account for ≥70% of the delta — investigate what they share.**"
        )
    else:
        print(
            "**No single segment dominates — the cause is likely system-wide "
            "(deploy, tracking, infra) rather than segment-specific.**"
        )

    return 0


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

#!/usr/bin/env python3
"""Compare a recent series to comparable prior periods, accounting for cycles.

Reads the JSON output of a `posthog:query-trends` (or similar) call on stdin or
as a file path. Auto-detects interval and picks the right cycle:

    minute → no cycle, rolling (last N vs preceding N)
    hour   → weekly cycle (168 buckets, weekday × hour-of-day)
    day    → weekly cycle (7 buckets, weekday)
    week   → no cycle, sequential
    month  → no cycle, sequential

Use after step 2.1 of SKILL.md to resolve the variance question (step 2.2).

Usage:
    python3 scripts/compare_to_prior_periods.py < query_result.json
    python3 scripts/compare_to_prior_periods.py query_result.json

Optional env:
    TOLERANCE=N.NN  Fraction outside prior min/max counted as still "in range"
                    (default 0.10 — i.e. 10% wiggle on each side of the band).
    TOP=N           For hourly output, how many most-deviated points to show
                    (default 10).
    RECENT=N        How many recent intervals to evaluate (default: one cycle).
"""

from __future__ import annotations

import json
import os
import sys
from datetime import datetime
from statistics import median


WEEKDAYS = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]


def load_input() -> dict:
    if len(sys.argv) > 1:
        with open(sys.argv[1]) as f:
            raw = f.read()
    else:
        raw = sys.stdin.read()
    parsed = json.loads(raw)
    if isinstance(parsed, list) and parsed and parsed[0].get("type") == "text":
        parsed = json.loads(parsed[0]["text"])
    return parsed


def extract_series(payload: dict) -> tuple[list[str], list[float], str, str]:
    """Return (days, data, label, interval)."""
    results = payload.get("results") or payload.get("result") or []
    if not results:
        raise SystemExit("No results in payload — is this a trends query response?")
    series = results[0]
    days = series.get("days") or []
    data = series.get("data") or []
    label = series.get("label") or series.get("custom_name") or "metric"

    interval = (series.get("filter") or {}).get("interval")
    if not interval:
        interval = (payload.get("query") or {}).get("interval")
    if not interval:
        interval = infer_interval_from_days(days)

    if len(days) != len(data):
        raise SystemExit(f"days ({len(days)}) and data ({len(data)}) length mismatch")
    return days, data, label, interval


def infer_interval_from_days(days: list[str]) -> str:
    """Best-effort interval detection from the gap between days[0] and days[1]."""
    if len(days) < 2:
        return "day"
    a = parse_dt(days[0])
    b = parse_dt(days[1])
    if a is None or b is None:
        return "day"
    delta = abs((b - a).total_seconds())
    if delta < 120:
        return "minute"
    if delta < 7200:
        return "hour"
    if delta < 60 * 60 * 30:  # < ~30 hours = day
        return "day"
    if delta < 60 * 60 * 24 * 10:
        return "week"
    return "month"


def parse_dt(s: str) -> datetime | None:
    s = s.replace("Z", "+00:00")
    for fmt in (None, "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
        try:
            return datetime.fromisoformat(s) if fmt is None else datetime.strptime(s, fmt)
        except ValueError:
            continue
    return None


def fmt(v: float) -> str:
    a = abs(v)
    sign = "-" if v < 0 else ""
    if a >= 1_000_000:
        return f"{sign}{a / 1_000_000:.2f}M"
    if a >= 1_000:
        return f"{sign}{a / 1_000:.1f}K"
    return f"{sign}{a:,.0f}"


def classify(value: float, prior: list[float], tolerance: float) -> tuple[str, float]:
    """Return (verdict, deviation_pct).

    deviation_pct is signed; 0 means inside the prior min/max band.
    """
    if not prior:
        return "no priors", 0.0
    lo, hi = min(prior), max(prior)
    lo_band = lo * (1 - tolerance)
    hi_band = hi * (1 + tolerance)
    if lo_band <= value <= hi_band:
        return "in range", 0.0
    if value < lo_band:
        pct = (value - lo) / lo * 100 if lo else 0
        return f"BELOW ({pct:+.0f}% vs min)", pct
    pct = (value - hi) / hi * 100 if hi else 0
    return f"ABOVE ({pct:+.0f}% vs max)", pct


def compare_cycle_keyed(
    points: list[tuple[datetime, float]],
    cycle_len: int,
    key_fn,
    tolerance: float,
) -> tuple[list[dict], int, int]:
    """Group prior points by cycle key, compare recent points to their bucket.

    Returns (results, in_range_count, out_of_range_count).
    """
    if len(points) < cycle_len * 2:
        return [], 0, 0
    recent = points[-cycle_len:]
    priors = points[:-cycle_len]

    grouped: dict = {}
    for dt, v in priors:
        grouped.setdefault(key_fn(dt), []).append(v)

    last_dt = recent[-1][0]
    out: list[dict] = []
    in_range = out_of_range = partial = 0
    for dt, v in recent:
        key = key_fn(dt)
        prior_vals = grouped.get(key, [])
        verdict, dev = classify(v, prior_vals, tolerance)
        # Soften: if this is the very last bucket and it's >=50% below prior min,
        # it's almost certainly a partial period (incomplete day / hour).
        is_last = dt == last_dt
        if (
            is_last
            and prior_vals
            and "BELOW" in verdict
            and v < min(prior_vals) * 0.5
        ):
            verdict = f"PARTIAL? ({(v - min(prior_vals)) / min(prior_vals) * 100:+.0f}% vs min)"
            partial += 1
        elif "in range" in verdict:
            in_range += 1
        elif prior_vals:
            out_of_range += 1
        out.append({
            "dt": dt,
            "key": key,
            "value": v,
            "prior_min": min(prior_vals) if prior_vals else None,
            "prior_max": max(prior_vals) if prior_vals else None,
            "prior_median": median(prior_vals) if prior_vals else None,
            "n_prior": len(prior_vals),
            "verdict": verdict,
            "deviation": abs(dev),
            "partial": "PARTIAL?" in verdict,
        })
    return out, in_range, out_of_range


def report_day_cycle(label: str, days: list[str], data: list[float], tolerance: float) -> None:
    points: list[tuple[datetime, float]] = [
        (dt, v) for d, v in zip(days, data) if (dt := parse_dt(d)) is not None
    ]

    results, in_range, out_of_range = compare_cycle_keyed(
        points, cycle_len=7, key_fn=lambda dt: dt.weekday(), tolerance=tolerance
    )

    print(f"# Same-day-of-week comparison — {label}")
    print()
    print(f"Window: {days[0]} → {days[-1]}  ({len(days)} days)")
    print(f"Cycle: weekly  (each weekday compared to prior {(len(days) // 7) - 1} same-weekdays)")
    print()
    if not results:
        print("Not enough history — need at least 2 full weeks. Widen the dateRange.")
        return

    print("| Day | Most recent | Prior median | Prior range | Verdict |")
    print("| --- | ---: | ---: | --- | --- |")
    for r in results:
        prior_range = (
            f"{fmt(r['prior_min'])} – {fmt(r['prior_max'])}" if r["prior_min"] is not None else "—"
        )
        print(
            f"| {WEEKDAYS[r['key']]} {r['dt'].date()} "
            f"| {fmt(r['value'])} "
            f"| {fmt(r['prior_median']) if r['prior_median'] is not None else '—'} "
            f"| {prior_range} "
            f"| {r['verdict']} |"
        )

    print()
    has_partial = any(r.get("partial") for r in results)
    if out_of_range == 0 and in_range > 0:
        msg = (
            f"**Verdict: every completed weekday is within ±{tolerance:.0%} of the "
            "prior weeks' range — likely normal seasonality.**"
        )
        if has_partial:
            msg += " The last day looks partial; ignore it for now and re-check after it completes."
        print(msg)
    elif out_of_range > 0:
        print(
            f"**Verdict: {out_of_range} completed weekday(s) outside the prior weeks' "
            f"range (±{tolerance:.0%} tolerance) — proceed with the playbook.**"
        )
        if has_partial:
            print("(The last day appears partial and was excluded from the count.)")


def report_hour_cycle(
    label: str, days: list[str], data: list[float], tolerance: float, top: int
) -> None:
    points: list[tuple[datetime, float]] = [
        (dt, v) for d, v in zip(days, data) if (dt := parse_dt(d)) is not None
    ]

    results, in_range, out_of_range = compare_cycle_keyed(
        points,
        cycle_len=168,
        key_fn=lambda dt: (dt.weekday(), dt.hour),
        tolerance=tolerance,
    )

    print(f"# Same-hour-of-week comparison — {label}")
    print()
    print(f"Window: {days[0]} → {days[-1]}  ({len(days)} hours, "
          f"~{len(days) / 168:.1f} weeks)")
    print(f"Cycle: weekly × hourly (168 buckets, each hour vs prior weeks' same weekday-hour)")
    print()
    if not results:
        print("Not enough history — need at least 2 full weeks of hourly data. Widen the dateRange.")
        return

    print(f"Last 168 hours: **{in_range} in range, {out_of_range} outside ±{tolerance:.0%}**")
    print()

    flagged = [
        r for r in results
        if "in range" not in r["verdict"] and not r.get("partial")
    ]
    flagged.sort(key=lambda r: r["deviation"], reverse=True)

    if flagged:
        print(f"## Top {min(top, len(flagged))} most deviated hours")
        print()
        print("| Time | Value | Prior median | Prior range | Verdict |")
        print("| --- | ---: | ---: | --- | --- |")
        for r in flagged[:top]:
            wd, hr = r["key"]
            prior_range = (
                f"{fmt(r['prior_min'])} – {fmt(r['prior_max'])}" if r["prior_min"] is not None else "—"
            )
            print(
                f"| {WEEKDAYS[wd]} {hr:02d}:00 ({r['dt'].strftime('%Y-%m-%d')}) "
                f"| {fmt(r['value'])} "
                f"| {fmt(r['prior_median']) if r['prior_median'] is not None else '—'} "
                f"| {prior_range} "
                f"| {r['verdict']} |"
            )
        print()

        # Aggregate where the flags concentrate
        by_weekday: dict[int, int] = {}
        by_hour: dict[int, int] = {}
        for r in flagged:
            wd, hr = r["key"]
            by_weekday[wd] = by_weekday.get(wd, 0) + 1
            by_hour[hr] = by_hour.get(hr, 0) + 1

        if by_weekday:
            print("## Flags by weekday")
            print()
            for wd in sorted(by_weekday, key=lambda k: -by_weekday[k]):
                print(f"- {WEEKDAYS[wd]}: {by_weekday[wd]}")
            print()
        if by_hour:
            print("## Flags by hour-of-day")
            print()
            for hr in sorted(by_hour, key=lambda k: -by_hour[k])[:10]:
                print(f"- {hr:02d}:00 — {by_hour[hr]}")
            print()

    print()
    if out_of_range == 0 and in_range > 0:
        print(
            f"**Verdict: every hour in the last week is within ±{tolerance:.0%} of "
            "prior weeks — likely normal seasonality.**"
        )
    elif out_of_range > 0:
        if out_of_range / max(1, in_range + out_of_range) > 0.5:
            print(
                f"**Verdict: >50% of hours are out of range — likely a sustained shift, "
                "not a localized incident. Proceed with the playbook.**"
            )
        else:
            print(
                f"**Verdict: {out_of_range} flagged hours concentrated in the table above. "
                "Check whether they cluster around a deploy time / incident — "
                "see SKILL.md step 2.3.**"
            )


def report_sequential(
    label: str,
    days: list[str],
    data: list[float],
    interval: str,
    recent_n: int,
    tolerance: float,
) -> None:
    """Sequential comparison for week / month / minute — no natural cycle."""
    if len(data) < recent_n + 3:
        print(f"# Sequential comparison — {label}", flush=True)
        print()
        print(
            f"Not enough history for {interval} interval — need at least "
            f"{recent_n + 3} points; have {len(data)}. Widen the dateRange."
        )
        return

    recent = list(zip(days[-recent_n:], data[-recent_n:]))
    priors = data[:-recent_n]
    prior_med = median(priors)
    lo, hi = min(priors), max(priors)

    print(f"# Sequential comparison — {label}")
    print()
    print(
        f"Window: {days[0]} → {days[-1]}  ({len(days)} {interval} intervals). "
        f"No natural cycle for {interval} — comparing last {recent_n} to "
        f"prior {len(priors)} values."
    )
    print()
    print(f"Prior median: {fmt(prior_med)}  |  range: {fmt(lo)} – {fmt(hi)}")
    print()
    print(f"| {interval.title()} | Value | Verdict |")
    print("| --- | ---: | --- |")
    out_of_range = 0
    in_range = 0
    for d, v in recent:
        verdict, _ = classify(v, priors, tolerance)
        if "in range" in verdict:
            in_range += 1
        else:
            out_of_range += 1
        print(f"| {d} | {fmt(v)} | {verdict} |")
    print()
    if out_of_range == 0:
        print(
            f"**Verdict: all recent {interval}s within ±{tolerance:.0%} of the prior "
            "range — within normal variance.**"
        )
    else:
        print(f"**Verdict: {out_of_range} of {recent_n} recent {interval}s outside the "
              f"prior range — proceed with the playbook.**")


def main() -> int:
    tolerance = float(os.environ.get("TOLERANCE", "0.10"))
    top = int(os.environ.get("TOP", "10"))
    recent_override = os.environ.get("RECENT")

    payload = load_input()
    days, data, label, interval = extract_series(payload)

    if not days:
        print("Empty series — nothing to compare.", file=sys.stderr)
        return 1

    if interval == "day":
        report_day_cycle(label, days, data, tolerance)
    elif interval == "hour":
        report_hour_cycle(label, days, data, tolerance, top)
    elif interval in {"minute", "second"}:
        recent_n = int(recent_override) if recent_override else 60
        report_sequential(label, days, data, interval, recent_n, tolerance)
    elif interval == "week":
        recent_n = int(recent_override) if recent_override else 4
        report_sequential(label, days, data, interval, recent_n, tolerance)
    elif interval == "month":
        recent_n = int(recent_override) if recent_override else 3
        report_sequential(label, days, data, interval, recent_n, tolerance)
    else:
        print(f"Unsupported interval '{interval}'. Treating as sequential.", file=sys.stderr)
        recent_n = int(recent_override) if recent_override else 7
        report_sequential(label, days, data, interval, recent_n, tolerance)

    return 0


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