SPB Git
spboucher.ai

spb/countryatlas

Public

Analytics API 1.1 + pipeline reliability and change detection 2.0

API: /pulse, /movers, /extremes, /scatter, /trajectory, /finder, /peers, /indicators/{slug}/{related,distribution,frames,quality}, /rankings/{id}/race, /regions/compare, /countries/{id}/{story,quality}, dna?reference=, /updates, search intents (type action); shared per-snapshot cross-section cache; 13 API tests.
Pipeline: structural_break / trend_reversal / volatility_spike detectors, similarity contributions carry value_a/value_b, deterministic retry/backoff honouring Retry-After, validation rules schema_change / impossible_year / null_spike / vintage_shift, ≥90 % observation-count integrity guard, meta.source_health + previous_observation_count, build logs; group descriptions in the registry; docs updated. 138 tests pass.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Simon-Pierre Boucher committed (Sep 12, 2026) parent d588bee Browse files

25 changed files +2,856 −52

25 changed files
Load all 25 files Unified Split
modified docs/API.md +20 −0
@@ -326,3 +326,23 @@ url, score: 1.0}`: `compare <c1> <c2> [<c3>…]` → `/compare/<slugs>`; `rank[i
326 326 `structural_break` (single mean shift with gain ≥ 0.5 and shift ≥ 1.5 × series sd, segments ≥ 5 years, break within the last 10 years for
327 327 `changes`), `trend_reversal` (three consecutive yearly moves of one sign after three of the opposite sign), `volatility_spike` (sd of the
328 328 last 5 yearly differences ≥ 3 × the sd of the previous 15). Same row shape; `detail` documents the parameters.
329 +
330 +### Implementation notes (as built, 2026-09-12 — deviations from the contract above)
331 +
332 +* `/pulse` takes `min_population` (default 1 000 000, `0` disables) that filters the **top movers only** (`top_up`/`top_down`); counts and
333 + shares use every reporting country. The response echoes `min_population` and `filter_note`. Pools need ≥ 30 reporting countries (not 20)
334 + so a thin indicator (Gini, 25 countries) never headlines; `year_reference` is the latest year of the annual pillars population / GDP /
335 + life expectancy (monthly/quarterly series already carry the current year). Convergence compares max-year vs max-year − 10.
336 +* `/movers` windows 5/10 report kinds `change_5y` / `change_10y` (direction in `direction`), keep moves with robust |z| ≥ 1.5 and rank by
337 + severity = min(1, |z|/4) × importance; `kind=record|reversal|acceleration|structural` are only meaningful for `window=1` and return an
338 + empty list otherwise. Window 1 reads the `changes` table (any kind, including the pipeline's structural_break / trend_reversal /
339 + volatility_spike).
340 +* `/extremes` facets each carry their own `from_year` / `to_year` (the indicator's latest common year and its window start); the top-level
341 + pair is the min/max across facets. Start values use the nearest observation within ±2 years of the window start.
342 +* `/indicators/{slug}/related`: `min_n` accepts ≥ 5 (default 40). Pearson is computed on the association scale (log10 for level series,
343 + flagged `log_x`/`log_y`), Spearman on raw values.
344 +* `/indicators/{slug}/distribution`: `highlight` also returns `year`; group medians need ≥ 2 members with data.
345 +* `/regions/compare` `history[slug]` also carries `kind`; `/updates` sources carry `runs {n, n_failed}` and `snapshot.reference_year`.
346 +* `/countries/{id}/story` text template is `"{Country}: {Indicator} rose|fell from {first} in {y0} to {last} in {y1} ({signed change}[, ±x % a year])."`
347 + (colon form keeps acronyms such as GDP / PPP intact); `/countries/{id}/dna` adds `note`.
348 +* `/search` intents: countries are resolved by ISO2/ISO3 code first, then exact single word, then (near-)exact 2–3-word names, then fuzzy.
modified docs/ARCHITECTURE.md +7 −0
@@ -264,6 +264,13 @@ No table or column of `schema.sql` was changed. The following precisions/deviati
264 264 * **`validation_issues`** are capped at 2 000 rows per rule per spec (5 000 per spec file) to stay small; counts are exact in
265 265 `import_runs.warnings/errors`.
266 266 * `ca sql "<query>"` was added to the CLI (read-only DuckDB); `ca normalize` re-reads the newest raw files per spec.
267 +* **Change detection 2.0 (2026-09-12)**: three more `kind` values in `changes`/`events` — `structural_break`, `trend_reversal`,
268 + `volatility_spike` — computed with `countryatlas.stats` (docs/PIPELINE.md §Build step 6). Same row shape; `detail` records the
269 + parameters (`gain`, `shift_ratio`, `run`, `ratio`…). Similarity `contributions[feature]` gain `value_a` / `value_b` (raw values).
270 +* **Reliability (2026-09-12)**: validation adds `impossible_year` (row quarantined), `schema_change` (dataset quarantined),
271 + `null_spike` and `vintage_shift` (warnings vs the previous staging file); the build refuses to publish a snapshot with < 90 % of
272 + the previous snapshot's observations, records `meta.previous_observation_count` and `meta.source_health`, and writes
273 + `logs/build-<run_id>.json`. HTTP retries are deterministic (2/4/8/16 s, 5 attempts, `Retry-After` honoured).
267 274
268 275 ## 8. API (FastAPI, `/api/v1`, OpenAPI at `/api/v1/openapi.json`, docs at `/api/v1/docs`)
269 276
modified docs/PIPELINE.md +29 −7
@@ -74,9 +74,21 @@ latest period ended more than `stale_after_days` ago; only that latest row) · `
74 74 or non-finite). Nothing is deleted. Quarantined rows stay in `observations` for inspection but are **excluded from every
75 75 derived table** (`latest`, `rankings`, `changes`, `events`, `similarity`, `insights`, `country_dna`).
76 76
77 −Dataset-level quarantine (`quarantine_dataset=True`): duplicates, unit ≠ registry unit, or fewer than 30 % of the rows
78 −of the previous staging file for the same spec. The new file is **not written**, the previous one is kept, and the
79 −`run.json` says `quarantined` with the reason. `ca status` lists these.
77 +Row statuses also cover `impossible_year` (year < 1750 or > current year + 10 → `quarantined`).
78 +
79 +Dataset-level quarantine (`quarantine_dataset=True`): `schema_change` (a required staging column is missing or has an
80 +incompatible dtype), duplicates, unit ≠ registry unit, or fewer than 30 % of the rows of the previous staging file for the
81 +same spec (`partial_download`). The new file is **not written**, the previous one is kept, and the `run.json` says
82 +`quarantined` with the reason. `ca status` lists these.
83 +
84 +Dataset-level **warnings** compared with the previous staging file of the same spec (`PreviousStats`, read by
85 +`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,
87 +stored in `validation_issues`, and the file is still written: suspicious values are flagged, never deleted.
88 +
89 +HTTP reliability (`connectors/base.py::Connector.get`): every request goes through one method with 5 attempts and a
90 +deterministic exponential backoff (2, 4, 8, 16 s, capped at 60 s, no jitter) on 408/425/429/5xx and transport errors
91 +(timeouts, resets); a `Retry-After` header (seconds or HTTP-date) is honoured before the backoff, capped at 120 s.
80 92
81 93 Extreme jumps: positive level series (`format` in currency/number/tonnes/kwh/per_*/km/ha with a lower bound ≥ 0) are
82 94 compared on log-differences, others on absolute differences; threshold = `jump_threshold` × 1.4826 × MAD of the
@@ -110,12 +122,22 @@ country's differences, with a floor of 10 % (relative) or 2 % of the series rang
110 122 monotone series and `cumulative`-tagged indicators are silent; severity = detector score × importance weight
111 123 (headline 1.0 / featured 0.9 / other 0.7). Q/M series are annualised first (one detection per year at most); index /
112 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) with
126 + gain ≥ 0.5 of the total sum of squares, |shift| ≥ 1.5 × sd, ≥ 5 years on each side, non-monotone series; a `change` when
127 + the break is within the last 10 years, always an `event` (one per series); `trend_reversal` — three yearly moves of one
128 + sign after three of the opposite sign, cumulative move ≥ floor (severity 0.55); `volatility_spike` — sd of the last 5
129 + 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.
113 131 7. `similarity` (5 modes, `registry/similarity.yaml`) and `country_dna` (9 percentile dimensions, `dna:` section of the
114 − same file). 8. `insights` (`registry/insights.yaml`, 16 templates, all numbers computed).
115 −9. `meta` (schema_version, build_run_id, built_at, counts, connectors, duration) → `CHECKPOINT`.
132 + 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).
134 +9. `meta` (schema_version, build_run_id, built_at, counts, `previous_observation_count`, `source_health` JSON = per
135 + connector ok / partial / failed / quarantined spec counts from `import_runs`, connectors, duration) → `CHECKPOINT`.
116 136 10. Integrity: ≥ 100 000 observations when World Bank staging exists; every headline indicator that has staging data
117 − has ≥ 100 countries in `latest`. Failure (or any exception) deletes the build file and leaves the live DB untouched.
118 −11. `os.replace` → `atlas.duckdb`, copy to `snapshots/`, prune to `CA_KEEP_SNAPSHOTS`.
137 + has ≥ 100 countries in `latest`; **the new snapshot has ≥ 90 % of the previous snapshot's observations** (a shrunken
138 + build is never published). Failure (or any exception) deletes the build file and leaves the live DB untouched.
139 +11. `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).
119 141
120 142 ## Adding a connector
121 143
modified registry/groups.yaml +12 −0
@@ -71,21 +71,25 @@ groups:
71 71 slug: asia
72 72 name: Asia
73 73 kind: continent
74 + description: Countries and territories on the Asian continent (registry `continent` field; transcontinental states follow their registry assignment).
74 75 members: CONTINENT:Asia
75 76 - id: africa
76 77 slug: africa
77 78 name: Africa
78 79 kind: continent
80 + description: Countries and territories on the African continent.
79 81 members: CONTINENT:Africa
80 82 - id: americas
81 83 slug: americas
82 84 name: Americas
83 85 kind: continent
86 + description: North, Central and South America and the Caribbean.
84 87 members: CONTINENT:Americas
85 88 - id: oceania
86 89 slug: oceania
87 90 name: Oceania
88 91 kind: continent
92 + description: Australia, New Zealand and the Pacific island countries and territories.
89 93 members: CONTINENT:Oceania
90 94
91 95 # ---- Income groups (World Bank FY2026 classification) -------------------------------------------
@@ -101,18 +105,21 @@ groups:
101 105 name: Upper middle income
102 106 kind: income
103 107 wb_code: UMC
108 + description: World Bank income classification (GNI per capita, Atlas method).
104 109 members: WB_INCOME:UMC
105 110 - id: lmc
106 111 slug: lower-middle-income
107 112 name: Lower middle income
108 113 kind: income
109 114 wb_code: LMC
115 + description: World Bank income classification (GNI per capita, Atlas method).
110 116 members: WB_INCOME:LMC
111 117 - id: lic
112 118 slug: low-income
113 119 name: Low income
114 120 kind: income
115 121 wb_code: LIC
122 + description: World Bank income classification (GNI per capita, Atlas method).
116 123 members: WB_INCOME:LIC
117 124
118 125 # ---- Organisations ------------------------------------------------------------------------------
@@ -179,6 +186,7 @@ groups:
179 186 slug: african-union
180 187 name: African Union
181 188 kind: org
189 + description: African Union — the 54 African states in the registry (all UN member states of the continent; the Sahrawi Republic is an AU member but is not a registry country).
182 190 members: CONTINENT:Africa
183 191 - id: opec
184 192 slug: opec
@@ -190,6 +198,7 @@ groups:
190 198 slug: commonwealth
191 199 name: Commonwealth of Nations
192 200 kind: org
201 + description: Commonwealth of Nations — 56 member states (Gabon and Togo joined in 2022).
193 202 members: [ATG, AUS, BHS, BGD, BRB, BLZ, BWA, BRN, CMR, CAN, CYP, DMA, SWZ, FJI, GAB, GMB, GHA, GRD, GUY, IND, JAM, KEN, KIR, LSO,
194 203 MWI, MYS, MDV, MLT, MUS, MOZ, NAM, NRU, NZL, NGA, PAK, PNG, RWA, KNA, LCA, VCT, WSM, SYC, SLE, SGP, SLB, ZAF, LKA,
195 204 TZA, TGO, TON, TTO, TUV, UGA, GBR, VUT, ZMB]
@@ -197,16 +206,19 @@ groups:
197 206 slug: mercosur
198 207 name: Mercosur
199 208 kind: org
209 + description: Southern Common Market — full members Argentina, Brazil, Paraguay, Uruguay and Bolivia (full member since 2024); Venezuela is suspended since 2016 and excluded.
200 210 members: [ARG, BRA, PRY, URY, BOL]
201 211 - id: gcc
202 212 slug: gulf-cooperation-council
203 213 name: Gulf Cooperation Council
204 214 kind: org
215 + description: Gulf Cooperation Council — Bahrain, Kuwait, Oman, Qatar, Saudi Arabia and the United Arab Emirates.
205 216 members: [BHR, KWT, OMN, QAT, SAU, ARE]
206 217 - id: small-island-developing-states
207 218 slug: small-island-developing-states
208 219 name: Small island developing states
209 220 kind: org
210 221 wb_code: SST
222 + description: UN list of Small Island Developing States present in the registry (World Bank aggregate code SST).
211 223 members: [ATG, BHS, BRB, BLZ, CPV, COM, CUB, DMA, DOM, FJI, GRD, GNB, GUY, HTI, JAM, KIR, MDV, MHL, MUS, FSM, NRU, PLW, PNG,
212 224 WSM, STP, SYC, SGP, SLB, KNA, LCA, VCT, SUR, TLS, TON, TTO, TUV, VUT]
modified src/countryatlas/api/main.py +8 −3
@@ -26,6 +26,7 @@ from countryatlas.api.errors import Problem, problem_body
26 26 from countryatlas.api.ratelimit import TokenBucketLimiter, client_ip
27 27 from countryatlas.api.routers import (
28 28 admin,
29 + analytics,
29 30 changes,
30 31 compare,
31 32 countries,
@@ -34,6 +35,7 @@ from countryatlas.api.routers import (
34 35 home,
35 36 indicators,
36 37 methodology,
38 + quality,
37 39 rankings,
38 40 regions,
39 41 search,
@@ -79,7 +81,7 @@ def create_app(db_path: str | Path | None = None, *, rate_limit_per_minute: int
79 81 app = FastAPI(
80 82 lifespan=lifespan,
81 83 title="CountryAtlas API",
82 − version="1.0.0",
84 + version="1.1.0",
83 85 summary="Country statistics with provenance on every value — www.countryatlas.co",
84 86 description=(
85 87 "Public data API of CountryAtlas. Countries (ISO3 or slug), indicators (slug), rankings, comparisons, groups, "
@@ -176,8 +178,11 @@ def create_app(db_path: str | Path | None = None, *, rate_limit_per_minute: int
176 178 return _problem(request, 500, "Internal server error", f"{type(exc).__name__}: {exc}")
177 179
178 180 # ------------------------------------------------------------------ routers
179 − for r in (health.router, countries.router, indicators.router, series.router, rankings.router, compare.router, regions.router,
180 − search.router, home.router, changes.router, sources.router, methodology.router, download.router, admin.router):
181 + # quality before countries/indicators: `/indicators/{slug}/quality` and `/countries/{id}/quality` are fixed-suffix routes that must
182 + # win over the `/{id}/{topic}`-style paths of the generic routers.
183 + for r in (health.router, quality.router, countries.router, indicators.router, series.router, rankings.router, compare.router,
184 + regions.router, search.router, home.router, changes.router, sources.router, methodology.router, download.router,
185 + analytics.router, admin.router):
181 186 app.include_router(r, prefix=API_PREFIX)
182 187
183 188 @app.get("/", include_in_schema=False)
added src/countryatlas/api/routers/analytics.py +887 −0
@@ -0,0 +1,887 @@
1 +"""Analytics endpoints (API 1.1): /pulse, /movers, /extremes, /scatter, /trajectory, /finder, /peers.
2 +
3 +All deterministic, country-only (kind='country'), canonical annual frequency, non-forecast, non-quarantined. Statistics come
4 +from countryatlas.stats; per-snapshot precomputations (the latest cross-section matrix) live in `snap._cached`. Wording is
5 +neutral: increase / decrease — "improvement" / "deterioration" only when the indicator declares `higher_is_better`.
6 +"""
7 +from __future__ import annotations
8 +
9 +import math
10 +from typing import Any
11 +
12 +import numpy as np
13 +from fastapi import APIRouter, Depends, Query
14 +
15 +from countryatlas import stats
16 +from countryatlas.api import schemas
17 +from countryatlas.api.common import (
18 + clean_float,
19 + country_card,
20 + indicator_card,
21 + merged_indicator,
22 + meta_block,
23 + parse_csv,
24 + resolve_country,
25 + resolve_group,
26 + resolve_indicator,
27 +)
28 +from countryatlas.api.db import Snapshot, get_snapshot
29 +from countryatlas.api.errors import bad_request
30 +from countryatlas.api.formatting import format_change, format_value
31 +from countryatlas.api.provenance import build_provenance
32 +from countryatlas.registry import topics as registry_topics
33 +
34 +router = APIRouter(tags=["analytics"])
35 +
36 +NEAREST_YEARS = 3
37 +LEVEL_FORMATS = {"currency", "number", "tonnes", "kwh", "per_1000", "per_100k", "per_million", "km", "ha"}
38 +CATEGORY_TOPICS: dict[str, list[str]] = {
39 + "economic": ["economy", "government", "trade", "income"],
40 + "demographic": ["population"],
41 + "health": ["health"],
42 + "energy": ["energy"],
43 + "climate": ["climate", "environment"],
44 + "digital": ["digital", "innovation"],
45 + "housing": ["housing"],
46 + "labor": ["labor"],
47 +}
48 +CATEGORIES = ["all", *CATEGORY_TOPICS]
49 +KIND_FILTERS = ["all", "improvement", "deterioration", "increase", "decrease", "record", "reversal", "acceleration", "structural"]
50 +UP_KINDS = {"yoy_jump", "record_high", "n_year_high", "accelerating", "change_5y_up", "change_10y_up"}
51 +DOWN_KINDS = {"yoy_drop", "record_low", "n_year_low", "decelerating", "change_5y_down", "change_10y_down"}
52 +KIND_GROUPS: dict[str, set[str]] = {
53 + "increase": {"yoy_jump", "record_high", "n_year_high", "accelerating"},
54 + "decrease": {"yoy_drop", "record_low", "n_year_low", "decelerating"},
55 + "record": {"record_high", "record_low"},
56 + "reversal": {"sign_flip", "trend_reversal"},
57 + "acceleration": {"accelerating", "decelerating"},
58 + "structural": {"structural_break", "volatility_spike"},
59 +}
60 +NOTE_CORRELATION = "Correlation does not imply causation. These are descriptive cross-sectional statistics."
61 +PEER_PAIRS = [
62 + {"x": "gdp-per-capita-ppp", "y": "life-expectancy", "label": "Life expectancy given income"},
63 + {"x": "gdp-per-capita-ppp", "y": "expected-years-of-schooling", "label": "Expected schooling given income"},
64 + {"x": "gdp-per-capita-ppp", "y": "co2-per-capita", "label": "CO₂ per capita given income"},
65 + {"x": "health-expenditure-per-capita", "y": "life-expectancy", "label": "Life expectancy given health spending"},
66 + {"x": "gdp-per-capita-ppp", "y": "internet-users", "label": "Internet adoption given income"},
67 + {"x": "gdp-per-capita-ppp", "y": "infant-mortality-rate", "label": "Infant mortality given income"},
68 +]
69 +
70 +
71 +# ------------------------------------------------------------------------------------------------ shared helpers
72 +def country_ids(snap: Snapshot) -> list[str]:
73 + """ISO3 of kind='country' rows, stable order (by id)."""
74 + return snap._cached("country_ids", lambda: sorted(c["id"] for c in snap.countries().values() if (c.get("kind") or "country") == "country"))
75 +
76 +
77 +def point_country(c: dict[str, Any]) -> dict[str, Any]:
78 + return {"id": c["id"], "slug": c.get("slug"), "name": c.get("short_name"), "flag": c.get("flag_emoji"), "region": c.get("region_wb"),
79 + "income": c.get("income_group")}
80 +
81 +
82 +def importance(ind: dict[str, Any], headline: set[str]) -> float:
83 + if ind["id"] in headline:
84 + return 1.0
85 + return 0.9 if ind.get("featured") else 0.7
86 +
87 +
88 +def headline_ids(snap: Snapshot) -> set[str]:
89 + return set(registry_topics()["headline"])
90 +
91 +
92 +def featured_ids(snap: Snapshot) -> list[str]:
93 + """Headline first, then featured (registry order), deduplicated, present in the snapshot."""
94 + out: list[str] = []
95 + for s in registry_topics()["headline"]:
96 + if s in snap.indicators() and s not in out:
97 + out.append(s)
98 + for i in snap.indicators().values():
99 + m = merged_indicator(i)
100 + if m.get("featured") and m["id"] not in out:
101 + out.append(m["id"])
102 + return out
103 +
104 +
105 +def is_level(ind: dict[str, Any]) -> bool:
106 + """Positive level series (GDP, population, emissions…): analysed on relative changes / log scale."""
107 + fmt = (ind.get("format") or "number").lower()
108 + lo = ind.get("bounds_min")
109 + return fmt in LEVEL_FORMATS and lo is not None and float(lo) >= 0
110 +
111 +
112 +def delta_mode(ind: dict[str, Any]) -> str:
113 + """'points' for percent-like, 'pct' for positive level series, 'abs' otherwise."""
114 + fmt = (ind.get("format") or "").lower()
115 + if fmt == "percent" or (ind.get("unit") or "").startswith("%"):
116 + return "points"
117 + if is_level(ind):
118 + return "pct"
119 + return "abs"
120 +
121 +
122 +def auto_log(ind: dict[str, Any], values: np.ndarray | None = None) -> bool:
123 + if not is_level(ind):
124 + return False
125 + if values is None:
126 + return True
127 + v = values[np.isfinite(values)]
128 + v = v[v > 0]
129 + return bool(len(v) >= 3 and v.max() / v.min() > 50)
130 +
131 +
132 +def interpretation(ind: dict[str, Any], direction: str) -> str | None:
133 + hib = ind.get("higher_is_better")
134 + if hib is None:
135 + return None
136 + return "improvement" if (direction == "up") == bool(hib) else "deterioration"
137 +
138 +
139 +class CrossSection:
140 + """Latest cross-section: countries × indicators value/year matrices (from `latest`, countries only)."""
141 +
142 + def __init__(self, snap: Snapshot) -> None:
143 + self.countries = country_ids(snap)
144 + self.idx = {c: i for i, c in enumerate(self.countries)}
145 + self.indicators = sorted(snap.indicators())
146 + self.iidx = {i: k for k, i in enumerate(self.indicators)}
147 + n, m = len(self.countries), len(self.indicators)
148 + self.values = np.full((n, m), np.nan)
149 + self.years = np.full((n, m), np.nan)
150 + rows = snap.query_rows(
151 + "SELECT l.country_id, l.indicator_id, l.value, l.year FROM latest l JOIN countries c ON c.id = l.country_id "
152 + "WHERE coalesce(c.kind, 'country') = 'country' AND l.value IS NOT NULL"
153 + )
154 + for cid, iid, v, y in rows:
155 + i, j = self.idx.get(cid), self.iidx.get(iid)
156 + if i is None or j is None:
157 + continue
158 + self.values[i, j] = float(v)
159 + self.years[i, j] = int(y)
160 + with np.errstate(invalid="ignore"):
161 + self.max_year = np.nanmax(self.years, axis=0)
162 +
163 + def col(self, indicator_id: str, within: int | None = NEAREST_YEARS) -> tuple[np.ndarray, np.ndarray]:
164 + """(values, years) for one indicator; values older than `within` years of the indicator's max year → nan."""
165 + j = self.iidx.get(indicator_id)
166 + if j is None:
167 + return np.full(len(self.countries), np.nan), np.full(len(self.countries), np.nan)
168 + v, y = self.values[:, j].copy(), self.years[:, j].copy()
169 + if within is not None and np.isfinite(self.max_year[j]):
170 + stale = y < self.max_year[j] - within
171 + v[stale] = np.nan
172 + y[stale] = np.nan
173 + return v, y
174 +
175 +
176 +def cross_section(snap: Snapshot) -> CrossSection:
177 + return snap._cached("cross_section", lambda: CrossSection(snap))
178 +
179 +
180 +def indicator_provenances(snap: Snapshot) -> dict[str, dict[str, Any]]:
181 + """indicator_id → provenance of its dominant source (most rows), cached per snapshot."""
182 +
183 + def load() -> dict[str, dict[str, Any]]:
184 + rows = snap.query(
185 + """SELECT indicator_id, source_id, source_dataset, source_series_code, max(retrieved_at) AS ra, max(source_updated_at) AS su,
186 + count(*) AS n
187 + FROM observations WHERE NOT is_forecast GROUP BY 1, 2, 3, 4 QUALIFY row_number() OVER (PARTITION BY indicator_id ORDER BY n DESC) = 1"""
188 + )
189 + return {r["indicator_id"]: build_provenance(snap, r["indicator_id"], r["source_id"], r["source_dataset"], r["source_series_code"],
190 + r["ra"], r["su"]) for r in rows}
191 +
192 + return snap._cached("indicator_provenances", load)
193 +
194 +
195 +def population_latest(snap: Snapshot) -> dict[str, float]:
196 + return snap._cached("population_latest", lambda: {r[0]: float(r[1]) for r in snap.query_rows(
197 + "SELECT country_id, value FROM latest WHERE indicator_id = 'population' AND value IS NOT NULL")})
198 +
199 +
200 +def filter_note(min_pop: int | None) -> str | None:
201 + if not min_pop:
202 + return None
203 + return f"Countries above {min_pop / 1_000_000:g}M inhabitants"
204 +
205 +
206 +def latest_common_year(snap: Snapshot, indicator_id: str, min_countries: int = 50) -> int | None:
207 + from countryatlas.api.routers.indicators import latest_common_year as _lcy
208 +
209 + y, _ = _lcy(snap, indicator_id, min_countries)
210 + return y
211 +
212 +
213 +def group_members_or_all(snap: Snapshot, group: str) -> tuple[dict[str, Any], list[str] | None]:
214 + g = resolve_group(snap, group)
215 + if g["id"] == "world":
216 + return g, None
217 + return g, snap.group_members(g["id"])
218 +
219 +
220 +def _mover_lite(snap: Snapshot, cid: str, ind: dict[str, Any], value: float | None, ref: float | None, year: int | None) -> dict[str, Any]:
221 + delta = None if value is None or ref is None else value - ref
222 + return {"country": country_card(snap.countries()[cid]), "value": value, "ref_value": ref, "delta": delta,
223 + "delta_pct": stats.yoy(value, ref), "formatted": format_value(value, ind), "year": year}
224 +
225 +
226 +def _pct_txt(v: float) -> str:
227 + return f"{v:.0f} %"
228 +
229 +
230 +# ------------------------------------------------------------------------------------------------ /pulse
231 +PULSE_MIN_COUNTRIES = 30
232 +
233 +
234 +@router.get("/pulse", response_model=schemas.PulseResponse, summary="World Pulse — what is changing globally (latest year vs previous)")
235 +def pulse(min_population: int | None = Query(1_000_000, ge=0, description="Population floor for the top movers only (counts use every country)"),
236 + snap: Snapshot = Depends(get_snapshot)) -> dict[str, Any]:
237 + ids = featured_ids(snap)
238 + pop = population_latest(snap)
239 + ph = ",".join("?" * len(ids))
240 + rows = snap.query(
241 + f"""SELECT l.indicator_id, l.country_id, l.value, l.prev_value, l.change_abs, l.change_pct, l.year
242 + FROM latest l JOIN countries c ON c.id = l.country_id
243 + WHERE coalesce(c.kind, 'country') = 'country' AND l.indicator_id IN ({ph}) AND l.value IS NOT NULL""",
244 + ids,
245 + )
246 + max_years = {r[0]: int(r[1]) for r in snap.query_rows(f"SELECT indicator_id, max(year) FROM latest WHERE indicator_id IN ({ph}) GROUP BY 1", ids)}
247 + # reference = latest year of the annual headline pillars (monthly/quarterly series may already carry the current year)
248 + year_reference = max([max_years.get(k, 0) for k in ("population", "gdp", "life-expectancy")] or [0]) or (max(max_years.values()) if max_years else 0)
249 + records = {(r[0], r[1]): int(r[2]) for r in snap.query_rows(
250 + f"""SELECT ch.indicator_id, ch.kind, count(*) FROM changes ch JOIN countries c ON c.id = ch.country_id
251 + WHERE coalesce(c.kind, 'country') = 'country' AND ch.kind IN ('record_high', 'record_low') AND ch.indicator_id IN ({ph})
252 + GROUP BY 1, 2""", ids)}
253 + n_changes = int(snap.scalar("SELECT count(*) FROM changes ch JOIN countries c ON c.id = ch.country_id WHERE coalesce(c.kind,'country') = 'country'") or 0)
254 + by_ind: dict[str, list[dict[str, Any]]] = {}
255 + for r in rows:
256 + by_ind.setdefault(r["indicator_id"], []).append(r)
257 + # convergence inputs: values 10 years before the indicator max year and at the max year
258 + conv_rows = snap.query(
259 + f"""SELECT o.indicator_id, o.country_id, o.year, o.value FROM observations o JOIN countries c ON c.id = o.country_id
260 + JOIN indicators i ON i.id = o.indicator_id
261 + WHERE o.indicator_id IN ({ph}) AND o.frequency = i.frequency AND NOT o.is_forecast AND o.value IS NOT NULL
262 + AND coalesce(o.status, 'imported') <> 'quarantined' AND coalesce(c.kind, 'country') = 'country'
263 + AND o.year IN (SELECT unnest([y, y - 10]) FROM (SELECT indicator_id AS ii, max(year) AS y FROM latest GROUP BY 1) WHERE ii = o.indicator_id)""",
264 + ids,
265 + )
266 + conv: dict[str, dict[str, dict[int, float]]] = {}
267 + for r in conv_rows:
268 + conv.setdefault(r["indicator_id"], {}).setdefault(r["country_id"], {})[int(r["year"])] = float(r["value"])
269 + provs = indicator_provenances(snap)
270 + items = []
271 + reporting: set[str] = set()
272 + for iid in ids:
273 + ind = merged_indicator(snap.indicators()[iid])
274 + my = max_years.get(iid)
275 + if my is None or my < year_reference - 1:
276 + continue
277 + recent = [r for r in by_ind.get(iid, []) if r["year"] is not None and int(r["year"]) >= my - 1 and r.get("change_abs") is not None]
278 + if len(recent) < PULSE_MIN_COUNTRIES:
279 + continue
280 + for r in recent:
281 + reporting.add(r["country_id"])
282 + big = np.array([not min_population or pop.get(r["country_id"], 0) >= min_population for r in recent])
283 + if big.sum() < 20:
284 + big = np.ones(len(recent), dtype=bool)
285 + mode = delta_mode(ind)
286 + key = "change_pct" if mode == "pct" else "change_abs"
287 + deltas = np.array([clean_float(r[key]) if clean_float(r[key]) is not None else np.nan for r in recent])
288 + absd = np.array([clean_float(r["change_abs"]) or 0.0 for r in recent])
289 + n = len(recent)
290 + n_up = int(np.sum(absd > 0))
291 + n_down = int(np.sum(absd < 0))
292 + n_flat = n - n_up - n_down
293 + share_up = round(100.0 * n_up / n, 1)
294 + share_down = round(100.0 * n_down / n, 1)
295 + hib = ind.get("higher_is_better")
296 + sem = "neutral" if hib is None else ("higher_is_better" if hib else "lower_is_better")
297 + dd = np.where(big, deltas, np.nan)
298 + order = np.argsort(np.nan_to_num(dd, nan=-np.inf))
299 + top_up = recent[int(order[-1])] if np.isfinite(dd[order[-1]]) and dd[order[-1]] > 0 else None
300 + order_lo = np.argsort(np.nan_to_num(dd, nan=np.inf))
301 + top_down = recent[int(order_lo[0])] if np.isfinite(dd[order_lo[0]]) and dd[order_lo[0]] < 0 else None
302 + rh, rl = records.get((iid, "record_high"), 0), records.get((iid, "record_low"), 0)
303 + name = ind.get("short_name") or ind.get("name")
304 + # headline template
305 + if rh >= 20 and rh >= n * 0.2 and share_up >= 50:
306 + headline = f"{name} hit a record high in {rh} countries"
307 + elif rl >= 20 and rl >= n * 0.2 and share_down >= 50:
308 + headline = f"{name} fell to a record low in {rl} countries"
309 + elif share_down >= 60:
310 + headline = f"{name} fell in {_pct_txt(share_down)} of {n} reporting countries"
311 + elif share_up >= 60:
312 + headline = f"{name} rose in {_pct_txt(share_up)} of {n} reporting countries"
313 + elif iid == "population" and n_down:
314 + headline = f"Population is shrinking in {n_down} countries"
315 + else:
316 + headline = f"{name} rose in {n_up} and fell in {n_down} of {n} reporting countries"
317 + # convergence over 10 years
318 + cv = None
319 + cd = conv.get(iid, {})
320 + s0 = [cd[c].get(my - 10) for c in cd]
321 + s1 = [cd[c].get(my) for c in cd]
322 + pairs = [(a, b) for a, b in zip(s0, s1, strict=True) if a is not None and b is not None]
323 + if len(pairs) >= 10:
324 + c = stats.convergence([p[0] for p in pairs], [p[1] for p in pairs])
325 + if c:
326 + cv = {"direction": c["direction"], "cv_start": c["cv_start"], "cv_end": c["cv_end"], "n": c["n"], "from_year": my - 10}
327 + items.append({
328 + "indicator": indicator_card(ind), "year": my, "n": n, "n_up": n_up, "n_down": n_down, "n_flat": n_flat,
329 + "share_up": share_up, "share_down": share_down,
330 + "median_change_abs": clean_float(np.nanmedian(absd)) if n else None,
331 + "median_change_pct": clean_float(np.nanmedian([clean_float(r["change_pct"]) if clean_float(r["change_pct"]) is not None else np.nan for r in recent])),
332 + "direction_semantics": sem, "record_highs": rh, "record_lows": rl, "headline": headline,
333 + "top_up": _mover_lite(snap, top_up["country_id"], ind, clean_float(top_up["value"]), clean_float(top_up["prev_value"]), top_up["year"]) if top_up else None,
334 + "top_down": _mover_lite(snap, top_down["country_id"], ind, clean_float(top_down["value"]), clean_float(top_down["prev_value"]), top_down["year"]) if top_down else None,
335 + "convergence": cv, "provenance": provs.get(iid),
336 + })
337 + return {
338 + "meta": meta_block(snap), "year_reference": year_reference, "min_population": min_population, "filter_note": filter_note(min_population),
339 + "summary": {"n_indicators": len(items), "n_countries_reporting": len(reporting),
340 + "n_record_highs": sum(i["record_highs"] for i in items), "n_record_lows": sum(i["record_lows"] for i in items),
341 + "n_changes": n_changes},
342 + "items": items,
343 + }
344 +
345 +
346 +# ------------------------------------------------------------------------------------------------ /movers
347 +def _mover_item(snap: Snapshot, r: dict[str, Any], ind: dict[str, Any], kind: str, headline: str | None, severity: float | None,
348 + year: int | None, ref_year: int | None, prov: dict[str, Any] | None) -> dict[str, Any]:
349 + value, ref = clean_float(r.get("value")), clean_float(r.get("ref_value"))
350 + delta = None if value is None or ref is None else value - ref
351 + direction = "up" if (delta or 0) >= 0 else "down"
352 + if kind in UP_KINDS:
353 + direction = "up"
354 + elif kind in DOWN_KINDS:
355 + direction = "down"
356 + return {
357 + "country": country_card(snap.countries()[r["country_id"]]), "indicator": indicator_card(ind), "kind": kind, "year": year,
358 + "ref_year": ref_year, "value": value, "ref_value": ref, "delta": delta, "delta_pct": stats.yoy(value, ref),
359 + "formatted": format_value(value, ind), "formatted_ref": format_value(ref, ind), "severity": clean_float(severity),
360 + "direction": direction, "interpretation": interpretation(ind, direction), "headline": headline, "provenance": prov,
361 + }
362 +
363 +
364 +def _kind_matches(kind: str, ind: dict[str, Any], flt: str, direction: str) -> bool:
365 + if flt == "all":
366 + return True
367 + if flt in ("improvement", "deterioration"):
368 + return interpretation(ind, direction) == flt
369 + if flt == "increase":
370 + return direction == "up" and kind not in KIND_GROUPS["structural"] | KIND_GROUPS["reversal"]
371 + if flt == "decrease":
372 + return direction == "down" and kind not in KIND_GROUPS["structural"] | KIND_GROUPS["reversal"]
373 + return kind in KIND_GROUPS.get(flt, set())
374 +
375 +
376 +@router.get("/movers", response_model=schemas.MoversResponse, summary="Biggest movers by window / category / kind")
377 +def movers(
378 + window: int = Query(1, description="1 (latest detections), 5 or 10 years"),
379 + category: str = Query("all"),
380 + kind: str = Query("all"),
381 + limit: int = Query(40, ge=1, le=300),
382 + min_population: int | None = Query(None, ge=0),
383 + snap: Snapshot = Depends(get_snapshot),
384 +) -> dict[str, Any]:
385 + if window not in (1, 5, 10):
386 + raise bad_request("window must be 1, 5 or 10.")
387 + if category not in CATEGORIES:
388 + raise bad_request(f"category must be one of {', '.join(CATEGORIES)}.")
389 + if kind not in KIND_FILTERS:
390 + raise bad_request(f"kind must be one of {', '.join(KIND_FILTERS)}.")
391 + topics = CATEGORY_TOPICS.get(category)
392 + pop = population_latest(snap)
393 + provs = indicator_provenances(snap)
394 + hl = headline_ids(snap)
395 + items: list[dict[str, Any]] = []
396 + if window == 1:
397 + where = ["coalesce(c.kind, 'country') = 'country'"]
398 + params: list[Any] = []
399 + if topics:
400 + where.append(f"i.topic IN ({','.join('?' * len(topics))})")
401 + params += topics
402 + rows = snap.query(
403 + f"""SELECT ch.country_id, ch.indicator_id, ch.kind, ch.year, ch.value, ch.ref_value, ch.severity, ch.headline, ch.window_years
404 + FROM changes ch JOIN countries c ON c.id = ch.country_id JOIN indicators i ON i.id = ch.indicator_id
405 + WHERE {' AND '.join(where)} ORDER BY ch.severity DESC NULLS LAST, ch.year DESC, ch.country_id LIMIT 4000""",
406 + params,
407 + )
408 + for r in rows:
409 + if min_population and pop.get(r["country_id"], 0) < min_population:
410 + continue
411 + ind = merged_indicator(snap.indicators()[r["indicator_id"]])
412 + direction = "up" if r["kind"] in UP_KINDS else "down" if r["kind"] in DOWN_KINDS else ("up" if (clean_float(r["value"]) or 0) >= (clean_float(r["ref_value"]) or 0) else "down")
413 + if not _kind_matches(r["kind"], ind, kind, direction):
414 + continue
415 + ref_year = (int(r["year"]) - 1) if r["kind"] in ("yoy_jump", "yoy_drop", "sign_flip") else None
416 + items.append(_mover_item(snap, r, ind, r["kind"], r["headline"], r["severity"], int(r["year"]) if r["year"] is not None else None,
417 + ref_year, provs.get(r["indicator_id"])))
418 + if len(items) >= limit:
419 + break
420 + else:
421 + if kind in ("record", "reversal", "acceleration", "structural"):
422 + return {"meta": meta_block(snap), "window": window, "category": category, "kind": kind, "min_population": min_population,
423 + "filter_note": filter_note(min_population), "categories": CATEGORIES, "kinds": KIND_FILTERS, "n": 0, "items": []}
424 + where = ["coalesce(c.kind, 'country') = 'country'", "l.value IS NOT NULL", "o.value IS NOT NULL", "l.year >= (SELECT max(year) - 2 FROM latest)"]
425 + params = [window]
426 + if topics:
427 + where.append(f"i.topic IN ({','.join('?' * len(topics))})")
428 + params += topics
429 + rows = snap.query(
430 + f"""SELECT l.indicator_id, l.country_id, l.year, l.value, o.value AS ref_value, o.year AS ref_year
431 + FROM latest l JOIN countries c ON c.id = l.country_id JOIN indicators i ON i.id = l.indicator_id
432 + JOIN observations o ON o.country_id = l.country_id AND o.indicator_id = l.indicator_id AND o.frequency = l.frequency
433 + AND o.year = l.year - ? AND NOT o.is_forecast AND coalesce(o.status, 'imported') <> 'quarantined'
434 + WHERE {' AND '.join(where)}""",
435 + params,
436 + )
437 + by_ind: dict[str, list[dict[str, Any]]] = {}
438 + for r in rows:
439 + by_ind.setdefault(r["indicator_id"], []).append(r)
440 + for iid, rs in by_ind.items():
441 + ind = merged_indicator(snap.indicators()[iid])
442 + if not ind.get("ranking_eligible", True) or len(rs) < 20:
443 + continue
444 + mode = delta_mode(ind)
445 + v = np.array([float(r["value"]) for r in rs])
446 + ref = np.array([float(r["ref_value"]) for r in rs])
447 + with np.errstate(divide="ignore", invalid="ignore"):
448 + d = np.where((v > 0) & (ref > 0), np.log(v) - np.log(ref), np.nan) if mode == "pct" else v - ref
449 + z = stats.robust_zscore(d.tolist())
450 + w = importance(ind, hl)
451 + for r, zi, di in zip(rs, z, d, strict=True):
452 + if zi is None or not np.isfinite(di) or abs(zi) < 1.5:
453 + continue
454 + if min_population and pop.get(r["country_id"], 0) < min_population:
455 + continue
456 + direction = "up" if di > 0 else "down"
457 + k = f"change_{window}y"
458 + if not _kind_matches(k, ind, kind, direction):
459 + continue
460 + verb = "rose" if direction == "up" else "fell"
461 + ch = format_change(float(r["value"]) - float(r["ref_value"]), stats.yoy(float(r["value"]), float(r["ref_value"])), ind)
462 + headline = (f"{ind.get('short_name') or ind.get('name')} {verb} {ch.replace('+', '').replace('−', '') if ch else ''} in "
463 + f"{snap.countries()[r['country_id']]['short_name']} between {int(r['ref_year'])} and {int(r['year'])} "
464 + f"(from {format_value(float(r['ref_value']), ind)} to {format_value(float(r['value']), ind)}).")
465 + items.append(_mover_item(snap, r, ind, k, headline, min(1.0, abs(zi) / 4.0) * w, int(r["year"]), int(r["ref_year"]), provs.get(iid)))
466 + items.sort(key=lambda x: -(x["severity"] or 0))
467 + items = items[:limit]
468 + return {"meta": meta_block(snap), "window": window, "category": category, "kind": kind, "min_population": min_population,
469 + "filter_note": filter_note(min_population), "categories": CATEGORIES, "kinds": KIND_FILTERS, "n": len(items), "items": items}
470 +
471 +
472 +# ------------------------------------------------------------------------------------------------ /extremes
473 +FACETS: list[tuple[str, str, str, str, str]] = [ # (id, title, indicator, direction, metric)
474 + ("aging", "Most rapidly ageing", "median-age", "up", "abs"),
475 + ("urbanizing", "Fastest urbanising", "urban-population-share", "up", "points"),
476 + ("fertility-decline", "Largest fertility declines", "fertility-rate", "down", "abs"),
477 + ("life-expectancy-gains", "Biggest gains in life expectancy", "life-expectancy", "up", "abs"),
478 + ("gdp-transformations", "Largest GDP per capita transformations", "gdp-per-capita-ppp", "up", "pct"),
479 + ("digital-adoption", "Fastest digital adoption", "internet-users", "up", "points"),
480 + ("renewable-transitions", "Largest renewable transitions", "renewable-electricity-share", "up", "points"),
481 + ("co2-reductions", "Largest CO₂ per capita reductions", "co2-per-capita", "down", "pct"),
482 + ("population-decline", "Fastest shrinking populations", "population", "down", "pct"),
483 + ("population-boom", "Fastest growing populations", "population", "up", "pct"),
484 + ("inflation-surges", "Largest inflation surges", "inflation", "up", "points"),
485 + ("debt-buildup", "Largest public debt build-ups", "general-government-gross-debt-pct-gdp", "up", "points"),
486 + ("unemployment-falls", "Largest falls in unemployment", "unemployment-rate", "down", "points"),
487 +]
488 +EXTREME_WINDOWS = {"1": 1, "5": 5, "10": 10, "25": 25, "since1990": None}
489 +
490 +
491 +@router.get("/extremes", response_model=schemas.ExtremesResponse, summary="Curated extremes facets over a window")
492 +def extremes(
493 + window: str = Query("10", description="1 | 5 | 10 | 25 | since1990"),
494 + topic: str | None = Query(None),
495 + min_population: int | None = Query(1_000_000, ge=0),
496 + rows: int = Query(10, ge=3, le=50),
497 + snap: Snapshot = Depends(get_snapshot),
498 +) -> dict[str, Any]:
499 + if window not in EXTREME_WINDOWS:
500 + raise bad_request("window must be 1, 5, 10, 25 or since1990.")
501 + n_years = EXTREME_WINDOWS[window]
502 + pop = population_latest(snap)
503 + provs = indicator_provenances(snap)
504 + facets = []
505 + from_years: list[int] = []
506 + to_years: list[int] = []
507 + for fid, title, iid, direction, metric in FACETS:
508 + ind_row = snap.indicators().get(iid)
509 + if ind_row is None:
510 + continue
511 + ind = merged_indicator(ind_row)
512 + if topic and ind.get("topic") != topic:
513 + continue
514 + to_year = latest_common_year(snap, iid)
515 + if to_year is None:
516 + continue
517 + from_year = 1990 if n_years is None else to_year - n_years
518 + if from_year >= to_year:
519 + continue
520 + rs = snap.query(
521 + """WITH o AS (
522 + SELECT o.country_id, o.year, o.value FROM observations o JOIN countries c ON c.id = o.country_id JOIN indicators i ON i.id = o.indicator_id
523 + WHERE o.indicator_id = ? AND o.frequency = i.frequency AND NOT o.is_forecast AND o.value IS NOT NULL
524 + AND coalesce(o.status, 'imported') <> 'quarantined' AND coalesce(c.kind, 'country') = 'country'),
525 + s AS (SELECT country_id, arg_min(value, abs(year - ?)) AS v0, arg_min(year, abs(year - ?)) AS y0 FROM o WHERE abs(year - ?) <= 2 GROUP BY 1),
526 + e AS (SELECT country_id, arg_max(value, year) AS v1, max(year) AS y1 FROM o WHERE year BETWEEN ? AND ? GROUP BY 1)
527 + SELECT s.country_id, s.v0, s.y0, e.v1, e.y1 FROM s JOIN e ON e.country_id = s.country_id WHERE e.y1 > s.y0""",
528 + [iid, from_year, from_year, from_year, to_year - 2, to_year],
529 + )
530 + cand = []
531 + for r in rs:
532 + if min_population and pop.get(r["country_id"], 0) < min_population:
533 + continue
534 + v0, v1 = float(r["v0"]), float(r["v1"])
535 + delta = v1 - v0
536 + pct = stats.yoy(v1, v0)
537 + score = pct if metric == "pct" else delta
538 + if score is None or not math.isfinite(score):
539 + continue
540 + if (direction == "up" and score <= 0) or (direction == "down" and score >= 0):
541 + continue
542 + cand.append((abs(score), {
543 + "country": country_card(snap.countries()[r["country_id"]]), "value_start": v0, "value_end": v1,
544 + "year_start": int(r["y0"]), "year_end": int(r["y1"]), "delta": delta, "delta_pct": pct,
545 + "formatted_start": format_value(v0, ind), "formatted_end": format_value(v1, ind)}))
546 + if len(rs) < 30:
547 + continue
548 + cand.sort(key=lambda x: -x[0])
549 + from_years.append(from_year)
550 + to_years.append(to_year)
551 + facets.append({"id": fid, "title": title, "indicator": indicator_card(ind), "direction": direction, "metric": metric,
552 + "rows": [c[1] for c in cand[:rows]], "n": len(cand), "from_year": from_year, "to_year": to_year, "provenance": provs.get(iid)})
553 + return {"meta": meta_block(snap), "window": window, "from_year": min(from_years) if from_years else None,
554 + "to_year": max(to_years) if to_years else None, "min_population": min_population, "filter_note": filter_note(min_population),
555 + "facets": facets}
556 +
557 +
558 +# ------------------------------------------------------------------------------------------------ /scatter
559 +def _nearest_values(snap: Snapshot, indicator_id: str, year: int, members: list[str] | None, within: int = NEAREST_YEARS) -> dict[str, tuple[float, int]]:
560 + """country → (value, year) using the value nearest to `year` within ±within years (later year wins ties)."""
561 + where = ["o.indicator_id = ?", "o.frequency = i.frequency", "NOT o.is_forecast", "o.value IS NOT NULL",
562 + "coalesce(o.status, 'imported') <> 'quarantined'", "coalesce(c.kind, 'country') = 'country'", "abs(o.year - ?) <= ?"]
563 + params: list[Any] = [indicator_id, year, within]
564 + if members is not None:
565 + if not members:
566 + return {}
567 + where.append(f"o.country_id IN ({','.join('?' * len(members))})")
568 + params += members
569 + rows = snap.query_rows(
570 + f"""SELECT o.country_id, arg_min(o.value, abs(o.year - ?) * 2 - (o.year > ?)::INT) AS v, arg_min(o.year, abs(o.year - ?) * 2 - (o.year > ?)::INT) AS y
571 + FROM observations o JOIN countries c ON c.id = o.country_id JOIN indicators i ON i.id = o.indicator_id
572 + WHERE {' AND '.join(where)} GROUP BY 1""",
573 + [year, year, year, year, *params],
574 + )
575 + return {r[0]: (float(r[1]), int(r[2])) for r in rows}
576 +
577 +
578 +def _parse_log(v: str | None, ind: dict[str, Any], values: np.ndarray | None) -> bool:
579 + if v is None or v.lower() == "auto":
580 + return auto_log(ind, values)
581 + return v.lower() in ("1", "true", "yes")
582 +
583 +
584 +def _size_indicator(snap: Snapshot, size: str | None) -> dict[str, Any] | None:
585 + if size is None:
586 + size = "population"
587 + if size.lower() in ("none", "", "0"):
588 + return None
589 + return resolve_indicator(snap, size)
590 +
591 +
592 +@router.get("/scatter", response_model=schemas.ScatterResponse, summary="Cross-section scatter with Pearson / Spearman / OLS")
593 +def scatter(
594 + x: str = Query(...), y: str = Query(...), size: str | None = Query("population"), year: int | None = Query(None, ge=1800, le=2100),
595 + group: str = Query("world"), log_x: str | None = Query("auto"), log_y: str | None = Query("auto"),
596 + snap: Snapshot = Depends(get_snapshot),
597 +) -> dict[str, Any]:
598 + xi, yi = resolve_indicator(snap, x), resolve_indicator(snap, y)
599 + if xi["id"] == yi["id"]:
600 + raise bad_request("x and y must differ.")
601 + si = _size_indicator(snap, size)
602 + g, members = group_members_or_all(snap, group)
603 + if year is None:
604 + yx, yy = latest_common_year(snap, xi["id"]), latest_common_year(snap, yi["id"])
605 + year_used = min([v for v in (yx, yy) if v is not None], default=None)
606 + else:
607 + year_used = year
608 + points: list[dict[str, Any]] = []
609 + xs: list[float] = []
610 + ys: list[float] = []
611 + if year_used is not None:
612 + vx = _nearest_values(snap, xi["id"], year_used, members)
613 + vy = _nearest_values(snap, yi["id"], year_used, members)
614 + vs = _nearest_values(snap, si["id"], year_used, members) if si else {}
615 + for cid in sorted(set(vx) & set(vy)):
616 + c = snap.countries().get(cid)
617 + if c is None:
618 + continue
619 + p = point_country(c)
620 + p.update({"x": vx[cid][0], "y": vy[cid][0], "size": vs.get(cid, (None, None))[0], "year_x": vx[cid][1], "year_y": vy[cid][1]})
621 + points.append(p)
622 + xs.append(vx[cid][0])
623 + ys.append(vy[cid][0])
624 + xa, ya = np.array(xs, dtype=float), np.array(ys, dtype=float)
625 + lx, ly = _parse_log(log_x, xi, xa), _parse_log(log_y, yi, ya)
626 + with np.errstate(divide="ignore", invalid="ignore"):
627 + tx = np.where(xa > 0, np.log10(xa), np.nan) if lx else xa
628 + ty = np.where(ya > 0, np.log10(ya), np.nan) if ly else ya
629 + r, n = stats.pearson(tx.tolist(), ty.tolist())
630 + rho, _ = stats.spearman(xa.tolist(), ya.tolist())
631 + fit = stats.ols(tx.tolist(), ty.tolist())
632 + provs = indicator_provenances(snap)
633 + return {
634 + "meta": meta_block(snap), "x": indicator_card(xi), "y": indicator_card(yi), "size": indicator_card(si) if si else None,
635 + "year": year, "year_used": year_used, "nearest_years": NEAREST_YEARS,
636 + "group": {"id": g["id"], "slug": g.get("slug"), "name": g.get("name"), "kind": g.get("kind"), "wb_code": g.get("wb_code"), "n_members": g.get("n_members")},
637 + "n": len(points), "points": points,
638 + "stats": {"n": n, "pearson": r, "spearman": rho, "ols": {"slope": fit["slope"], "intercept": fit["intercept"], "r2": fit["r2"]} if fit else None,
639 + "log_x": lx, "log_y": ly},
640 + "note": NOTE_CORRELATION,
641 + "provenance": [p for p in (provs.get(xi["id"]), provs.get(yi["id"]), provs.get(si["id"]) if si else None) if p],
642 + }
643 +
644 +
645 +# ------------------------------------------------------------------------------------------------ /trajectory
646 +@router.get("/trajectory", response_model=schemas.TrajectoryResponse, summary="Gapminder-style frames for two indicators")
647 +def trajectory(
648 + x: str = Query("gdp-per-capita-ppp"), y: str = Query("life-expectancy"), size: str | None = Query("population"),
649 + from_: int | None = Query(None, alias="from", ge=1800, le=2100), to: int | None = Query(None, ge=1800, le=2100),
650 + group: str = Query("world"), snap: Snapshot = Depends(get_snapshot),
651 +) -> dict[str, Any]:
652 + xi, yi = resolve_indicator(snap, x), resolve_indicator(snap, y)
653 + if xi["id"] == yi["id"]:
654 + raise bad_request("x and y must differ.")
655 + si = _size_indicator(snap, size)
656 + g, members = group_members_or_all(snap, group)
657 + ids = [xi["id"], yi["id"]] + ([si["id"]] if si else [])
658 + where = [f"o.indicator_id IN ({','.join('?' * len(ids))})", "o.frequency = 'A'", "NOT o.is_forecast", "o.value IS NOT NULL",
659 + "coalesce(o.status, 'imported') <> 'quarantined'", "coalesce(c.kind, 'country') = 'country'"]
660 + params: list[Any] = list(ids)
661 + if members is not None:
662 + if not members:
663 + where.append("FALSE")
664 + else:
665 + where.append(f"o.country_id IN ({','.join('?' * len(members))})")
666 + params += members
667 + rows = snap.query_rows(f"SELECT o.indicator_id, o.country_id, o.year, o.value FROM observations o JOIN countries c ON c.id = o.country_id WHERE {' AND '.join(where)}", params)
668 + data: dict[str, dict[str, dict[int, float]]] = {i: {} for i in ids}
669 + for iid, cid, yr, v in rows:
670 + data[iid].setdefault(cid, {})[int(yr)] = float(v)
671 + # years with ≥ 50 countries on both axes
672 + count_by_year: dict[int, int] = {}
673 + for cid in set(data[xi["id"]]) & set(data[yi["id"]]):
674 + for yr in set(data[xi["id"]][cid]) & set(data[yi["id"]][cid]):
675 + count_by_year[yr] = count_by_year.get(yr, 0) + 1
676 + good = sorted(y_ for y_, n in count_by_year.items() if n >= 50)
677 + if not good:
678 + good = sorted(count_by_year)
679 + y0 = from_ if from_ is not None else (good[0] if good else None)
680 + y1 = to if to is not None else (good[-1] if good else None)
681 + years = list(range(y0, y1 + 1)) if y0 is not None and y1 is not None and y0 <= y1 else []
682 + countries_out: list[dict[str, Any]] = []
683 + series: dict[str, Any] = {}
684 + dom = {k: [math.inf, -math.inf] for k in ("x", "y", "size")}
685 + for cid in sorted(set(data[xi["id"]]) | set(data[yi["id"]])):
686 + c = snap.countries().get(cid)
687 + if c is None or not years:
688 + continue
689 + sx = [data[xi["id"]].get(cid, {}).get(yr) for yr in years]
690 + sy = [data[yi["id"]].get(cid, {}).get(yr) for yr in years]
691 + both = sum(1 for a, b in zip(sx, sy, strict=True) if a is not None and b is not None)
692 + if both < 0.3 * len(years) or both == 0:
693 + continue
694 + ss = [data[si["id"]].get(cid, {}).get(yr) for yr in years] if si else [None] * len(years)
695 + for k, arr in (("x", sx), ("y", sy), ("size", ss)):
696 + for v in arr:
697 + if v is not None:
698 + dom[k][0] = min(dom[k][0], v)
699 + dom[k][1] = max(dom[k][1], v)
700 + countries_out.append(point_country(c))
701 + series[cid] = {"x": sx, "y": sy, "size": ss}
702 + xs_all = np.array([v for s in series.values() for v in s["x"] if v is not None] or [np.nan])
703 + ys_all = np.array([v for s in series.values() for v in s["y"] if v is not None] or [np.nan])
704 + provs = indicator_provenances(snap)
705 +
706 + def dom_or_none(k: str) -> list[float] | None:
707 + return None if dom[k][0] == math.inf else dom[k]
708 +
709 + return {
710 + "meta": meta_block(snap), "x": indicator_card(xi), "y": indicator_card(yi), "size": indicator_card(si) if si else None,
711 + "group": {"id": g["id"], "slug": g.get("slug"), "name": g.get("name"), "kind": g.get("kind"), "wb_code": g.get("wb_code"), "n_members": g.get("n_members")},
712 + "years": years, "countries": countries_out, "series": series,
713 + "domains": {"x": dom_or_none("x") or [0, 1], "y": dom_or_none("y") or [0, 1], "size": dom_or_none("size")},
714 + "log_x": auto_log(xi, xs_all), "log_y": auto_log(yi, ys_all),
715 + "provenance": [p for p in (provs.get(i) for i in ids) if p],
716 + }
717 +
718 +
719 +# ------------------------------------------------------------------------------------------------ /finder
720 +FINDER_OPS = ("gt", "gte", "lt", "lte", "eq", "between")
721 +
722 +
723 +def _parse_filter(snap: Snapshot, raw: str) -> dict[str, Any]:
724 + parts = raw.split(":")
725 + if len(parts) != 3:
726 + raise bad_request(f"Filter '{raw}' must look like slug:op:value (ops: {', '.join(FINDER_OPS)}).")
727 + slug, op, val = parts[0].strip().lower(), parts[1].strip().lower(), parts[2].strip()
728 + if op not in FINDER_OPS:
729 + raise bad_request(f"Unknown operator '{op}' in '{raw}'.")
730 + ind = resolve_indicator(snap, slug)
731 + try:
732 + if op == "between":
733 + a, b = val.split("..")
734 + v1, v2 = float(a), float(b)
735 + if v1 > v2:
736 + v1, v2 = v2, v1
737 + else:
738 + v1, v2 = float(val), None
739 + except ValueError as e:
740 + raise bad_request(f"Bad value in filter '{raw}'.") from e
741 + return {"indicator": ind, "op": op, "value": v1, "value2": v2}
742 +
743 +
744 +def _apply(op: str, v: np.ndarray, a: float, b: float | None) -> np.ndarray:
745 + with np.errstate(invalid="ignore"):
746 + if op == "gt":
747 + return v > a
748 + if op == "gte":
749 + return v >= a
750 + if op == "lt":
751 + return v < a
752 + if op == "lte":
753 + return v <= a
754 + if op == "eq":
755 + return np.isclose(v, a)
756 + return (v >= a) & (v <= (b if b is not None else a))
757 +
758 +
759 +@router.get("/finder", response_model=schemas.FinderResponse, summary="Structured country finder over latest values")
760 +def finder(
761 + f: list[str] | None = Query(None, description="Repeatable filter slug:op:value (ops gt gte lt lte eq between a..b)"),
762 + mode: str = Query("and", pattern="^(and|or)$"),
763 + region: str | None = Query(None), income: str | None = Query(None), sort: str | None = Query(None),
764 + limit: int = Query(218, ge=1, le=300), snap: Snapshot = Depends(get_snapshot),
765 +) -> dict[str, Any]:
766 + filters = [_parse_filter(snap, raw) for raw in (f or []) if raw]
767 + if not filters:
768 + raise bad_request("Provide at least one filter, e.g. f=gdp-per-capita:gt:40000.")
769 + cs = cross_section(snap)
770 + n = len(cs.countries)
771 + pool = np.ones(n, dtype=bool)
772 + for gslug in (region, income):
773 + if gslug:
774 + g = resolve_group(snap, gslug)
775 + if g["id"] != "world":
776 + mem = set(snap.group_members(g["id"]))
777 + pool &= np.array([c in mem for c in cs.countries])
778 + matched_matrix = np.zeros((n, len(filters)), dtype=bool)
779 + evaluated = np.zeros(n, dtype=bool)
780 + for k, flt in enumerate(filters):
781 + v, _y = cs.col(flt["indicator"]["id"], within=None)
782 + evaluated |= np.isfinite(v)
783 + matched_matrix[:, k] = _apply(flt["op"], v, flt["value"], flt["value2"]) & np.isfinite(v)
784 + ok = matched_matrix.all(axis=1) if mode == "and" else matched_matrix.any(axis=1)
785 + ok &= pool
786 + # sort
787 + sort_slug, sort_dir = filters[0]["indicator"]["id"], "desc"
788 + if sort:
789 + parts = sort.split(":")
790 + sort_slug = resolve_indicator(snap, parts[0])["id"]
791 + if len(parts) > 1 and parts[1].lower() in ("asc", "desc"):
792 + sort_dir = parts[1].lower()
793 + sv, _ = cs.col(sort_slug, within=None)
794 + idx = [i for i in range(n) if ok[i]]
795 + idx.sort(key=lambda i: ((0, -sv[i] if sort_dir == "desc" else sv[i]) if np.isfinite(sv[i]) else (1, 0)))
796 + slugs = list(dict.fromkeys([flt["indicator"]["id"] for flt in filters] + [sort_slug]))
797 + provs = indicator_provenances(snap)
798 + items = []
799 + for i in idx[:limit]:
800 + cid = cs.countries[i]
801 + vals: dict[str, Any] = {}
802 + for s in slugs:
803 + v, y = cs.col(s, within=None)
804 + ind = merged_indicator(snap.indicators()[s])
805 + vals[s] = {"value": clean_float(v[i]), "year": int(y[i]) if np.isfinite(y[i]) else None,
806 + "formatted": format_value(clean_float(v[i]), ind), "provenance": provs.get(s)}
807 + items.append({"country": country_card(snap.countries()[cid]), "matched": [flt["indicator"]["id"] for k, flt in enumerate(filters) if matched_matrix[i, k]],
808 + "values": vals})
809 + return {
810 + "meta": meta_block(snap), "mode": mode,
811 + "filters": [{"indicator": indicator_card(flt["indicator"]), "op": flt["op"], "value": flt["value"], "value2": flt["value2"],
812 + "year_used": int(cs.max_year[cs.iidx[flt["indicator"]["id"]]]) if flt["indicator"]["id"] in cs.iidx and np.isfinite(cs.max_year[cs.iidx[flt["indicator"]["id"]]]) else None}
813 + for flt in filters],
814 + "n_matching": int(ok.sum()), "n_evaluated": int((evaluated & pool).sum()), "items": items,
815 + }
816 +
817 +
818 +# ------------------------------------------------------------------------------------------------ /peers
819 +PEERS_METHODOLOGY = (
820 + "A robust line (Theil–Sen: median of pairwise slopes; or ordinary least squares) is fitted across countries for the chosen year, "
821 + "y on x (x on a log10 scale when the indicator spans orders of magnitude). Each country's residual is its value minus the fitted "
822 + "value at its x; residual_z divides the residual by 1.4826 × MAD of all residuals. 'Above' and 'below' describe position relative "
823 + "to the fitted line for that year only — a descriptive cross-sectional deviation, not a causal or normative judgement."
824 +)
825 +
826 +
827 +@router.get("/peers", response_model=schemas.PeersResponse, summary="Above / below expected given another indicator (robust fit)")
828 +def peers(
829 + y: str = Query("life-expectancy"), x: str = Query("gdp-per-capita-ppp"), year: int | None = Query(None, ge=1800, le=2100),
830 + method: str = Query("theil-sen", pattern="^(theil-sen|ols)$"), log_x: str | None = Query("auto"), top: int = Query(12, ge=1, le=50),
831 + snap: Snapshot = Depends(get_snapshot),
832 +) -> dict[str, Any]:
833 + xi, yi = resolve_indicator(snap, x), resolve_indicator(snap, y)
834 + if xi["id"] == yi["id"]:
835 + raise bad_request("x and y must differ.")
836 + if year is None:
837 + yx, yy = latest_common_year(snap, xi["id"]), latest_common_year(snap, yi["id"])
838 + year_used = min([v for v in (yx, yy) if v is not None], default=None)
839 + else:
840 + year_used = year
841 + pts: list[dict[str, Any]] = []
842 + if year_used is not None:
843 + vx = _nearest_values(snap, xi["id"], year_used, None)
844 + vy = _nearest_values(snap, yi["id"], year_used, None)
845 + for cid in sorted(set(vx) & set(vy)):
846 + c = snap.countries().get(cid)
847 + if c is None:
848 + continue
849 + p = point_country(c)
850 + p.update({"x": vx[cid][0], "y": vy[cid][0]})
851 + pts.append(p)
852 + xa = np.array([p["x"] for p in pts], dtype=float)
853 + ya = np.array([p["y"] for p in pts], dtype=float)
854 + lx = _parse_log(log_x, xi, xa)
855 + with np.errstate(divide="ignore", invalid="ignore"):
856 + tx = np.where(xa > 0, np.log10(xa), np.nan) if lx else xa
857 + fit = stats.theil_sen(tx.tolist(), ya.tolist()) if method == "theil-sen" else stats.ols(tx.tolist(), ya.tolist())
858 + if fit:
859 + expected = fit["intercept"] + fit["slope"] * tx
860 + resid = ya - expected
861 + scale = fit.get("residual_mad") if method == "theil-sen" else None
862 + if not scale:
863 + fr = resid[np.isfinite(resid)]
864 + scale = float(np.median(np.abs(fr - np.median(fr))) * stats.MAD_SCALE) if len(fr) else None
865 + for p, e, r in zip(pts, expected, resid, strict=True):
866 + ok = bool(np.isfinite(e))
867 + p["expected"] = float(e) if ok else None
868 + p["residual"] = float(r) if ok else None
869 + p["residual_z"] = float(r / scale) if ok and scale else None
870 + else:
871 + scale = None
872 + for p in pts:
873 + p.update({"expected": None, "residual": None, "residual_z": None})
874 + ranked = sorted([p for p in pts if p["residual_z"] is not None], key=lambda p: p["residual_z"])
875 + return {
876 + "meta": meta_block(snap), "x": indicator_card(xi), "y": indicator_card(yi), "year_used": year_used, "n": len(pts), "method": method,
877 + "fit": {"slope": fit["slope"], "intercept": fit["intercept"], "r2": fit["r2"], "log_x": lx, "residual_scale": scale} if fit else None,
878 + "points": pts, "above": list(reversed(ranked[-top:])) if ranked else [], "below": ranked[:top],
879 + "pairs": [p for p in PEER_PAIRS if p["x"] in snap.indicators() and p["y"] in snap.indicators()],
880 + "note": "Countries above the fitted line have a higher value than the cross-country relationship would suggest for their x; "
881 + "below, a lower one. " + NOTE_CORRELATION,
882 + "methodology": PEERS_METHODOLOGY,
883 + }
884 +
885 +
886 +__all__ = ["CrossSection", "auto_log", "country_ids", "cross_section", "delta_mode", "indicator_provenances", "is_level", "latest_common_year",
887 + "parse_csv", "point_country", "population_latest", "resolve_country", "router"]
modified src/countryatlas/api/routers/countries.py +133 −6
@@ -416,19 +416,146 @@ DNA_LABELS = {
416 416 }
417 417
418 418
419 −@router.get("/{id}/dna", response_model=schemas.DNAResponse, summary="Country DNA (9 percentile dimensions)")
420 −def get_country_dna(id: str, snap: Snapshot = Depends(get_snapshot)) -> dict[str, Any]:
419 +def _dna_dims(snap: Snapshot, country_id: str) -> tuple[dict[str, float | None], int | None]:
421 420 import json
422 421
423 − c = resolve_country(snap, id)
424 − row = snap.one("SELECT * FROM country_dna WHERE country_id = ?", [c["id"]])
422 + row = snap.one("SELECT * FROM country_dna WHERE country_id = ?", [country_id])
425 423 dims: dict[str, float | None] = {}
426 424 if row and row.get("dims") is not None:
427 425 raw = row["dims"]
428 426 if isinstance(raw, str):
429 427 raw = json.loads(raw)
430 428 dims = {k: clean_float(v) for k, v in dict(raw).items()}
429 + return dims, (row.get("year_ref") if row else None)
430 +
431 +
432 +def _dna_reference(snap: Snapshot, c: dict[str, Any], reference: str | None, dims: dict[str, float | None]) -> dict[str, Any] | None:
433 + """world → 50 everywhere; region/income → median of the members' dims; a country → that country's dims."""
434 + import json
435 +
436 + import numpy as np
437 +
438 + if not reference:
439 + return None
440 + ref = reference.strip().lower()
441 + if ref == "world":
442 + return {"kind": "world", "id": "world", "label": "World median", "dims": {k: 50.0 for k in dims}}
443 + if ref in ("region", "income"):
444 + g = None
445 + for grp in snap.groups().values():
446 + if grp.get("kind") == ref and (grp.get("wb_code") or "").upper() == ((c.get("region_wb") if ref == "region" else c.get("income_group")) or "").upper():
447 + g = grp
448 + break
449 + if g is None:
450 + return None
451 + members = [m for m in snap.group_members(g["id"]) if m != c["id"]]
452 + if not members:
453 + return None
454 + rows = snap.query(f"SELECT dims FROM country_dna WHERE country_id IN ({','.join('?' * len(members))})", members)
455 + acc: dict[str, list[float]] = {}
456 + for r in rows:
457 + raw = r["dims"]
458 + d = json.loads(raw) if isinstance(raw, str) else dict(raw)
459 + for k, v in d.items():
460 + if v is not None:
461 + acc.setdefault(k, []).append(float(v))
462 + return {"kind": ref, "id": g["id"], "label": f"{g.get('name')} median",
463 + "dims": {k: (float(np.median(acc[k])) if acc.get(k) else None) for k in dims}}
464 + other = resolve_country(snap, ref)
465 + od, _ = _dna_dims(snap, other["id"])
466 + return {"kind": "country", "id": other["id"], "label": other.get("short_name"), "dims": {k: od.get(k) for k in dims}}
467 +
468 +
469 +@router.get("/{id}/dna", response_model=schemas.DNAResponse, summary="Country DNA (9 percentile dimensions) with optional reference profile")
470 +def get_country_dna(id: str, reference: str | None = Query(None, description="world | region | income | <ISO3 or slug>"),
471 + snap: Snapshot = Depends(get_snapshot)) -> dict[str, Any]:
472 + c = resolve_country(snap, id)
473 + dims, year_ref = _dna_dims(snap, c["id"])
431 474 dimensions = [{"id": k, "label": DNA_LABELS.get(k, (k.replace("_", " ").title(), None))[0],
432 475 "indicator": DNA_LABELS.get(k, (None, None))[1], "value": v} for k, v in dims.items()]
433 − return {"meta": meta_block(snap), "country": country_card(c), "dims": dims, "year_ref": row.get("year_ref") if row else None,
434 − "dimensions": dimensions}
476 + return {"meta": meta_block(snap), "country": country_card(c), "dims": dims, "year_ref": year_ref, "dimensions": dimensions,
477 + "reference": _dna_reference(snap, c, reference, dims), "note": "Percentile profile among all countries, not an overall country score."}
478 +
479 +
480 +# ------------------------------------------------------------------------------------------------ story (API 1.1)
481 +STORY_INDICATORS = ["population", "gdp-per-capita-ppp", "life-expectancy", "fertility-rate", "urban-population-share", "co2-per-capita",
482 + "internet-users", "renewable-electricity-share", "general-government-gross-debt-pct-gdp", "median-age",
483 + "energy-use-per-capita", "unemployment-rate"]
484 +STORY_FALLBACK = {"gdp-per-capita-ppp": "gdp-per-capita"}
485 +STORY_MIN_POINTS = 10
486 +STORY_MAX_ITEMS = 10
487 +
488 +
489 +@router.get("/{id}/story", response_model=schemas.StoryResponse, summary="How a country changed: long-run indicators, templated text")
490 +def get_country_story(id: str, snap: Snapshot = Depends(get_snapshot)) -> dict[str, Any]:
491 + from countryatlas import stats
492 + from countryatlas.api.formatting import format_change
493 + from countryatlas.api.routers.analytics import delta_mode
494 +
495 + c = resolve_country(snap, id)
496 + wanted: list[str] = []
497 + for s in STORY_INDICATORS:
498 + if s in snap.indicators():
499 + wanted.append(s)
500 + elif STORY_FALLBACK.get(s) in snap.indicators():
501 + wanted.append(STORY_FALLBACK[s])
502 + ph = ",".join("?" * len(wanted))
503 + rows = snap.query(
504 + f"""SELECT o.indicator_id, o.year, o.value, o.source_id, o.source_dataset, o.source_series_code, o.retrieved_at, o.source_updated_at
505 + FROM observations o JOIN indicators i ON i.id = o.indicator_id
506 + WHERE o.country_id = ? AND o.indicator_id IN ({ph}) AND o.frequency = 'A' AND NOT o.is_forecast AND o.value IS NOT NULL
507 + AND coalesce(o.status, 'imported') <> 'quarantined' ORDER BY o.indicator_id, o.year""",
508 + [c["id"], *wanted],
509 + )
510 + by_ind: dict[str, list[dict[str, Any]]] = {}
511 + for r in rows:
512 + by_ind.setdefault(r["indicator_id"], []).append(r)
513 + ranks = {(r[0], int(r[1])): (int(r[2]), int(r[3])) for r in snap.query_rows(
514 + f"SELECT indicator_id, year, rank, n FROM rankings WHERE country_id = ? AND indicator_id IN ({ph})", [c["id"], *wanted])}
515 + items = []
516 + since: int | None = None
517 + for iid in wanted:
518 + rs = by_ind.get(iid, [])
519 + if len(rs) < STORY_MIN_POINTS:
520 + continue
521 + ind = merged_indicator(snap.indicators()[iid])
522 + vals = [float(r["value"]) for r in rs]
523 + years = [int(r["year"]) for r in rs]
524 + # start from the first year with data (≥ 1960 preferred when the series is longer)
525 + start = next((i for i, y in enumerate(years) if y >= 1960), 0)
526 + vals, years, rs = vals[start:], years[start:], rs[start:]
527 + if len(vals) < STORY_MIN_POINTS:
528 + continue
529 + first, last = vals[0], vals[-1]
530 + y0, y1 = years[0], years[-1]
531 + since = y0 if since is None else min(since, y0)
532 + delta = last - first
533 + pct = stats.yoy(last, first)
534 + cg = stats.cagr(first, last, y1 - y0) if delta_mode(ind) == "pct" else None
535 + peak_i, trough_i = max(range(len(vals)), key=lambda i: vals[i]), min(range(len(vals)), key=lambda i: vals[i])
536 + mode = delta_mode(ind)
537 + name = ind.get("short_name") or ind.get("name")
538 + verb = "rose" if delta > 0 else "fell" if delta < 0 else "was unchanged"
539 + ch = format_change(delta, pct, ind)
540 + if mode == "points":
541 + tail = f"({ch})" if ch else ""
542 + else:
543 + cg_txt = f", {'+' if cg >= 0 else '−'}{abs(cg):.1f} % a year" if cg is not None else ""
544 + tail = f"({ch}{cg_txt})" if ch else ""
545 + text = f"{c['short_name']}: {name or iid} {verb} from {format_value(first, ind)} in {y0} to {format_value(last, ind)} in {y1} {tail}".strip() + "."
546 + prov = build_provenance(snap, iid, rs[-1].get("source_id"), rs[-1].get("source_dataset"), rs[-1].get("source_series_code"),
547 + rs[-1].get("retrieved_at"), rs[-1].get("source_updated_at"), c.get("iso2"))
548 + rf, rl = ranks.get((iid, y0)), ranks.get((iid, y1))
549 + items.append({
550 + "indicator": indicator_card(ind),
551 + "first": {"year": y0, "value": first, "formatted": format_value(first, ind)},
552 + "last": {"year": y1, "value": last, "formatted": format_value(last, ind)},
553 + "change_abs": delta, "change_pct": pct, "cagr": cg,
554 + "peak": {"year": years[peak_i], "value": vals[peak_i]}, "trough": {"year": years[trough_i], "value": vals[trough_i]},
555 + "rank_first": {"rank": rf[0], "n": rf[1], "year": y0} if rf else None,
556 + "rank_last": {"rank": rl[0], "n": rl[1], "year": y1} if rl else None,
557 + "series": [[y, v] for y, v in zip(years, vals, strict=True)], "text": text, "provenance": prov,
558 + })
559 + if len(items) >= STORY_MAX_ITEMS:
560 + break
561 + return {"meta": meta_block(snap), "country": country_card(c), "since": since, "items": items}
modified src/countryatlas/api/routers/health.py +1 −1
@@ -11,7 +11,7 @@ from countryatlas.api.db import get_database
11 11
12 12 router = APIRouter(tags=["health"])
13 13
14 −API_VERSION = "0.1.0"
14 +API_VERSION = "1.1.0"
15 15
16 16
17 17 @router.get("/health", response_model=schemas.HealthResponse, summary="Health / snapshot status")
modified src/countryatlas/api/routers/indicators.py +157 −0
@@ -3,6 +3,7 @@ from __future__ import annotations
3 3
4 4 from typing import Any
5 5
6 +import numpy as np
6 7 from fastapi import APIRouter, Depends, Query
7 8
8 9 from countryatlas.api import schemas
@@ -14,6 +15,7 @@ from countryatlas.api.common import (
14 15 merged_indicator,
15 16 meta_block,
16 17 quantile_breaks,
18 + resolve_country,
17 19 resolve_group,
18 20 resolve_indicator,
19 21 )
@@ -339,3 +341,158 @@ def get_indicator_trend(
339 341 r["retrieved_at"], r["source_updated_at"]), n_values=r["n"]) for r in prov_rows],
340 342 "note": "Aggregates are computed across member countries present in the snapshot (no World Bank aggregate series is stored).",
341 343 }
344 +
345 +
346 +# ------------------------------------------------------------------------------------------------ analytics (API 1.1)
347 +from countryatlas import stats as _stats
348 +from countryatlas.api.errors import bad_request as _bad_request
349 +from countryatlas.api.routers import analytics as _an
350 +
351 +NOTE_CORRELATION = "Correlation does not imply causation. Descriptive cross-sectional statistics on latest values."
352 +
353 +
354 +def _log_matrix(snap: Snapshot):
355 + """Per-snapshot: cross-section values transformed for association (log10 for level series), aligned with CrossSection."""
356 +
357 + def load():
358 + cs = _an.cross_section(snap)
359 + m = np.full_like(cs.values, np.nan)
360 + logs: dict[str, bool] = {}
361 + for iid, j in cs.iidx.items():
362 + ind = merged_indicator(snap.indicators()[iid])
363 + v, _ = cs.col(iid)
364 + use_log = _an.auto_log(ind, v)
365 + logs[iid] = use_log
366 + with np.errstate(divide="ignore", invalid="ignore"):
367 + m[:, j] = np.where(v > 0, np.log10(v), np.nan) if use_log else v
368 + return m, logs
369 +
370 + return snap._cached("assoc_matrix", load)
371 +
372 +
373 +@router.get("/{slug}/related", response_model=schemas.RelatedResponse, summary="Statistically related indicators (descriptive)")
374 +def get_indicator_related(slug: str, limit: int = Query(12, ge=1, le=60), min_n: int = Query(40, ge=5),
375 + snap: Snapshot = Depends(get_snapshot)) -> dict[str, Any]:
376 + ind = resolve_indicator(snap, slug)
377 + cs = _an.cross_section(snap)
378 + m, logs = _log_matrix(snap)
379 + j = cs.iidx.get(ind["id"])
380 + items = []
381 + if j is not None:
382 + x = m[:, j]
383 + raw_x, _ = cs.col(ind["id"])
384 + for other, k in cs.iidx.items():
385 + if other == ind["id"]:
386 + continue
387 + y = m[:, k]
388 + ok = np.isfinite(x) & np.isfinite(y)
389 + if ok.sum() < min_n:
390 + continue
391 + raw_y, years_y = cs.col(other)
392 + rho, n = _stats.spearman(raw_x[ok].tolist(), raw_y[ok].tolist())
393 + r, _ = _stats.pearson(x[ok].tolist(), y[ok].tolist())
394 + if rho is None:
395 + continue
396 + o = merged_indicator(snap.indicators()[other])
397 + yrs = years_y[ok]
398 + items.append({"indicator": indicator_card(o), "pearson": r, "spearman": rho, "n": n,
399 + "year": int(np.nanmax(yrs)) if len(yrs) else None, "log_x": logs[ind["id"]], "log_y": logs[other],
400 + "direction": "positive" if rho >= 0 else "negative"})
401 + items.sort(key=lambda i: -abs(i["spearman"] or 0))
402 + year_used = int(cs.max_year[j]) if j is not None and np.isfinite(cs.max_year[j]) else None
403 + return {"meta": meta_block(snap), "indicator": indicator_card(ind), "year_used": year_used, "n_candidates": len(items),
404 + "items": items[:limit], "note": NOTE_CORRELATION}
405 +
406 +
407 +@router.get("/{slug}/distribution", response_model=schemas.DistributionResponse, summary="Distribution across countries for a year")
408 +def get_indicator_distribution(slug: str, year: int | None = Query(None, ge=1800, le=2100), highlight: str | None = Query(None),
409 + bins: int = Query(20, ge=5, le=60), snap: Snapshot = Depends(get_snapshot)) -> dict[str, Any]:
410 + ind = resolve_indicator(snap, slug)
411 + year_used = year if year is not None else latest_common_year(snap, ind["id"])[0]
412 + vals = _an._nearest_values(snap, ind["id"], year_used, None) if year_used is not None else {}
413 + ids = sorted(vals)
414 + arr = np.array([vals[c][0] for c in ids], dtype=float)
415 + use_log = _an.auto_log(ind, arr)
416 + fin = arr[np.isfinite(arr)]
417 +
418 + def pct(q: float) -> float | None:
419 + return float(np.percentile(fin, q)) if len(fin) else None
420 +
421 + st = {"min": float(fin.min()) if len(fin) else None, "p10": pct(10), "p25": pct(25), "median": pct(50),
422 + "mean": float(fin.mean()) if len(fin) else None, "p75": pct(75), "p90": pct(90), "max": float(fin.max()) if len(fin) else None}
423 + # group medians
424 + def group_medians(kind: str) -> list[dict[str, Any]]:
425 + out = []
426 + for g in snap.groups().values():
427 + if g.get("kind") != kind:
428 + continue
429 + mem = set(snap.group_members(g["id"]))
430 + gv = [vals[c][0] for c in ids if c in mem]
431 + if len(gv) >= 2:
432 + out.append({"group": {"id": g["id"], "slug": g.get("slug"), "name": g.get("name"), "kind": g.get("kind"), "wb_code": g.get("wb_code"),
433 + "n_members": g.get("n_members")}, "median": float(np.median(gv)), "n": len(gv)})
434 + out.sort(key=lambda x: -(x["median"] or 0))
435 + return out
436 +
437 + by_region, by_income = group_medians("region"), group_medians("income")
438 + hl = None
439 + if highlight:
440 + c = resolve_country(snap, highlight)
441 + v = vals.get(c["id"])
442 + if v is not None:
443 + pr = _stats.percentile_rank(arr.tolist())
444 + i = ids.index(c["id"])
445 + desc = ind.get("higher_is_better") is not False
446 + rk = _stats.rank(arr.tolist(), descending=desc)[i]
447 + reg = next((g for g in by_region if g["group"]["wb_code"] == c.get("region_wb")), None)
448 + inc = next((g for g in by_income if (g["group"]["wb_code"] or "").upper() == (c.get("income_group") or "").upper()), None)
449 + hl = {"country": country_card(c), "value": v[0], "year": v[1], "percentile": pr[i], "rank": rk, "n": len(fin),
450 + "region": reg["group"] if reg else None, "region_median": reg["median"] if reg else None,
451 + "income": inc["group"] if inc else None, "income_median": inc["median"] if inc else None}
452 + else:
453 + hl = {"country": country_card(c), "value": None, "year": None, "percentile": None, "rank": None, "n": len(fin), "region": None,
454 + "region_median": None, "income": None, "income_median": None}
455 + return {"meta": meta_block(snap), "indicator": indicator_card(ind), "year": year, "year_used": year_used, "n": len(fin), "log": use_log,
456 + "histogram": _stats.histogram(fin.tolist(), bins=bins, log=use_log), "stats": st, "highlight": hl,
457 + "by_region": by_region, "by_income": by_income, "provenance": _an.indicator_provenances(snap).get(ind["id"])}
458 +
459 +
460 +@router.get("/{slug}/frames", response_model=schemas.FramesResponse, summary="Multi-year map frames (time machine)")
461 +def get_indicator_frames(slug: str, from_: int | None = Query(None, alias="from", ge=1750, le=2100), to: int | None = Query(None, ge=1750, le=2100),
462 + step: int = Query(1, ge=1, le=10), group: str = Query("world"), snap: Snapshot = Depends(get_snapshot)) -> dict[str, Any]:
463 + ind = resolve_indicator(snap, slug)
464 + g, members = _an.group_members_or_all(snap, group)
465 + where = ["o.indicator_id = ?", "o.frequency = 'A'", "NOT o.is_forecast", "o.value IS NOT NULL", "coalesce(o.status, 'imported') <> 'quarantined'",
466 + "coalesce(c.kind, 'country') = 'country'"]
467 + params: list[Any] = [ind["id"]]
468 + if members is not None:
469 + if not members:
470 + where.append("FALSE")
471 + else:
472 + where.append(f"o.country_id IN ({','.join('?' * len(members))})")
473 + params += members
474 + if from_ is not None:
475 + where.append("o.year >= ?")
476 + params.append(from_)
477 + if to is not None:
478 + where.append("o.year <= ?")
479 + params.append(to)
480 + rows = snap.query_rows(f"SELECT o.country_id, o.year, o.value FROM observations o JOIN countries c ON c.id = o.country_id WHERE {' AND '.join(where)}", params)
481 + by_year: dict[int, dict[str, float]] = {}
482 + for cid, yr, v in rows:
483 + by_year.setdefault(int(yr), {})[cid] = float(v)
484 + years = sorted(y for y, d in by_year.items() if len(d) >= 20)
485 + if step > 1:
486 + years = [y for y in years if (y - years[0]) % step == 0] if years else []
487 + years = years[-80:]
488 + ids = sorted({c for y in years for c in by_year[y]})
489 + values = {c: [by_year[y].get(c) for y in years] for c in ids}
490 + pooled = [v for y in years for v in by_year[y].values()]
491 + breaks = quantile_breaks(pooled, 6)
492 + if from_ is not None and to is not None and from_ > to:
493 + raise _bad_request("`from` must be <= `to`.")
494 + return {"meta": meta_block(snap), "indicator": indicator_card(ind),
495 + "group": {"id": g["id"], "slug": g.get("slug"), "name": g.get("name"), "kind": g.get("kind"), "wb_code": g.get("wb_code"), "n_members": g.get("n_members")},
496 + "years": years, "values": values,
497 + "legend": {"min": min(pooled) if pooled else None, "max": max(pooled) if pooled else None, "breaks": breaks, "n_classes": len(breaks) + 1},
498 + "n_by_year": [len(by_year[y]) for y in years], "provenance": _an.indicator_provenances(snap).get(ind["id"])}
added src/countryatlas/api/routers/quality.py +202 −0
@@ -0,0 +1,202 @@
1 +"""Data quality and freshness (API 1.1): /indicators/{slug}/quality, /countries/{id}/quality, /updates.
2 +
3 +Badges vocabulary (shared): fresh · historical · sparse · limited-coverage · stale · flagged · forecast. No file paths, hosts
4 +or secrets are ever returned.
5 +"""
6 +from __future__ import annotations
7 +
8 +from typing import Any
9 +
10 +import numpy as np
11 +from fastapi import APIRouter, Depends, Query
12 +
13 +from countryatlas.api import schemas
14 +from countryatlas.api.common import country_card, indicator_card, merged_indicator, meta_block, resolve_country, resolve_indicator
15 +from countryatlas.api.db import Snapshot, get_snapshot
16 +from countryatlas.api.provenance import _iso
17 +from countryatlas.api.routers.analytics import country_ids, indicator_provenances, latest_common_year
18 +from countryatlas.api.routers.indicators import indicator_summary
19 +
20 +router = APIRouter(tags=["quality"])
21 +
22 +SPARSE_POINTS = 10
23 +FRESH_LAG = 1
24 +STALE_LAG = 3
25 +FLAGGED_SHARE = 0.05
26 +
27 +
28 +def reference_year(snap: Snapshot) -> int:
29 + return snap._cached("reference_year", lambda: int(snap.scalar(
30 + "SELECT max(l.year) FROM latest l JOIN countries c ON c.id = l.country_id WHERE coalesce(c.kind, 'country') = 'country' "
31 + "AND l.indicator_id IN ('population', 'gdp', 'life-expectancy')") or 0))
32 +
33 +
34 +def _badges(*, latest_year: int | None, first_year: int | None, points: float | None, coverage_pct: float | None, flagged_share: float,
35 + forecast: bool, ref: int) -> list[str]:
36 + out: list[str] = []
37 + if latest_year is not None and latest_year >= ref - FRESH_LAG:
38 + out.append("fresh")
39 + if first_year is not None and first_year <= 1970:
40 + out.append("historical")
41 + if points is not None and points < SPARSE_POINTS:
42 + out.append("sparse")
43 + if coverage_pct is not None and coverage_pct < 50:
44 + out.append("limited-coverage")
45 + if latest_year is not None and latest_year <= ref - STALE_LAG:
46 + out.append("stale")
47 + if flagged_share > FLAGGED_SHARE:
48 + out.append("flagged")
49 + if forecast:
50 + out.append("forecast")
51 + return out
52 +
53 +
54 +@router.get("/indicators/{slug}/quality", response_model=schemas.IndicatorQualityResponse, summary="Coverage / freshness / continuity of an indicator")
55 +def indicator_quality(slug: str, snap: Snapshot = Depends(get_snapshot)) -> dict[str, Any]:
56 + ind = resolve_indicator(snap, slug)
57 + total = len(country_ids(snap))
58 + ref = reference_year(snap)
59 + per_country = snap.query(
60 + """SELECT o.country_id, count(*) AS n, min(o.year) AS y0, max(o.year) AS y1,
61 + sum(CASE WHEN o.status = 'warning' THEN 1 ELSE 0 END) AS n_warn,
62 + max(CASE WHEN o.is_forecast THEN 1 ELSE 0 END) AS has_fc
63 + FROM observations o JOIN countries c ON c.id = o.country_id JOIN indicators i ON i.id = o.indicator_id
64 + WHERE o.indicator_id = ? AND o.frequency = i.frequency AND coalesce(c.kind, 'country') = 'country' AND o.value IS NOT NULL
65 + GROUP BY 1""",
66 + [ind["id"]],
67 + )
68 + actual = [r for r in per_country]
69 + n_countries = len(actual)
70 + years_rows = snap.query_rows(
71 + """SELECT o.year, count(*) FROM observations o JOIN countries c ON c.id = o.country_id JOIN indicators i ON i.id = o.indicator_id
72 + WHERE o.indicator_id = ? AND o.frequency = i.frequency AND NOT o.is_forecast AND o.value IS NOT NULL
73 + AND coalesce(c.kind, 'country') = 'country' GROUP BY 1 ORDER BY 1""",
74 + [ind["id"]],
75 + )
76 + latest_years = [int(r["y1"]) for r in actual]
77 + points = [int(r["n"]) for r in actual]
78 + n_warn = sum(int(r["n_warn"]) for r in actual)
79 + n_obs = sum(points)
80 + first_year = min((int(r["y0"]) for r in actual), default=None)
81 + last_year = max(latest_years, default=None)
82 + coverage = round(100.0 * n_countries / max(total, 1), 1)
83 + med_points = float(np.median(points)) if points else None
84 + prov = indicator_provenances(snap).get(ind["id"])
85 + src_rows = snap.query(
86 + "SELECT source_id, source_dataset, source_series_code, max(retrieved_at) AS ra, max(source_updated_at) AS su, count(*) AS n "
87 + "FROM observations WHERE indicator_id = ? GROUP BY 1, 2, 3 ORDER BY n DESC", [ind["id"]])
88 + from countryatlas.api.provenance import build_provenance
89 +
90 + sources = [dict(build_provenance(snap, ind["id"], r["source_id"], r["source_dataset"], r["source_series_code"], r["ra"], r["su"]), n_values=r["n"])
91 + for r in src_rows]
92 + return {
93 + "meta": meta_block(snap), "indicator": indicator_card(ind), "n_countries": n_countries, "n_countries_total": total, "coverage_pct": coverage,
94 + "first_year": first_year, "last_year": last_year, "latest_common_year": latest_common_year(snap, ind["id"]),
95 + "n_years": len(years_rows), "years_with_50plus": sum(1 for _, n in years_rows if n >= 50),
96 + "median_points_per_country": med_points, "sparse_countries": sum(1 for p in points if p < SPARSE_POINTS),
97 + "stale_countries": sum(1 for y in latest_years if y <= ref - STALE_LAG), "flagged_values": n_warn,
98 + "sources": sources,
99 + "badges": _badges(latest_year=last_year, first_year=first_year, points=med_points, coverage_pct=coverage,
100 + flagged_share=(n_warn / n_obs) if n_obs else 0.0, forecast=any(int(r["has_fc"]) for r in actual), ref=ref),
101 + "provenance": prov,
102 + }
103 +
104 +
105 +@router.get("/countries/{id}/quality", response_model=schemas.CountryQualityResponse, summary="Per-indicator data quality for a country")
106 +def country_quality(id: str, snap: Snapshot = Depends(get_snapshot)) -> dict[str, Any]:
107 + c = resolve_country(snap, id)
108 + ref = reference_year(snap)
109 + rows = snap.query(
110 + """SELECT o.indicator_id, count(*) AS n, min(o.year) AS y0, max(o.year) AS y1, count(DISTINCT o.year) AS ny,
111 + sum(CASE WHEN o.status = 'warning' THEN 1 ELSE 0 END) AS n_warn,
112 + arg_max(o.status, o.period) AS status, arg_max(o.source_id, o.period) AS source,
113 + max(o.source_updated_at) AS su, max(o.retrieved_at) AS ra, max(CASE WHEN o.is_forecast THEN 1 ELSE 0 END) AS has_fc
114 + FROM observations o JOIN indicators i ON i.id = o.indicator_id
115 + WHERE o.country_id = ? AND o.frequency = i.frequency AND NOT o.is_forecast AND o.value IS NOT NULL GROUP BY 1 ORDER BY 1""",
116 + [c["id"]],
117 + )
118 + items = []
119 + n_fresh = n_stale = n_sparse = n_flagged = 0
120 + latest = None
121 + for r in rows:
122 + ind = merged_indicator(snap.indicators()[r["indicator_id"]])
123 + y0, y1, n = int(r["y0"]), int(r["y1"]), int(r["n"])
124 + expected = y1 - y0 + 1 if ind.get("frequency", "A") == "A" else n
125 + missing = max(0, expected - int(r["ny"]))
126 + cont = round(100.0 * int(r["ny"]) / expected, 1) if expected else None
127 + badges = _badges(latest_year=y1, first_year=y0, points=n, coverage_pct=None, flagged_share=int(r["n_warn"]) / n if n else 0.0,
128 + forecast=bool(int(r["has_fc"])), ref=ref)
129 + n_fresh += "fresh" in badges
130 + n_stale += "stale" in badges
131 + n_sparse += "sparse" in badges
132 + n_flagged += "flagged" in badges
133 + latest = y1 if latest is None else max(latest, y1)
134 + items.append({"indicator": indicator_card(ind), "latest_year": y1, "first_year": y0, "n_points": n, "expected_points": expected,
135 + "missing_years": missing, "continuity_pct": cont, "status": r["status"], "source": r["source"],
136 + "source_updated_at": _iso(r["su"]), "retrieved_at": _iso(r["ra"]), "badges": badges})
137 + total = len(snap.indicators())
138 + return {"meta": meta_block(snap), "country": country_card(c),
139 + "summary": {"n_indicators": total, "n_with_data": len(items), "coverage_pct": round(100.0 * len(items) / max(total, 1), 1),
140 + "latest_year": latest, "n_fresh": n_fresh, "n_stale": n_stale, "n_sparse": n_sparse, "n_flagged": n_flagged},
141 + "items": items}
142 +
143 +
144 +@router.get("/updates", response_model=schemas.UpdatesResponse, summary="Freshness dashboard: sources, runs, changed values")
145 +def updates(runs_limit: int = Query(20, ge=1, le=100), snap: Snapshot = Depends(get_snapshot)) -> dict[str, Any]:
146 + run_id = snap.run_id
147 + ref = reference_year(snap)
148 + changed = {r[0]: (int(r[1]), int(r[2])) for r in snap.query_rows(
149 + "SELECT coalesce(new_source_id, old_source_id) AS s, count(*), count(DISTINCT country_id) FROM observation_revisions WHERE run_id = ? GROUP BY 1",
150 + [run_id])}
151 + src_stats = {r["source_id"]: r for r in snap.query(
152 + """SELECT source_id, count(*) AS n_obs, count(DISTINCT indicator_id) AS n_ind, count(DISTINCT source_dataset) AS n_ds,
153 + max(CASE WHEN NOT is_forecast THEN year END) AS latest_year, max(retrieved_at) AS ra, max(source_updated_at) AS su
154 + FROM observations GROUP BY 1""")}
155 + run_status = {r["connector"]: r for r in snap.query(
156 + """SELECT connector, arg_max(status, coalesce(finished_at, started_at)) AS status, max(finished_at) AS finished,
157 + sum(CASE WHEN status IN ('failed', 'quarantined') THEN 1 ELSE 0 END) AS n_failed, count(*) AS n
158 + FROM import_runs GROUP BY 1""")}
159 + sources = []
160 + for sid, s in snap.sources().items():
161 + st = src_stats.get(sid)
162 + rs = run_status.get(sid)
163 + if st is None and rs is None:
164 + continue
165 + latest_year = int(st["latest_year"]) if st and st.get("latest_year") is not None else None
166 + if rs is None:
167 + status = "unknown"
168 + elif rs["status"] == "failed" or (rs["n"] and rs["n_failed"] == rs["n"]):
169 + status = "failed"
170 + elif rs["n_failed"] or rs["status"] in ("partial", "quarantined"):
171 + status = "partial"
172 + elif latest_year is not None and latest_year <= ref - STALE_LAG:
173 + status = "stale"
174 + else:
175 + status = "ok"
176 + ch = changed.get(sid, (0, 0))
177 + sources.append({
178 + "source": {"id": sid, "name": s.get("name"), "organization": s.get("organization"), "url": s.get("url"), "licence": s.get("licence")},
179 + "status": status, "last_success_at": _iso(s.get("last_success_at")), "last_retrieved_at": _iso(st["ra"]) if st else None,
180 + "source_updated_at": _iso(st["su"]) if st else None, "n_datasets": int(st["n_ds"]) if st else 0,
181 + "n_indicators": int(st["n_ind"]) if st else 0, "n_observations": int(st["n_obs"]) if st else 0, "latest_year": latest_year,
182 + "values_changed": ch[0], "countries_affected": ch[1],
183 + "runs": {"n": int(rs["n"]) if rs else 0, "n_failed": int(rs["n_failed"]) if rs else 0},
184 + })
185 + sources.sort(key=lambda s: -s["n_observations"])
186 + runs = snap.query(
187 + "SELECT run_id, connector, dataset, started_at, finished_at, status, rows_valid, warnings, errors, message FROM import_runs "
188 + "ORDER BY coalesce(finished_at, started_at) DESC NULLS LAST LIMIT ?", [runs_limit])
189 + for r in runs:
190 + r["started_at"], r["finished_at"] = _iso(r.get("started_at")), _iso(r.get("finished_at"))
191 + if r.get("message"):
192 + r["message"] = str(r["message"])[:200]
193 + inds = [merged_indicator(i) for i in snap.indicators().values() if i.get("latest_source_updated_at") is not None]
194 + inds.sort(key=lambda i: _iso(i.get("latest_source_updated_at")) or "", reverse=True)
195 + total_changed = int(snap.scalar("SELECT count(*) FROM observation_revisions WHERE run_id = ?", [run_id]) or 0)
196 + return {
197 + "meta": meta_block(snap),
198 + "snapshot": {"run_id": run_id, "built_at": snap.built_at, "observations": int(snap.meta.get("observation_count") or snap.table_count("observations")),
199 + "indicators": len(snap.indicators()), "countries": len(country_ids(snap)), "values_changed": total_changed,
200 + "values_changed_by_source": {k: v[0] for k, v in changed.items()}, "reference_year": ref},
201 + "sources": sources, "recent_runs": runs, "indicators_recently_updated": [indicator_summary(snap, i) for i in inds[:12]],
202 + }
modified src/countryatlas/api/routers/rankings.py +46 −0
Large diff (49 lines) is not loaded by default to keep the page fast.Load diff
modified src/countryatlas/api/routers/regions.py +94 −0
Large diff (106 lines) is not loaded by default to keep the page fast.Load diff
modified src/countryatlas/api/routers/search.py +115 −0
Large diff (131 lines) is not loaded by default to keep the page fast.Load diff
modified src/countryatlas/api/schemas.py +336 −0
Large diff (339 lines) is not loaded by default to keep the page fast.Load diff
modified src/countryatlas/connectors/base.py +35 −10
Large diff (86 lines) is not loaded by default to keep the page fast.Load diff
modified src/countryatlas/pipeline/build.py +66 −3
Large diff (131 lines) is not loaded by default to keep the page fast.Load diff
modified src/countryatlas/pipeline/changes.py +109 −1
Large diff (143 lines) is not loaded by default to keep the page fast.Load diff
modified src/countryatlas/pipeline/fetch.py +6 −5
Large diff (30 lines) is not loaded by default to keep the page fast.Load diff
modified src/countryatlas/pipeline/similarity.py +6 −1
Large diff (25 lines) is not loaded by default to keep the page fast.Load diff
modified src/countryatlas/pipeline/staging.py +12 −0
Large diff (18 lines) is not loaded by default to keep the page fast.Load diff
modified src/countryatlas/pipeline/validate.py +112 −15
Large diff (211 lines) is not loaded by default to keep the page fast.Load diff
added tests/api/test_analytics.py +207 −0
Large diff (207 lines) is not loaded by default to keep the page fast.Load diff
modified tests/test_changes.py +73 −0
Large diff (76 lines) is not loaded by default to keep the page fast.Load diff
added tests/test_pipeline_reliability.py +125 −0
Large diff (125 lines) is not loaded by default to keep the page fast.Load diff
modified tests/test_validate.py +58 −0
Large diff (62 lines) is not loaded by default to keep the page fast.Load diff
Menu

Theme
↑↓ navigate ↵ open esc close / or ⌘K anywhere