SPB Git forge
spboucher.ai ↗

spb/countryatlas

Public
20commits 1branches 0releases
268.3 MBsize
maindefault branch
12 days agolast push
TypeScript 57% Python 38.6% JavaScript 3.6% CSS 0.6%

Pipeline: canonical-frequency derived tables, rank-year fallback, integrity warnings

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Simon-Pierre Boucher committed 13 days ago (Sep 11, 2026) parent b9cbc47

15 changed files +169 −41

modified docs/ARCHITECTURE.md +11 −4
@@ -199,6 +199,11 @@ No table or column of `schema.sql` was changed. The following precisions/deviati
199 199 `series_fresher_source`. Consequence: forecast rows appear in `observations` only when the chosen source itself publishes
200 200 them (IMF-only indicators, or IMF chosen as fresher); the complete IMF series (history + forecasts) is always available in
201 201 `observations_alt` for alternative views. Series with forecast-only data fall back to plain priority.
202 +* **Canonical frequency only in derived tables**: `latest`, `rankings`, `changes`, `events`, `similarity`, `insights`,
203 + `country_dna` and `coverage` use only observations whose `frequency` equals the indicator's registry `frequency`
204 + (inflation A, policy-rate M, real-house-price-index Q…). Higher-frequency series (e.g. FRED monthly CPI for the USA) stay in
205 + `observations` for the series/chart endpoints only. Integrity emits warnings when a `latest` row has a non-canonical
206 + frequency or a headline indicator is ranked in a pool of fewer than 50 countries.
202 207 * **Quarantined rows stay in `observations`** (never deleted) but are excluded from every derived table (`latest`,
203 208 `rankings`, `changes`, `events`, `similarity`, `insights`, `country_dna`). A lower-priority source is *not* promoted when
204 209 a row is quarantined (the value is flagged, not replaced). `observations_alt` = every row of every non-chosen source,
@@ -210,10 +215,12 @@ No table or column of `schema.sql` was changed. The following precisions/deviati
210 215 compared on log-differences; others on absolute differences. Threshold = `jump_threshold` × 1.4826·MAD with a floor
211 216 (10 % relative, or 2 % of the country's series range) to avoid flagging smooth series; ≥ 5 points required. ~5 % of WDI
212 217 rows carry `warning`.
213 −* **`latest` ranks** are computed among *all* `kind='country'` values of the same indicator and the same **year** as the
214 − country's latest observation (not only among countries whose latest year coincides); `latest.rank_year` = that year, the
215 − API compares it with the indicator's max year. Q/M series use the last period of each year. `rank_income` is NULL when the
216 − country has no income group. `change_10y_*` compares with the observation exactly 10 years earlier (same frequency).
218 +* **`latest` ranks** are computed among *all* `kind='country'` values of the same indicator within one **year**: the most
219 + recent year ≤ the country's latest year in which the country has a value AND at least 20 countries are ranked
220 + (`MIN_COUNTRIES_FOR_RANKING`). `latest.rank_year` = that year — it may be older than `latest.year` when only a handful of
221 + countries already report the newest year (e.g. CHN internet-users: value 2025, ranks from 2024); ranks are NULL when no
222 + such year exists. Q/M series use the last period of each year. `rank_income` is NULL when the country has no income group.
223 + `change_10y_*` compares with the observation exactly 10 years earlier (same frequency).
217 224 * **Ranking direction**: rank 1 = lowest value when `higher_is_better = false`, otherwise the highest value — i.e. "best"
218 225 when `higher_is_better` is set, "highest" when it is null. `rankings.pct_rank = 1 − (rank − 1)/(n − 1)` (1.0 = rank 1).
219 226 `rankings` includes every year with ≥ 20 countries for `ranking_eligible` indicators.
modified docs/PIPELINE.md +3 −1
@@ -96,7 +96,9 @@ country's differences, with a floor of 10 % (relative) or 2 % of the series rang
96 96 3. `observation_revisions`: the previous `atlas.duckdb` is attached read-only; every key whose value or source changed
97 97 is recorded with the new `run_id`; the previous revisions table is copied over.
98 98 4. `import_runs` (latest `run.json` per spec) and `validation_issues` (issues sidecars).
99 −5. Derived tables (`pipeline/derived.py`, SQL): `latest` (prev period, 10-year change, ranks within the same year among
99 +5. Derived tables (`pipeline/derived.py`, SQL) are computed from `obs_ok` = non-forecast, non-quarantined observations at
100 + the indicator's **canonical frequency** only (a monthly FRED series never becomes a country's "latest" for an annual
101 + indicator): `latest` (prev period, 10-year change, ranks within the same year among
100 102 `kind='country'` — world / WB region / income group), `rankings` (ranking-eligible indicators, years with ≥ 20
101 103 countries), `coverage`, indicator/source coverage columns, `search_index`.
102 104 **Rank direction:** rank 1 = lowest value when `higher_is_better = false`, otherwise highest value ("best" when
modified src/countryatlas/api/common.py +3 −2
@@ -8,7 +8,7 @@ from typing import Any
8 8
9 9 from countryatlas.api.db import Snapshot
10 10 from countryatlas.api.errors import bad_request, not_found
11 −from countryatlas.api.formatting import format_change, format_value
11 +from countryatlas.api.formatting import format_change, format_short, format_value
12 12 from countryatlas.api.provenance import build_provenance
13 13 from countryatlas.registry import indicators_by_id as registry_indicators
14 14 from countryatlas.registry import topics as registry_topics
@@ -264,6 +264,7 @@ def metric_from_latest(snap: Snapshot, row: dict[str, Any], ind: dict[str, Any],
264 264 "has_data": value is not None,
265 265 "value": value,
266 266 "formatted": format_value(value, ind),
267 + "formatted_short": format_short(value, ind),
267 268 "period": row.get("period"),
268 269 "year": row.get("year"),
269 270 "frequency": row.get("frequency"),
@@ -305,7 +306,7 @@ def metric_from_latest(snap: Snapshot, row: dict[str, Any], ind: dict[str, Any],
305 306
306 307 def empty_metric(ind: dict[str, Any]) -> dict[str, Any]:
307 308 return {"indicator": ind["id"], "indicator_name": ind.get("short_name") or ind.get("name"), "has_data": False,
308 − "value": None, "formatted": "—", "unit": ind.get("unit"), "unit_short": ind.get("unit_short"),
309 + "value": None, "formatted": "—", "formatted_short": "—", "unit": ind.get("unit"), "unit_short": ind.get("unit_short"),
309 310 "format": ind.get("format"), "higher_is_better": ind.get("higher_is_better"), "sparkline": [], "provenance": None}
310 311
311 312
modified src/countryatlas/api/db.py +13 −0
@@ -39,6 +39,18 @@ class DataNotBuilt(Exception):
39 39 """Raised by data endpoints when no snapshot exists yet."""
40 40
41 41
42 +def invalidate_registry_caches() -> None:
43 + """Drop the lru_cache of every registry loader so YAML edits (topics.yaml headline, indicators.yaml…) are re-read when a
44 + new snapshot is opened. Registry-derived lists are therefore effectively cached per run_id."""
45 + from countryatlas import registry
46 +
47 + for name in ("countries", "countries_by_id", "countries_by_slug", "groups", "groups_by_id", "indicators", "indicators_by_id",
48 + "topics", "lookup"):
49 + fn = getattr(registry, name, None)
50 + if fn is not None and hasattr(fn, "cache_clear"):
51 + fn.cache_clear()
52 +
53 +
42 54 class Snapshot:
43 55 """One open read-only connection to one physical DuckDB file, plus lazily loaded static lookups."""
44 56
@@ -214,6 +226,7 @@ class Database:
214 226 log.error("cannot open %s: %s", self.path, e)
215 227 return None
216 228 self._snap = new
229 + invalidate_registry_caches()
217 230 log.info("opened snapshot run_id=%s built_at=%s (%s)", new.run_id, new.built_at, self.path)
218 231 return new
219 232
modified src/countryatlas/api/formatting.py +11 −3
@@ -1,7 +1,8 @@
1 1 """Server-side number formatting (search hints, OpenGraph, headlines). The web app formats too — keep both in sync.
2 2
3 −`format_value(53372.1, gdp_per_capita)` → "53.4k"; `format_value(1.23e12, gdp)` → "1.2T"; percent → "3.4 %";
4 −years → "82.1 yrs"; tonnes → "5.2 t"; per_1000 → "3.2 per 1,000"; per_100k → "1.2 per 100k".
3 +`format_value(53372.1, gdp_per_capita)` → "US$53.4k"; `format_value(1.23e12, gdp)` → "US$1.2T" (prefix = registry `unit_short`:
4 +`US$`, `intl $`, `PPS`…); percent → "3.4 %"; years → "82.1 yrs"; tonnes → "5.2 t"; per_1000 → "3.2 per 1,000"; per_100k → "1.2 per 100k".
5 +`format_short(value, indicator)` is the same without unit/prefix ("53.4k", "3.4").
5 6 """
6 7 from __future__ import annotations
7 8
@@ -58,7 +59,9 @@ def format_value(value: float | None, indicator: Any = None, *, with_unit: bool
58 59 v = float(value)
59 60
60 61 if fmt == "currency":
61 − return compact_number(v, precision=1)
62 + s = compact_number(v, precision=1)
63 + prefix = unit_short or "US$"
64 + return f"{prefix}{s}" if with_unit else s
62 65 if fmt == "percent":
63 66 s = f"{v:.{max(precision, 1)}f}"
64 67 return f"{s} %" if with_unit else s
@@ -91,6 +94,11 @@ def format_value(value: float | None, indicator: Any = None, *, with_unit: bool
91 94 return s
92 95
93 96
97 +def format_short(value: float | None, indicator: Any = None) -> str:
98 + """Compact number without unit prefix/suffix (for tight UI cells; the unit is shown elsewhere)."""
99 + return format_value(value, indicator, with_unit=False)
100 +
101 +
94 102 def format_change(change_abs: float | None, change_pct: float | None, indicator: Any = None) -> str | None:
95 103 """Human change: percent-like indicators → '+1.2 pts', otherwise '+3.4 %'."""
96 104 fmt = _get(indicator, "format", "number")
modified src/countryatlas/api/routers/home.py +17 −9
@@ -8,24 +8,31 @@ from fastapi import APIRouter, Depends
8 8 from countryatlas.api import schemas
9 9 from countryatlas.api.common import clean_float, country_card, indicator_card, merged_indicator, meta_block
10 10 from countryatlas.api.db import Snapshot, get_snapshot
11 −from countryatlas.api.formatting import format_value
11 +from countryatlas.api.formatting import format_short, format_value
12 12 from countryatlas.api.provenance import _iso, build_provenance
13 13 from countryatlas.api.routers.countries import change_item, query_changes
14 14 from countryatlas.api.routers.indicators import indicator_summary
15 15
16 16 router = APIRouter(tags=["home"])
17 17
18 −# (key, title, indicator, order, min_population, extra description)
18 +# (key, title, indicator, order, min_population, description)
19 19 CURATED = [
20 20 ("largest_economies", "Largest economies", "gdp", "desc", None, "GDP, current US$"),
21 − ("fastest_population_growth", "Fastest population growth", "population-growth", "desc", 1_000_000, "Countries above 1 M inhabitants"),
22 − ("highest_life_expectancy", "Highest life expectancy", "life-expectancy", "desc", None, "Years at birth"),
23 − ("energy_transition_leaders", "Energy transition leaders", "renewable-electricity-share", "desc", None, "Share of electricity from renewables"),
21 + ("fastest_gdp_growth", "Fastest GDP growth", "gdp-growth", "desc", 1_000_000, "Real GDP growth, latest year"),
22 + ("fastest_population_growth", "Fastest population growth", "population-growth", "desc", 1_000_000, "Annual population growth"),
23 + ("highest_life_expectancy", "Highest life expectancy", "life-expectancy", "desc", 1_000_000, "Years at birth"),
24 + ("energy_transition_leaders", "Energy transition leaders", "renewable-electricity-share", "desc", 1_000_000, "Share of electricity from renewables"),
24 25 ("highest_gdp_per_capita_ppp", "Highest GDP per capita (PPP)", "gdp-per-capita-ppp", "desc", None, "International $"),
25 − ("lowest_unemployment", "Lowest unemployment", "unemployment-rate", "asc", 5_000_000, "Countries above 5 M inhabitants"),
26 + ("lowest_unemployment", "Lowest unemployment", "unemployment-rate", "asc", 5_000_000, "Unemployment, % of labour force"),
26 27 ]
27 28
28 29
30 +def filter_note(min_pop: int | None) -> str | None:
31 + if not min_pop:
32 + return None
33 + return f"Countries above {min_pop // 1_000_000}M inhabitants"
34 +
35 +
29 36 def curated_list(snap: Snapshot, indicator_id: str, order: str, min_pop: int | None, n: int = 8) -> dict[str, Any] | None:
30 37 ind_row = snap.indicators().get(indicator_id)
31 38 if ind_row is None:
@@ -47,12 +54,13 @@ def curated_list(snap: Snapshot, indicator_id: str, order: str, min_pop: int | N
47 54 for i, r in enumerate(rows):
48 55 c = snap.countries().get(r["country_id"], {"id": r["country_id"]})
49 56 v = clean_float(r["value"])
50 − items.append({"rank": i + 1, "country": country_card(c), "value": v, "formatted": format_value(v, ind), "year": r["year"],
57 + items.append({"rank": i + 1, "country": country_card(c), "value": v, "formatted": format_value(v, ind), "formatted_short": format_short(v, ind),
58 + "year": r["year"],
51 59 "change_pct": clean_float(r.get("change_pct")), "change_abs": clean_float(r.get("change_abs")),
52 60 "rank_world": r.get("rank_world"), "n_world": r.get("n_world"),
53 61 "provenance": build_provenance(snap, indicator_id, r["source_id"], r.get("source_dataset"), r.get("source_series_code"),
54 62 r.get("retrieved_at"), r.get("source_updated_at"), c.get("iso2"))})
55 − return {"indicator": indicator_card(ind), "sort": order, "rows": items}
63 + return {"indicator": indicator_card(ind), "sort": order, "min_population": min_pop, "filter_note": filter_note(min_pop), "rows": items}
56 64
57 65
58 66 def global_snapshot(snap: Snapshot) -> dict[str, Any]:
@@ -73,7 +81,7 @@ def global_snapshot(snap: Snapshot) -> dict[str, Any]:
73 81 return {
74 82 "world_population": clean_float(row.get("pop")), "world_population_formatted": format_value(clean_float(row.get("pop")), {"format": "number"}),
75 83 "world_population_year": row.get("pop_year"),
76 − "world_gdp": clean_float(row.get("gdp")), "world_gdp_formatted": format_value(clean_float(row.get("gdp")), {"format": "currency"}),
84 + "world_gdp": clean_float(row.get("gdp")), "world_gdp_formatted": format_value(clean_float(row.get("gdp")), {"format": "currency", "unit_short": "US$"}),
77 85 "world_gdp_year": row.get("gdp_year"),
78 86 "median_life_expectancy": clean_float(row.get("le")), "median_life_expectancy_year": row.get("le_year"),
79 87 "n_countries": n_countries, "n_territories": len(snap.countries()) - n_countries,
modified src/countryatlas/api/routers/indicators.py +3 −1
@@ -18,7 +18,7 @@ from countryatlas.api.common import (
18 18 resolve_indicator,
19 19 )
20 20 from countryatlas.api.db import Snapshot, get_snapshot
21 −from countryatlas.api.formatting import format_value
21 +from countryatlas.api.formatting import format_short, format_value
22 22 from countryatlas.api.provenance import _iso, build_provenance, provenance_from_row, source_url
23 23 from countryatlas.registry import topics as registry_topics
24 24
@@ -132,6 +132,7 @@ def world_latest(snap: Snapshot, ind: dict[str, Any]) -> dict[str, Any] | None:
132 132 "kind": kind,
133 133 "value": value,
134 134 "formatted": format_value(value, ind),
135 + "formatted_short": format_short(value, ind),
135 136 "year": year,
136 137 "n": row.get("n"),
137 138 "median": clean_float(row.get("med")),
@@ -161,6 +162,7 @@ def ranked_extremes(snap: Snapshot, ind: dict[str, Any], year: int | None, n: in
161 162 def pack(r: dict[str, Any], rank: int) -> dict[str, Any]:
162 163 c = snap.countries().get(r["country_id"], {"id": r["country_id"]})
163 164 return {"country": country_card(c), "value": clean_float(r["value"]), "formatted": format_value(clean_float(r["value"]), ind),
165 + "formatted_short": format_short(clean_float(r["value"]), ind),
164 166 "year": r["year"], "rank": rank, "provenance": provenance_from_row(snap, r, ind["id"], c.get("iso2"))}
165 167
166 168 top = [pack(r, i + 1) for i, r in enumerate(ordered[:n])]
modified src/countryatlas/api/routers/rankings.py +2 −2
@@ -20,7 +20,7 @@ from countryatlas.api.common import (
20 20 sparklines_for_countries,
21 21 )
22 22 from countryatlas.api.db import Snapshot, get_snapshot
23 −from countryatlas.api.formatting import format_change, format_value
23 +from countryatlas.api.formatting import format_change, format_short, format_value
24 24 from countryatlas.api.provenance import provenance_from_row
25 25 from countryatlas.api.routers.indicators import indicator_summary
26 26
@@ -134,7 +134,7 @@ def get_ranking(
134 134 ch10["formatted"] = format_change(ch10["abs"], ch10["pct"], ind)
135 135 out_rows.append({
136 136 "rank": int(r["rank_in_group"]), "rank_world": r.get("rank_world"), "n_world": r.get("n_world"), "pct_rank": clean_float(r.get("pct_rank")),
137 − "country": country_card(c), "value": v, "formatted": format_value(v, ind), "year": r["year"],
137 + "country": country_card(c), "value": v, "formatted": format_value(v, ind), "formatted_short": format_short(v, ind), "year": r["year"],
138 138 "change_1y": ch1, "change_10y": ch10, "sparkline": sparks.get(r["country_id"], []),
139 139 "provenance": provenance_from_row(snap, r, ind["id"], c.get("iso2")),
140 140 })
modified src/countryatlas/api/routers/regions.py +6 −3
@@ -17,7 +17,7 @@ from countryatlas.api.common import (
17 17 resolve_indicator,
18 18 )
19 19 from countryatlas.api.db import Snapshot, get_snapshot
20 −from countryatlas.api.formatting import format_value
20 +from countryatlas.api.formatting import format_short, format_value
21 21 from countryatlas.api.provenance import build_provenance
22 22 from countryatlas.registry import topics as registry_topics
23 23
@@ -62,6 +62,7 @@ def _group_stats(snap: Snapshot, members: list[str]) -> dict[str, Any]:
62 62 ind = merged_indicator(ind_row)
63 63 val = clean_float({"sum": r["total"], "median": r["med"], "weighted_mean": r["wmean"] or r["med"]}[agg])
64 64 out[iid] = {"indicator": indicator_card(ind), "kind": agg, "label": label, "value": val, "formatted": format_value(val, ind),
65 + "formatted_short": format_short(val, ind),
65 66 "n": r["n"], "year": r["year"]}
66 67 return out
67 68
@@ -117,7 +118,8 @@ def get_region(slug: str, indicator: str = Query("gdp-per-capita", description="
117 118 if r is None or r.get("value") is None:
118 119 vals[iid] = None
119 120 continue
120 − vals[iid] = {"value": clean_float(r["value"]), "formatted": format_value(clean_float(r["value"]), ind), "year": r["year"],
121 + vals[iid] = {"value": clean_float(r["value"]), "formatted": format_value(clean_float(r["value"]), ind),
122 + "formatted_short": format_short(clean_float(r["value"]), ind), "year": r["year"],
121 123 "rank_world": r.get("rank_world"), "n_world": r.get("n_world"),
122 124 "provenance": build_provenance(snap, iid, r["source_id"], r.get("source_dataset"), r.get("source_series_code"),
123 125 r.get("retrieved_at"), r.get("source_updated_at"), c.get("iso2"))}
@@ -141,7 +143,8 @@ def get_region(slug: str, indicator: str = Query("gdp-per-capita", description="
141 143 for i, r in enumerate(rows):
142 144 c = snap.countries().get(r["country_id"], {"id": r["country_id"]})
143 145 ranking_rows.append({"rank": i + 1, "country": country_card(c), "value": clean_float(r["value"]),
144 − "formatted": format_value(clean_float(r["value"]), ind), "year": r["year"], "rank_world": r.get("rank_world"),
146 + "formatted": format_value(clean_float(r["value"]), ind), "formatted_short": format_short(clean_float(r["value"]), ind),
147 + "year": r["year"], "rank_world": r.get("rank_world"),
145 148 "n_world": r.get("n_world"), "change_pct": clean_float(r.get("change_pct")), "change_abs": clean_float(r.get("change_abs")),
146 149 "provenance": build_provenance(snap, ind["id"], r["source_id"], r.get("source_dataset"), r.get("source_series_code"),
147 150 r.get("retrieved_at"), r.get("source_updated_at"), c.get("iso2"))})
modified src/countryatlas/pipeline/build.py +14 −0
@@ -32,6 +32,7 @@ SCHEMA_VERSION = 1
32 32 SCHEMA_SQL = Path(__file__).resolve().parents[1] / "storage" / "schema.sql"
33 33 MIN_OBSERVATIONS_WITH_WB = 100_000
34 34 HEADLINE_MIN_COUNTRIES = 100
35 +HEADLINE_MIN_N_WORLD = 50 # warning when a headline indicator's latest row is ranked in a pool smaller than this
35 36
36 37 # Public facts about sources whose connector module is not implemented yet (so `sources` is complete for the API).
37 38 SOURCE_DEFAULTS: dict[str, dict[str, str]] = {
@@ -370,6 +371,19 @@ def _integrity(con: duckdb.DuckDBPyConnection, files: list[Path], strict: bool)
370 371 problems.append(f"headline indicator {slug} has only {n} countries in latest (< {HEADLINE_MIN_COUNTRIES})")
371 372 if problems and strict:
372 373 raise IntegrityError("; ".join(problems))
374 + # soft checks (warnings only): frequency purity and thin ranking pools on headline indicators
375 + n_mismatch = con.execute(
376 + "SELECT count(*) FROM latest l JOIN indicators i ON i.id = l.indicator_id WHERE l.frequency <> i.frequency"
377 + ).fetchone()[0]
378 + if n_mismatch:
379 + problems.append(f"{n_mismatch} latest rows whose frequency differs from the indicator's canonical frequency")
380 + thin = con.execute(
381 + f"""SELECT indicator_id, min(n_world) FROM latest
382 + WHERE indicator_id IN (SELECT unnest($h::TEXT[])) AND n_world < {HEADLINE_MIN_N_WORLD} GROUP BY 1""",
383 + {"h": list(registry.topics()["headline"])},
384 + ).fetchall()
385 + for slug, n in thin:
386 + problems.append(f"headline indicator {slug} has latest rows ranked among only {n} countries (n_world < {HEADLINE_MIN_N_WORLD})")
373 387 return problems
374 388
375 389
modified src/countryatlas/pipeline/derived.py +13 −4
@@ -29,15 +29,19 @@ _ORDER = (
29 29
30 30
31 31 def create_base_views(con: duckdb.DuckDBPyConnection) -> None:
32 − """obs_ok = observations usable for derived tables (non-forecast, non-quarantined, finite value)."""
32 + """obs_ok = observations usable for derived tables: non-forecast, non-quarantined, finite value, and ONLY the
33 + indicator's canonical frequency (registry `frequency`). Higher-frequency series (e.g. FRED monthly inflation for the
34 + USA) stay in `observations` for series/chart endpoints but never enter latest/rankings/changes/events/similarity."""
33 35 con.execute(
34 36 """
35 37 CREATE OR REPLACE TEMP VIEW obs_ok AS
36 38 SELECT o.*
37 39 FROM observations o
40 + JOIN indicators i ON i.id = o.indicator_id
38 41 WHERE NOT coalesce(o.is_forecast, false)
39 42 AND coalesce(o.status, 'imported') <> 'quarantined'
40 43 AND o.value IS NOT NULL AND isfinite(o.value)
44 + AND o.frequency = i.frequency
41 45 """
42 46 )
43 47
@@ -74,7 +78,7 @@ def build_ranks(con: duckdb.DuckDBPyConnection) -> None:
74 78
75 79 def build_latest(con: duckdb.DuckDBPyConnection) -> int:
76 80 con.execute(
77 − """
81 + f"""
78 82 CREATE OR REPLACE TEMP TABLE lat AS
79 83 WITH ordered AS (
80 84 SELECT *,
@@ -92,12 +96,17 @@ def build_latest(con: duckdb.DuckDBPyConnection) -> int:
92 96 CASE WHEN l.prev_value IS NULL OR l.prev_value = 0 THEN NULL
93 97 ELSE (l.value - l.prev_value) / abs(l.prev_value) * 100 END AS change_pct,
94 98 r.rank_world, r.n_world, r.rank_region, r.n_region, r.rank_income, r.n_income,
95 − l.year AS rank_year, l.source_id, l.is_forecast, l.is_estimate, l.status,
99 + r.year AS rank_year, l.source_id, l.is_forecast, l.is_estimate, l.status,
96 100 t.value AS value_10y_ago,
97 101 l.value - t.value AS change_10y_abs,
98 102 CASE WHEN t.value IS NULL OR t.value = 0 THEN NULL ELSE (l.value - t.value) / abs(t.value) * 100 END
99 103 FROM lat l
100 − LEFT JOIN ranks_all r ON r.country_id = l.country_id AND r.indicator_id = l.indicator_id AND r.year = l.year
104 + -- ranks come from the most recent year (≤ latest) in which the country has a value AND ≥ MIN_COUNTRIES_FOR_RANKING
105 + -- countries are ranked; rank_year records that year (may be older than `year`); NULL when no such year exists
106 + LEFT JOIN (
107 + SELECT *, row_number() OVER (PARTITION BY country_id, indicator_id ORDER BY year DESC) AS rn
108 + FROM ranks_all WHERE n_world >= {MIN_COUNTRIES_FOR_RANKING}
109 + ) r ON r.country_id = l.country_id AND r.indicator_id = l.indicator_id AND r.rn = 1
101 110 LEFT JOIN obs_ok t ON t.country_id = l.country_id AND t.indicator_id = l.indicator_id
102 111 AND t.period = CAST(l.period - INTERVAL 10 YEAR AS DATE) AND t.frequency = l.frequency;
103 112 """
modified tests/api/test_countries.py +11 −3
@@ -41,14 +41,22 @@ def test_country_overview(get):
41 41 assert "g7" in {g["id"] for g in body["groups"]} and "world" in {g["id"] for g in body["groups"]}
42 42 assert body["coverage"]["n_indicators"] == 14
43 43 assert body["freshness"]["source_updated_at"].startswith("2026-07-01") and body["freshness"]["built_at"]
44 + from countryatlas.registry import topics as registry_topics
45 +
44 46 headline = {m["indicator"]: m for m in body["headline"]}
45 − assert list(headline) == ["population", "gdp", "gdp-per-capita", "gdp-growth", "inflation", "unemployment-rate", "life-expectancy",
46 − "median-age", "government-debt-pct-gdp", "co2-per-capita", "renewable-electricity-share", "internet-users"]
47 + assert list(headline) == registry_topics()["headline"] # read from topics.yaml at request time (cache cleared per snapshot)
48 + assert list(headline)[:3] == ["population", "gdp", "gdp-per-capita"]
49 + # indicators in the headline but absent from the snapshot still render as "no data"
50 + absent = [m for m in body["headline"] if m["indicator"] not in ("population", "gdp", "gdp-per-capita", "gdp-growth", "inflation",
51 + "unemployment-rate", "life-expectancy", "median-age", "co2-per-capita", "renewable-electricity-share", "internet-users",
52 + "government-debt-pct-gdp")]
53 + assert all(m["has_data"] is False and m["formatted"] == "—" and m["provenance"] is None for m in absent)
47 54 gpc = headline["gdp-per-capita"]
48 55 assert gpc["has_data"] and gpc["year"] == 2024 and gpc["value"] == pytest.approx(value("CAN", "gdp-per-capita", 2024))
49 56 assert gpc["change"]["pct"] == pytest.approx(3.0) and gpc["change"]["formatted"] == "+3.0 %"
50 57 assert gpc["rank_world"] and gpc["n_world"] == 8 and gpc["rank_region"] and gpc["n_region"] == 2
51 − assert gpc["formatted"].endswith("k")
58 + assert gpc["formatted"].startswith("US$") and gpc["formatted"].endswith("k") and gpc["formatted_short"] == gpc["formatted"][3:]
59 + assert headline["inflation"]["formatted"].endswith(" %") and headline["life-expectancy"]["formatted"].endswith(" yrs")
52 60 assert len(gpc["sparkline"]) == 30 and gpc["sparkline"][-1][0] == 2024 and isinstance(gpc["sparkline"][-1][0], int)
53 61 assert_provenance(gpc["provenance"])
54 62 assert gpc["provenance"]["url"] == "https://data.worldbank.org/indicator/NY.GDP.PCAP.CD?locations=CA"
modified tests/api/test_infra.py +24 −4
@@ -10,7 +10,8 @@ import duckdb
10 10 from fastapi.testclient import TestClient
11 11
12 12 from countryatlas.api.cache import ResponseCache
13 −from countryatlas.api.formatting import compact_number, format_change, format_value
13 +from countryatlas.api.db import invalidate_registry_caches
14 +from countryatlas.api.formatting import compact_number, format_change, format_short, format_value
14 15 from countryatlas.api.main import create_app
15 16 from countryatlas.api.provenance import source_url
16 17 from countryatlas.config import settings
@@ -170,9 +171,11 @@ def test_unknown_route_and_validation(client):
170 171
171 172
172 173 def test_formatting():
173 − assert format_value(53372.1, {"format": "currency"}) == "53.4k"
174 − assert format_value(1.23e12, {"format": "currency"}) == "1.2T"
175 − assert format_value(45.3e9, {"format": "currency"}) == "45.3B"
174 + assert format_value(53372.1, {"format": "currency", "unit_short": "US$"}) == "US$53.4k"
175 + assert format_value(1.23e12, {"format": "currency"}) == "US$1.2T" # default prefix
176 + assert format_value(45.3e9, {"format": "currency", "unit_short": "intl $"}) == "intl $45.3B"
177 + assert format_short(53372.1, {"format": "currency", "unit_short": "US$"}) == "53.4k"
178 + assert format_short(3.44, {"format": "percent"}) == "3.4"
176 179 assert format_value(3.44, {"format": "percent", "precision": 1}) == "3.4 %"
177 180 assert format_value(82.13, {"format": "years"}) == "82.1 yrs"
178 181 assert format_value(5.234, {"format": "tonnes", "precision": 2}) == "5.23 t"
@@ -186,6 +189,23 @@ def test_formatting():
186 189 assert format_change(-100.0, -3.4, {"format": "currency"}) == "−3.4 %"
187 190
188 191
192 +def test_registry_caches_cleared_on_snapshot_open(tmp_path: Path, fixture_db: Path):
193 + from countryatlas import registry
194 +
195 + registry.topics()
196 + assert registry.topics.cache_info().currsize == 1
197 + invalidate_registry_caches()
198 + assert registry.topics.cache_info().currsize == 0 and registry.indicators.cache_info().currsize == 0
199 + # opening a new snapshot (new run_id) clears the loaders again
200 + registry.topics()
201 + db_path = tmp_path / "atlas.duckdb"
202 + shutil.copy(fixture_db, db_path)
203 + app = create_app(db_path, rate_limit_per_minute=0, cache=ResponseCache())
204 + with TestClient(app) as c:
205 + assert c.get("/api/v1/health").json()["status"] == "ok"
206 + assert registry.topics.cache_info().currsize == 0
207 +
208 +
189 209 def test_source_urls():
190 210 assert source_url("worldbank", "WDI", "NY.GDP.PCAP.CD", "CA") == "https://data.worldbank.org/indicator/NY.GDP.PCAP.CD?locations=CA"
191 211 assert source_url("owid", "grapher", "median-age") == "https://ourworldindata.org/grapher/median-age"
modified tests/api/test_misc_routers.py +8 −4
@@ -59,15 +59,19 @@ def test_home(get):
59 59 snap = body["snapshot"]
60 60 assert snap["n_countries"] == 8 and snap["n_indicators"] == 14 and snap["n_observations"] > 3000 and snap["built_at"]
61 61 assert snap["world_population"] == pytest.approx(sum(value(c, "population", 2024) for c in COUNTRIES))
62 − assert snap["world_gdp_formatted"].endswith("T") and snap["median_life_expectancy"] > 60
62 + assert snap["world_gdp_formatted"].startswith("US$") and snap["world_gdp_formatted"].endswith("T") and snap["median_life_expectancy"] > 60
63 63 lists = body["lists"]
64 − assert set(lists) == {"largest_economies", "fastest_population_growth", "highest_life_expectancy", "energy_transition_leaders",
65 − "highest_gdp_per_capita_ppp", "lowest_unemployment"}
64 + assert set(lists) == {"largest_economies", "fastest_gdp_growth", "fastest_population_growth", "highest_life_expectancy",
65 + "energy_transition_leaders", "highest_gdp_per_capita_ppp", "lowest_unemployment"}
66 66 le = lists["largest_economies"]
67 67 assert le["rows"][0]["country"]["id"] == "USA" and le["rows"][0]["rank"] == 1 and len(le["rows"]) == 8
68 + assert le["filter_note"] is None and le["rows"][0]["formatted"].startswith("US$") and le["rows"][0]["formatted_short"]
68 69 assert_provenance(le["rows"][0]["provenance"])
69 − assert lists["lowest_unemployment"]["rows"][0]["country"]["id"] == "JPN"
70 + assert lists["lowest_unemployment"]["rows"][0]["country"]["id"] == "JPN" and lists["lowest_unemployment"]["filter_note"] == "Countries above 5M inhabitants"
70 71 assert lists["energy_transition_leaders"]["rows"][0]["country"]["id"] == "BRA"
72 + for k in ("fastest_gdp_growth", "fastest_population_growth", "highest_life_expectancy", "energy_transition_leaders"):
73 + assert lists[k]["filter_note"] == "Countries above 1M inhabitants" and lists[k]["min_population"] == 1_000_000
74 + assert lists["fastest_gdp_growth"]["indicator"]["id"] == "gdp-growth" and lists["fastest_gdp_growth"]["rows"][0]["year"] == 2024
71 75 assert len(body["recent_changes"]) == 12 and body["recent_changes"][0]["country"]["id"] and body["recent_changes"][0]["headline"]
72 76 assert body["recently_updated"] and body["featured_indicators"] and body["trending"]
73 77 assert all(i["featured"] for i in body["featured_indicators"])
modified tests/test_build.py +30 −1
@@ -52,6 +52,26 @@ def _tiny_staging() -> None:
52 52 _stage(ma, "years", 25, 1.005, range(2000, 2024))
53 53 fr = IndicatorSourceSpec(indicator_id="fertility-rate", connector="worldbank", dataset="WDI", code="SP.DYN.TFRT.IN")
54 54 _stage(fr, "births per woman", 3.0, 0.99, range(2000, 2024))
55 + # annual inflation for everyone + a MONTHLY FRED series for the USA (must stay out of latest/rankings)
56 + infl = IndicatorSourceSpec(indicator_id="inflation", connector="worldbank", dataset="WDI", code="FP.CPI.TOTL.ZG")
57 + _stage(infl, "annual %", 2.0, 1.01, range(2000, 2026))
58 + # CAN alone already has 2026 → its rank must fall back to 2025 (the last year with ≥ 20 ranked countries)
59 + import polars as pl
60 +
61 + p = spec_paths(infl)["parquet"]
62 + df = pl.read_parquet(p)
63 + extra = df.filter((pl.col("country_id") == "CAN") & (pl.col("year") == 2025)).with_columns(
64 + pl.lit(date(2026, 1, 1)).alias("period"), pl.lit(2026, dtype=pl.Int32).alias("year"), (pl.col("value") + 1.0).alias("value"))
65 + write_parquet_atomic(pl.concat([df, extra]), p)
66 + fred = IndicatorSourceSpec(indicator_id="inflation", connector="fred", dataset="FRED", code="CPIAUCSL", priority=2,
67 + countries=["USA"], frequency="M")
68 + now = datetime.now(UTC)
69 + monthly = [NormalizedObservation(country_id="USA", indicator_id="inflation", period=date(2026, m, 1), year=2026, frequency="M",
70 + value=3.0 + m / 10, unit="annual %", source_id="fred", source_dataset="FRED",
71 + source_series_code="CPIAUCSL", retrieved_at=now) for m in range(1, 8)]
72 + write_parquet_atomic(rows_to_frame(monthly), spec_paths(fred)["parquet"])
73 + write_run(fred, ImportRun(run_id="T1", connector="fred", dataset="FRED", started_at=now, finished_at=now, status="ok",
74 + rows_norm=len(monthly), rows_valid=len(monthly)))
55 75
56 76
57 77 @pytest.mark.usefixtures("data_dir")
@@ -94,8 +114,17 @@ def test_build_tiny_staging_produces_all_tables() -> None:
94 114 assert con.execute("SELECT count(*) FROM search_index WHERE type='country'").fetchone()[0] >= 200
95 115 assert con.execute("SELECT count(*) FROM insights").fetchone()[0] > 0
96 116 assert con.execute("SELECT count(*) FROM country_dna").fetchone()[0] > 0
97 − assert con.execute("SELECT count(*) FROM import_runs").fetchone()[0] == 6
117 + assert con.execute("SELECT count(*) FROM import_runs").fetchone()[0] == 8
98 118 assert r.counts["series_fresher_source"] == 1
119 + # canonical frequency only: the USA monthly FRED series is in observations but not in latest/rankings
120 + assert con.execute("SELECT count(*) FROM observations WHERE frequency='M'").fetchone()[0] == 7
121 + usa = con.execute("SELECT year, frequency, source_id, n_world FROM latest WHERE country_id='USA' AND indicator_id='inflation'").fetchone()
122 + assert usa == (2025, "A", "worldbank", len(COUNTRIES))
123 + assert con.execute("SELECT count(*) FROM latest l JOIN indicators i ON i.id=l.indicator_id WHERE l.frequency<>i.frequency").fetchone()[0] == 0
124 + assert con.execute("SELECT count(*) FROM rankings WHERE indicator_id='inflation' AND year=2026").fetchone()[0] == 0
125 + # rank fallback: CAN's latest inflation is 2026 (only country) → ranked within 2025 among all 24
126 + can = con.execute("SELECT year, rank_year, n_world FROM latest WHERE country_id='CAN' AND indicator_id='inflation'").fetchone()
127 + assert can[0] == 2026 and can[1] == 2025 and can[2] == len(COUNTRIES)
99 128 meta = dict(con.execute("SELECT key, value FROM meta").fetchall())
100 129 assert meta["schema_version"] == "1" and meta["build_run_id"] == "20260101T000000Z"
101 130 assert int(meta["observation_count"]) == n_obs
102 131