spb/countryatlas

1# CountryAtlas data pipeline (`ca`)23The pipeline turns external statistical series into one read-only DuckDB snapshot (`atlas.duckdb`) that the API4serves. It is deterministic, idempotent and never edits the live database in place (ARCHITECTURE §2 snapshot swap).56```7registry/*.yaml ──┐8                  ▼9 connectors ──► raw/ (gzip, forever) ──► normalize ──► validate ──► staging/<connector>/<spec>.parquet10                                                                          │11                       build ◄────────────────────────────────────────────┘12                         │  registry tables → merge by priority → revisions → latest/rankings/coverage13                         │  → changes/events → similarity/DNA → insights → search_index → meta → integrity14                         ▼15        build/atlas-<run_id>.duckdb ──os.replace──► atlas.duckdb  (+ snapshots/atlas-<run_id>.duckdb, keep 7)16```1718## Commands1920| Command | What it does |21|---|---|22| `ca registry validate` | YAML sanity (unique slugs, topics, connectors, `similarity.yaml` / `insights.yaml` references). Exit 1 on hard errors. |23| `ca fetch [-c worldbank] [-i gdp]` | Download → store raw → normalize → validate → staging, one unit of work per source spec, concurrently (`CA_HTTP_CONCURRENCY`, default 6). |24| `ca normalize [-c X]` | Same, but re-reads the newest raw files instead of downloading (re-run after fixing a connector). |25| `ca validate [-c X]` | Re-applies the generic rules to every staging file and rewrites statuses + `.issues.json`. |26| `ca build [--no-strict] [--no-swap]` | New snapshot from staging; derived tables; integrity checks; atomic swap + snapshot copy. |27| `ca refresh [-c X]` | `fetch` + `build` under one run id (what the scheduler runs). |28| `ca schedule [--now]` | Loop: refresh daily at `CA_REFRESH_HOUR:CA_REFRESH_MINUTE` (03:15 `America/Toronto`); `kill -USR1 <pid>` refreshes now; heartbeat in `data_dir/scheduler.json`; logs in `logs/refresh-<date>.log`. |29| `ca status` | Connectors (implemented, specs, staging files, last run, ok/failed), failed specs, snapshot `meta`, scheduler heartbeat. |30| `ca export indicator <slug> -f csv\|json\|parquet` / `ca export country <ISO3>` | Files under `exports/`; the API reuses `countryatlas.pipeline.export`. |31| `ca sql "select …"` / `ca sql` | Read-only DuckDB query / mini REPL on the live snapshot. |3233Everything runs with `.venv/bin/ca …`. Data lives in `CA_DATA_DIR` (default `~/countryatlas-data`).3435## Where files live (`CA_DATA_DIR`)3637```38raw/<connector>/<dataset>/<YYYY-MM-DD>/<code>-<sha1[:10]>.{json|csv}.gz   payload exactly as served (kept forever)39                                       <code>-<sha1[:10]>.{json|csv}.meta.json   url, retrieved_at, source_updated_at, pages, meta40staging/<connector>/<dataset>__<code>__<indicator>.parquet    NormalizedObservation columns (+ status), one file per spec41staging/<connector>/<…>.run.json                              ImportRun of the LAST attempt (ok|partial|failed|quarantined)42staging/<connector>/<…>.issues.json                           validation issues of the last successful write (capped)43staging/<connector>/<…>.meta.json                             source_url, notes (sourceNote…), source_updated_at, licence44build/atlas-<run_id>.duckdb                                   work in progress (deleted on failure)45atlas.duckdb                                                  live snapshot (API opens read-only)46snapshots/atlas-<run_id>.duckdb                               last CA_KEEP_SNAPSHOTS (7) successful builds47exports/indicators/<slug>.<fmt>, exports/countries/<ISO3>.<fmt>48logs/refresh-<date>.log, scheduler.json49```5051Shared OWID CSVs (`co2`, `energy`) are downloaded once per run and stored once (raw code `owid-co2-data` /52`owid-energy-data`); `store_raw` de-duplicates by content hash, so re-running on the same day is free.5354## The unit of work: one `IndicatorSourceSpec`5556Each entry of `indicators[].sources` (plus `registry/sources/<connector>.yaml`) is processed in isolation57(`pipeline/fetch.py::process_spec`):58591. `connector.fetch(spec)` → one or more `RawPayload` (pagination inside the connector) → `store_raw`.602. `connector.normalize(raw, spec)` → `list[NormalizedObservation]` (ISO3 via `registry.lookup()`, aggregates dropped,61   `spec.transform` applied, `is_forecast` / `is_estimate` set).623. `connector.validate(rows)` — duplicates → dataset quarantined.634. Generic rules (`pipeline/validate.py`, below) → `status` per row, issues, possibly dataset quarantine.645. Atomic write of the parquet + sidecars.6566Any exception → `run.json` with `status: failed` and the message; the previous parquet stays. A connector whose module67is not implemented yet is skipped with a warning (its specs are simply not staged). Retired World Bank codes are reported68as `failed` with the API message and a "RETIRED … fix the registry mapping" hint.6970## Validation & quarantine7172Row statuses (`observations.status`): `imported` (default) · `warning` (extreme jump, kept) · `stale` (the country's73latest period ended more than `stale_after_days` ago; only that latest row) · `quarantined` (outside registry `bounds`74or non-finite). Nothing is deleted. Quarantined rows stay in `observations` for inspection but are **excluded from every75derived table** (`latest`, `rankings`, `changes`, `events`, `similarity`, `insights`, `country_dna`).7677Row statuses also cover `impossible_year` (year < 1750 or > current year + 10 → `quarantined`).7879Dataset-level quarantine (`quarantine_dataset=True`): `schema_change` (a required staging column is missing or has an80incompatible dtype), duplicates, unit ≠ registry unit, or fewer than 30 % of the rows of the previous staging file for the81same spec (`partial_download`). The new file is **not written**, the previous one is kept, and the `run.json` says82`quarantined` with the reason. `ca status` lists these.8384Dataset-level **warnings** compared with the previous staging file of the same spec (`PreviousStats`, read by85`fetch.py` before validation): `null_spike` (share of null values > 3 × the previous share and > 5 %) and `vintage_shift`86(median |relative change| over ≥ 20 overlapping keys > 25 % — a source revision or a changed definition). Both are logged,87stored in `validation_issues`, and the file is still written: suspicious values are flagged, never deleted.8889HTTP reliability (`connectors/base.py::Connector.get`): every request goes through one method with 5 attempts and a90deterministic exponential backoff (2, 4, 8, 16 s, capped at 60 s, no jitter) on 408/425/429/5xx and transport errors91(timeouts, resets); a `Retry-After` header (seconds or HTTP-date) is honoured before the backoff, capped at 120 s.9293Extreme jumps: positive level series (`format` in currency/number/tonnes/kwh/per_*/km/ha with a lower bound ≥ 0) are94compared on log-differences, others on absolute differences; threshold = `jump_threshold` × 1.4826 × MAD of the95country's differences, with a floor of 10 % (relative) or 2 % of the series range (absolute); at least 5 points.9697## Build (`pipeline/build.py`, ≈ 10 s for 2.3 M staging rows)98991. `schema.sql` → registry tables (`countries`, `groups`, `group_members`, `sources`, `indicators`,100   `indicator_sources` incl. `source_url`/`notes` from the staging sidecars).1012. All staging parquet files whose spec still exists in the registry → `staging_all` (joined with102   `indicator_sources.priority`); files of removed/renamed specs are **orphans**, ignored with a warning (delete them or103   re-map the spec; `ca validate` lists them too). Source selection is per104   **series** `(country, indicator, frequency)`: the highest-priority source with non-forecast data for that country wins the105   whole series, unless a lower-priority source is > 3 years fresher (`metadata.merge_reason` = `priority` | `fresher`).106   Sources are never spliced inside a series; every other source's complete series goes to `observations_alt`. Forecast107   rows of the chosen source are kept (dashed on charts) but never enter derived tables.1083. `observation_revisions`: the previous `atlas.duckdb` is attached read-only; every key whose value or source changed109   is recorded with the new `run_id`; the previous revisions table is copied over.1104. `import_runs` (latest `run.json` per spec) and `validation_issues` (issues sidecars).1115. Derived tables (`pipeline/derived.py`, SQL) are computed from `obs_ok` = non-forecast, non-quarantined observations at112   the indicator's **canonical frequency** only (a monthly FRED series never becomes a country's "latest" for an annual113   indicator): `latest` (prev period, 10-year change, ranks within the same year among114   `kind='country'` — world / WB region / income group), `rankings` (ranking-eligible indicators, years with ≥ 20115   countries), `coverage`, indicator/source coverage columns, `search_index`.116   **Rank direction:** rank 1 = lowest value when `higher_is_better = false`, otherwise highest value ("best" when117   `higher_is_better` is set, "highest" when null). `pct_rank = 1 − (rank−1)/(n−1)`.1186. `changes` / `events` (`pipeline/changes.py`, numpy per series, ≈ 45 k series in ~3 s) — see ARCHITECTURE §7/§7.1 and119   the module docstring; headlines are English templates such as120   *"Inflation fell 3.4 points to 3.4 % in 2025 (largest drop since 2009)."* `changes` only keep detections at a series'121   latest period when that period is recent (≤ 2 years behind the indicator's max year and ≤ 3 years behind today);122   monotone series and `cumulative`-tagged indicators are silent; severity = detector score × importance weight123   (headline 1.0 / featured 0.9 / other 0.7). Q/M series are annualised first (one detection per year at most); index /124   non-rankable series need a ≥ 10 % move; `events` keep severity ≥ 0.35 and ≤ 3 rows per (country, indicator, year).125   **Change detection 2.0** (`countryatlas.stats`): `structural_break` — single mean shift (binary segmentation) with126   gain ≥ 0.5 of the total sum of squares, |shift| ≥ 1.5 × sd, ≥ 5 years on each side, non-monotone series; a `change` when127   the break is within the last 10 years, always an `event` (one per series); `trend_reversal` — three yearly moves of one128   sign after three of the opposite sign, cumulative move ≥ floor (severity 0.55); `volatility_spike` — sd of the last 5129   yearly differences ≥ 3 × sd of the previous 15 and ≥ floor (severity 0.45–0.7). Real snapshot (2026-09-12): 3 355 breaks,130   1 799 volatility spikes (mostly life expectancy 2019–2024), 546 reversals in `changes`; 19 593 break events.1317. `similarity` (5 modes, `registry/similarity.yaml`) and `country_dna` (9 percentile dimensions, `dna:` section of the132   same file). Each contribution carries `z_a`, `z_b`, `weight`, `contribution` **and the raw latest values `value_a` /133   `value_b`** (after an optional `per` ratio, before the log transform) so the UI can quote real numbers. 8. `insights` (`registry/insights.yaml`, 16 templates, all numbers computed).1349. `meta` (schema_version, build_run_id, built_at, counts, `previous_observation_count`, `source_health` JSON = per135   connector ok / partial / failed / quarantined spec counts from `import_runs`, connectors, duration) → `CHECKPOINT`.13610. Integrity: ≥ 100 000 observations when World Bank staging exists; every headline indicator that has staging data137    has ≥ 100 countries in `latest`; **the new snapshot has ≥ 90 % of the previous snapshot's observations** (a shrunken138    build is never published). Failure (or any exception) deletes the build file and leaves the live DB untouched.13911. `os.replace` → `atlas.duckdb`, copy to `snapshots/`, prune to `CA_KEEP_SNAPSHOTS`. Every build (published or failed)140    writes `logs/build-<run_id>.json` (counts, warnings, error, staging files used per connector).141142## Adding a connector1431441. Create `src/countryatlas/connectors/<id>.py` with a subclass of `countryatlas.connectors.base.Connector`:145146   ```python147   class FooConnector(Connector):148       id = "foo"; name = "Foo Stats"; organization = "…"; url = "…"; licence = "…"; attribution = "…"149       api_base = "https://…"; rate_per_minute = 60; country_codes = "iso3"150151       def fetch(self, spec: IndicatorSourceSpec) -> RawPayload | list[RawPayload]:152           r = self.get(f"{self.api_base}/{spec.code}", params=spec.params)      # retries/backoff/rate limit built in153           return self.payload(r, dataset=spec.dataset, code=spec.code, source_url=…, notes=…)154155       def normalize(self, raw, spec) -> list[NormalizedObservation]:156           lk = lookup(); unit = indicators_by_id()[spec.indicator_id].unit157           … iso3 = lk.from_iso3(code) / from_iso2 / from_name …  (None → drop)158           … value = self.apply_transform(float(v), spec.transform) …159   ```160   Put `source_url` / `notes` in `RawPayload.meta` — the pipeline copies them into `indicator_sources`. Raise161   `countryatlas.connectors._util.ConnectorError` for non-transient problems (unknown code, empty dataset). If one raw162   file serves many specs, cache it in the instance and implement `raw_code_for(spec)` so `ca normalize` can find it.1632. Add one line to `CONNECTORS` in `src/countryatlas/connectors/__init__.py` (`"foo": "countryatlas.connectors.foo:FooConnector"`).164   The package is also scanned for `Connector` subclasses, so a missing entry is not fatal.1653. Map indicators: `sources:` entries in `registry/indicators.yaml` **or** a `registry/sources/foo.yaml` file166   (`sources: [{indicator, dataset, code, params, priority, transform, countries, frequency, notes}]`). Add `foo` to167   `registry.CONNECTORS` if it is a new id.1684. `ca fetch -c foo -i <one-indicator>` → check the staging parquet, then `ca build` and `ca status`.169170Tests: `.venv/bin/python -m pytest tests/` (fixtures in `tests/fixtures/` are small recorded subsets of real payloads).171