# CountryAtlas data pipeline (`ca`)

The pipeline turns external statistical series into one read-only DuckDB snapshot (`atlas.duckdb`) that the API
serves. It is deterministic, idempotent and never edits the live database in place (ARCHITECTURE §2 snapshot swap).

```
registry/*.yaml ──┐
                  ▼
 connectors ──► raw/ (gzip, forever) ──► normalize ──► validate ──► staging/<connector>/<spec>.parquet
                                                                          │
                       build ◄────────────────────────────────────────────┘
                         │  registry tables → merge by priority → revisions → latest/rankings/coverage
                         │  → changes/events → similarity/DNA → insights → search_index → meta → integrity
                         ▼
        build/atlas-<run_id>.duckdb ──os.replace──► atlas.duckdb  (+ snapshots/atlas-<run_id>.duckdb, keep 7)
```

## Commands

| Command | What it does |
|---|---|
| `ca registry validate` | YAML sanity (unique slugs, topics, connectors, `similarity.yaml` / `insights.yaml` references). Exit 1 on hard errors. |
| `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). |
| `ca normalize [-c X]` | Same, but re-reads the newest raw files instead of downloading (re-run after fixing a connector). |
| `ca validate [-c X]` | Re-applies the generic rules to every staging file and rewrites statuses + `.issues.json`. |
| `ca build [--no-strict] [--no-swap]` | New snapshot from staging; derived tables; integrity checks; atomic swap + snapshot copy. |
| `ca refresh [-c X]` | `fetch` + `build` under one run id (what the scheduler runs). |
| `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`. |
| `ca status` | Connectors (implemented, specs, staging files, last run, ok/failed), failed specs, snapshot `meta`, scheduler heartbeat. |
| `ca export indicator <slug> -f csv\|json\|parquet` / `ca export country <ISO3>` | Files under `exports/`; the API reuses `countryatlas.pipeline.export`. |
| `ca sql "select …"` / `ca sql` | Read-only DuckDB query / mini REPL on the live snapshot. |

Everything runs with `.venv/bin/ca …`. Data lives in `CA_DATA_DIR` (default `~/countryatlas-data`).

## Where files live (`CA_DATA_DIR`)

```
raw/<connector>/<dataset>/<YYYY-MM-DD>/<code>-<sha1[:10]>.{json|csv}.gz   payload exactly as served (kept forever)
                                       <code>-<sha1[:10]>.{json|csv}.meta.json   url, retrieved_at, source_updated_at, pages, meta
