# 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

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).
8. 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.
9. 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.
10. 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:

   pythonCopy

   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).