Spotlighting Your Project
Create a compelling data-driven section showing the impact of [project name]: downloads, users helped, issues resolved, and community growth statistics.
Checks CSV time series (metrics, sensor readings, sales by hour) for data problems before you chart or alert on them: missing intervals, duplicate or out-of-order timestamps, empty values, flatlines from stuck exporters, robust median/MAD spikes, and counter resets. Use when a dashboard looks wrong, before training a forecast, or when asking "is this data complete?". Includes a stdlib Python checker with --self-test.
---
name: time-series-gap-anomaly-checker
description: Checks CSV time series (metrics, sensor readings, sales by hour) for data problems before you chart or alert on them: missing intervals, duplicate or out-of-order timestamps, empty values, flatlines from stuck exporters, robust median/MAD spikes, and counter resets. Use when a dashboard looks wrong, before training a forecast, or when asking "is this data complete?". Includes a stdlib Python checker with --self-test.
---
# Time Series Gap and Anomaly Checker
You help analysts, SREs, and data engineers decide whether a time series can be trusted. You find where data is missing, repeated, stuck, or suspicious, explain the likely cause of each problem, and separate collection problems (the pipeline) from real events (the system being measured). You never silently fill gaps or delete outliers; you propose a treatment and say what it changes.
## Files in this skill
- `scripts/ts_check.py` - reads a CSV, infers the sampling interval, and reports gaps, duplicates, out-of-order rows, missing values, flatlines, spikes, and counter resets as markdown or JSON (Python 3 standard library only)
- `references/anomaly-methods.md` - how the checks work, how to tune thresholds, and when each method fails
- `references/gap-handling-playbook.md` - what to do with each kind of problem: fill, interpolate, drop, or annotate
- `templates/data-quality-report.md` - a report for a dashboard fix, incident review, or dataset hand-off
- `examples/api-metrics.csv` - three hours of 5-minute API metrics with a scrape outage, a duplicate, a stuck value, a latency spike, and a counter reset
## Workflow
### 1. Understand the series
Ask what each column measures, its unit, the expected sampling interval, and whether any column is a monotonic counter (requests_total, bytes_sent). Ask which time zone naive timestamps use; the script treats them as UTC.
### 2. Run the checker
```
python3 scripts/ts_check.py examples/api-metrics.csv --counter requests_total
python3 scripts/ts_check.py data.csv --time-col ts --cols temp_c,humidity --interval 10m
python3 scripts/ts_check.py data.csv --z 5 --window 12 --flatline-points 12 --json > report.json
python3 scripts/ts_check.py --self-test
```
Timestamps can be ISO 8601 or Unix epoch seconds or milliseconds. Exit code 1 when any issue is found, 2 on bad input, so it can run as a pipeline gate.
### 3. Read the results
- **Completeness** = rows present / rows expected over the range. Below 95% usually means aggregates and averages are biased.
- **gap**: spacing over 1.5x the interval. Long gaps at regular times often mean deploys or cron jobs; one long gap usually means an outage of the collector.
- **duplicate / out_of_order**: double scrapes, retries, or merged files.
- **flatline**: the same value repeated; real for on/off signals, suspicious for noisy measurements.
- **spike**: robust z-score against a rolling median. Check whether neighbouring columns spike at the same time (a real event) or not (a glitch).
- **reset**: a counter that went down, usually a process restart; rate calculations must handle it.
### 4. Recommend treatment
Use `references/gap-handling-playbook.md`. Tune thresholds with `references/anomaly-methods.md` if the data is seasonal, bursty, or very sparse.
### 5. Report
Fill `templates/data-quality-report.md` with the summary numbers, the top issues with times in the user's time zone, the suspected cause, and the agreed treatment.
## Rules
- Report problems; do not alter the source data. Write cleaned data to a new file and document every change.
- Do not call a spike an incident without corroboration (other metrics, logs, or a human).
- State the interval and thresholds you used so results are reproducible.
- Convert UTC times to the user's time zone when presenting them.
FILE:references/anomaly-methods.md
# Anomaly Methods and Tuning
## Interval inference
The expected interval is the median spacing between sorted unique timestamps. This survives a few gaps and jitter. Pass `--interval` when the series is mostly gaps (sparse events) or when you know the scrape interval.
A gap is any spacing above 1.5x the interval, so 5-minute data with a 7-minute spacing is not a gap but a 10-minute spacing is (1 missing point).
## Robust spikes (rolling median + MAD)
For each point, take up to `--window` neighbours on each side (excluding the point), compute their median and the median absolute deviation (MAD). The robust z-score is:
z = |value - median| / (1.4826 * MAD)
1.4826 makes MAD comparable to a standard deviation for normal data. Default threshold 4 is roughly a 1-in-15,000 event for normal noise. When more than half the neighbours are identical (MAD = 0), the mean absolute deviation is used, and for a perfectly flat neighbourhood 1% of the level.
Why not mean and standard deviation? One large spike inflates the standard deviation and hides itself. The median and MAD ignore it.
## Tuning
| Data | Suggestion |
|---|---|
| Noisy metrics (latency, CPU) | `--z 5`, `--window 12` |
| Daily or weekly seasonality | Check one season at a time, or compare to the same hour last week before trusting spikes |
| Bursty business data (orders) | Higher `--z` (6-8); spikes are often real |
| Integer or low-resolution values | Expect MAD = 0; raise `--z` or aggregate first |
| On/off or status signals | Ignore flatline; check only gaps |
| Counters | Always pass `--counter`; check rates after handling resets |
## Known limits
- Gradual drifts and level shifts are not spikes; plot the series or compare week-over-week medians.
- The first and last few points have fewer neighbours, so their scores are less reliable.
- Irregular event streams (logins, purchases) have no fixed interval; aggregate into buckets first.
FILE:references/gap-handling-playbook.md
# Gap and Anomaly Handling Playbook
| Problem | Likely causes | Default treatment | Avoid |
|---|---|---|---|
| Short gap (1-2 points) | scrape timeout, network blip | Linear interpolation for smooth signals; leave null for counts | Filling counts with zero |
| Long gap | collector down, deploy, host off | Leave null and annotate the chart; exclude from SLO math or count as unknown | Interpolating across hours |
| Duplicate timestamps | double scrape, retried batch, merged files | Keep one row (first, or the one with most fields); check whether values differ | Summing duplicates |
| Out-of-order rows | late-arriving data, file concatenation | Sort by time before any windowed calculation | Assuming file order is time order |
| Missing values | sensor dropout, parse errors, "N/A" strings | Fix the parser first; then treat as a gap | Treating blank as 0 |
| Flatline | stuck exporter, cached value, sensor freeze | Confirm with a second source; mark as suspect | Trusting it for alerts |
| Spike (single point) | glitch, unit error, real burst | Check related columns and logs; cap or remove only with a documented reason | Deleting without a note |
| Counter reset | process restart, overflow | Compute rates per segment (rate after reset = new value / interval) | Negative rates |
## Before modeling or forecasting
1. Sort, de-duplicate, and regularize to the expected interval (insert nulls for missing slots).
2. Decide gap policy per column and record it.
3. Keep a mask column (`is_imputed`) so models and readers know what was filled.
4. Re-run the checker on the cleaned file and attach both reports.
## When to escalate
- Completeness under 90% for a metric used in an SLO or invoice.
- Gaps that repeat at the same time every day (a job is killing the collector).
- Spikes that appear in all hosts at once (a real incident, not a data issue).
FILE:templates/data-quality-report.md
# Time Series Data Quality - <dataset or dashboard>
**Source:** <file / query> **Range:** <start> to <end> (<time zone>) **Interval:** <expected>
**Rows:** <n> **Completeness:** <percent> **Thresholds:** z=<z>, window=<w>, flatline=<n>
## Summary
<two sentences: can this data be trusted for the intended use, and what must change first>
## Issues
| Kind | When (<time zone>) | Column | Detail | Suspected cause | Treatment |
|---|---|---|---|---|---|
| | | | | | |
## Decisions
- Gap policy: <interpolate short gaps / leave null / exclude>
- Outlier policy: <keep / cap / remove with note>
- Counter handling: <segment rates at resets>
## Follow-ups
- [ ] <fix collector / parser / alert> - owner <name> - due <date>
Generated with scripts/ts_check.py. Source data was not modified.
FILE:examples/api-metrics.csv
timestamp,p95_latency_ms,error_rate,requests_total
2026-10-18T22:00:00Z,182,0.4,1200
2026-10-18T22:05:00Z,176,0.3,2435
2026-10-18T22:10:00Z,190,0.5,3705
2026-10-18T22:15:00Z,185,0.4,5010
2026-10-18T22:20:00Z,179,0.3,6210
2026-10-18T22:25:00Z,188,0.4,7445
2026-10-18T22:30:00Z,181,0.5,8715
2026-10-18T22:35:00Z,193,0.4,10020
2026-10-18T22:35:00Z,193,0.4,10020
2026-10-18T22:40:00Z,177,0.3,11220
2026-10-18T22:45:00Z,184,0.4,12455
2026-10-18T22:50:00Z,186,0.4,13725
2026-10-18T22:55:00Z,180,0.4,15030
2026-10-18T23:00:00Z,191,0.4,16230
2026-10-18T23:05:00Z,,0.4,17465
2026-10-18T23:10:00Z,178,0.4,18735
2026-10-18T23:15:00Z,187,0.4,20040
2026-10-18T23:20:00Z,182,0.4,21240
2026-10-18T23:25:00Z,189,0.4,22475
2026-10-18T23:30:00Z,612,2.1,23745
2026-10-18T23:35:00Z,185,0.5,25050
2026-10-18T23:40:00Z,180,0.3,26250
2026-10-18T23:45:00Z,184,0.4,27485
2026-10-19T00:10:00Z,188,0.5,28755
2026-10-19T00:15:00Z,183,0.4,30060
2026-10-19T00:20:00Z,177,0.3,31260
2026-10-19T00:25:00Z,185,0.4,32495
2026-10-19T00:30:00Z,182,0.5,1180
2026-10-19T00:35:00Z,189,0.4,2485
2026-10-19T00:40:00Z,184,0.3,3685
2026-10-19T00:45:00Z,180,0.4,4920
2026-10-19T00:50:00Z,186,0.5,6190
2026-10-19T00:55:00Z,183,0.4,7495
FILE:scripts/ts_check.py
#!/usr/bin/env python3
"""Time series gap and anomaly checker for CSV metrics (Python 3 standard library only).
Reads a CSV with a timestamp column and one or more numeric columns, infers the sampling
interval (median spacing) unless --interval is given, and reports:
gap missing samples (spacing > 1.5 x interval), with how many points are missing
duplicate repeated timestamps
out_of_order timestamps that go backwards
missing empty, NaN, or non-numeric values
flatline the same value repeated for >= --flatline-points samples (stuck sensor/exporter)
spike robust outliers: |value - rolling median| / (1.4826 * MAD) >= --z (default 4);
if MAD is 0 the mean absolute deviation is used instead
reset a counter column that decreases (--counter col1,col2)
Timestamps: ISO 8601 (with Z or offset; naive = UTC) or Unix epoch seconds/milliseconds.
Usage:
python3 ts_check.py metrics.csv
python3 ts_check.py metrics.csv --time-col ts --cols cpu,latency_ms --interval 5m --z 5
python3 ts_check.py metrics.csv --counter requests_total --json
python3 ts_check.py --self-test
Exit code: 0 clean, 1 issues found, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import io
import json
import math
import re
import statistics
import sys
class InputError(ValueError):
pass
def parse_ts(raw):
s = (raw or "").strip()
if not s:
raise InputError("empty timestamp")
if re.fullmatch(r"-?\d+(\.\d+)?", s):
v = float(s)
return v / 1000.0 if v > 1e11 else v
try:
t = dt.datetime.fromisoformat(s.replace("Z", "+00:00").replace(" ", "T", 1))
except ValueError:
raise InputError(f"unparseable timestamp {s!r}")
if t.tzinfo is None:
t = t.replace(tzinfo=dt.timezone.utc)
return t.timestamp()
def parse_interval(s):
m = re.fullmatch(r"(\d+(?:\.\d+)?)\s*(ms|s|m|h|d)?", s.strip())
if not m:
raise InputError(f"bad interval {s!r}; use e.g. 30s, 5m, 1h")
mult = {"ms": 0.001, "s": 1, None: 1, "m": 60, "h": 3600, "d": 86400}[m.group(2)]
return float(m.group(1)) * mult
def fmt_ts(ts):
return dt.datetime.fromtimestamp(ts, dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def fmt_dur(sec):
for unit, size in (("d", 86400), ("h", 3600), ("m", 60)):
if sec >= size:
v = sec / size
return f"{v:.0f}{unit}" if v == int(v) else f"{v:.1f}{unit}"
return f"{sec:g}s"
def to_float(raw):
s = (raw or "").strip()
if s == "" or s.lower() in ("nan", "null", "none", "na", "n/a", "-"):
return None
try:
v = float(s)
except ValueError:
return None
return None if math.isnan(v) or math.isinf(v) else v
def load(text, time_col=None, cols=None):
reader = csv.DictReader(io.StringIO(text))
if not reader.fieldnames:
raise InputError("CSV has no header row")
fields = [f.strip() for f in reader.fieldnames]
reader.fieldnames = fields
if time_col is None:
cands = [f for f in fields if f.lower() in ("timestamp", "time", "ts", "date", "datetime")]
time_col = cands[0] if cands else fields[0]
if time_col not in fields:
raise InputError(f"time column {time_col!r} not in header {fields}")
value_cols = cols or [f for f in fields if f != time_col]
for c in value_cols:
if c not in fields:
raise InputError(f"column {c!r} not in header {fields}")
rows = []
for i, r in enumerate(reader, start=2):
rows.append({"line": i, "ts": parse_ts(r.get(time_col)),
"raw": {c: r.get(c) for c in value_cols}})
if len(rows) < 3:
raise InputError("need at least 3 data rows")
return time_col, value_cols, rows
def rolling_robust(values, idx, window):
lo, hi = max(0, idx - window), min(len(values), idx + window + 1)
neigh = [values[j] for j in range(lo, hi) if j != idx and values[j] is not None]
if len(neigh) < 4:
return None, None
med = statistics.median(neigh)
mad = statistics.median(abs(v - med) for v in neigh)
scale = 1.4826 * mad
if scale == 0: # more than half the neighbours equal the median: fall back to mean absolute deviation
scale = 1.2533 * statistics.fmean(abs(v - med) for v in neigh)
if scale == 0: # perfectly flat neighbourhood: 1% of the level
scale = max(abs(med) * 0.01, 1e-9)
return med, scale
def check(text, time_col=None, cols=None, interval=None, z=4.0, window=6, flatline_points=6, counters=()):
time_col, value_cols, rows = load(text, time_col, cols)
issues = []
def add(kind, line, ts, col, detail):
issues.append({"kind": kind, "line": line, "time": fmt_ts(ts), "column": col, "detail": detail})
# order and duplicates (in file order)
seen = {}
for prev, cur in zip(rows, rows[1:]):
if cur["ts"] < prev["ts"]:
add("out_of_order", cur["line"], cur["ts"], time_col, f"goes back {fmt_dur(prev['ts'] - cur['ts'])} from line {prev['line']}")
for r in rows:
if r["ts"] in seen:
add("duplicate", r["line"], r["ts"], time_col, f"same timestamp as line {seen[r['ts']]}")
else:
seen[r["ts"]] = r["line"]
srt = sorted({r["ts"]: r for r in reversed(rows)}.values(), key=lambda r: r["ts"]) # first occurrence wins
diffs = [b["ts"] - a["ts"] for a, b in zip(srt, srt[1:]) if b["ts"] > a["ts"]]
if interval is None:
if not diffs:
raise InputError("cannot infer interval: all timestamps identical")
interval = statistics.median(diffs)
expected = int(round((srt[-1]["ts"] - srt[0]["ts"]) / interval)) + 1
missing_points = 0
for a, b in zip(srt, srt[1:]):
d = b["ts"] - a["ts"]
if d > 1.5 * interval:
n = int(round(d / interval)) - 1
missing_points += n
add("gap", b["line"], a["ts"], time_col, f"{fmt_dur(d)} without data after {fmt_ts(a['ts'])} (~{n} missing)")
stats = {}
for c in value_cols:
vals = [to_float(r["raw"][c]) for r in srt]
for r, v in zip(srt, vals):
if v is None:
add("missing", r["line"], r["ts"], c, f"value {r['raw'][c]!r} is empty or not numeric")
# flatline runs
run_start = 0
for i in range(1, len(vals) + 1):
if i < len(vals) and vals[i] is not None and vals[i] == vals[run_start]:
continue
length = i - run_start
if vals[run_start] is not None and length >= flatline_points:
r0, r1 = srt[run_start], srt[i - 1]
add("flatline", r0["line"], r0["ts"], c, f"value {vals[run_start]:g} repeated {length}x until {fmt_ts(r1['ts'])}")
run_start = i
if c in counters:
for i in range(1, len(vals)):
if vals[i] is not None and vals[i - 1] is not None and vals[i] < vals[i - 1]:
add("reset", srt[i]["line"], srt[i]["ts"], c, f"counter dropped {vals[i - 1]:g} -> {vals[i]:g} (restart or wrap)")
else:
for i, v in enumerate(vals):
if v is None:
continue
med, scale = rolling_robust(vals, i, window)
if med is None:
continue
score = abs(v - med) / scale
if score >= z:
add("spike", srt[i]["line"], srt[i]["ts"], c, f"{v:g} vs local median {med:g} (robust z {score:.1f})")
good = [v for v in vals if v is not None]
stats[c] = {"count": len(good), "min": min(good) if good else None, "max": max(good) if good else None,
"median": statistics.median(good) if good else None}
order = {"out_of_order": 0, "duplicate": 1, "gap": 2, "reset": 3, "missing": 4, "flatline": 5, "spike": 6}
issues.sort(key=lambda x: (order[x["kind"]], x["time"], x["column"]))
counts = {}
for it in issues:
counts[it["kind"]] = counts.get(it["kind"], 0) + 1
return {
"rows": len(rows), "time_column": time_col, "columns": value_cols,
"start": fmt_ts(srt[0]["ts"]), "end": fmt_ts(srt[-1]["ts"]),
"interval": fmt_dur(interval), "interval_seconds": interval,
"expected_points": expected, "missing_points": missing_points,
"completeness": round(100.0 * len(srt) / expected, 1) if expected else 100.0,
"stats": stats, "counts": counts, "issues": issues,
}
def to_markdown(r):
lines = ["# Time series check", "",
f"Rows: {r['rows']} Range: {r['start']} to {r['end']} Interval: {r['interval']}",
f"Completeness: {r['completeness']}% ({r['missing_points']} of {r['expected_points']} expected points missing)", "",
"| Column | Count | Min | Median | Max |", "|---|---|---|---|---|"]
for c, s in r["stats"].items():
f = lambda v: "-" if v is None else f"{v:g}"
lines.append(f"| {c} | {s['count']} | {f(s['min'])} | {f(s['median'])} | {f(s['max'])} |")
lines += ["", "## Issues"]
if not r["issues"]:
lines.append("No issues found.")
else:
lines.append("Totals: " + ", ".join(f"{k} {v}" for k, v in r["counts"].items()))
lines += ["", "| Kind | Time (UTC) | Column | Line | Detail |", "|---|---|---|---|---|"]
lines += [f"| {i['kind']} | {i['time']} | {i['column']} | {i['line']} | {i['detail']} |" for i in r["issues"]]
return "\n".join(lines)
def self_test():
cases = []
def ok(name, cond):
cases.append((name, bool(cond)))
base = 1760860800 # 2025-10-19T08:00:00Z
def csv_of(points, header="timestamp,value"):
return header + "\n" + "\n".join(f"{t},{v}" for t, v in points) + "\n"
wave = [10 + (i % 3) for i in range(30)]
clean = csv_of([(base + 60 * i, v) for i, v in enumerate(wave)])
r = check(clean)
ok("clean series has no issues", r["issues"] == [])
ok("interval inferred as 1m", r["interval"] == "1m")
ok("completeness 100%", r["completeness"] == 100.0)
gap_pts = [(base + 60 * i, v) for i, v in enumerate(wave) if not 10 <= i < 15]
r = check(csv_of(gap_pts))
ok("gap detected", r["counts"].get("gap") == 1)
ok("gap counts 5 missing points", r["missing_points"] == 5)
spiky = list(wave); spiky[20] = 95
r = check(csv_of([(base + 60 * i, v) for i, v in enumerate(spiky)]))
ok("spike detected", [i["line"] for i in r["issues"] if i["kind"] == "spike"] == [22])
flat = list(wave); flat[5:13] = [7] * 8
r = check(csv_of([(base + 60 * i, v) for i, v in enumerate(flat)]), z=100)
ok("flatline detected", r["counts"].get("flatline") == 1)
miss = [(base + 60 * i, ("" if i == 4 else "NaN" if i == 9 else v)) for i, v in enumerate(wave)]
ok("missing values detected", check(csv_of(miss)).get("counts").get("missing") == 2)
dup = [(base + 60 * i, v) for i, v in enumerate(wave)]
dup.insert(6, dup[5])
ok("duplicate detected", check(csv_of(dup))["counts"].get("duplicate") == 1)
ooo = [(base + 60 * i, v) for i, v in enumerate(wave)]
ooo[7], ooo[8] = ooo[8], ooo[7]
ok("out of order detected", check(csv_of(ooo))["counts"].get("out_of_order") == 1)
counter = [(base + 60 * i, 100 * i if i < 20 else 100 * (i - 20)) for i in range(30)]
r = check(csv_of(counter, "ts,requests_total"), counters=("requests_total",))
ok("counter reset detected", r["counts"].get("reset") == 1 and "spike" not in r["counts"])
iso_rows = "time,cpu\n" + "\n".join(f"2026-10-19T08:{i:02d}:00Z,{40 + (i % 2)}" for i in range(0, 60, 5)) + "\n"
r = check(iso_rows)
ok("ISO timestamps and 5m interval", r["interval"] == "5m" and r["issues"] == [])
ms_rows = csv_of([((base + 30 * i) * 1000, v) for i, v in enumerate(wave)])
ok("epoch milliseconds accepted", check(ms_rows)["interval"] == "30s")
r = check(clean, interval=30.0)
ok("explicit 30s interval flags every 1m step as a gap", r["counts"].get("gap") == 29 and r["completeness"] < 60)
jitter = csv_of([(base + 60 * i + (3 if i % 2 else -2), v) for i, v in enumerate(wave)])
ok("small timestamp jitter is not a gap", "gap" not in check(jitter)["counts"])
multi = "timestamp,a,b\n" + "\n".join(f"{base + 60 * i},{wave[i]},{wave[i] * 2 if i != 12 else 500}" for i in range(30)) + "\n"
r = check(multi, cols=["a", "b"])
ok("multi-column spike only in b", {i["column"] for i in r["issues"]} == {"b"})
for bad, label in (("timestamp,value\n1,2\n", "too few rows rejected"),
("timestamp,value\nyesterday,1\ntoday,2\nnow,3\n", "bad timestamp rejected")):
try:
check(bad)
ok(label, False)
except InputError:
ok(label, True)
ok("markdown renders", "## Issues" in to_markdown(check(clean)))
passed = sum(c for _, c in cases)
for name, c in cases:
print(("PASS " if c else "FAIL ") + name)
print(f"{passed}/{len(cases)} passed")
return 0 if passed == len(cases) else 1
def main(argv=None):
ap = argparse.ArgumentParser(description="Find gaps, duplicates, flatlines, spikes, and counter resets in CSV metrics.")
ap.add_argument("csv", nargs="?", help="CSV file, or '-' for stdin")
ap.add_argument("--time-col")
ap.add_argument("--cols", help="comma-separated value columns (default: all except the time column)")
ap.add_argument("--interval", help="expected spacing, e.g. 30s, 5m, 1h (default: median spacing)")
ap.add_argument("--z", type=float, default=4.0, help="robust z-score threshold for spikes (default 4)")
ap.add_argument("--window", type=int, default=6, help="neighbours on each side for the rolling median (default 6)")
ap.add_argument("--flatline-points", type=int, default=6)
ap.add_argument("--counter", default="", help="comma-separated monotonic counter columns (checked for resets, not spikes)")
ap.add_argument("--json", action="store_true")
ap.add_argument("--self-test", action="store_true")
a = ap.parse_args(argv)
if a.self_test:
return self_test()
if not a.csv:
ap.error("csv path required (or --self-test)")
try:
text = sys.stdin.read() if a.csv == "-" else open(a.csv, encoding="utf-8-sig").read()
r = check(text, a.time_col, [c.strip() for c in a.cols.split(",")] if a.cols else None,
parse_interval(a.interval) if a.interval else None, a.z, a.window, a.flatline_points,
tuple(c.strip() for c in a.counter.split(",") if c.strip()))
except (OSError, InputError) as e:
print(f"error: {e}", file=sys.stderr)
return 2
print(json.dumps(r, indent=2) if a.json else to_markdown(r))
return 1 if r["issues"] else 0
if __name__ == "__main__":
sys.exit(main())Turns request counts into SLO answers: SLI, error budget for a 28d or 30d window, budget consumed and remaining, burn rates over 5m to 3d, multiwindow burn-rate alerts (page at 14.4x and 6x, ticket at 3x and 1x, scaled to the window), and time until the budget runs out. Use when setting an SLO, tuning alerts, deciding on a release freeze, or writing an incident summary. Includes a stdlib Python calculator with --self-test.
---
name: slo-error-budget-burn-rate-calculator
description: Turns request counts into SLO answers: SLI, error budget for a 28d or 30d window, budget consumed and remaining, burn rates over 5m to 3d, multiwindow burn-rate alerts (page at 14.4x and 6x, ticket at 3x and 1x, scaled to the window), and time until the budget runs out. Use when setting an SLO, tuning alerts, deciding on a release freeze, or writing an incident summary. Includes a stdlib Python calculator with --self-test.
---
# SLO Error Budget and Burn Rate Calculator
You help engineers and on-call teams reason about reliability with numbers instead of gut feeling. You turn good/total request counts into an SLI, translate the SLO into an error budget, show how fast the budget is burning in several windows, say which alerts would page or open a ticket, and estimate when the budget runs out at the current pace. You explain every number in plain language so product owners can follow the decision.
## Files in this skill
- `scripts/slo_budget.py` - reads a CSV of request counts per interval and prints SLI, budget consumed and remaining, burn rates, alert states, exhaustion times and the worst interval; `--allowance` prints allowed downtime and alert thresholds for any SLO (Python 3 standard library only)
- `references/slo-math.md` - formulas for SLI, error budget, burn rate, projected consumption, and worked numbers for common targets
- `references/burn-rate-alerting.md` - the multiwindow, multi-burn-rate alert policy, how thresholds scale with the window, and tuning advice
- `templates/slo-status-report.md` - a weekly SLO status or incident budget summary
- `examples/checkout-api-requests.csv` - two days of 15-minute request counts for a checkout API with a small blip and a 2-hour incident at the end
## Workflow
### 1. Define the SLI and SLO
Ask what a "good" event is (status < 500, latency under 300 ms, successful job run), where the counts come from, the target (99%, 99.5%, 99.9%, 99.95%), and the window (rolling 28 or 30 days). Push back on targets the system has never met: show the historical SLI first.
### 2. Get the data
Export good and total (or bad and total) counts per interval, for example 1-minute, 5-minute or 15-minute buckets, as CSV with `timestamp,total,good`. The timestamp marks the start of each interval. Use UTC.
### 3. Run the calculator
```
python3 scripts/slo_budget.py examples/checkout-api-requests.csv --slo 99.9 --window 30d
python3 scripts/slo_budget.py requests.csv --slo 99.5 --window 28d --json > slo.json
python3 scripts/slo_budget.py requests.csv --now 2026-10-19T22:00:00Z
python3 scripts/slo_budget.py --allowance --slo 99.95 --window 30d
python3 scripts/slo_budget.py --self-test
```
Exit code 1 means a page alert is firing or the budget is exhausted; 2 means bad input (missing columns, bad > total, unparseable timestamps).
### 4. Interpret
- Burn rate 1.0 spends exactly the budget over the window; 14.4 for an hour spends 2% of a 30-day budget.
- Budget consumed is projected against the traffic seen so far; with less than a full window of data, say so.
- An alert fires only when both its long and short window exceed the threshold; the short window makes it stop quickly after recovery. See `references/burn-rate-alerting.md`.
- Windows shorter than the data step are rounded up to one step; windows longer than the data are "not evaluable".
### 5. Recommend and report
Tie the numbers to a decision: page now, open a ticket, slow down risky releases while remaining budget is low, or loosen an SLO that is always violated by design. Fill `templates/slo-status-report.md`.
## Example result
For `examples/checkout-api-requests.csv` at 99.9% over 30d: SLI 99.8768%, 8.2% of the budget consumed, 1h burn 26.46 and 5m burn 28.06 so the fast page fires, 6h burn 5.83 just under 6.0, the 3d ticket is not evaluable with two days of data, and at the last hour's pace the budget is gone in about 1d 16h. Exit code 1.
## Rules
- State the SLO, window, data range and any missing intervals with every result.
- Do not mix good-event definitions in one SLI; compute separate SLOs for availability and latency.
- An SLO is a decision tool, not a blame tool: describe impact on users, not on people.
- Do not invent traffic numbers; if counts are missing, ask for them or say which result cannot be computed.
FILE:references/slo-math.md
# SLO Math
## Definitions
- **SLI** (service level indicator) = good events / total events, as a percentage.
- **SLO** (objective) = target SLI over a window, e.g. 99.9% of requests succeed over a rolling 30 days.
- **Error budget** = 1 - SLO = the share of events allowed to fail. 99.9% -> 0.1%.
- **Budget in events** = (1 - SLO) x total events in the window.
- **Error rate** over a window = bad / total in that window.
- **Burn rate** = error rate / (1 - SLO). 1.0 means the budget would be exactly used up at the end of the window.
## Projected consumption with partial data
With D seconds of data and W seconds of SLO window:
- expected requests in window = (requests in data / D) x W
- allowed bad = (1 - SLO) x expected requests
- consumed = bad in data / allowed bad
- time to exhaustion at a recent rate = (allowed bad - bad so far) / (bad per second in that recent window)
## Allowed full downtime
| SLO | per 30 days | per 28 days | per week | per day |
|---|---|---|---|---|
| 99% | 7h 12m | 6h 43m 12s | 1h 40m 48s | 14m 24s |
| 99.5% | 3h 36m | 3h 21m 36s | 50m 24s | 7m 12s |
| 99.9% | 43m 12s | 40m 19s | 10m 5s | 1m 26s |
| 99.95% | 21m 36s | 20m 10s | 5m 2s | 43s |
| 99.99% | 4m 19s | 4m 2s | 1m 0s | 9s |
Request-based SLOs do not care about clock time directly: 10 minutes of 50% errors costs the same as 5 minutes of full downtime at equal traffic, and an outage at 3 a.m. costs less budget than one at peak.
## Worked example
SLO 99.9%, 30d window, 1,000,000 requests per day.
- Budget: 0.001 x 30,000,000 = 30,000 failed requests.
- A 2% error rate for one hour at 41,667 req/h burns 833 failures = 2.8% of the budget, burn rate 20.
- At burn rate 20 the whole budget lasts 30d / 20 = 36 hours.
## Pitfalls
- Counting retries as separate good events hides user-facing failures.
- Very low traffic (night, new region) makes error rates noisy; use a minimum request count or a longer short window.
- Synthetic probes measure the probe, not users; label them separately.
- Changing the SLO mid-window: recompute the whole window with the new target and say so.
FILE:references/burn-rate-alerting.md
# Burn-Rate Alerting
## Default policy (multiwindow, multi-burn-rate)
| Severity | Budget share | Long window | Short window | Burn threshold (30d) | Burn threshold (28d) |
|---|---|---|---|---|---|
| page | 2% | 1h | 5m | 14.4 | 13.44 |
| page | 5% | 6h | 30m | 6.0 | 5.6 |
| ticket | 10% | 1d | 2h | 3.0 | 2.8 |
| ticket | 10% | 3d | 6h | 1.0 | 0.93 |
Threshold = budget share x SLO window / long window. Example: 0.02 x 720h / 1h = 14.4.
## Why two windows
- The **long window** proves the problem is significant (enough budget spent).
- The **short window** (about 1/12 of the long one) proves it is still happening, so the alert resets within minutes after recovery instead of staying red for an hour.
- Fire only when both exceed the threshold.
## Example alert rule (Prometheus style)
```
(
sum(rate(http_requests_errors_total[1h])) / sum(rate(http_requests_total[1h])) > (14.4 * 0.001)
and
sum(rate(http_requests_errors_total[5m])) / sum(rate(http_requests_total[5m])) > (14.4 * 0.001)
)
```
## Tuning
- Low-traffic services: require a minimum number of requests in the short window, or page on 6h only.
- Too many pages: check whether the SLO is stricter than users need before raising thresholds.
- Missed incidents: slow burns below 1.0 never alert by design; review the budget weekly.
- Data step coarser than the short window (e.g. 15-minute buckets): the short window becomes one bucket and recovery detection is slower.
## Budget policy ideas
- Over 50% consumed: review risky changes, prefer canaries.
- Over 75%: release freeze for non-fixes, reliability work first.
- Exhausted: postmortem required; freeze until the rolling window recovers.
FILE:templates/slo-status-report.md
# SLO Status - <service> - <period or incident>
**SLI:** <what counts as good> **SLO:** <target>% over rolling <28d/30d>
**Data:** <start> to <end> UTC, step <interval>, <n> missing intervals
**Calculator:** `slo_budget.py --slo <target> --window <window>` (exit <code>)
## Budget
| Metric | Value |
|---|---|
| SLI in period | |
| Budget consumed | |
| Budget remaining | |
| Allowed full downtime per window | |
## Burn rates
| 5m | 30m | 1h | 2h | 6h | 1d | 3d |
|---|---|---|---|---|---|---|
| | | | | | | |
## Alerts
| Policy | Status | Long / short burn |
|---|---|---|
| page 1h/5m | | |
| page 6h/30m | | |
| ticket 1d/2h | | |
| ticket 3d/6h | | |
## What happened
<worst interval, user impact, cause if known>
## Decision
- [ ] Page / incident open
- [ ] Release policy: <normal / caution / freeze> because <remaining budget>
- [ ] SLO or alert tuning: <change and reason>
## Summary for stakeholders
> <service> met <SLI>% against a <target>% target. <x>% of this window's error budget is used; at the current pace it lasts <time>. Next step: <decision>.
FILE:examples/checkout-api-requests.csv
timestamp,total,good
2026-10-18T00:00:00Z,4757,4755
2026-10-18T00:15:00Z,4525,4523
2026-10-18T00:30:00Z,4313,4311
2026-10-18T00:45:00Z,4122,4120
2026-10-18T01:00:00Z,3951,3949
2026-10-18T01:15:00Z,3803,3801
2026-10-18T01:30:00Z,3678,3676
2026-10-18T01:45:00Z,3577,3576
2026-10-18T02:00:00Z,3500,3498
2026-10-18T02:15:00Z,3448,3446
2026-10-18T02:30:00Z,3421,3419
2026-10-18T02:45:00Z,3019,3018
2026-10-18T03:00:00Z,3044,3043
2026-10-18T03:15:00Z,3093,3092
2026-10-18T03:30:00Z,3169,3168
2026-10-18T03:45:00Z,3270,3268
2026-10-18T04:00:00Z,3396,3394
2026-10-18T04:15:00Z,3547,3545
2026-10-18T04:30:00Z,3722,3720
2026-10-18T04:45:00Z,3921,3919
2026-10-18T05:00:00Z,4143,4141
2026-10-18T05:15:00Z,4388,4386
2026-10-18T05:30:00Z,4253,4251
2026-10-18T05:45:00Z,4539,4537
2026-10-18T06:00:00Z,4845,4843
2026-10-18T06:15:00Z,5168,5166
2026-10-18T06:30:00Z,5509,5507
2026-10-18T06:45:00Z,5865,5863
2026-10-18T07:00:00Z,6236,6234
2026-10-18T07:15:00Z,6619,6616
2026-10-18T07:30:00Z,7013,7009
2026-10-18T07:45:00Z,7418,7414
2026-10-18T08:00:00Z,7831,7827
2026-10-18T08:15:00Z,7850,7847
2026-10-18T08:30:00Z,8274,8271
2026-10-18T08:45:00Z,8702,8699
2026-10-18T09:00:00Z,9132,9127
2026-10-18T09:15:00Z,9561,9556
2026-10-18T09:30:00Z,9989,9984
2026-10-18T09:45:00Z,10413,10408
2026-10-18T10:00:00Z,10832,10827
2026-10-18T10:15:00Z,11245,11240
2026-10-18T10:30:00Z,11650,11645
2026-10-18T10:45:00Z,12044,12038
2026-10-18T11:00:00Z,12028,12022
2026-10-18T11:15:00Z,12398,12392
2026-10-18T11:30:00Z,12754,12748
2026-10-18T11:45:00Z,13095,13089
2026-10-18T12:00:00Z,13418,13412
2026-10-18T12:15:00Z,13724,13719
2026-10-18T12:30:00Z,14010,14003
2026-10-18T12:45:00Z,14275,14268
2026-10-18T13:00:00Z,14520,14513
2026-10-18T13:15:00Z,14742,14735
2026-10-18T13:30:00Z,14941,14934
2026-10-18T13:45:00Z,14716,14710
2026-10-18T14:00:00Z,14867,14861
2026-10-18T14:15:00Z,14993,14985
2026-10-18T14:30:00Z,15094,15086
2026-10-18T14:45:00Z,15170,15163
2026-10-18T15:00:00Z,15220,15213
2026-10-18T15:15:00Z,15244,15237
2026-10-18T15:30:00Z,15242,15236
2026-10-18T15:45:00Z,15215,15209
2026-10-18T16:00:00Z,15163,15155
2026-10-18T16:15:00Z,14686,14679
2026-10-18T16:30:00Z,14585,14578
2026-10-18T16:45:00Z,14460,14453
2026-10-18T17:00:00Z,14312,14306
2026-10-18T17:15:00Z,14141,14135
2026-10-18T17:30:00Z,13950,13866
2026-10-18T17:45:00Z,13738,13656
2026-10-18T18:00:00Z,13506,13499
2026-10-18T18:15:00Z,13257,13251
2026-10-18T18:30:00Z,12990,12984
2026-10-18T18:45:00Z,12708,12702
2026-10-18T19:00:00Z,12012,12007
2026-10-18T19:15:00Z,11702,11697
2026-10-18T19:30:00Z,11382,11376
2026-10-18T19:45:00Z,11051,11045
2026-10-18T20:00:00Z,10712,10707
2026-10-18T20:15:00Z,10367,10362
2026-10-18T20:30:00Z,10017,10013
2026-10-18T20:45:00Z,9663,9659
2026-10-18T21:00:00Z,9308,9304
2026-10-18T21:15:00Z,8952,8947
2026-10-18T21:30:00Z,8598,8594
2026-10-18T21:45:00Z,7848,7844
2026-10-18T22:00:00Z,7503,7500
2026-10-18T22:15:00Z,7164,7161
2026-10-18T22:30:00Z,6833,6830
2026-10-18T22:45:00Z,6513,6510
2026-10-18T23:00:00Z,6204,6201
2026-10-18T23:15:00Z,5907,5904
2026-10-18T23:30:00Z,5625,5622
2026-10-18T23:45:00Z,5358,5356
2026-10-19T00:00:00Z,5109,5107
2026-10-19T00:15:00Z,4877,4875
2026-10-19T00:30:00Z,4265,4263
2026-10-19T00:45:00Z,4074,4072
2026-10-19T01:00:00Z,3903,3901
2026-10-19T01:15:00Z,3755,3753
2026-10-19T01:30:00Z,3630,3628
2026-10-19T01:45:00Z,3529,3527
2026-10-19T02:00:00Z,3452,3451
2026-10-19T02:15:00Z,3400,3399
2026-10-19T02:30:00Z,3373,3371
2026-10-19T02:45:00Z,3371,3369
2026-10-19T03:00:00Z,3396,3394
2026-10-19T03:15:00Z,3045,3044
2026-10-19T03:30:00Z,3121,3120
2026-10-19T03:45:00Z,3222,3221
2026-10-19T04:00:00Z,3348,3347
2026-10-19T04:15:00Z,3499,3497
2026-10-19T04:30:00Z,3674,3672
2026-10-19T04:45:00Z,3873,3871
2026-10-19T05:00:00Z,4095,4093
2026-10-19T05:15:00Z,4340,4338
2026-10-19T05:30:00Z,4605,4603
2026-10-19T05:45:00Z,4491,4489
2026-10-19T06:00:00Z,4797,4795
2026-10-19T06:15:00Z,5120,5117
2026-10-19T06:30:00Z,5461,5458
2026-10-19T06:45:00Z,5817,5814
2026-10-19T07:00:00Z,6188,6185
2026-10-19T07:15:00Z,6571,6568
2026-10-19T07:30:00Z,6965,6962
2026-10-19T07:45:00Z,7370,7366
2026-10-19T08:00:00Z,7783,7779
2026-10-19T08:15:00Z,8202,8198
2026-10-19T08:30:00Z,8226,8222
2026-10-19T08:45:00Z,8654,8650
2026-10-19T09:00:00Z,9084,9080
2026-10-19T09:15:00Z,9513,9509
2026-10-19T09:30:00Z,9941,9936
2026-10-19T09:45:00Z,10365,10360
2026-10-19T10:00:00Z,10784,10779
2026-10-19T10:15:00Z,11197,11192
2026-10-19T10:30:00Z,11602,11597
2026-10-19T10:45:00Z,11996,11991
2026-10-19T11:00:00Z,12380,12375
2026-10-19T11:15:00Z,12350,12344
2026-10-19T11:30:00Z,12706,12700
2026-10-19T11:45:00Z,13047,13041
2026-10-19T12:00:00Z,13370,13364
2026-10-19T12:15:00Z,13676,13670
2026-10-19T12:30:00Z,13962,13956
2026-10-19T12:45:00Z,14227,14221
2026-10-19T13:00:00Z,14472,14464
2026-10-19T13:15:00Z,14694,14687
2026-10-19T13:30:00Z,14893,14886
2026-10-19T13:45:00Z,15068,15061
2026-10-19T14:00:00Z,14819,14812
2026-10-19T14:15:00Z,14945,14939
2026-10-19T14:30:00Z,15046,15040
2026-10-19T14:45:00Z,15122,15114
2026-10-19T15:00:00Z,15172,15164
2026-10-19T15:15:00Z,15196,15189
2026-10-19T15:30:00Z,15194,15187
2026-10-19T15:45:00Z,15167,15160
2026-10-19T16:00:00Z,15115,15109
2026-10-19T16:15:00Z,15038,15032
2026-10-19T16:30:00Z,14937,14929
2026-10-19T16:45:00Z,14412,14405
2026-10-19T17:00:00Z,14264,14257
2026-10-19T17:15:00Z,14093,14087
2026-10-19T17:30:00Z,13902,13896
2026-10-19T17:45:00Z,13690,13684
2026-10-19T18:00:00Z,13458,13453
2026-10-19T18:15:00Z,13209,13202
2026-10-19T18:30:00Z,12942,12936
2026-10-19T18:45:00Z,12660,12654
2026-10-19T19:00:00Z,12364,12358
2026-10-19T19:15:00Z,11654,11649
2026-10-19T19:30:00Z,11334,11329
2026-10-19T19:45:00Z,11003,10999
2026-10-19T20:00:00Z,10664,10658
2026-10-19T20:15:00Z,10319,10314
2026-10-19T20:30:00Z,9969,9964
2026-10-19T20:45:00Z,9615,9611
2026-10-19T21:00:00Z,9260,9256
2026-10-19T21:15:00Z,8904,8900
2026-10-19T21:30:00Z,8550,8547
2026-10-19T21:45:00Z,8200,8196
2026-10-19T22:00:00Z,7455,7298
2026-10-19T22:15:00Z,7116,6959
2026-10-19T22:30:00Z,6785,6629
2026-10-19T22:45:00Z,6465,6310
2026-10-19T23:00:00Z,6156,6002
2026-10-19T23:15:00Z,5859,5707
2026-10-19T23:30:00Z,5577,5426
2026-10-19T23:45:00Z,5310,5161
FILE:scripts/slo_budget.py
#!/usr/bin/env python3
"""SLO error budget and burn rate calculator (Python 3 standard library only).
Input: CSV with one row per interval, the timestamp marking the START of the interval, and request
counts. Accepted columns (case-insensitive): timestamp|time|ts, total|requests|count, and either
good|success|ok or bad|errors|failed. Timestamps: ISO 8601 (Z or offset) or Unix seconds.
Math (see references/slo-math.md):
error budget (fraction) = 1 - SLO e.g. 99.9% -> 0.001
error rate over window = sum(bad) / sum(total)
burn rate = error rate / (1 - SLO) 1.0 = spends exactly the budget over the SLO window
budget consumed so far = sum(bad) / ((1 - SLO) * expected requests in the SLO window)
alert threshold = budget share * window hours / long window hours (14.4 = 2% of 30d in 1h)
Default multiwindow, multi-burn-rate policy (Google SRE Workbook): page 2% in 1h (short 5m),
page 5% in 6h (short 30m), ticket 10% in 1d (short 2h), ticket 10% in 3d (short 6h). An alert fires
only when BOTH the long and the short window burn faster than the threshold. Windows shorter than
the data step are rounded up to one step; windows longer than the data are reported as not evaluable.
Usage:
python3 slo_budget.py requests.csv --slo 99.9 --window 30d
python3 slo_budget.py requests.csv --slo 99.5 --window 28d --json
python3 slo_budget.py --allowance --slo 99.95 --window 30d # downtime allowed, no CSV needed
python3 slo_budget.py --self-test
Exit code: 0 OK, 1 a page alert fires or the budget is exhausted, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import json
import re
import statistics
import sys
POLICY = [
# (severity, long window, short window, budget share)
("page", "1h", "5m", 0.02),
("page", "6h", "30m", 0.05),
("ticket", "1d", "2h", 0.10),
("ticket", "3d", "6h", 0.10),
]
ALIASES = {
"ts": ("timestamp", "time", "ts", "date", "datetime"),
"total": ("total", "requests", "count", "events"),
"good": ("good", "success", "ok", "successes"),
"bad": ("bad", "errors", "failed", "failures", "error"),
}
class SLOError(ValueError):
pass
def parse_duration(s):
m = re.fullmatch(r"\s*(\d+(?:\.\d+)?)\s*([smhdw])\s*", str(s).lower())
if not m:
raise SLOError(f"bad duration {s!r} (use e.g. 5m, 1h, 28d)")
return float(m.group(1)) * {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}[m.group(2)]
def fmt_dur(sec):
sec = int(round(sec))
if sec >= 86400:
d, h = divmod(sec // 3600, 24)
return f"{d}d {h}h" if h else f"{d}d"
if sec >= 3600:
h, m = divmod(sec // 60, 60)
return f"{h}h {m}m" if m else f"{h}h"
if sec >= 60:
m, s = divmod(sec, 60)
return f"{m}m {s}s" if s else f"{m}m"
return f"{sec}s"
def parse_ts(v):
v = v.strip()
if re.fullmatch(r"\d+(\.\d+)?", v):
return dt.datetime.fromtimestamp(float(v), dt.timezone.utc)
t = dt.datetime.fromisoformat(v.replace("Z", "+00:00").replace(" ", "T"))
if t.tzinfo is None:
t = t.replace(tzinfo=dt.timezone.utc)
return t.astimezone(dt.timezone.utc)
def load_rows(lines):
rd = csv.DictReader(lines)
if not rd.fieldnames:
raise SLOError("empty CSV")
cols = {}
lower = {f.strip().lower(): f for f in rd.fieldnames}
for key, names in ALIASES.items():
for n in names:
if n in lower:
cols[key] = lower[n]
break
if "ts" not in cols or "total" not in cols or not ({"good", "bad"} & cols.keys()):
raise SLOError(f"need timestamp, total and good or bad columns; got {rd.fieldnames}")
rows = []
for i, r in enumerate(rd, 2):
try:
ts = parse_ts(r[cols["ts"]])
total = int(float(r[cols["total"]]))
bad = int(float(r[cols["bad"]])) if "bad" in cols else total - int(float(r[cols["good"]]))
except (ValueError, TypeError) as e:
raise SLOError(f"line {i}: {e}")
if total < 0 or bad < 0 or bad > total:
raise SLOError(f"line {i}: counts must satisfy 0 <= bad <= total (total={total}, bad={bad})")
rows.append((ts, total, bad))
if len(rows) < 2:
raise SLOError("need at least two rows")
rows.sort(key=lambda r: r[0])
return rows
def analyze(rows, slo_pct, window, now=None, policy=POLICY):
if not 0 < slo_pct < 100:
raise SLOError("SLO must be between 0 and 100 (exclusive), e.g. 99.9")
budget = 1 - slo_pct / 100
win = parse_duration(window)
diffs = [(b[0] - a[0]).total_seconds() for a, b in zip(rows, rows[1:])]
step = statistics.median(diffs)
if step <= 0:
raise SLOError("duplicate timestamps")
first = rows[0][0]
end = (now or rows[-1][0] + dt.timedelta(seconds=step))
covered = (end - first).total_seconds()
expected_rows = int(round(covered / step))
missing = max(0, expected_rows - len(rows))
def window_stats(seconds):
seconds = max(seconds, step)
if seconds > covered + 1e-6:
return None
start = end - dt.timedelta(seconds=seconds)
tot = sum(t for ts, t, b in rows if start <= ts < end)
bad = sum(b for ts, t, b in rows if start <= ts < end)
rate = bad / tot if tot else 0.0
return {"seconds": seconds, "total": tot, "bad": bad, "error_rate": rate, "burn_rate": rate / budget}
total = sum(r[1] for r in rows)
bad = sum(r[2] for r in rows)
req_per_sec = total / covered
expected_window_total = req_per_sec * win
allowed_bad = budget * expected_window_total
consumed = bad / allowed_bad if allowed_bad else 0.0
alerts = []
for sev, lw, sw, share in policy:
thr = share * win / parse_duration(lw)
L, S = window_stats(parse_duration(lw)), window_stats(parse_duration(sw))
if L is None or S is None:
alerts.append({"severity": sev, "long": lw, "short": sw, "threshold": thr, "status": "not evaluable",
"long_burn": None, "short_burn": None})
continue
fires = L["burn_rate"] >= thr and S["burn_rate"] >= thr
alerts.append({"severity": sev, "long": lw, "short": sw, "threshold": thr,
"status": "FIRING" if fires else "ok",
"long_burn": L["burn_rate"], "short_burn": S["burn_rate"]})
burn = {}
for w in ("5m", "30m", "1h", "2h", "6h", "1d", "3d"):
s = window_stats(parse_duration(w))
burn[w] = None if s is None else s["burn_rate"]
remaining_bad = allowed_bad - bad
exhaust = {}
for w in ("1h", "6h", "1d"):
s = window_stats(parse_duration(w))
if s and s["bad"] > 0 and remaining_bad > 0:
exhaust[w] = remaining_bad / (s["bad"] / s["seconds"])
elif s:
exhaust[w] = 0.0 if remaining_bad <= 0 else None
worst = max(rows, key=lambda r: (r[2] / r[1] if r[1] else 0, r[0]))
return {
"slo_percent": slo_pct, "window": window, "window_seconds": win, "step_seconds": step,
"data_start": first.strftime("%Y-%m-%dT%H:%M:%SZ"), "data_end": end.strftime("%Y-%m-%dT%H:%M:%SZ"),
"rows": len(rows), "missing_intervals": missing,
"total": total, "bad": bad, "sli_percent": 100 * (1 - bad / total) if total else 100.0,
"allowed_bad_in_window": allowed_bad, "budget_consumed": consumed, "budget_remaining": 1 - consumed,
"allowed_downtime_seconds": budget * win,
"burn_rates": burn, "alerts": alerts, "exhaustion_seconds": exhaust,
"worst_interval": {"start": worst[0].strftime("%Y-%m-%dT%H:%M:%SZ"), "total": worst[1], "bad": worst[2],
"error_rate": worst[2] / worst[1] if worst[1] else 0.0},
}
def allowance_table(slo_pct, window):
win = parse_duration(window)
b = 1 - slo_pct / 100
return {"slo_percent": slo_pct, "window": window, "allowed_downtime_seconds": b * win,
"per_day_seconds": b * 86400, "per_week_seconds": b * 604800,
"thresholds": [{"severity": s, "long": l, "short": sh, "share": sha, "burn_rate": sha * win / parse_duration(l)}
for s, l, sh, sha in POLICY]}
def to_markdown(r):
pct = lambda x: f"{100 * x:.1f}%"
out = [f"# SLO error budget report", "",
f"**SLO:** {r['slo_percent']}% over {r['window']} **Data:** {r['data_start']} to {r['data_end']} "
f"({r['rows']} rows, step {fmt_dur(r['step_seconds'])}, {r['missing_intervals']} missing)",
f"**SLI over data:** {r['sli_percent']:.4f}% ({r['bad']:,} bad of {r['total']:,})",
f"**Error budget:** {r['allowed_bad_in_window']:,.0f} bad requests per {r['window']} at current traffic "
f"(= {fmt_dur(r['allowed_downtime_seconds'])} of full downtime)",
f"**Consumed so far:** {pct(r['budget_consumed'])} **Remaining:** {pct(r['budget_remaining'])}", "",
"## Burn rates (1.0 = on budget)", "", "| Window | Burn rate |", "|---|---|"]
for w, v in r["burn_rates"].items():
out.append(f"| {w} | {'n/a' if v is None else f'{v:.2f}'} |")
out += ["", "## Alert policy", "", "| Severity | Long / short window | Threshold | Long burn | Short burn | Status |",
"|---|---|---|---|---|---|"]
for a in r["alerts"]:
lb = "n/a" if a["long_burn"] is None else f"{a['long_burn']:.2f}"
sb = "n/a" if a["short_burn"] is None else f"{a['short_burn']:.2f}"
out.append(f"| {a['severity']} | {a['long']} / {a['short']} | {a['threshold']:.1f} | {lb} | {sb} | {a['status']} |")
out += ["", "## Time to exhaustion at recent burn"]
for w, s in r["exhaustion_seconds"].items():
out.append(f"- at last {w} rate: " + ("budget already exhausted" if s == 0 else ("no errors in window" if s is None else fmt_dur(s))))
wi = r["worst_interval"]
out += ["", f"Worst interval: {wi['start']} with {wi['bad']:,} bad of {wi['total']:,} ({100 * wi['error_rate']:.2f}%)"]
return "\n".join(out)
def self_test():
cases = []
def ok(name, cond):
cases.append((name, bool(cond)))
t0 = dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc)
def series(n, step_min, total, bad_fn):
return [(t0 + dt.timedelta(minutes=step_min * i), total, bad_fn(i)) for i in range(n)]
ok("duration parse", parse_duration("28d") == 28 * 86400 and parse_duration("5m") == 300)
ok("duration format", fmt_dur(43.2 * 60) == "43m 12s" and fmt_dur(86400 * 3) == "3d")
a = allowance_table(99.9, "30d")
ok("99.9% of 30d allows 43m 12s", round(a["allowed_downtime_seconds"]) == 2592)
ok("page threshold 14.4 for 30d", abs(a["thresholds"][0]["burn_rate"] - 14.4) < 1e-9)
ok("28d threshold scales to 13.44", abs(allowance_table(99.9, "28d")["thresholds"][0]["burn_rate"] - 13.44) < 1e-9)
flat = series(4 * 24 * 4, 15, 10000, lambda i: 10) # 0.1% errors at 99.9 -> burn 1.0
r = analyze(flat, 99.9, "30d")
ok("steady burn rate is 1.0", abs(r["burn_rates"]["1h"] - 1.0) < 1e-9 and abs(r["burn_rates"]["1d"] - 1.0) < 1e-9)
ok("no alert at burn 1.0", all(x["status"] != "FIRING" for x in r["alerts"][:3]))
ok("burn 1.0 over 4 of 30 days consumes 4/30", abs(r["budget_consumed"] - 4 / 30) < 1e-9)
ok("SLI computed", abs(r["sli_percent"] - 99.9) < 1e-9)
spike = series(4 * 24 * 4, 15, 10000, lambda i: 200 if i >= 4 * 24 * 4 - 4 else 5) # last hour 2% errors
r = analyze(spike, 99.9, "30d")
ok("fast burn pages (1h/5m)", r["alerts"][0]["status"] == "FIRING")
ok("short window rounded up to step", abs(r["burn_rates"]["5m"] - 20.0) < 1e-9)
ok("6h page not firing for 1h spike", r["alerts"][1]["status"] == "ok")
recovered = spike[:-2] + [(spike[-2][0], 10000, 5), (spike[-1][0], 10000, 5)]
r = analyze(recovered, 99.9, "30d")
ok("short window resets page after recovery", r["alerts"][0]["status"] == "ok")
short = series(8, 15, 1000, lambda i: 0)
r = analyze(short, 99.9, "30d")
ok("long windows not evaluable on 2h of data", r["alerts"][3]["status"] == "not evaluable" and r["burn_rates"]["3d"] is None)
gap = flat[:10] + flat[12:]
ok("missing intervals counted", analyze(gap, 99.9, "30d")["missing_intervals"] == 2)
ok("exhaustion time at burn 1", abs(analyze(flat, 99.9, "30d")["exhaustion_seconds"]["1d"] - 26 * 86400) < 1)
rows = load_rows(["time,requests,errors", "2026-01-01T00:00:00Z,100,1", "2026-01-01T00:05:00Z,100,0"])
ok("aliases requests/errors", rows[0][1] == 100 and rows[0][2] == 1)
rows = load_rows(["ts,total,good", "1767225600,50,49", "1767225900,50,50"])
ok("unix timestamps and good column", rows[0][2] == 1 and rows[0][0] == t0)
for data, label in [(["ts,total", "x,1"], "missing good/bad rejected"),
(["ts,total,bad", "2026-01-01T00:00:00Z,10,11", "2026-01-01T00:05:00Z,10,0"], "bad > total rejected"),
(["ts,total,bad", "nope,10,1", "2026-01-01T00:05:00Z,10,0"], "bad timestamp rejected")]:
try:
load_rows(data)
ok(label, False)
except SLOError:
ok(label, True)
try:
analyze(flat, 100, "30d")
ok("SLO 100 rejected", False)
except SLOError:
ok("SLO 100 rejected", True)
ok("markdown has alert table", "## Alert policy" in to_markdown(analyze(flat, 99.9, "30d")))
passed = sum(c for _, c in cases)
for name, c in cases:
print(("PASS " if c else "FAIL ") + name)
print(f"{passed}/{len(cases)} passed")
return 0 if passed == len(cases) else 1
def main(argv=None):
ap = argparse.ArgumentParser(description="Compute SLO error budget consumption and multiwindow burn-rate alerts.")
ap.add_argument("file", nargs="?", help="CSV of request counts per interval ('-' for stdin)")
ap.add_argument("--slo", type=float, default=99.9, help="target in percent (default 99.9)")
ap.add_argument("--window", default="30d", help="SLO window, e.g. 28d or 30d (default 30d)")
ap.add_argument("--now", help="evaluate as of this ISO time (default: end of the last interval)")
ap.add_argument("--allowance", action="store_true", help="print allowed downtime and alert thresholds only")
ap.add_argument("--json", action="store_true")
ap.add_argument("--self-test", action="store_true")
a = ap.parse_args(argv)
if a.self_test:
return self_test()
try:
if a.allowance:
t = allowance_table(a.slo, a.window)
if a.json:
print(json.dumps(t, indent=2))
else:
print(f"SLO {a.slo}% over {a.window}: {fmt_dur(t['allowed_downtime_seconds'])} of full downtime "
f"({fmt_dur(t['per_day_seconds'])}/day, {fmt_dur(t['per_week_seconds'])}/week)")
for x in t["thresholds"]:
print(f" {x['severity']:6} {x['share'] * 100:.0f}% of budget in {x['long']} (short {x['short']}): burn rate >= {x['burn_rate']:.2f}")
return 0
if not a.file:
ap.error("CSV file required (or --allowance / --self-test)")
lines = sys.stdin.read().splitlines() if a.file == "-" else open(a.file, encoding="utf-8").read().splitlines()
r = analyze(load_rows(lines), a.slo, a.window, parse_ts(a.now) if a.now else None)
except (OSError, SLOError, ValueError) as e:
print(f"error: {e}", file=sys.stderr)
return 2
print(json.dumps(r, indent=2) if a.json else to_markdown(r))
paging = any(x["severity"] == "page" and x["status"] == "FIRING" for x in r["alerts"])
return 1 if paging or r["budget_remaining"] <= 0 else 0
if __name__ == "__main__":
sys.exit(main())