staging/<connector>/<dataset>__<code>__<indicator>.parquet    NormalizedObservation columns (+ status), one file per spec
staging/<connector>/<…>.run.json                              ImportRun of the LAST attempt (ok|partial|failed|quarantined)
staging/<connector>/<…>.issues.json                           validation issues of the last successful write (capped)
staging/<connector>/<…>.meta.json                             source_url, notes (sourceNote…), source_updated_at, licence
build/atlas-<run_id>.duckdb                                   work in progress (deleted on failure)
atlas.duckdb                                                  live snapshot (API opens read-only)
snapshots/atlas-<run_id>.duckdb                               last CA_KEEP_SNAPSHOTS (7) successful builds
exports/indicators/<slug>.<fmt>, exports/countries/<ISO3>.<fmt>
logs/refresh-<date>.log, scheduler.json
```

Shared OWID CSVs (`co2`, `energy`) are downloaded once per run and stored once (raw code `owid-co2-data` /
`owid-energy-data`); `store_raw` de-duplicates by content hash, so re-running on the same day is free.

## The unit of work: one `IndicatorSourceSpec`

Each entry of `indicators[].sources` (plus `registry/sources/<connector>.yaml`) is processed in isolation
(`pipeline/fetch.py::process_spec`):

1. `connector.fetch(spec)` → one or more `RawPayload` (pagination inside the connector) → `store_raw`.
2. `connector.normalize(raw, spec)` → `list[NormalizedObservation]` (ISO3 via `registry.lookup()`, aggregates dropped,
   `spec.transform` applied, `is_forecast` / `is_estimate` set).
3. `connector.validate(rows)` — duplicates → dataset quarantined.
4. Generic rules (`pipeline/validate.py`, below) → `status` per row, issues, possibly dataset quarantine.
5. Atomic write of the parquet + sidecars.

Any exception → `run.json` with `status: failed` and the message; the previous parquet stays. A connector whose module
is not implemented yet is skipped with a warning (its specs are simply not staged). Retired World Bank codes are reported
as `failed` with the API message and a "RETIRED … fix the registry mapping" hint.

## Validation & quarantine

Row statuses (`observations.status`): `imported` (default) · `warning` (extreme jump, kept) · `stale` (the country's
latest period ended more than `stale_after_days` ago; only that latest row) · `quarantined` (outside registry `bounds`
or non-finite). Nothing is deleted. Quarantined rows stay in `observations` for inspection but are **excluded from every
derived table** (`latest`, `rankings`, `changes`, `events`, `similarity`, `insights`, `country_dna`).

Row statuses also cover `impossible_year` (year < 1750 or > current year + 10 → `quarantined`).

Dataset-level quarantine (`quarantine_dataset=True`): `schema_change` (a required staging column is missing or has an
incompatible dtype), duplicates, unit ≠ registry unit, or fewer than 30 % of the rows of the previous staging file for the
same spec (`partial_download`). The new file is **not written**, the previous one is kept, and the `run.json` says
`quarantined` with the reason. `ca status` lists these.

Dataset-level **warnings** compared with the previous staging file of the same spec (`PreviousStats`, read by
`fetch.py` before validation): `null_spike` (share of null values > 3 × the previous share and > 5 %) and `vintage_shift`
(median |relative change| over ≥ 20 overlapping keys > 25 % — a source revision or a changed definition). Both are logged,
stored in `validation_issues`, and the file is still written: suspicious values are flagged, never deleted.

HTTP reliability (`connectors/base.py::Connector.get`): every request goes through one method with 5 attempts and a
deterministic exponential backoff (2, 4, 8, 16 s, capped at 60 s, no jitter) on 408/425/429/5xx and transport errors
(timeouts, resets); a `Retry-After` header (seconds or HTTP-date) is honoured before the backoff, capped at 120 s.

Extreme jumps: positive level series (`format` in currency/number/tonnes/kwh/per_*/km/ha with a lower bound ≥ 0) are
compared on log-differences, others on absolute differences; threshold = `jump_threshold` × 1.4826 × MAD of the
country's differences, with a floor of 10 % (relative) or 2 % of the series range (absolute); at least 5 points.

## Build (`pipeline/build.py`, ≈ 10 s for 2.3 M staging rows)

1. `schema.sql` → registry tables (`countries`, `groups`, `group_members`, `sources`, `indicators`,
   `indicator_sources` incl. `source_url`/`notes` from the staging sidecars).
2. All staging parquet files whose spec still exists in the registry → `staging_all` (joined with
   `indicator_sources.priority`); files of removed/renamed specs are **orphans**, ignored with a warning (delete them or
   re-map the spec; `ca validate` lists them too). Source selection is per
   **series** `(country, indicator, frequency)`: the highest-priority source with non-forecast data for that country wins the
   whole series, unless a lower-priority source is > 3 years fresher (`metadata.merge_reason` = `priority` | `fresher`).
   Sources are never spliced inside a series; every other source's complete series goes to `observations_alt`. Forecast
   rows of the chosen source are kept (dashed on charts) but never enter derived tables.
3. `observation_revisions`: the previous `atlas.duckdb` is attached read-only; every key whose value or source changed
   is recorded with the new `run_id`; the previous revisions table is copied over.
4. `import_runs` (latest `run.json` per spec) and `validation_issues` (issues sidecars).
5. Derived tables (`pipeline/derived.py`, SQL) are computed from `obs_ok` = non-forecast, non-quarantined observations at
   the indicator's **canonical frequency** only (a monthly FRED series never becomes a country's "latest" for an annual
   indicator): `latest` (prev period, 10-year change, ranks within the same year among
   `kind='country'` — world / WB region / income group), `rankings` (ranking-eligible indicators, years with ≥ 20
   countries), `coverage`, indicator/source coverage columns, `search_index`.
   **Rank direction:** rank 1 = lowest value when `higher_is_better = false`, otherwise highest value ("best" when
   `higher_is_better` is set, "highest" when null). `pct_rank = 1 − (rank−1)/(n−1)`.
6. `changes` / `events` (`pipeline/changes.py`, numpy per series, ≈ 45 k series in ~3 s) — see ARCHITECTURE §7/§7.1 and
   the module docstring; headlines are English templates such as
   *"Inflation fell 3.4 points to 3.4 % in 2025 (largest drop since 2009)."* `changes` only keep detections at a series'
   latest period when that period is recent (≤ 2 years behind the indicator's max year and ≤ 3 years behind today);
   monotone series and `cumulative`-tagged indicators are silent; severity = detector score × importance weight
   (headline 1.0 / featured 0.9 / other 0.7). Q/M series are annualised first (one detection per year at most); index /
   non-rankable series need a ≥ 10 % move; `events` keep severity ≥ 0.35 and ≤ 3 rows per (country, indicator, year).
   **Change detection 2.0** (`countryatlas.stats`): `structural_break` — single mean shift (binary segmentation) with
   gain ≥ 0.5 of the total sum of squares, |shift| ≥ 1.5 × sd, ≥ 5 years on each side, non-monotone series; a `change` when
   the break is within the last 10 years, always an `event` (one per series); `trend_reversal` — three yearly moves of one
   sign after three of the opposite sign, cumulative move ≥ floor (severity 0.55); `volatility_spike` — sd of the last 5
   yearly differences ≥ 3 × sd of the previous 15 and ≥ floor (severity 0.45–0.7). Real snapshot (2026-09-12): 3 355 breaks,
   1 799 volatility spikes (mostly life expectancy 2019–2024), 546 reversals in `changes`; 19 593 break events.
7. `similarity` (5 modes, `registry/similarity.yaml`) and `country_dna` (9 percentile dimensions, `dna:` section of the
   same file). Each contribution carries `z_a`, `z_b`, `weight`, `contribution` **and the raw latest values `value_a` /
   `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).
