"""Build a small synthetic DuckDB snapshot for API tests (NEVER shipped). 8 countries × 12 indicators × 1990–2024, values follow obvious deterministic formulas (so any test can recompute them), plus IMF forecast rows (2025–2026), alternative-source rows, derived tables (latest, rankings, changes, events, similarity, insights, country_dna, coverage), import_runs, validation_issues, search_index and meta — all from `schema.sql`. Usage: `python tests/fixtures/make_fixture_db.py /tmp/atlas-fixture.duckdb` """ from __future__ import annotations import json import math import sys from datetime import date, datetime from pathlib import Path import duckdb ROOT = Path(__file__).resolve().parents[2] SCHEMA = ROOT / "src" / "countryatlas" / "storage" / "schema.sql" COUNTRIES = ["CAN", "USA", "FRA", "DEU", "JPN", "IND", "BRA", "NGA"] INDICATORS = ["population", "gdp", "gdp-per-capita", "gdp-growth", "inflation", "unemployment-rate", "life-expectancy", "median-age", "government-debt-pct-gdp", "co2-per-capita", "renewable-electricity-share", "internet-users", "gdp-per-capita-ppp", "population-growth"] YEARS = list(range(1990, 2025)) RUN_ID = "fixture-20260911T000000" BUILT_AT = "2026-09-11T00:00:00Z" RETRIEVED = datetime(2026, 9, 10, 3, 20, 11) SOURCE_UPDATED = datetime(2026, 7, 1, 0, 0, 0) # synthetic anchors (1990 population in millions, gdp per capita 1990 in US$, life expectancy 1990) ANCHOR = { "CAN": dict(pop=27.7, gpc=21000, le=77.4, growth=1.0, base_unemp=8.0, ren=60, inet_year=1996), "USA": dict(pop=249.6, gpc=23900, le=75.2, growth=1.0, base_unemp=5.6, ren=11, inet_year=1995), "FRA": dict(pop=58.0, gpc=21800, le=76.7, growth=0.5, base_unemp=9.0, ren=15, inet_year=1997), "DEU": dict(pop=79.4, gpc=22200, le=75.3, growth=0.2, base_unemp=6.5, ren=4, inet_year=1997), "JPN": dict(pop=123.5, gpc=25400, le=78.8, growth=0.1, base_unemp=2.1, ren=12, inet_year=1996), "IND": dict(pop=870.5, gpc=370, le=57.9, growth=1.8, base_unemp=5.5, ren=25, inet_year=2002), "BRA": dict(pop=150.7, gpc=3100, le=65.3, growth=1.4, base_unemp=7.0, ren=93, inet_year=2000), "NGA": dict(pop=95.2, gpc=560, le=45.8, growth=2.6, base_unemp=4.0, ren=20, inet_year=2005), } def value(country: str, indicator: str, year: int) -> float | None: a = ANCHOR[country] t = year - 1990 pop = a["pop"] * 1e6 * (1 + a["growth"] / 100) ** t gpc = a["gpc"] * (1.03 if country in ("CAN", "USA", "FRA", "DEU", "JPN") else 1.06) ** t if indicator == "population": return round(pop) if indicator == "population-growth": return a["growth"] - 0.01 * t if indicator == "gdp": return pop * gpc if indicator == "gdp-per-capita": return gpc if indicator == "gdp-per-capita-ppp": return gpc * (1.1 if country in ("CAN", "USA", "FRA", "DEU", "JPN") else 3.2) if indicator == "gdp-growth": base = 2.0 if country in ("CAN", "USA", "FRA", "DEU", "JPN") else 5.0 if year == 2009: return base - 5.0 # global recession → sign flip if year == 2020: return base - 8.0 return base + 0.8 * math.sin(t / 2.0) if indicator == "inflation": if year == 2022: return 8.0 if country != "JPN" else 2.5 return 2.0 + 0.5 * math.cos(t / 3.0) + (3.0 if country in ("IND", "BRA", "NGA") else 0) if indicator == "unemployment-rate": return a["base_unemp"] + (3.0 if year == 2020 else 0) + 0.5 * math.sin(t / 4.0) if indicator == "life-expectancy": return a["le"] + 0.22 * t - (0.8 if year in (2020, 2021) else 0) if indicator == "median-age": return (28 if country in ("IND", "NGA", "BRA") else 33) + 0.3 * t - (8 if country == "NGA" else 0) if indicator == "government-debt-pct-gdp": if year < 1995: return None # deliberately missing early years return 40 + 1.2 * t + (60 if country == "JPN" else 0) + (10 if year >= 2020 else 0) if indicator == "co2-per-capita": base = {"CAN": 16, "USA": 20, "FRA": 6.5, "DEU": 11, "JPN": 9, "IND": 0.7, "BRA": 1.5, "NGA": 0.4}[country] return base * (0.99 ** t if base > 5 else 1.02 ** t) if indicator == "renewable-electricity-share": return min(99.0, a["ren"] + 0.6 * t) if indicator == "internet-users": if year < a["inet_year"]: return None return min(98.0, 100 / (1 + math.exp(-(year - a["inet_year"] - 8) / 2.5))) raise KeyError(indicator) def build(path: Path) -> Path: from countryatlas.registry import countries as reg_countries from countryatlas.registry import groups as reg_groups from countryatlas.registry import indicators as reg_indicators from countryatlas.registry import topics as reg_topics path = Path(path) if path.exists(): path.unlink() con = duckdb.connect(str(path)) con.execute(SCHEMA.read_text()) # ---- countries cs = [c for c in reg_countries() if c.id in COUNTRIES] cols = ["id", "iso2", "iso3", "iso_numeric", "slug", "short_name", "official_name", "capital", "continent", "region_wb", "region_wb_name", "subregion", "income_group", "income_group_name", "currency_code", "currency_name", "area_km2", "latitude", "longitude", "flag_emoji", "un_member", "independent", "landlocked", "borders", "languages", "demonym", "status", "kind"] con.executemany(f"INSERT INTO countries ({', '.join(cols)}) VALUES ({', '.join('?' * len(cols))})", [[getattr(c, k) for k in cols] for c in cs]) # ---- groups + members (restricted to fixture countries) members_rows = [] for g in reg_groups(): m = [x for x in g.members if x in COUNTRIES] con.execute("INSERT INTO groups VALUES (?, ?, ?, ?, ?, ?, ?)", [g.id, g.slug, g.name, g.kind, g.description, g.wb_code, len(m)]) members_rows += [[g.id, x] for x in m] con.executemany("INSERT INTO group_members VALUES (?, ?)", members_rows) # ---- sources sources = [ ("worldbank", "World Bank", "World Bank Group", "https://data.worldbank.org/", "CC BY 4.0", "World Development Indicators, The World Bank", "https://api.worldbank.org/v2"), ("imf", "IMF", "International Monetary Fund", "https://data.imf.org/", "IMF Terms of Use", "World Economic Outlook, IMF", "https://www.imf.org/external/datamapper/api/v1"), ("owid", "Our World in Data", "Global Change Data Lab", "https://ourworldindata.org/", "CC BY 4.0", "Our World in Data", "https://ourworldindata.org/grapher"), ("who", "WHO", "World Health Organization", "https://www.who.int/data/gho", "CC BY-NC-SA 3.0 IGO", "WHO Global Health Observatory", "https://ghoapi.azureedge.net/api"), ] con.executemany("INSERT INTO sources (id, name, organization, url, licence, attribution, api_base, last_success_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", [[*s, RETRIEVED] for s in sources]) # ---- indicators + indicator_sources inds = {i.slug: i for i in reg_indicators() if i.slug in INDICATORS} for slug in INDICATORS: i = inds[slug] con.execute( """INSERT INTO indicators (id, slug, name, short_name, description, topic, subtopic, unit, unit_short, frequency, precision, aggregation, higher_is_better, ranking_eligible, featured, format, scale, bounds_min, bounds_max, methodology, tags, per_capita_of) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", [i.slug, i.slug, i.name, i.short_name, i.description, i.topic, i.subtopic, i.unit, i.unit_short, i.frequency, i.precision, i.aggregation, i.higher_is_better, i.ranking_eligible, i.featured, i.format, i.scale, i.bounds[0], i.bounds[1] if len(i.bounds) > 1 else None, i.methodology, i.tags, i.per_capita_of]) for s in i.sources: if s.connector not in {x[0] for x in sources}: continue con.execute("INSERT INTO indicator_sources (indicator_id, source_id, dataset, series_code, params, priority, transform, countries, notes) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", [i.slug, s.connector, s.dataset, s.code, json.dumps(s.params), s.priority, s.transform, s.countries, s.notes]) # ---- observations (primary source = first registry source; alt = second when present) obs_rows, alt_rows = [], [] for slug in INDICATORS: i = inds[slug] primary = i.sources[0] alt = i.sources[1] if len(i.sources) > 1 and i.sources[1].connector in {x[0] for x in sources} else None for c in COUNTRIES: for y in YEARS: v = value(c, slug, y) if v is None: continue status = "verified" if slug == "inflation" and y == 2022: status = "warning" obs_rows.append([c, slug, date(y, 1, 1), y, "A", v, i.unit, primary.connector, primary.dataset, primary.code, False, False, 0, RETRIEVED, SOURCE_UPDATED, status, None]) if alt is not None: alt_rows.append([c, slug, date(y, 1, 1), y, "A", v * 1.01, i.unit, alt.connector, alt.dataset, alt.code, False, False, 0, RETRIEVED, SOURCE_UPDATED, "imported", None]) # IMF forecasts for gdp / gdp-per-capita / gdp-growth / inflation if alt is not None and alt.connector == "imf": for y in (2025, 2026): v = value(c, slug, 2024) * (1.03 ** (y - 2024)) if slug != "gdp-growth" else 2.2 obs_rows.append([c, slug, date(y, 1, 1), y, "A", v, i.unit, "imf", "WEO", alt.code, True, True, 0, RETRIEVED, SOURCE_UPDATED, "imported", json.dumps({"forecast": True})]) ins = "INSERT INTO {t} VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" con.executemany(ins.format(t="observations"), sorted(obs_rows, key=lambda r: (r[1], r[0], r[2]))) con.executemany(ins.format(t="observations_alt"), alt_rows) # ---- latest (last non-forecast value; ranks within fixture countries) con.execute(""" INSERT INTO latest WITH nf AS (SELECT * FROM observations WHERE NOT is_forecast AND value IS NOT NULL), cur AS (SELECT * FROM nf QUALIFY row_number() OVER (PARTITION BY country_id, indicator_id ORDER BY period DESC) = 1), prev AS (SELECT * FROM nf QUALIFY row_number() OVER (PARTITION BY country_id, indicator_id ORDER BY period DESC) = 2), ten AS (SELECT n.country_id, n.indicator_id, n.value FROM nf n JOIN cur ON cur.country_id = n.country_id AND cur.indicator_id = n.indicator_id AND n.year = cur.year - 10), ranked AS ( SELECT cur.*, c.region_wb, c.income_group, rank() OVER (PARTITION BY cur.indicator_id, cur.year ORDER BY CASE WHEN i.higher_is_better = false THEN cur.value END ASC, CASE WHEN i.higher_is_better = false THEN NULL ELSE cur.value END DESC) AS rw, count(*) OVER (PARTITION BY cur.indicator_id, cur.year) AS nw, rank() OVER (PARTITION BY cur.indicator_id, cur.year, c.region_wb ORDER BY CASE WHEN i.higher_is_better = false THEN cur.value END ASC, CASE WHEN i.higher_is_better = false THEN NULL ELSE cur.value END DESC) AS rr, count(*) OVER (PARTITION BY cur.indicator_id, cur.year, c.region_wb) AS nr, rank() OVER (PARTITION BY cur.indicator_id, cur.year, c.income_group ORDER BY CASE WHEN i.higher_is_better = false THEN cur.value END ASC, CASE WHEN i.higher_is_better = false THEN NULL ELSE cur.value END DESC) AS ri, count(*) OVER (PARTITION BY cur.indicator_id, cur.year, c.income_group) AS ni FROM cur JOIN countries c ON c.id = cur.country_id JOIN indicators i ON i.id = cur.indicator_id) SELECT r.country_id, r.indicator_id, r.period, r.year, r.frequency, r.value, p.period, p.value, r.value - p.value, CASE WHEN p.value <> 0 THEN (r.value - p.value) / abs(p.value) * 100 END, r.rw, r.nw, r.rr, r.nr, r.ri, r.ni, r.year, r.source_id, r.is_forecast, r.is_estimate, r.status, t.value, r.value - t.value, CASE WHEN t.value <> 0 THEN (r.value - t.value) / abs(t.value) * 100 END FROM ranked r LEFT JOIN prev p ON p.country_id = r.country_id AND p.indicator_id = r.indicator_id LEFT JOIN ten t ON t.country_id = r.country_id AND t.indicator_id = r.indicator_id """) # ---- rankings (every year, ranking-eligible indicators) con.execute(""" INSERT INTO rankings SELECT o.indicator_id, o.year, o.country_id, o.value, rank() OVER (PARTITION BY o.indicator_id, o.year ORDER BY CASE WHEN i.higher_is_better = false THEN o.value END ASC, CASE WHEN i.higher_is_better = false THEN NULL ELSE o.value END DESC) AS rank, count(*) OVER (PARTITION BY o.indicator_id, o.year) AS n, percent_rank() OVER (PARTITION BY o.indicator_id, o.year ORDER BY o.value) AS pct_rank FROM observations o JOIN indicators i ON i.id = o.indicator_id WHERE NOT o.is_forecast AND o.value IS NOT NULL AND i.ranking_eligible """) # ---- changes / events (simple deterministic detectors on the synthetic series) now = datetime(2026, 9, 11, 0, 0, 0) changes, events = [], [] for c in COUNTRIES: for slug in INDICATORS: series = [(y, value(c, slug, y)) for y in YEARS if value(c, slug, y) is not None] if len(series) < 3: continue vals = [v for _, v in series] y, v = series[-1] py, pv = series[-2] d = v - pv dp = d / abs(pv) * 100 if pv else None if v == max(vals): changes.append([f"{c}:{slug}:{y}:record_high", c, slug, "record_high", date(y, 1, 1), y, v, max(vals[:-1]), d, dp, len(vals), 0.6, f"{c} reached a record high for {slug} in {y}.", json.dumps({"n_years": len(vals)}), now]) if slug in ("inflation", "unemployment-rate", "gdp-growth") and abs(d) >= 1.0: sev = min(1.0, abs(d) / 5.0) changes.append([f"{c}:{slug}:{y}:yoy", c, slug, "yoy_jump" if d > 0 else "yoy_drop", date(y, 1, 1), y, v, pv, d, dp, 1, sev, f"{slug} moved {d:+.1f} pts in {c} in {y}.", json.dumps({"floor": 1.0}), now]) for (y0, v0), (y1, v1) in zip(series, series[1:]): if slug == "gdp-growth" and (v0 > 0 > v1 or v0 < 0 < v1): events.append([f"{c}:{slug}:{y1}:sign_flip", c, slug, "sign_flip", date(y1, 1, 1), y1, v1, v0, v1 - v0, None, 1, 0.8, f"{c}'s GDP growth turned {'negative' if v1 < 0 else 'positive'} in {y1}.", None]) if slug == "inflation" and v1 - v0 > 3: events.append([f"{c}:{slug}:{y1}:yoy_jump", c, slug, "yoy_jump", date(y1, 1, 1), y1, v1, v0, v1 - v0, None, 1, 0.7, f"Inflation jumped {v1 - v0:+.1f} pts in {c} in {y1}.", None]) con.executemany("INSERT INTO changes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", changes) con.executemany("INSERT INTO events VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", events) # ---- similarity (distance on log gdp pc + life expectancy) feats = {c: (math.log(value(c, "gdp-per-capita", 2024)), value(c, "life-expectancy", 2024) / 10) for c in COUNTRIES} sim_rows = [] for mode in ("overall", "economic", "demographic"): for c in COUNTRIES: peers = [] for p in COUNTRIES: if p == c: continue d = math.dist(feats[c], feats[p]) peers.append((p, 100 * math.exp(-d / 1.5), d)) peers.sort(key=lambda x: -x[1]) for rank, (p, score, d) in enumerate(peers, 1): contrib = {"gdp-per-capita": {"z_a": feats[c][0], "z_b": feats[p][0], "weight": 1, "contribution": abs(feats[c][0] - feats[p][0]) / d if d else 0}, "life-expectancy": {"z_a": feats[c][1], "z_b": feats[p][1], "weight": 1, "contribution": abs(feats[c][1] - feats[p][1]) / d if d else 0}} sim_rows.append([c, mode, p, score, rank, json.dumps(contrib)]) con.executemany("INSERT INTO similarity VALUES (?, ?, ?, ?, ?, ?)", sim_rows) # ---- insights ins_rows = [] for c in COUNTRIES: p0, p1 = value(c, "population", 1990), value(c, "population", 2024) pct = (p1 / p0 - 1) * 100 ins_rows.append([f"{c}:pop_growth_since", c, "pop_growth_since", f"{c}'s population grew {pct:.0f}% since 1990.", json.dumps({"pct": pct, "y0": 1990}), ["population"], now]) le = value(c, "life-expectancy", 2024) ins_rows.append([f"{c}:life_expectancy", c, "life_expectancy_level", f"Life expectancy in {c} is {le:.1f} years.", json.dumps({"value": le}), ["life-expectancy"], now]) con.executemany("INSERT INTO insights VALUES (?, ?, ?, ?, ?, ?, ?)", ins_rows) # ---- country_dna (percentile ranks) def pct_rank(slug: str, c: str) -> float: vals = sorted(value(x, slug, 2024) for x in COUNTRIES) return 100.0 * vals.index(value(c, slug, 2024)) / (len(vals) - 1) for c in COUNTRIES: dims = {"income": pct_rank("gdp-per-capita-ppp", c), "demographics": pct_rank("median-age", c), "emissions": pct_rank("co2-per-capita", c), "energy": pct_rank("renewable-electricity-share", c), "public_spending": pct_rank("government-debt-pct-gdp", c), "urbanization": None, "trade": None, "innovation": None, "education": pct_rank("internet-users", c)} con.execute("INSERT INTO country_dna VALUES (?, ?, ?)", [c, json.dumps(dims), 2024]) # ---- coverage con.execute(""" INSERT INTO coverage SELECT country_id, count(DISTINCT indicator_id), count(*), max(year), 100.0 * count(DISTINCT indicator_id) / (SELECT count(*) FROM indicators), ? FROM observations WHERE NOT is_forecast GROUP BY country_id""", [now]) # ---- indicator coverage columns + sources counts con.execute(""" UPDATE indicators SET n_countries = (SELECT count(DISTINCT country_id) FROM observations o WHERE o.indicator_id = indicators.id AND NOT is_forecast), n_observations = (SELECT count(*) FROM observations o WHERE o.indicator_id = indicators.id AND NOT is_forecast), first_year = (SELECT min(year) FROM observations o WHERE o.indicator_id = indicators.id), last_year = (SELECT max(year) FROM observations o WHERE o.indicator_id = indicators.id AND NOT is_forecast), latest_source_updated_at = (SELECT max(source_updated_at) FROM observations o WHERE o.indicator_id = indicators.id), primary_source_id = (SELECT source_id FROM observations o WHERE o.indicator_id = indicators.id GROUP BY source_id ORDER BY count(*) DESC LIMIT 1) """) con.execute(""" UPDATE indicator_sources SET n_observations = (SELECT count(*) FROM observations o WHERE o.indicator_id = indicator_sources.indicator_id AND o.source_id = indicator_sources.source_id), n_countries = (SELECT count(DISTINCT country_id) FROM observations o WHERE o.indicator_id = indicator_sources.indicator_id AND o.source_id = indicator_sources.source_id), last_year = (SELECT max(year) FROM observations o WHERE o.indicator_id = indicator_sources.indicator_id AND o.source_id = indicator_sources.source_id), last_run_id = ?, last_status = 'ok'""", [RUN_ID]) con.execute("UPDATE sources SET n_indicators = (SELECT count(DISTINCT indicator_id) FROM observations o WHERE o.source_id = sources.id), " "n_observations = (SELECT count(*) FROM observations o WHERE o.source_id = sources.id)") # ---- import runs + validation issues runs = [ [RUN_ID, "worldbank", "WDI", datetime(2026, 9, 10, 3, 15), datetime(2026, 9, 10, 3, 20), "ok", 12000, 11800, 11790, 3, 0, None, "raw/worldbank/WDI/2026-09-10"], [RUN_ID, "imf", "WEO", datetime(2026, 9, 10, 3, 20), datetime(2026, 9, 10, 3, 22), "ok", 4000, 3900, 3900, 0, 0, None, "raw/imf/WEO/2026-09-10"], [RUN_ID, "owid", "co2", datetime(2026, 9, 10, 3, 22), datetime(2026, 9, 10, 3, 23), "ok", 3000, 2900, 2900, 0, 0, None, "raw/owid/co2/2026-09-10"], [RUN_ID, "owid", "energy", datetime(2026, 9, 10, 3, 23), datetime(2026, 9, 10, 3, 24), "partial", 3000, 2000, 2000, 5, 0, "2 pages failed", "raw/owid/energy/2026-09-10"], [RUN_ID, "who", "GHO", datetime(2026, 9, 10, 3, 24), datetime(2026, 9, 10, 3, 25), "failed", 0, 0, 0, 0, 1, "HTTP 503", None], ["fixture-20260910T000000", "worldbank", "WDI", datetime(2026, 9, 9, 3, 15), datetime(2026, 9, 9, 3, 20), "ok", 12000, 11800, 11790, 1, 0, None, "raw/worldbank/WDI/2026-09-09"], ] con.executemany("INSERT INTO import_runs VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", runs) con.executemany("INSERT INTO validation_issues VALUES (?, ?, ?, ?, ?, ?, ?, ?)", [ [RUN_ID, "worldbank", "inflation", "NGA", date(2022, 1, 1), "warning", "extreme_jump", "|Δ| = 6.0 > 4 × MAD"], [RUN_ID, "worldbank", "inflation", "BRA", date(2022, 1, 1), "warning", "extreme_jump", "|Δ| = 6.0 > 4 × MAD"], [RUN_ID, "owid", "renewable-electricity-share", None, None, "info", "partial_download", "2 pages failed, previous kept"], [RUN_ID, "who", "life-expectancy", None, None, "error", "schema_change", "HTTP 503"], ]) # ---- search index si = [] for c in cs: si.append(["country", c.id, c.slug, c.short_name, f"{c.official_name} {c.iso2} {c.iso3} {c.capital}", f"Country · {c.region_wb_name}", 1.0]) tnames = {t["id"]: t["name"] for t in reg_topics()["topics"]} for slug in INDICATORS: i = inds[slug] si.append(["indicator", slug, slug, i.name, f"{i.short_name or ''} {' '.join(i.tags)}", f"Indicator · {tnames.get(i.topic, i.topic)} · {i.unit}", 0.9 if i.featured else 0.7]) for t in reg_topics()["topics"]: si.append(["topic", t["id"], t["id"], t["name"], t.get("short", ""), f"Topic · {len(t['indicators'])} indicators", 0.6]) for g in reg_groups(): si.append(["region", g.id, g.slug, g.name, g.wb_code or "", f"Region · {g.kind}", 0.6]) for s in sources: si.append(["source", s[0], s[0], s[1], s[2], "Source", 0.4]) con.executemany("INSERT INTO search_index VALUES (?, ?, ?, ?, ?, ?, ?)", si) # ---- meta n_obs = con.execute("SELECT count(*) FROM observations").fetchone()[0] con.executemany("INSERT INTO meta VALUES (?, ?)", [ ["build_run_id", RUN_ID], ["built_at", BUILT_AT], ["schema_version", "1"], ["indicator_count", str(len(INDICATORS))], ["country_count", str(len(COUNTRIES))], ["observation_count", str(n_obs)], ["fixture", "true"], ]) con.close() return path if __name__ == "__main__": out = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("/tmp/atlas-fixture.duckdb") print(build(out))