#!/usr/bin/env python3 """The ODPC Record — analysis script. Single source of every statistic. Reads the data spine (products/research_wing/data/decisions.json, Kenya records), writes figures.json, charts/*.svg+png, and renders odpc-record.md from odpc-record.src.md. No statistic is hand-typed. Usage: python3 analyze.py # figures + report render /tmp/env/bin/python analyze.py --charts # also rebuild charts """ import json import re import statistics import sys from collections import Counter, defaultdict from datetime import date from pathlib import Path HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE.parent)) from lib_report import (fmt_int, fmt_usd, fmt_kes, pct, # noqa: E402 render_template, tracker_url) DATA = HERE.parent.parent / "data" CHARTS = HERE / "charts" BRAND_STYLE = HERE.parent.parent.parent.parent / "brand" / "charts" / "lawlab.mplstyle" TEMPLATE = HERE / "odpc-record.src.md" REPORT = HERE / "odpc-record.md" H1_LO, H1_HI = "2026-01", "2026-06" def ym(rec): return str(rec["date_decided"])[:7] def quarter(rec): y, m = ym(rec).split("-") return f"{y}-Q{(int(m) - 1) // 3 + 1}" def filing_year(rec): m = re.search(r"No\.?\s*\d+\s+of\s+(\d{4})", rec.get("case_number") or "") return int(m.group(1)) if m else None def filing_seq(rec): m = re.search(r"No\.?\s*(\d+)\s+of\s+\d{4}", rec.get("case_number") or "") return int(m.group(1)) if m else None def load(): decisions = json.loads((DATA / "decisions.json").read_text()) ke = [x for x in decisions if x["jurisdiction"] == "KE"] return decisions, ke def compute(decisions, ke): f = {} today = date.today().isoformat() f["build_date"] = today f["build_year"] = today[:4] odpc = [x for x in ke if x["regulator"] == "ODPC"] dets = [x for x in odpc if x["decision_type"] == "determination"] hc = [x for x in ke if x["regulator"] == "High Court"] # ------------------------------------------------------------- corpus f["n_all"] = fmt_int(len(decisions)) f["n_ke"] = fmt_int(len(ke)) f["pct_ke_of_all"] = pct(len(ke), len(decisions)) f["n_odpc"] = fmt_int(len(odpc)) f["n_dets"] = fmt_int(len(dets)) f["n_hc"] = fmt_int(len(hc)) f["n_investigations"] = fmt_int( sum(1 for x in odpc if x["decision_type"] == "investigation")) dates = sorted(str(x["date_decided"]) for x in dets) f["first_det_ym"] = dates[0] # 2023-06-12 f["last_det_ym"] = dates[-1] def human(d): months = ["January", "February", "March", "April", "May", "June", "July", "August", "September", "October", "November", "December"] parts = str(d).split("-") if len(parts) == 3: return f"{int(parts[2])} {months[int(parts[1]) - 1]} {parts[0]}" return f"{months[int(parts[1]) - 1]} {parts[0]}" f["first_det_h"] = human(dates[0]) f["last_det_h"] = human(dates[-1]) f["n_month_precision"] = fmt_int( sum(1 for x in ke if x.get("date_precision") == "month")) years = Counter(ym(x)[:4] for x in ke) for y in ("2023", "2024", "2025", "2026"): f[f"n_{y}"] = fmt_int(years.get(y, 0)) f["n_h1_2026"] = fmt_int(sum(1 for x in ke if H1_LO <= ym(x) <= H1_HI)) q = Counter(quarter(x) for x in ke) busiest, busiest_n = q.most_common(1)[0] f["busiest_quarter"] = busiest.replace("-", " ") f["n_busiest_quarter"] = fmt_int(busiest_n) # ------------------------------------------------- outcomes and orders upheld = [x for x in ke if "complaint upheld" in x["outcome"]] dismissed = [x for x in ke if "complaint dismissed" in x["outcome"]] f["n_upheld"] = fmt_int(len(upheld)) f["n_dismissed"] = fmt_int(len(dismissed)) adjudicated = len(upheld) + len(dismissed) f["n_adjudicated"] = fmt_int(adjudicated) f["pct_upheld"] = pct(len(upheld), adjudicated) f["n_resolved_parties"] = fmt_int( sum(1 for x in ke if "resolved between the parties" in x["outcome"])) f["n_enf_notices"] = fmt_int( sum(1 for x in ke if "enforcement notice issued" in x["outcome"])) f["pct_enf_of_upheld"] = pct( sum(1 for x in upheld if "enforcement notice issued" in x["outcome"]), len(upheld)) f["n_prosecution"] = fmt_int( sum(1 for x in ke if "prosecution" in x["outcome"])) f["n_prosecution_dismissed"] = fmt_int( sum(1 for x in dismissed if "prosecution" in x["outcome"])) rtypes = Counter(x["respondent_type"] for x in ke) f["n_resp_private"] = fmt_int(rtypes["private"]) f["n_resp_public"] = fmt_int(rtypes["public"]) f["n_resp_individual"] = fmt_int(rtypes["individual"]) f["pct_resp_private"] = pct(rtypes["private"], len(ke)) # ------------------------------------------------------------ pipeline with_fy = [(x, filing_year(x)) for x in dets] with_fy = [(x, fy) for x, fy in with_fy if fy] f["n_with_case_number"] = fmt_int( sum(1 for x in ke if x.get("case_number"))) f["n_with_filing_year"] = fmt_int(len(with_fy)) lag = Counter(int(ym(x)[:4]) - fy for x, fy in with_fy) f["n_same_year"] = fmt_int(lag.get(0, 0)) f["n_plus_one"] = fmt_int(lag.get(1, 0)) f["n_plus_two"] = fmt_int(lag.get(2, 0)) f["pct_same_year"] = pct(lag.get(0, 0), len(with_fy)) f["pct_within_one"] = pct(lag.get(0, 0) + lag.get(1, 0), len(with_fy)) # complaint-sequence floors: highest complaint number seen per filing year floors = defaultdict(int) filed = Counter() for x, fy in with_fy: floors[fy] = max(floors[fy], filing_seq(x) or 0) filed[fy] += 1 rows = [] for y in sorted(floors): if y in (2022, 2026): # single record / partial numbering, noisy continue rows.append("| {} | at least {:,} | {} | {}% |".format( y, floors[y], filed[y], round(100 * filed[y] / floors[y], 1))) f["floor_table"] = "\n".join(rows) f["floor_2023"] = fmt_int(floors[2023]) f["floor_2024"] = fmt_int(floors[2024]) f["floor_2025"] = fmt_int(floors[2025]) f["filed_2023"] = fmt_int(filed[2023]) f["filed_2024"] = fmt_int(filed[2024]) f["filed_2025"] = fmt_int(filed[2025]) total_floor = floors[2023] + floors[2024] + floors[2025] total_pub = filed[2023] + filed[2024] + filed[2025] f["floor_3yr"] = fmt_int(total_floor) f["pub_3yr"] = fmt_int(total_pub) f["pct_pub_of_floor"] = f"{round(100 * total_pub / total_floor, 1)}%" # participation part = [(x.get("extras") or {}).get("respondent_participated") for x in ke] f["n_participated"] = fmt_int(sum(1 for p in part if p is True)) n_absent = sum(1 for p in part if p is False) f["n_absent"] = fmt_int(n_absent) f["pct_absent"] = pct(n_absent, len(ke)) absent = [x for x in ke if (x.get("extras") or {}).get("respondent_participated") is False] a_up = sum(1 for x in absent if "complaint upheld" in x["outcome"]) a_dis = sum(1 for x in absent if "complaint dismissed" in x["outcome"]) f["n_absent_upheld"] = fmt_int(a_up) f["n_absent_dismissed"] = fmt_int(a_dis) f["pct_absent_upheld"] = pct(a_up, a_up + a_dis) f["n_site_visits"] = fmt_int(sum( 1 for x in ke if (x.get("extras") or {}).get("site_visit_conducted") is True)) ccounts = [(x.get("extras") or {}).get("complainant_count") for x in ke] ccounts = [c for c in ccounts if isinstance(c, int)] f["n_multi_complainant"] = fmt_int(sum(1 for c in ccounts if c > 1)) f["max_complainants"] = fmt_int(max(ccounts)) # ------------------------------------------------------- compensation comp = [x for x in ke if x.get("compensation_amount")] amts = sorted(x["compensation_amount"] for x in comp) usd = sorted(x["compensation_usd"] for x in comp if x.get("compensation_usd")) f["n_comp"] = fmt_int(len(comp)) f["pct_comp_of_upheld"] = pct( sum(1 for x in upheld if x.get("compensation_amount")), len(upheld)) f["comp_median_kes"] = fmt_kes(statistics.median(amts)) f["comp_median_usd"] = fmt_usd(statistics.median(usd)) f["comp_mean_kes"] = fmt_kes(statistics.mean(amts)) f["comp_total_kes"] = fmt_kes(sum(amts)) f["comp_total_usd"] = fmt_usd(sum(usd)) f["comp_max_kes"] = fmt_kes(max(amts)) f["comp_min_kes"] = fmt_kes(min(amts)) top5 = sorted(comp, key=lambda x: -x["compensation_amount"])[:5] ex_top5 = amts[:-5] f["comp_mean_ex_top5"] = fmt_kes(statistics.mean(ex_top5)) f["comp_median_ex_top5"] = fmt_kes(statistics.median(ex_top5)) f["pct_top5_of_total"] = pct( sum(x["compensation_amount"] for x in top5), sum(amts)) rows = [] for x in top5: rows.append("| [{}]({}) | {} | KES {:,} | {} | {} |".format( x["respondent"], tracker_url(x), str(x["date_decided"])[:4], x["compensation_amount"], fmt_usd(x["compensation_usd"]), ", ".join(x["violations"][:2]))) f["top_awards_table"] = "\n".join(rows) def med_for(violation): vals = [x["compensation_amount"] for x in comp if violation in (x.get("violations") or [])] return vals img = med_for("unauthorised image/photo use") debt = med_for("unlawful debt collection") f["n_comp_image"] = fmt_int(len(img)) f["comp_median_image"] = fmt_kes(statistics.median(img)) f["n_comp_debt"] = fmt_int(len(debt)) f["comp_median_debt"] = fmt_kes(statistics.median(debt)) # ---------------------------------------------------------- violations viols = Counter() for x in ke: for v in x.get("violations") or []: viols[v] += 1 f["top_violations_table"] = "\n".join( f"| {v} | {n} | {pct(n, len(ke))} |" for v, n in viols.most_common(10)) keyv = { "n_viol_unlawful_processing": "unlawful processing", "n_viol_consent": "lack of consent", "n_viol_erasure": "failure to respond to erasure request", "n_viol_debt": "unlawful debt collection", "n_viol_image": "unauthorised image/photo use", "n_viol_commercial": "commercial use of personal data", "n_viol_disclosure": "unlawful data disclosure", "n_viol_security": "failure to implement security measures", "n_viol_registration": "failure to register with odpc", "n_viol_obstruction": "obstruction of data commissioner", "n_viol_dsar": "failure to respond to dsar", "n_viol_breach": "failure to prevent data breach", } for token, v in keyv.items(): f[token] = fmt_int(viols[v]) # provisions, rolled to parent section provs = Counter() for x in ke: for p in x.get("provisions_cited") or []: m = re.match(r"(s\.\d+)", p) if m: provs[m.group(1)] += 1 f["top_provisions_table"] = "\n".join( f"| {p} DPA 2019 | {n} |" for p, n in provs.most_common(10)) for sec in ("s.25", "s.26", "s.28", "s.29", "s.30", "s.32", "s.37", "s.40", "s.56", "s.65", "s.72"): f[f"n_prov_{sec.replace('s.', 's')}"] = fmt_int(provs.get(sec, 0)) # violation -> sections cross-tab (top violations x top co-cited sections) rows = [] for v, n in viols.most_common(8): secs = Counter() for x in ke: if v in (x.get("violations") or []): for p in x.get("provisions_cited") or []: m = re.match(r"(s\.\d+)", p) if m: secs[m.group(1)] += 1 top3 = ", ".join(s for s, _ in secs.most_common(3)) rows.append(f"| {v} | {n} | {top3} |") f["taxonomy_table"] = "\n".join(rows) # ------------------------------------------------------------- sectors sectors = Counter(x.get("sector") or "unclassified" for x in ke) n_general = sectors.pop("general", 0) f["n_sector_general"] = fmt_int(n_general) f["top_sectors_table"] = "\n".join( f"| {s} | {n} |" for s, n in sectors.most_common(10)) for token, s in (("n_sector_lending", "digital lending"), ("n_sector_banking", "banking"), ("n_sector_tech", "technology"), ("n_sector_education", "education"), ("n_sector_retail", "retail"), ("n_sector_health", "healthcare")): f[token] = fmt_int(sectors.get(s, 0)) lending = [x for x in ke if x.get("sector") == "digital lending"] l_up = sum(1 for x in lending if "complaint upheld" in x["outcome"]) l_dis = sum(1 for x in lending if "complaint dismissed" in x["outcome"]) f["pct_lending_upheld"] = pct(l_up, l_up + l_dis) def sector_viols(sector, skip=("unlawful processing",)): c = Counter() for x in ke: if x.get("sector") == sector: for v in x.get("violations") or []: if v not in skip: c[v] += 1 return ", ".join(f"{v} ({n})" for v, n in c.most_common(3)) f["lending_viols"] = sector_viols("digital lending") f["banking_viols"] = sector_viols("banking") f["education_viols"] = sector_viols("education") # percentage tokens for prose (never hand-typed) f["pct_viol_unlawful_processing"] = pct( viols["unlawful processing"], len(ke)) f["pct_viol_erasure"] = pct( viols["failure to respond to erasure request"], len(ke)) f["pct_viol_image"] = pct(viols["unauthorised image/photo use"], len(ke)) # ------------------------------------------------------------- status f["n_final"] = fmt_int(sum(1 for x in ke if x["status"] == "final")) f["n_ongoing"] = fmt_int(sum(1 for x in ke if x["status"] == "ongoing")) f["n_appealed"] = fmt_int(sum(1 for x in ke if x["status"] == "appealed")) # fx documentation (tools/migrate.py KES_USD table) f["kes_usd_2023"], f["kes_usd_2024"] = "140", "132" f["kes_usd_2025"], f["kes_usd_2026"] = "129", "129" return f # ------------------------------------------------------------------ charts def build_charts(decisions, ke): import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt if BRAND_STYLE.exists(): plt.style.use(str(BRAND_STYLE)) CHARTS.mkdir(exist_ok=True) cobalt, sky, gold = "#2168F2", "#7FB3FF", "#C9962D" def save(fig, name): for ext in ("svg", "png"): fig.savefig(CHARTS / f"{name}.{ext}", bbox_inches="tight", facecolor="white") plt.close(fig) print(f" chart: {name}.svg/.png") # 1 — quarterly volume q = Counter(quarter(x) for x in ke) labels = [f"{y}-Q{i}" for y in (2023, 2024, 2025, 2026) for i in (1, 2, 3, 4)][1:14] # 2023-Q2 .. 2026-Q2 vals = [q.get(l, 0) for l in labels] fig, ax = plt.subplots(figsize=(9, 4.2)) colors = [sky if l >= "2026-Q2" else cobalt for l in labels] ax.bar(labels, vals, color=colors, width=0.62) for i, v in enumerate(vals): ax.text(i, v + 1, str(v), ha="center", fontsize=8.5, color="#5A5A5A") ax.set_title("Kenya: published enforcement actions per quarter, 2023–H1 2026") ax.set_ylabel("Actions published") ax.tick_params(axis="x", rotation=45) fig.text(0.12, -0.14, "Q2 2026 (lighter) reflects ODPC batch-publication lag, not a fall in activity.\n" "Source: decisions.json, Law Lab Africa Research, " "research.lawlab.africa/tracker.", fontsize=7.5, color="#8A93A6") save(fig, "fig1-quarterly") # 2 — filings floor vs published, per filing year floors, filed = defaultdict(int), Counter() for x in ke: fy, seq = filing_year(x), filing_seq(x) if fy and seq and fy in (2023, 2024, 2025): floors[fy] = max(floors[fy], seq) filed[fy] += 1 yrs = [2023, 2024, 2025] fig, ax = plt.subplots(figsize=(8, 4)) xpos = range(len(yrs)) ax.bar([x - 0.2 for x in xpos], [floors[y] for y in yrs], width=0.38, color="#CADEFC", label="Complaints filed (floor, from ODPC case numbering)") ax.bar([x + 0.2 for x in xpos], [filed[y] for y in yrs], width=0.38, color=cobalt, label="Published determinations from that filing year") for i, y in enumerate(yrs): ax.text(i - 0.2, floors[y] + 40, f"{floors[y]:,}", ha="center", fontsize=9, color="#5A5A5A") ax.text(i + 0.2, filed[y] + 40, str(filed[y]), ha="center", fontsize=9, color="#1A1A1A") ax.set_xticks(list(xpos)) ax.set_xticklabels([str(y) for y in yrs]) ax.set_title("The visibility gap: complaints filed against determinations published") ax.legend(loc="upper right", fontsize=8.5, frameon=False) fig.text(0.12, -0.1, "Floor = highest complaint number the ODPC's own case numbering assigns that year, as cited\n" "in published determinations. Source: decisions.json case_number fields.", fontsize=7.5, color="#8A93A6") save(fig, "fig2-visibility-gap") # 3 — every compensation award, log scale comp = sorted([x for x in ke if x.get("compensation_amount")], key=lambda x: x["compensation_amount"]) vals = [x["compensation_amount"] for x in comp] med = statistics.median(vals) fig, ax = plt.subplots(figsize=(9, 3.4)) ax.scatter(vals, [0.5] * len(vals), s=70, color=gold, alpha=0.75, zorder=3) ax.axvline(med, color=cobalt, lw=1.2, ls="--", alpha=0.7) ax.text(med * 1.1, 0.82, f"median KES {round(med):,}", fontsize=9, color=cobalt) top = comp[-1] ax.annotate(f"{top['respondent']}\nKES {top['compensation_amount']:,}", (top["compensation_amount"], 0.5), textcoords="offset points", xytext=(-10, -38), fontsize=8, color="#8a6a1d", ha="right") ax.set_xscale("log") ax.set_yticks([]) ax.set_ylim(0, 1) ax.set_xlabel("KES, log scale") ax.set_title(f"Every ODPC compensation award in the published record (n={len(vals)})") fig.text(0.12, -0.08, "Source: decisions.json, Law Lab Africa Research, " "research.lawlab.africa/tracker.", fontsize=7.5, color="#8A93A6") save(fig, "fig3-awards") # 4 — violations viols = Counter() for x in ke: for v in x.get("violations") or []: viols[v] += 1 pairs = viols.most_common(10) fig, ax = plt.subplots(figsize=(8.6, 4.4)) names = [p[0] for p in pairs][::-1] counts = [p[1] for p in pairs][::-1] ax.barh(names, counts, color=cobalt, height=0.6) for i, v in enumerate(counts): ax.text(v + 2, i, str(v), va="center", fontsize=9, color="#5A5A5A") ax.set_title("Most-cited violation categories, Kenya (a record can cite several)") ax.set_xlim(0, max(counts) * 1.12) fig.text(0.12, -0.05, "Source: decisions.json, Law Lab Africa Research.", fontsize=7.5, color="#8A93A6") save(fig, "fig4-violations") # 5 — provisions provs = Counter() for x in ke: for p in x.get("provisions_cited") or []: m = re.match(r"(s\.\d+)", p) if m: provs[m.group(1)] += 1 pairs = provs.most_common(10) fig, ax = plt.subplots(figsize=(8, 4.4)) names = [f"{p[0]} DPA 2019" for p in pairs][::-1] counts = [p[1] for p in pairs][::-1] ax.barh(names, counts, color=cobalt, height=0.6) for i, v in enumerate(counts): ax.text(v + 2, i, str(v), va="center", fontsize=9, color="#5A5A5A") ax.set_title("Most-cited provisions of the Data Protection Act, 2019") ax.set_xlim(0, max(counts) * 1.12) fig.text(0.12, -0.05, "Subsection citations rolled into their parent section. " "Source: decisions.json.", fontsize=7.5, color="#8A93A6") save(fig, "fig5-provisions") # 6 — sectors sectors = Counter(x.get("sector") for x in ke if x.get("sector") and x["sector"] != "general") pairs = sectors.most_common(10) fig, ax = plt.subplots(figsize=(8, 4.2)) names = [p[0] for p in pairs][::-1] counts = [p[1] for p in pairs][::-1] ax.barh(names, counts, color=cobalt, height=0.6) for i, v in enumerate(counts): ax.text(v + 0.5, i, str(v), va="center", fontsize=9, color="#5A5A5A") ax.set_title("Kenyan enforcement actions by respondent sector (top 10 classified)") ax.set_xlim(0, max(counts) * 1.15) n_gen = sum(1 for x in ke if x.get("sector") == "general") fig.text(0.12, -0.05, f"{n_gen} further records are classified 'general' (no dominant " "sector stated in the source document). Source: decisions.json.", fontsize=7.5, color="#8A93A6") save(fig, "fig6-sectors") # 7 — what an adjudication produces (order types) upheld = sum(1 for x in ke if "complaint upheld" in x["outcome"]) dism = sum(1 for x in ke if "complaint dismissed" in x["outcome"]) enf = sum(1 for x in ke if "enforcement notice issued" in x["outcome"]) comp_n = sum(1 for x in ke if x.get("compensation_amount")) pros = sum(1 for x in ke if "prosecution" in x["outcome"]) pairs = [("Complaint upheld", upheld), ("Complaint dismissed", dism), ("Enforcement notice issued", enf), ("Compensation ordered", comp_n), ("Prosecution recommended", pros)] fig, ax = plt.subplots(figsize=(8, 3.4)) names = [p[0] for p in pairs][::-1] counts = [p[1] for p in pairs][::-1] colors = [gold, gold, gold, "#B0B7C3", cobalt] ax.barh(names, counts, color=colors, height=0.58) for i, v in enumerate(counts): ax.text(v + 3, i, str(v), va="center", fontsize=9, color="#5A5A5A") ax.set_title("What the record contains (orders overlap; one matter can produce several)") ax.set_xlim(0, max(counts) * 1.14) fig.text(0.12, -0.07, "Source: decisions.json outcome and compensation fields, " "Law Lab Africa Research.", fontsize=7.5, color="#8A93A6") save(fig, "fig7-orders") def main(): decisions, ke = load() figures = compute(decisions, ke) (HERE / "figures.json").write_text( json.dumps(figures, indent=1, ensure_ascii=False) + "\n") print(f"figures.json: {len(figures)} figures") if "--charts" in sys.argv: build_charts(decisions, ke) if TEMPLATE.exists(): render_template(TEMPLATE, REPORT, figures) else: print("template missing; skipping render") if __name__ == "__main__": main()