9. `meta` (schema_version, build_run_id, built_at, counts, `previous_observation_count`, `source_health` JSON = per
   connector ok / partial / failed / quarantined spec counts from `import_runs`, connectors, duration) → `CHECKPOINT`.
10. Integrity: ≥ 100 000 observations when World Bank staging exists; every headline indicator that has staging data
    has ≥ 100 countries in `latest`; **the new snapshot has ≥ 90 % of the previous snapshot's observations** (a shrunken
    build is never published). Failure (or any exception) deletes the build file and leaves the live DB untouched.
11. `os.replace` → `atlas.duckdb`, copy to `snapshots/`, prune to `CA_KEEP_SNAPSHOTS`. Every build (published or failed)
    writes `logs/build-<run_id>.json` (counts, warnings, error, staging files used per connector).

## Adding a connector

1. Create `src/countryatlas/connectors/<id>.py` with a subclass of `countryatlas.connectors.base.Connector`:

   ```python
   class FooConnector(Connector):
       id = "foo"; name = "Foo Stats"; organization = "…"; url = "…"; licence = "…"; attribution = "…"
       api_base = "https://…"; rate_per_minute = 60; country_codes = "iso3"

       def fetch(self, spec: IndicatorSourceSpec) -> RawPayload | list[RawPayload]:
           r = self.get(f"{self.api_base}/{spec.code}", params=spec.params)      # retries/backoff/rate limit built in
           return self.payload(r, dataset=spec.dataset, code=spec.code, source_url=…, notes=…)

       def normalize(self, raw, spec) -> list[NormalizedObservation]:
           lk = lookup(); unit = indicators_by_id()[spec.indicator_id].unit
           … iso3 = lk.from_iso3(code) / from_iso2 / from_name …  (None → drop)
           … value = self.apply_transform(float(v), spec.transform) …
   ```
   Put `source_url` / `notes` in `RawPayload.meta` — the pipeline copies them into `indicator_sources`. Raise
   `countryatlas.connectors._util.ConnectorError` for non-transient problems (unknown code, empty dataset). If one raw
   file serves many specs, cache it in the instance and implement `raw_code_for(spec)` so `ca normalize` can find it.
2. Add one line to `CONNECTORS` in `src/countryatlas/connectors/__init__.py` (`"foo": "countryatlas.connectors.foo:FooConnector"`).
   The package is also scanned for `Connector` subclasses, so a missing entry is not fatal.
3. Map indicators: `sources:` entries in `registry/indicators.yaml` **or** a `registry/sources/foo.yaml` file
   (`sources: [{indicator, dataset, code, params, priority, transform, countries, frequency, notes}]`). Add `foo` to
   `registry.CONNECTORS` if it is a new id.
4. `ca fetch -c foo -i <one-indicator>` → check the staging parquet, then `ca build` and `ca status`.

Tests: `.venv/bin/python -m pytest tests/` (fixtures in `tests/fixtures/` are small recorded subsets of real payloads).