spb/countryatlas

1from __future__ import annotations23from datetime import UTC, date, datetime45import numpy as np6import polars as pl7import pytest89from countryatlas.pipeline.changes import Series, compute_changes_and_events, detect_changes, detect_events10from countryatlas.registry import indicators_by_id111213def _series(ind: str, values: list[float], start: int = 2000, country: str = "CAN") -> Series:14    years = np.arange(start, start + len(values), dtype=np.int32)15    return Series(country, ind, [date(int(y), 1, 1) for y in years], years, np.array(values, dtype=float))161718def test_inflation_drop_headline_and_since() -> None:19    ind = indicators_by_id()["inflation"]20    vals = [2.0, 2.1, 1.9, 2.2, 2.0, 2.3, 8.1, 5.9, 5.5, 3.4]  # 2015: 5.5 → 2016: 3.4 (−2.1 pts) hmm: last drop 2.121    vals = [2.0, 2.1, 1.9, 2.2, 2.0, 2.3, 2.1, 2.0, 6.8, 3.4]22    out = detect_changes(_series("inflation", vals, 2016), ind)23    kinds = {r["kind"]: r for r in out}24    assert "yoy_drop" in kinds25    h = kinds["yoy_drop"]["headline"]26    assert h.startswith("Inflation fell 3.4 points to 3.4 % in 2025")27    assert "largest drop" in h28    assert 0 < kinds["yoy_drop"]["severity"] <= 1293031def test_floor_blocks_small_moves() -> None:32    ind = indicators_by_id()["inflation"]  # change_floor 2 points33    vals = [2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 3.0]  # +1 pt: huge z but below the floor34    out = detect_changes(_series("inflation", vals), ind)35    assert not any(r["kind"] in ("yoy_jump", "yoy_drop") for r in out)363738def test_record_high_and_relative_headline() -> None:39    ind = indicators_by_id()["gdp-per-capita"]40    vals = [1000 * 1.03**k for k in range(15)]41    vals[5] *= 0.9  # a dip: the series must not be monotone, otherwise records are (rightly) not newsworthy42    vals[-1] = vals[-2] * 1.2543    out = detect_changes(_series("gdp-per-capita", vals), ind)44    kinds = {r["kind"] for r in out}45    assert "record_high" in kinds and "yoy_jump" in kinds46    jump = next(r for r in out if r["kind"] == "yoy_jump")47    assert "25.0 %" in jump["headline"] and "US$" in jump["headline"]484950def test_sign_flip_and_events_history() -> None:51    ind = indicators_by_id()["gdp-growth"]52    vals = [3.0, 2.5, 3.1, 2.8, 3.0, 2.9, -4.5, 5.0, 2.0, -1.0]53    ch = detect_changes(_series("gdp-growth", vals), ind)54    assert any(r["kind"] == "sign_flip" for r in ch)55    ev = detect_events(_series("gdp-growth", vals), ind)56    flips = [r for r in ev if r["kind"] == "sign_flip"]57    assert len(flips) == 3  # 2006, 2007, 200958    assert any(r["kind"] == "yoy_drop" and r["year"] == 2006 for r in ev)596061def test_monotone_and_cumulative_series_are_silent() -> None:62    inds = indicators_by_id()63    vals = [100.0 * 1.02**k for k in range(30)]  # strictly increasing: every point is a "record"64    ch = detect_changes(_series("population", vals), inds["population"])65    assert not any(r["kind"] in ("record_high", "n_year_high") for r in ch)66    rows = [{"country_id": "CAN", "indicator_id": "cumulative-co2", "period": date(1990 + k, 1, 1), "year": 1990 + k, "value": v}67            for k, v in enumerate(vals)]68    ch2, ev2 = compute_changes_and_events(pl.DataFrame(rows), inds)69    assert ch2.height == 0 and ev2.height == 0707172def test_changes_are_recent_and_weighted() -> None:73    inds = indicators_by_id()74    old = [2.0, 2.1, 1.9, 2.2, 2.0, 2.3, 2.1, 2.0, 6.8, 3.4]  # series ends in 200875    rows = [{"country_id": "CAN", "indicator_id": "inflation", "period": date(1999 + k, 1, 1), "year": 1999 + k, "value": v}76            for k, v in enumerate(old)]77    rows += [{"country_id": "FRA", "indicator_id": "inflation", "period": date(2016 + k, 1, 1), "year": 2016 + k, "value": v}78             for k, v in enumerate(old)]79    from datetime import UTC, datetime8081    ch, ev = compute_changes_and_events(pl.DataFrame(rows), inds, headline_ids={"inflation"}, now=datetime(2026, 9, 1, tzinfo=UTC))82    assert set(ch["country_id"].to_list()) == {"FRA"}  # Canada's 2008 drop is an event, not a change83    assert "CAN" in set(ev["country_id"].to_list())84    import json8586    d = json.loads(ch["detail"][0])87    assert d["weight"] == 1.0 and "raw_severity" in d88    ch_w, _ = compute_changes_and_events(pl.DataFrame(rows), inds, headline_ids=set(), now=datetime(2026, 9, 1, tzinfo=UTC))89    assert json.loads(ch_w["detail"][0])["weight"] == 0.9  # inflation is featured but not headline here90    assert ch_w["severity"][0] == pytest.approx(min(1.0, d["raw_severity"] * 0.9), rel=1e-3)919293def test_index_series_need_10pct_and_quarterly_is_annualised() -> None:94    inds = indicators_by_id()95    ind = inds["real-house-price-index"]  # format index, ranking_eligible false96    assert ind.format == "index" and not ind.ranking_eligible97    smooth = [100 + k for k in range(12)]98    smooth[-1] = smooth[-2] * 1.05  # +5 %: big z, but below the 10 % rule99    assert not any(r["kind"] in ("yoy_jump", "yoy_drop") for r in detect_changes(_series("real-house-price-index", smooth), ind))100    big = list(smooth)101    big[-1] = big[-2] * 1.2  # +20 %102    assert any(r["kind"] == "yoy_jump" for r in detect_changes(_series("real-house-price-index", big), ind))103    # quarterly rows collapse to one value per year (last value for an index) → at most one detection per year104    rows = []105    for y in range(2015, 2026):106        for q, month in enumerate((1, 4, 7, 10)):107            v = 100 + (y - 2015) * 2 + q * 0.3108            if y == 2025:109                v = 150.0 + q  # +~25 % jump in 2025110            rows.append({"country_id": "CAN", "indicator_id": "real-house-price-index", "period": date(y, month, 1), "year": y,111                         "frequency": "Q", "value": v})112    ch, ev = compute_changes_and_events(pl.DataFrame(rows), inds, now=datetime(2026, 1, 1, tzinfo=UTC))113    assert ch.filter(pl.col("kind") == "yoy_jump").height == 1114    assert ev.group_by(["indicator_id", "year"]).len()["len"].max() <= 3115    assert ev["severity"].min() >= 0.35116    assert ch.filter(pl.col("kind") == "yoy_jump")["value"][0] == 153.0  # last quarter of 2025117118119def test_driver_over_frame() -> None:120    inds = indicators_by_id()121    rows = []122    for c in ("CAN", "FRA"):123        for k, v in enumerate([2.0, 2.1, 1.9, 2.2, 2.0, 2.3, 2.1, 2.0, 6.8, 3.4]):124            rows.append({"country_id": c, "indicator_id": "inflation", "period": date(2016 + k, 1, 1), "year": 2016 + k, "value": v})125    df = pl.DataFrame(rows)126    ch, ev = compute_changes_and_events(df, inds)127    assert ch.filter(pl.col("kind") == "yoy_drop").height == 2128    assert set(ch.columns) >= {"id", "country_id", "indicator_id", "kind", "headline", "severity", "detected_at"}129    assert ev.height >= 2130131132# ------------------------------------------------------------------------------------------------ change detection 2.0133def test_structural_break_in_changes_and_events() -> None:134    ind = indicators_by_id()["unemployment-rate"]135    # 12 years around 5 % (with small noise so the series is not monotone), then 8 years around 12 % → break in 2012136    vals = [5.0, 5.2, 4.9, 5.1, 5.0, 5.3, 4.8, 5.1, 5.0, 5.2, 4.9, 5.1, 12.0, 12.2, 11.9, 12.1, 12.0, 12.3, 11.8, 12.1]137    ch = detect_changes(_series("unemployment-rate", vals, 2000), ind)  # last year 2019, break 2012 → within 10 years138    brk = [r for r in ch if r["kind"] == "structural_break"]139    assert len(brk) == 1140    b = brk[0]141    assert b["year"] == 2012 and "shifted to a higher level around 2012" in b["headline"]142    assert 0.45 <= b["severity"] <= 0.7143    import json as _json144145    d = b["detail"] if isinstance(b["detail"], dict) else _json.loads(b["detail"])146    assert d["break_year"] == 2012 and d["gain"] >= 0.5 and d["shift_ratio"] >= 1.5 and d["min_segment"] == 5147    ev = detect_events(_series("unemployment-rate", vals, 2000), ind)148    assert len([r for r in ev if r["kind"] == "structural_break"]) == 1149    # an old break (25 years ago) stays an event but is not a recent change150    old = vals + [12.0, 12.2, 11.9, 12.1, 12.0, 12.3, 11.8, 12.1, 12.0, 12.2, 11.9, 12.1, 12.0, 12.3, 11.8]151    ch_old = detect_changes(_series("unemployment-rate", old, 1988), ind)  # last year 2022, break 2000 → 22 years ago152    assert not any(r["kind"] == "structural_break" for r in ch_old)153    assert any(r["kind"] == "structural_break" for r in detect_events(_series("unemployment-rate", old, 1988), ind))154    # a smooth trend is not a level shift155    trend = [5.0 + 0.3 * k for k in range(20)]156    assert not any(r["kind"] == "structural_break" for r in detect_changes(_series("unemployment-rate", trend, 2000), ind))157158159def test_trend_reversal() -> None:160    ind = indicators_by_id()["unemployment-rate"]  # floor 1 point161    vals = [8.0, 8.0, 8.0, 9.0, 8.0, 7.0, 6.0, 7.0, 8.0, 9.0]  # 3 falls (9→6) then 3 rises (6→9): +3 points ≥ floor162    ch = detect_changes(_series("unemployment-rate", vals, 2010), ind)163    rev = [r for r in ch if r["kind"] == "trend_reversal"]164    assert len(rev) == 1165    assert rev[0]["headline"].startswith("Unemployment has risen for three consecutive years after falling for three")166    assert rev[0]["severity"] == pytest.approx(0.55) and rev[0]["ref_value"] == 6.0167    down = [3.0, 3.0, 3.0, 2.0, 3.0, 4.0, 5.0, 4.0, 3.0, 2.0]168    r2 = [r for r in detect_changes(_series("unemployment-rate", down, 2010), ind) if r["kind"] == "trend_reversal"]169    assert len(r2) == 1 and "has fallen for three consecutive years after rising for three" in r2[0]["headline"]170    # too small a swing (0.3 points) is silent171    tiny = [8.0, 8.0, 8.0, 8.3, 8.2, 8.1, 8.0, 8.1, 8.2, 8.3]172    assert not any(r["kind"] == "trend_reversal" for r in detect_changes(_series("unemployment-rate", tiny, 2010), ind))173174175def test_volatility_spike() -> None:176    ind = indicators_by_id()["inflation"]  # floor 2 points177    calm = [2.0 + 0.1 * ((-1) ** k) for k in range(16)]  # 15 tiny diffs (±0.2)178    wild = [2.0, 9.0, 1.0, 8.0, 0.0]  # 5 large swings179    ch = detect_changes(_series("inflation", calm + wild, 2005), ind)180    vol = [r for r in ch if r["kind"] == "volatility_spike"]181    assert len(vol) == 1182    assert "unusually volatile" in vol[0]["headline"] and 0.45 <= vol[0]["severity"] <= 0.7183    import json as _json184185    d = vol[0]["detail"] if isinstance(vol[0]["detail"], dict) else _json.loads(vol[0]["detail"])186    assert d["ratio"] >= 3.0 and d["recent_years"] == 5 and d["baseline_years"] == 15187    # same pattern but the recent swings are below the 2-point floor → silent188    small = calm + [2.0, 2.5, 2.0, 2.6, 2.1]189    assert not any(r["kind"] == "volatility_spike" for r in detect_changes(_series("inflation", small, 2005), ind))190    # too short a history → silent191    assert not any(r["kind"] == "volatility_spike" for r in detect_changes(_series("inflation", wild * 2, 2015), ind))192193194def test_new_kinds_flow_through_driver_with_weights() -> None:195    inds = indicators_by_id()196    vals = [5.0, 5.2, 4.9, 5.1, 5.0, 5.3, 4.8, 5.1, 5.0, 5.2, 4.9, 5.1, 12.0, 12.2, 11.9, 12.1, 12.0, 12.3, 11.8, 12.1]197    rows = [{"country_id": "CAN", "indicator_id": "unemployment-rate", "period": date(2006 + k, 1, 1), "year": 2006 + k, "value": v}198            for k, v in enumerate(vals)]199    ch, ev = compute_changes_and_events(pl.DataFrame(rows), inds, headline_ids={"unemployment-rate"}, now=datetime(2026, 9, 1, tzinfo=UTC))200    kinds = set(ch["kind"].to_list())201    assert "structural_break" in kinds202    assert ev.filter(pl.col("kind") == "structural_break").height == 1203