diff --git a/.gitignore b/.gitignore index 7cbbed1..57241a7 100644 --- a/.gitignore +++ b/.gitignore @@ -38,6 +38,9 @@ logs/ build/ output/ +# Generated synthetic datasets — reproduce from the generator (seeded), don't commit +tracks/customer-analytics/telco-customer-intelligence/data-model/data/ + # Claude agent instructions (internal — this repo is destined to be public) CLAUDE.md .claude/ diff --git a/README.md b/README.md index 71656de..343ebb4 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,7 @@ something under active maintenance. |---|---| | [`data-governance/`](tracks/data-governance) | Catalog-aware agents: reading metadata, and using it as context before touching data | | [`governed-ai/`](tracks/governed-ai) | LLM agents with grounded access to data — SQL, RAG, classification | -| [`customer-analytics/`](tracks/customer-analytics) | *(empty)* segmentation, churn, next-best-offer, ARPU, incrementality | +| [`customer-analytics/`](tracks/customer-analytics) | Segmentation, churn, next-best-offer, ARPU, incrementality — on one synthetic, causal, reproducible telco data model | | [`data-privacy/`](tracks/data-privacy) | *(empty)* consent, minimisation, subject rights | ## Projects @@ -41,6 +41,7 @@ something under active maintenance. | [`agent-data-analyst`](tracks/data-governance/agent-data-analyst) | data-governance | Analyses the *data* using the catalog as governance context: who owns the table, did it pass quality, what the codes mean | `reference` · paused | 47 commits · 2026-02 → 2026-06 | | [`agent-sql-khipu_ai`](tracks/governed-ai/agent-sql-khipu_ai) | governed-ai | LLM agent with SQL access and a RAG knowledge base, local models | `archived snapshot` | 8 commits · 2025-01 → 2025-11 | | [`agent_text_classification`](tracks/governed-ai/agent_text_classification) | governed-ai | Sentiment classification with a local LLM | `archived snapshot` | 8 commits · 2025-02 | +| [`telco-customer-intelligence`](tracks/customer-analytics/telco-customer-intelligence) | customer-analytics | Synthetic, causal, reproducible telco data model + the analytics cases it feeds (segmentation, churn, NBO, ARPU, incrementality) | `work in progress` | data model built & tested; cases in progress | **The two agents in `data-governance/` are a progression, not a duplicate.** `openmetadata-mcp-agent` talks *to the catalog*. `agent-data-analyst` analyses *the data*, @@ -91,13 +92,17 @@ database document what they need in their own `.env.example` — never real valu Stated plainly rather than hidden behind a green badge: -- **No test suite yet** in any track. CI runs the linter; the pytest job activates itself - the moment a `tracks///tests/` directory appears. The `test_*.py` files +- **Tests exist only for `telco-customer-intelligence`** so far (schema, referential + integrity, no-leakage, causal signal — standard-library only, so CI runs them without + installing anything). The migrated agent tracks still have none; CI's pytest job collects + every `tracks///tests/` directory as it appears. The `test_*.py` files under some `scripts/` folders are **manual integration scripts** — they need a live catalog, a database and credentials — so CI deliberately does not collect them. -- `customer-analytics/` and `data-privacy/` are empty. -- Dependencies in the migrated projects are declared with ranges, not pinned. A fresh - install today may not resolve to what the author ran. +- `data-privacy/` is empty; `customer-analytics/` has its data model built and its cases + in progress (see [#64](https://github.com/ronaldmego/site-ronaldmego/issues/64)). +- Dependencies in the migrated agent projects are declared with ranges, not pinned. A fresh + install today may not resolve to what the author ran. (The customer-analytics data model + has no third-party dependencies — standard library only.) ## License diff --git a/tracks/customer-analytics/telco-customer-intelligence/README.md b/tracks/customer-analytics/telco-customer-intelligence/README.md new file mode 100644 index 0000000..57c9399 --- /dev/null +++ b/tracks/customer-analytics/telco-customer-intelligence/README.md @@ -0,0 +1,48 @@ +# Telco Customer Intelligence + +Customer-analytics evidence on a single **synthetic, causal, reproducible** +telco dataset: one data model, feeding a season of cases — segmentation, churn, +next-best-offer, ARPU and campaign incrementality. + +> **Maturity: `work in progress`.** The data model (below) is built, tested and +> reproducible. The five cases are being added on top of it, one at a time. +> Everything here is synthetic — no employer or customer data. See +> [`data-model/DATA_CARD.md`](data-model/DATA_CARD.md). + +## Why this track + +The public brand lists **Customer & Digital Analytics** as a pillar; this track +is its open reference implementation. The hard part in customer intelligence +isn't the algorithm — it's the *data model and the governance around the +decision*: churn without leakage, next-best-offer with consent and eligibility, +uplift that isn't confounded by who you targeted. So the foundation is a dataset +built to make those problems real. + +## The foundation: [`data-model/`](data-model/) + +A seeded generator emits ~12 related tables with an explicit causal structure, +so the cases are demonstrable rather than circular. See +[`data-model/README.md`](data-model/README.md) for the schema, the causal design +and the no-leakage guarantee. + +```bash +cd data-model && uv run generate.py # reproducible CSVs, byte-for-byte by seed +``` + +Tested on every push (schema, referential integrity, no-leakage, causal signal). + +## Cases (roadmap) + +Each case will produce a reproducible notebook/pipeline, a visible result, and a +permanent write-up — the reusable evidence. (Distribution posts/video are a +separate, downstream concern.) + +| # | Case | The real problem it shows | +|---|---|---| +| 01 | Actionable segmentation | segments carry behaviour, need, risk, eligible offer, consent and a suggested action — not just RFM clusters | +| 02 | Churn without leakage | temporal split, calibration, explainable drivers, prioritisation by value — accuracy alone isn't success | +| 03 | Governed next-best-offer | propensity/uplift **and** eligibility, consent, exclusions, contact policy | +| 04 | ARPU / value decomposition | where revenue per user comes from and moves | +| 05 | Campaign incrementality | true uplift vs the confounded naive read (targeting was not random) | + +Tracked in [`ronaldmego/site-ronaldmego#64`](https://github.com/ronaldmego/site-ronaldmego/issues/64). diff --git a/tracks/customer-analytics/telco-customer-intelligence/data-model/DATA_CARD.md b/tracks/customer-analytics/telco-customer-intelligence/data-model/DATA_CARD.md new file mode 100644 index 0000000..6300032 --- /dev/null +++ b/tracks/customer-analytics/telco-customer-intelligence/data-model/DATA_CARD.md @@ -0,0 +1,44 @@ +# Data card — telco customer-intelligence (synthetic) + +**This dataset is 100% synthetic.** It is generated by a seeded program +(`generate.py`) from statistical assumptions. It contains **no real customers, +no employer data, and no data derived from any real telecom operator.** Any +resemblance to a real person or account is coincidental. + +## Why it exists + +To demonstrate customer-intelligence methods (segmentation, churn, next-best- +offer, ARPU, incrementality) on a shared, realistic-but-open dataset — without +exposing any employer's or customer's data. The narrative is grounded in the +kind of work the author has done; the *data* is fabricated on purpose. + +## What it is + +- **Population:** configurable (default 5,000 customers), a mix of prepaid and + postpaid plans across four fictional regions. +- **History:** configurable (default 24 months) of usage, billing, digital + engagement, support and campaign exposure — all before a single observation + cutoff. +- **Target:** a 90-day churn label living *after* the cutoff. + +## How it was made + +A deterministic generator with an explicit causal structure (documented in +`README.md`). Churn is caused by observable trajectories plus unobserved +satisfaction and noise; retention campaigns carry a true but confounded uplift. + +## Known limitations — read before using + +- **Not a real distribution.** Base rates are chosen to be *plausible and + instructive*, not to match any specific market. Do not cite any figure from + it as a real-world statistic. +- **Simplifications:** usage/billing/engagement are aggregated to monthly grain; + one active subscription spell per customer; a single churn cutoff. +- **The causal parameters are known** (they live in `config.py`). That is a + feature for teaching — you can check whether a model recovers them — but it + means the dataset is not a blind benchmark. + +## License & reuse + +Synthetic data, generated by code in this repository, under the repository's +MIT license. Reuse freely; keep this card with it. diff --git a/tracks/customer-analytics/telco-customer-intelligence/data-model/README.md b/tracks/customer-analytics/telco-customer-intelligence/data-model/README.md new file mode 100644 index 0000000..ee52512 --- /dev/null +++ b/tracks/customer-analytics/telco-customer-intelligence/data-model/README.md @@ -0,0 +1,99 @@ +# Telco customer-intelligence — data model + +A **synthetic, causal, reproducible** dataset that the five customer-analytics +cases (segmentation, churn, next-best-offer, ARPU, incrementality) all read +from. One data model, many cases — not a fresh dataset per notebook. + +Standard library only, no numpy/pandas: the generator stays auditable and CI can +validate it without installing anything. Output is plain CSV. + +## Generate + +```bash +# from this directory +uv run generate.py # defaults: 5000 customers, 24 months, seed 42 +uv run generate.py --customers 2000 --seed 7 --out data/ +python generate.py --customers 300 --months 12 # stdlib only — no deps required +``` + +Same seed → **byte-for-byte identical** CSVs on any machine. The `data/` output +is git-ignored; reproduce it from the generator rather than committing it. + +## Why monthly, not daily + +`#64` sketched `usage_daily`. This model aggregates usage, billing and digital +engagement to **customer × month**. Churn, ARPU and RFM are modelled at monthly +grain in practice; daily rows would multiply the dataset ~30× and bloat the CSVs +without adding analytical value. The grain is a deliberate modelling decision, +documented here rather than hidden. + +## Tables + +Reference dimensions are tiny and fixed; facts scale with the population. + +| Table | Grain | Key | Notes | +|---|---|---|---| +| `plans` | one row per plan | `plan_id` | prepaid/postpaid, fee, data/voice caps, tier | +| `offers` | one row per offer | `offer_id` | discount / data_bundle / upgrade | +| `campaigns` | one row per campaign | `campaign_id` | objective (retention/upsell/crosssell) → `offer_id` | +| `customers` | one row per customer | `customer_id` | signup, region, channel, age band, current plan, tenure | +| `subscriptions` | one spell per customer | `subscription_id` | active at cutoff (churn is modelled as future) | +| `usage_monthly` | customer × month | (`customer_id`,`period_month`) | data_gb, voice, sms, active_days | +| `billing` | customer × month | `invoice_id` | billed/paid, paid_date, days_late, status (paid/late/failed) | +| `digital_monthly` | customer × month | (`customer_id`,`period_month`) | app_logins, self_service, occasional NPS | +| `support_interactions` | one ticket | `ticket_id` | reason, channel, escalated, resolved, csat | +| `consent` | customer × channel | (`customer_id`,`channel`) | opt-in per email/sms/push/call | +| `campaign_exposures` | customer × campaign | `exposure_id` | **exposed vs control**, responded | +| `churn_labels` | one row per customer | `customer_id` | `churned_next_90d`, `observation_cutoff`, `churn_date` | + +## The causal structure (what makes the cases real) + +The label is **generated from observable trajectories** plus unobserved +satisfaction and noise — not sprinkled at random. That is what lets a churn +model recover signal, never fit perfectly, and never leak: + +- **usage decline** — a downward usage trend raises churn odds; +- **payment problems** — failed/late invoices raise churn odds; +- **unresolved escalations** — the strongest single driver; +- **weak digital engagement** — low app logins raise churn odds; +- **early life** — short-tenure customers churn more; +- **plan misfit** — paying for a plan that doesn't fit; +- **retention response** — a genuine responder to a retention campaign churns + *less* (a real, negative uplift); +- **latent satisfaction + gaussian noise** — deliberately unobserved, so the + achievable model performance is bounded (no leakage, no perfect fit). + +**No leakage by construction:** every fact stops at the observation cutoff (the +last month of history); the churn outcome lives in the *next 90 days*. There is +no post-outcome fact to leak — features are strictly pre-cutoff. The test suite +enforces this. + +**Incrementality has real confounding:** retention campaigns are *targeted at +high-risk customers* (selection bias), so a naive "responders churn less" read +is wrong — responders and the held-out **control** both sit above the base rate +because both were selected for risk. The control group is what lets the +incrementality case recover the true uplift instead of the confounded one. + +Reference signals from the default run (seed 42, 5000 customers) — **re-derive, +don't quote**; they move with the seed: + +- 90-day churn base rate ≈ 13% +- churn with an unresolved escalation ≈ 28% vs ≈ 10% without +- retention responders ≈ 20% vs held-out control ≈ 31% (both > base → confounding) + +## Layout + +``` +data-model/ +├── generate.py # CLI entrypoint +├── telco/ +│ ├── config.py # seed, scale, base rates (every knob, documented) +│ ├── model.py # the causal generator, table by table +│ └── writer.py # dict-of-tables → CSV +├── README.md # this file +└── DATA_CARD.md # synthetic-data disclosure +``` + +Tests live one level up in `../tests/` and run on the standard library, so CI +validates schema, referential integrity, the no-leakage property and the +presence of causal signal on every push. diff --git a/tracks/customer-analytics/telco-customer-intelligence/data-model/generate.py b/tracks/customer-analytics/telco-customer-intelligence/data-model/generate.py new file mode 100644 index 0000000..649dc9c --- /dev/null +++ b/tracks/customer-analytics/telco-customer-intelligence/data-model/generate.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python3 +"""Generate the synthetic telco customer-intelligence dataset. + +Reproducible: same seed -> byte-for-byte identical CSVs on any machine. + + uv run generate.py # defaults: 5000 customers, 24 months + uv run generate.py --customers 2000 --seed 7 --out data/ + python generate.py --customers 300 --months 12 # stdlib only, no deps needed +""" + +from __future__ import annotations + +import argparse +from pathlib import Path + +from telco import Config, generate, write_tables + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--customers", type=int, default=Config.n_customers, help="population size") + ap.add_argument("--months", type=int, default=Config.n_months, help="months of history before the cutoff") + ap.add_argument("--seed", type=int, default=Config.seed, help="RNG seed (determinism)") + ap.add_argument("--out", type=Path, default=Path(__file__).parent / "data", help="output directory") + args = ap.parse_args() + + cfg = Config(seed=args.seed, n_customers=args.customers, n_months=args.months) + tables = generate(cfg) + counts = write_tables(tables, args.out) + + churn = tables["churn_labels"] + churn_rate = sum(r["churned_next_90d"] for r in churn) / max(1, len(churn)) + print(f"Wrote {len(counts)} tables to {args.out}/ (seed={cfg.seed}, " + f"{cfg.n_customers} customers, {cfg.n_months} months)") + width = max(len(n) for n in counts) + for name, n in counts.items(): + print(f" {name:<{width}} {n:>9,} rows") + print(f"90-day churn base rate: {churn_rate:.1%}") + + +if __name__ == "__main__": + main() diff --git a/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/__init__.py b/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/__init__.py new file mode 100644 index 0000000..3198771 --- /dev/null +++ b/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/__init__.py @@ -0,0 +1,11 @@ +"""Synthetic telco customer-intelligence data model. + +The shared, causal, reproducible dataset that the five customer-analytics cases +(segmentation, churn, next-best-offer, ARPU, incrementality) all read from. +""" + +from .config import Config +from .model import generate +from .writer import write_tables + +__all__ = ["Config", "generate", "write_tables"] diff --git a/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/config.py b/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/config.py new file mode 100644 index 0000000..b023e7b --- /dev/null +++ b/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/config.py @@ -0,0 +1,85 @@ +"""Generation parameters for the synthetic telco data model. + +Everything the generator needs to be *reproducible* lives here: a single seed, +the size of the population, the length of history, and the base rates that shape +the causal structure. Change the seed and you get a different-but-consistent +world; keep it and the output is byte-for-byte identical on any machine. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Config: + """All knobs for one generation run. + + The defaults produce a mid-sized, analytically interesting dataset + (~5k customers over 24 months). Tests override ``n_customers`` and + ``n_months`` down to keep CI fast. + """ + + seed: int = 42 + n_customers: int = 5000 + n_months: int = 24 # months of history, all <= the churn observation cutoff + + # Calendar anchor for month 0. Kept fixed so dates are reproducible without + # depending on the machine clock. + start_year: int = 2024 + start_month_of_year: int = 1 # January + + # --- Base rates that drive the causal structure ----------------------- + # These are the levers the downstream cases are meant to *recover*, so they + # live in one place and are documented in data-model/README.md. + + # Fraction of customers on prepaid (vs postpaid) plans. + prepaid_share: float = 0.55 + + # Baseline probability a billed invoice is paid late / fails, before the + # per-customer price-sensitivity adjustment. + base_late_rate: float = 0.08 + base_fail_rate: float = 0.03 + + # Support: expected tickets per active year, and the chance a ticket ends + # escalated-and-unresolved (the strongest churn driver). + tickets_per_year: float = 1.4 + unresolved_escalation_rate: float = 0.12 + + # Churn: the intercept sets the overall base rate; the weights set how much + # each observed driver moves the log-odds. Downstream churn models should be + # able to recover the *sign and rough magnitude* of these — not the exact + # value, because latent satisfaction and noise are deliberately unobserved. + churn_intercept: float = -3.6 # tuned to a ~13% 90-day base rate (realistic for telco) + w_usage_decline: float = 1.6 + w_payment_problems: float = 1.1 + w_unresolved_support: float = 1.4 + w_low_engagement: float = 0.9 + w_early_life: float = 0.8 + w_plan_misfit: float = 0.7 + w_retention_response: float = 1.5 # subtracted: a real retention uplift + churn_noise_sd: float = 0.6 + + # Campaigns: size of the true retention uplift (reduction in churn log-odds + # for a genuine responder) and the selection bias that makes the naive + # estimate wrong — retention campaigns preferentially target high-risk + # customers, so a naive "responders churn less" read is confounded. + retention_selection_bias: float = 1.8 # how strongly targeting favours risk + control_share: float = 0.30 # held-out control per campaign (for uplift) + + def output_tables(self) -> tuple[str, ...]: + """Canonical table order — also the order they are written and tested.""" + return ( + "plans", + "offers", + "campaigns", + "customers", + "subscriptions", + "usage_monthly", + "billing", + "digital_monthly", + "support_interactions", + "consent", + "campaign_exposures", + "churn_labels", + ) diff --git a/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/model.py b/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/model.py new file mode 100644 index 0000000..62a4b2e --- /dev/null +++ b/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/model.py @@ -0,0 +1,462 @@ +"""Synthetic telco customer-intelligence data model. + +A deterministic, seeded generator that emits ~12 related tables with an +**explicit causal structure**. The point of modelling the causality — rather +than sprinkling random labels — is that the downstream cases become real: + +* churn is *caused* by observable trajectories (usage decline, payment + problems, unresolved support, weak engagement) plus unobserved satisfaction + and noise, so a churn model can recover signal but never fit perfectly and + never leaks; +* retention campaigns carry a *true* uplift, but are *targeted at high-risk + customers*, so the naive "responders churn less" read is confounded — which + is exactly what the incrementality case has to untangle. + +Pure standard library on purpose: no numpy/pandas. The output is plain CSV that +anyone can diff, and CI can validate without installing project dependencies. +Every value is synthetic — see ``DATA_CARD.md``. No employer data is used. +""" + +from __future__ import annotations + +import math +from datetime import date, timedelta +from random import Random + +from .config import Config + +# --- Reference dimensions (small, fixed) --------------------------------- + +PLANS = [ + # plan_id, name, family, monthly_fee, data_gb, voice_min, tier + ("PP_S", "Prepaid S", "prepaid", 8.0, 3, 200, 1), + ("PP_M", "Prepaid M", "prepaid", 12.0, 8, 400, 2), + ("PP_L", "Prepaid L", "prepaid", 18.0, 20, 800, 3), + ("PO_S", "Postpaid S", "postpaid", 20.0, 15, 600, 2), + ("PO_M", "Postpaid M", "postpaid", 30.0, 40, 1200, 3), + ("PO_L", "Postpaid L", "postpaid", 45.0, 100, 3000, 4), + ("PO_XL", "Postpaid XL", "postpaid", 65.0, 300, 5000, 5), +] + +OFFERS = [ + # offer_id, name, type, value, eligible_family + ("OF_DISC10", "10% loyalty discount", "discount", 0.10, "any"), + ("OF_DATA5", "+5 GB data bundle", "data_bundle", 5, "any"), + ("OF_UP_M", "Upgrade to M", "upgrade", 1, "any"), + ("OF_UP_L", "Upgrade to L", "upgrade", 1, "postpaid"), + ("OF_WAIVE", "Late-fee waiver", "discount", 0.05, "any"), +] + +CAMPAIGNS = [ + # campaign_id, name, channel, objective, offer_id, month_index + ("CMP_RET_Q1", "Q1 Retention Save", "call", "retention", "OF_DISC10", 5), + ("CMP_RET_Q3", "Q3 Retention Save", "sms", "retention", "OF_WAIVE", 17), + ("CMP_UP_MID", "Mid-year Upsell", "email", "upsell", "OF_UP_M", 11), + ("CMP_XSELL", "Data Cross-sell", "push", "crosssell", "OF_DATA5", 14), +] + +REGIONS = [ + ("North", 0.28), + ("Central", 0.34), + ("South", 0.22), + ("Coast", 0.16), +] +ACQ_CHANNELS = [ + ("retail_store", 0.34), + ("online", 0.30), + ("call_center", 0.14), + ("reseller", 0.14), + ("referral", 0.08), +] +AGE_BANDS = [("18-24", 0.18), ("25-34", 0.30), ("35-44", 0.24), ("45-59", 0.19), ("60+", 0.09)] +CONSENT_CHANNELS = ("email", "sms", "push", "call") +TICKET_REASONS = ("billing", "coverage", "device", "plan_change", "roaming", "other") + +_PLAN_BY_ID = {p[0]: p for p in PLANS} + + +# --- Small helpers -------------------------------------------------------- + + +def _sigmoid(z: float) -> float: + if z < -60: + return 0.0 + if z > 60: + return 1.0 + return 1.0 / (1.0 + math.exp(-z)) + + +def _clamp(x: float, lo: float, hi: float) -> float: + return lo if x < lo else hi if x > hi else x + + +def _weighted(rng: Random, items: list[tuple]) -> tuple: + """Pick one (value, weight) pair by weight and return the value part(s).""" + r = rng.random() + acc = 0.0 + for item in items: + acc += item[-1] + if r <= acc: + return item + return items[-1] + + +def _month_start(cfg: Config, idx: int) -> date: + y = cfg.start_year + (cfg.start_month_of_year - 1 + idx) // 12 + m = (cfg.start_month_of_year - 1 + idx) % 12 + 1 + return date(y, m, 1) + + +# --- Per-customer latent world ------------------------------------------- + + +class _Customer: + """The full latent + observed state of one customer, held in memory while + the row-level tables are emitted. Latent fields never reach the CSVs.""" + + __slots__ = ( + "cid", "signup_idx", "region", "acq_channel", "age_band", + "plan_id", "family", "contract_type", + "satisfaction", "price_sensitivity", "engagement_level", "plan_fit", + "usage_base", "usage_decline", "active_from", "active_to", + # accumulated observed scores used by the churn model + "payment_problem_score", "unresolved_support_score", + "retention_response", "targeted_risk_at_selection", + ) + + def __init__(self, cid: str): + self.cid = cid + self.payment_problem_score = 0.0 + self.unresolved_support_score = 0.0 + self.retention_response = 0 + self.targeted_risk_at_selection = 0.0 + + +def _make_customers(cfg: Config, rng: Random) -> list[_Customer]: + customers: list[_Customer] = [] + for i in range(cfg.n_customers): + c = _Customer(f"C{i:06d}") + + # Signup skewed towards older tenure: more customers joined before the + # window opened (signup_idx negative) so tenure varies realistically. + c.signup_idx = int(rng.triangular(-36, cfg.n_months - 1, -6)) + c.active_from = max(0, c.signup_idx) + + c.region = _weighted(rng, REGIONS)[0] + c.acq_channel = _weighted(rng, ACQ_CHANNELS)[0] + c.age_band = _weighted(rng, AGE_BANDS)[0] + + family = "prepaid" if rng.random() < cfg.prepaid_share else "postpaid" + plan = rng.choice([p for p in PLANS if p[2] == family]) + c.plan_id = plan[0] + c.family = family + c.contract_type = family + + # Latent traits (never written out). + c.satisfaction = rng.gauss(0.0, 1.0) + c.price_sensitivity = _clamp(rng.gauss(0.0, 1.0), -3, 3) + c.engagement_level = _clamp(rng.gauss(0.0, 1.0), -3, 3) + c.plan_fit = _clamp(rng.gauss(0.4, 0.8), -3, 3) # slightly positive on avg + + # Usage baseline scaled by plan tier, with an individual multiplier. + tier = plan[6] + c.usage_base = max(0.2, rng.gauss(0.4 + 0.18 * tier, 0.5)) + # Some customers drift down over the window (an observable churn cause). + c.usage_decline = _clamp(rng.gauss(0.0, 0.5) - 0.15, -1.5, 1.0) + + c.active_to = cfg.n_months - 1 # all history is pre-cutoff; churn is future + customers.append(c) + return customers + + +# --- Table emitters ------------------------------------------------------- + + +def _emit_reference() -> dict[str, list[dict]]: + plans = [ + dict(plan_id=p[0], name=p[1], family=p[2], monthly_fee=p[3], + data_gb=p[4], voice_min=p[5], tier=p[6]) + for p in PLANS + ] + offers = [ + dict(offer_id=o[0], name=o[1], type=o[2], value=o[3], eligible_family=o[4]) + for o in OFFERS + ] + campaigns = [ + dict(campaign_id=c[0], name=c[1], channel=c[2], objective=c[3], + offer_id=c[4], month_index=c[5]) + for c in CAMPAIGNS + ] + return {"plans": plans, "offers": offers, "campaigns": campaigns} + + +def _emit_customers_and_subscriptions(cfg: Config, customers: list[_Customer]): + cust_rows, sub_rows = [], [] + for c in customers: + signup = _month_start(cfg, c.signup_idx) + tenure = cfg.n_months - c.active_from # months observed in window + cust_rows.append(dict( + customer_id=c.cid, + signup_date=signup.isoformat(), + region=c.region, + acquisition_channel=c.acq_channel, + age_band=c.age_band, + contract_type=c.contract_type, + current_plan_id=c.plan_id, + tenure_months=tenure, + )) + sub_rows.append(dict( + subscription_id=f"S{c.cid[1:]}_0", + customer_id=c.cid, + plan_id=c.plan_id, + start_date=signup.isoformat(), + end_date="", # still active at cutoff; churn happens after + status="active", + )) + return cust_rows, sub_rows + + +def _emit_usage(cfg: Config, customers: list[_Customer], rng: Random): + """Customer x month usage. Aggregated to *monthly* grain on purpose: churn, + ARPU and RFM are modelled monthly, and daily would bloat the CSVs without + adding analytical value. The schema doc states this deviation explicitly.""" + rows = [] + for c in customers: + plan = _PLAN_BY_ID[c.plan_id] + cap = plan[4] + n_active = cfg.n_months - c.active_from + for k in range(n_active): + idx = c.active_from + k + # Trend across the customer's active life, in [0,1]. + progress = k / max(1, n_active - 1) + trend = 1.0 + c.usage_decline * progress + level = max(0.05, c.usage_base * trend * (1 + rng.gauss(0, 0.12))) + data_gb = round(_clamp(level * cap * 0.55, 0.05, cap * 1.4), 2) + voice = int(max(0, level * plan[5] * 0.5 * (1 + rng.gauss(0, 0.2)))) + sms = int(max(0, rng.gauss(20 * level, 8))) + active_days = int(_clamp(round(18 + 8 * level + rng.gauss(0, 3)), 1, 31)) + rows.append(dict( + customer_id=c.cid, + period_month=_month_start(cfg, idx).isoformat(), + data_gb_used=data_gb, + voice_min_used=voice, + sms_count=sms, + active_days=active_days, + )) + return rows + + +def _emit_billing(cfg: Config, customers: list[_Customer], rng: Random): + rows = [] + for c in customers: + plan = _PLAN_BY_ID[c.plan_id] + fee = plan[3] + n_active = cfg.n_months - c.active_from + problems = 0 + for k in range(n_active): + idx = c.active_from + k + overage = round(max(0.0, rng.gauss(0, fee * 0.15)), 2) + billed = round(fee + overage, 2) + # Payment behaviour worsens with price sensitivity. + p_late = _clamp(cfg.base_late_rate + 0.04 * c.price_sensitivity, 0.0, 0.6) + p_fail = _clamp(cfg.base_fail_rate + 0.02 * c.price_sensitivity, 0.0, 0.4) + r = rng.random() + if r < p_fail: + status, days_late, paid = "failed", 0, 0.0 + problems += 1 + elif r < p_fail + p_late: + days_late = rng.randint(5, 45) + status, paid = "late", billed + problems += 0.5 + else: + status, days_late, paid = "paid", 0, billed + period = _month_start(cfg, idx) + paid_date = "" if status == "failed" else (period + timedelta(days=days_late + 3)).isoformat() + rows.append(dict( + invoice_id=f"INV_{c.cid[1:]}_{idx:02d}", + customer_id=c.cid, + period_month=period.isoformat(), + amount_billed=billed, + amount_paid=round(paid, 2), + paid_date=paid_date, + days_late=days_late, + status=status, + )) + c.payment_problem_score = problems / max(1, n_active) + return rows + + +def _emit_digital(cfg: Config, customers: list[_Customer], rng: Random): + rows = [] + for c in customers: + n_active = cfg.n_months - c.active_from + for k in range(n_active): + idx = c.active_from + k + base = 4 + 3 * c.engagement_level + logins = int(_clamp(rng.gauss(base, 2), 0, 60)) + self_service = int(_clamp(rng.gauss(base * 0.4, 1.5), 0, 40)) + rows.append(dict( + customer_id=c.cid, + period_month=_month_start(cfg, idx).isoformat(), + app_logins=logins, + self_service_actions=self_service, + nps=("" if rng.random() > 0.15 else int(_clamp(round(7 + 1.2 * c.satisfaction), 0, 10))), + )) + return rows + + +def _emit_support(cfg: Config, customers: list[_Customer], rng: Random): + rows = [] + for c in customers: + n_active = cfg.n_months - c.active_from + expected = cfg.tickets_per_year * (n_active / 12.0) + # More tickets for lower satisfaction. + expected *= _clamp(1.0 - 0.25 * c.satisfaction, 0.4, 2.0) + n_tickets = 0 + acc = 0.0 + # Simple Poisson-ish draw via successive intervals. + while True: + acc += rng.expovariate(max(0.05, expected / max(1, n_active))) + if acc >= n_active: + break + n_tickets += 1 + idx = c.active_from + int(acc) + escalated = rng.random() < 0.22 + resolved = rng.random() < (0.85 - 0.25 * (1 if escalated else 0)) + unresolved_esc = escalated and not resolved + if unresolved_esc: + c.unresolved_support_score += 1 + rows.append(dict( + ticket_id=f"TK_{c.cid[1:]}_{n_tickets:02d}", + customer_id=c.cid, + period_month=_month_start(cfg, idx).isoformat(), + reason=rng.choice(TICKET_REASONS), + channel=rng.choice(("call", "chat", "store", "app")), + escalated=int(escalated), + resolved=int(resolved), + csat=("" if not resolved else int(_clamp(round(4 + c.satisfaction), 1, 5))), + )) + return rows + + +def _emit_consent(customers: list[_Customer], rng: Random): + rows = [] + for c in customers: + for ch in CONSENT_CHANNELS: + # Engaged customers opt in more; call is opted-in least. + base = 0.6 + 0.1 * c.engagement_level - (0.25 if ch == "call" else 0.0) + consent = int(rng.random() < _clamp(base, 0.05, 0.95)) + rows.append(dict( + customer_id=c.cid, + channel=ch, + consent=consent, + )) + return rows + + +def _risk_proxy(cfg: Config, c: _Customer) -> float: + """Observable-ish risk used both for campaign targeting (confounder) and as + the backbone of the churn propensity. Kept as a log-odds contribution.""" + usage_decline_score = _clamp(-c.usage_decline, 0.0, 1.5) # steeper decline -> higher + return ( + cfg.w_usage_decline * usage_decline_score + + cfg.w_payment_problems * c.payment_problem_score + + cfg.w_unresolved_support * _clamp(c.unresolved_support_score, 0, 3) + + cfg.w_low_engagement * _clamp(-c.engagement_level, 0.0, 2.0) + + cfg.w_early_life * (1.0 if (cfg.n_months - c.active_from) <= 6 else 0.0) + + cfg.w_plan_misfit * _clamp(-c.plan_fit, 0.0, 2.0) + ) + + +def _emit_exposures(cfg: Config, customers: list[_Customer], rng: Random): + """Assign campaign audiences. Retention campaigns are *targeted at risk* + (selection bias) and carry a *true* uplift for genuine responders, split + into exposed vs a held-out control so incrementality is recoverable.""" + rows = [] + exp_n = 0 + for camp in CAMPAIGNS: + camp_id, _, channel, objective, offer_id, _mo = camp + for c in customers: + risk = _risk_proxy(cfg, c) + if objective == "retention": + # Target high-risk customers (the confounder). + p_target = _sigmoid(-1.4 + cfg.retention_selection_bias * (risk - 2.0)) + elif objective in ("upsell", "crosssell"): + # Target engaged, higher-tier customers. + p_target = _sigmoid(-1.6 + 0.7 * c.engagement_level + 0.3 * _PLAN_BY_ID[c.plan_id][6]) + else: + p_target = 0.2 + if rng.random() >= p_target: + continue + exposed = int(rng.random() >= cfg.control_share) # control held out + # Response depends on offer relevance; true effect only if exposed. + base_resp = 0.12 + 0.05 * c.engagement_level + if objective == "retention": + base_resp += 0.10 * _clamp(risk - 1.5, 0, 3) # risk customers value a save + responded = int(exposed and rng.random() < _clamp(base_resp, 0.02, 0.7)) + if objective == "retention" and responded: + c.retention_response = 1 # feeds the true churn uplift + c.targeted_risk_at_selection = risk + exp_n += 1 + rows.append(dict( + exposure_id=f"EXP_{exp_n:07d}", + campaign_id=camp_id, + customer_id=c.cid, + exposed=exposed, + responded=responded, + channel=channel, + )) + return rows + + +def _emit_churn_labels(cfg: Config, customers: list[_Customer], rng: Random): + """The modelling target. Computed at the cutoff (last month of history) and + predicting the *next 90 days*. Because every fact table stops at the cutoff, + there is nothing post-outcome to leak: features are strictly pre-cutoff.""" + cutoff = _month_start(cfg, cfg.n_months - 1) + rows = [] + for c in customers: + z = cfg.churn_intercept + _risk_proxy(cfg, c) + z -= 0.4 * c.satisfaction # latent, unobserved -> irreducible error + z -= cfg.w_retention_response * c.retention_response # the true uplift + z += rng.gauss(0.0, cfg.churn_noise_sd) + p = _sigmoid(z) + churned = int(rng.random() < p) + churn_date = "" + if churned: + churn_date = (cutoff + timedelta(days=rng.randint(1, 90))).isoformat() + rows.append(dict( + customer_id=c.cid, + observation_cutoff=cutoff.isoformat(), + churned_next_90d=churned, + churn_date=churn_date, + )) + return rows + + +# --- Public entrypoint ---------------------------------------------------- + + +def generate(cfg: Config) -> dict[str, list[dict]]: + """Build the full data model and return it as a dict of table_name -> rows. + + Order matters: billing/support/exposures accumulate the observed scores that + the churn label reads, so churn_labels is generated last. + """ + rng = Random(cfg.seed) + tables = _emit_reference() + + customers = _make_customers(cfg, rng) + cust_rows, sub_rows = _emit_customers_and_subscriptions(cfg, customers) + tables["customers"] = cust_rows + tables["subscriptions"] = sub_rows + + tables["usage_monthly"] = _emit_usage(cfg, customers, rng) + tables["billing"] = _emit_billing(cfg, customers, rng) # sets payment scores + tables["digital_monthly"] = _emit_digital(cfg, customers, rng) + tables["support_interactions"] = _emit_support(cfg, customers, rng) # sets support scores + tables["consent"] = _emit_consent(customers, rng) + tables["campaign_exposures"] = _emit_exposures(cfg, customers, rng) # sets retention_response + tables["churn_labels"] = _emit_churn_labels(cfg, customers, rng) + + return tables diff --git a/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/writer.py b/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/writer.py new file mode 100644 index 0000000..acd4f9a --- /dev/null +++ b/tracks/customer-analytics/telco-customer-intelligence/data-model/telco/writer.py @@ -0,0 +1,30 @@ +"""Write the generated tables to CSV. + +CSV, not parquet, on purpose: the files are meant to be *auditable* — a reviewer +can open them, and ``git diff`` catches any accidental change to the generator's +output. Headers come from the union of keys in the first row of each table. +""" + +from __future__ import annotations + +import csv +from pathlib import Path + + +def write_tables(tables: dict[str, list[dict]], out_dir: str | Path) -> dict[str, int]: + """Write each table to ``out_dir/.csv``. Returns row counts per table.""" + out = Path(out_dir) + out.mkdir(parents=True, exist_ok=True) + counts: dict[str, int] = {} + for name, rows in tables.items(): + path = out / f"{name}.csv" + with path.open("w", newline="", encoding="utf-8") as fh: + if not rows: + fh.write("") + counts[name] = 0 + continue + writer = csv.DictWriter(fh, fieldnames=list(rows[0].keys())) + writer.writeheader() + writer.writerows(rows) + counts[name] = len(rows) + return counts diff --git a/tracks/customer-analytics/telco-customer-intelligence/pyproject.toml b/tracks/customer-analytics/telco-customer-intelligence/pyproject.toml new file mode 100644 index 0000000..a2460b5 --- /dev/null +++ b/tracks/customer-analytics/telco-customer-intelligence/pyproject.toml @@ -0,0 +1,35 @@ +[project] +name = "telco-customer-intelligence" +version = "0.1.0" +description = "Synthetic, causal, reproducible telco customer-intelligence data model + the analytics cases it feeds" +requires-python = ">=3.10" +# The data model itself is standard-library only — deliberately, so it stays +# auditable and CI can validate it without installing anything. The individual +# case notebooks (churn, NBO, ...) will declare their own heavier deps +# (pandas/scikit-learn) as they are added. + +[tool.ruff] +target-version = "py310" +line-length = 120 + +[tool.ruff.lint] +# An active track, so a real style bar (not just the monorepo correctness floor). +select = [ + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # pyflakes + "I", # isort + "B", # flake8-bugbear + "UP", # pyupgrade +] +ignore = [ + "E501", # line length handled pragmatically +] + +[tool.ruff.lint.per-file-ignores] +"__init__.py" = ["F401"] +# Tests import the package after a sys.path insert, so imports are not at top. +"tests/*" = ["E402"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/tracks/customer-analytics/telco-customer-intelligence/tests/test_data_model.py b/tracks/customer-analytics/telco-customer-intelligence/tests/test_data_model.py new file mode 100644 index 0000000..0d063ef --- /dev/null +++ b/tracks/customer-analytics/telco-customer-intelligence/tests/test_data_model.py @@ -0,0 +1,162 @@ +"""Contract tests for the synthetic telco data model. + +These run on the standard library alone (no numpy/pandas), so CI's ``uvx pytest`` +executes them without installing project dependencies. They assert three things: + +1. **Schema & referential integrity** — keys are unique, foreign keys resolve, + required fields are populated. +2. **No leakage** — every fact is dated on-or-before the churn observation + cutoff, so a churn model built from these tables cannot see the future. +3. **Causal signal is present** — churn actually rises with the drivers, and a + held-out control exists for the retention campaigns. A dataset that passed + the schema checks but had a flat, signal-free label would make the downstream + cases meaningless, so it is a test failure here. +""" + +from __future__ import annotations + +import sys +from collections import defaultdict +from datetime import date +from pathlib import Path + +# Make the data-model package importable regardless of pytest's rootdir. +_DM = Path(__file__).resolve().parent.parent / "data-model" +sys.path.insert(0, str(_DM)) + +import pytest # noqa: E402 +from telco import Config, generate # noqa: E402 + + +@pytest.fixture(scope="module") +def tables(): + # Small but non-trivial: enough customers for base rates to be stable. + cfg = Config(seed=123, n_customers=800, n_months=18) + return cfg, generate(cfg) + + +def test_all_expected_tables_present(tables): + cfg, t = tables + assert set(t.keys()) == set(cfg.output_tables()) + for name in cfg.output_tables(): + assert t[name], f"table {name} is empty" + + +def test_primary_keys_unique(tables): + _, t = tables + pks = { + "plans": "plan_id", + "offers": "offer_id", + "campaigns": "campaign_id", + "customers": "customer_id", + "subscriptions": "subscription_id", + "invoice": None, + "campaign_exposures": "exposure_id", + "churn_labels": "customer_id", + } + for table, key in pks.items(): + if key is None: + continue + ids = [r[key] for r in t[table]] + assert len(ids) == len(set(ids)), f"{table}.{key} has duplicates" + inv = [r["invoice_id"] for r in t["billing"]] + assert len(inv) == len(set(inv)), "billing.invoice_id has duplicates" + + +def test_foreign_keys_resolve(tables): + _, t = tables + customer_ids = {r["customer_id"] for r in t["customers"]} + plan_ids = {r["plan_id"] for r in t["plans"]} + offer_ids = {r["offer_id"] for r in t["offers"]} + + for r in t["subscriptions"]: + assert r["customer_id"] in customer_ids + assert r["plan_id"] in plan_ids + for r in t["campaigns"]: + assert r["offer_id"] in offer_ids + # Every fact table's customer_id must be a known customer. + for name in ("usage_monthly", "billing", "digital_monthly", + "support_interactions", "consent", "campaign_exposures", "churn_labels"): + for r in t[name]: + assert r["customer_id"] in customer_ids, f"{name} references unknown customer" + + +def test_every_customer_has_churn_label(tables): + _, t = tables + labeled = {r["customer_id"] for r in t["churn_labels"]} + all_customers = {r["customer_id"] for r in t["customers"]} + assert labeled == all_customers + + +def test_no_leakage_facts_are_pre_cutoff(tables): + """No fact may be dated after the observation cutoff — otherwise a churn + model would be trained on the future it is meant to predict.""" + _, t = tables + cutoff = date.fromisoformat(t["churn_labels"][0]["observation_cutoff"]) + for name in ("usage_monthly", "billing", "digital_monthly", "support_interactions"): + for r in t[name]: + period = date.fromisoformat(r["period_month"]) + assert period <= cutoff, f"{name} has a fact dated after the cutoff" + # Churn events, by contrast, must all be in the future (after the cutoff). + for r in t["churn_labels"]: + if r["churned_next_90d"] == 1: + assert date.fromisoformat(r["churn_date"]) > cutoff + + +def test_consent_covers_every_customer_and_channel(tables): + _, t = tables + n_customers = len(t["customers"]) + assert len(t["consent"]) == n_customers * 4 + for r in t["consent"]: + assert r["consent"] in (0, 1) + assert r["channel"] in ("email", "sms", "push", "call") + + +def test_churn_base_rate_is_plausible(tables): + _, t = tables + churn = [r["churned_next_90d"] for r in t["churn_labels"]] + rate = sum(churn) / len(churn) + # Realistic telco 90-day window; wide bounds so it is robust to the seed. + assert 0.05 < rate < 0.25, f"implausible churn base rate {rate:.1%}" + + +def test_causal_signal_unresolved_support_raises_churn(tables): + """The strongest designed driver must show up: customers with an unresolved + escalation churn materially more than those without.""" + _, t = tables + churn = {r["customer_id"]: r["churned_next_90d"] for r in t["churn_labels"]} + unresolved = defaultdict(int) + for r in t["support_interactions"]: + if r["escalated"] == 1 and r["resolved"] == 0: + unresolved[r["customer_id"]] += 1 + with_esc = [churn[c] for c in churn if unresolved[c] > 0] + without = [churn[c] for c in churn if unresolved[c] == 0] + assert with_esc and without + rate_with = sum(with_esc) / len(with_esc) + rate_without = sum(without) / len(without) + assert rate_with > rate_without + 0.05, ( + f"unresolved-support signal too weak: {rate_with:.1%} vs {rate_without:.1%}" + ) + + +def test_retention_campaigns_have_a_held_out_control(tables): + """Incrementality is only recoverable if a control group exists — assert + the retention campaigns actually withhold exposure from some targets.""" + _, t = tables + exposed = control = 0 + for r in t["campaign_exposures"]: + if r["campaign_id"].startswith("CMP_RET"): + if r["exposed"] == 1: + exposed += 1 + else: + control += 1 + assert exposed > 0 and control > 0, "retention campaign lacks exposed/control split" + + +def test_generation_is_deterministic(tables): + cfg, t = tables + again = generate(cfg) + # Same seed -> identical churn labels and identical row counts everywhere. + assert {k: len(v) for k, v in t.items()} == {k: len(v) for k, v in again.items()} + assert [r["churned_next_90d"] for r in t["churn_labels"]] == \ + [r["churned_next_90d"] for r in again["churn_labels"]]