spb/countryatlas

1from __future__ import annotations23from datetime import date45import numpy as np6import polars as pl78from countryatlas.pipeline.changes import Series, compute_changes_and_events, detect_changes, detect_events9from countryatlas.registry import indicators_by_id101112def _series(ind: str, values: list[float], start: int = 2000, country: str = "CAN") -> Series:13    years = np.arange(start, start + len(values), dtype=np.int32)14    return Series(country, ind, [date(int(y), 1, 1) for y in years], years, np.array(values, dtype=float))151617def test_inflation_drop_headline_and_since() -> None:18    ind = indicators_by_id()["inflation"]19    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.120    vals = [2.0, 2.1, 1.9, 2.2, 2.0, 2.3, 2.1, 2.0, 6.8, 3.4]21    out = detect_changes(_series("inflation", vals, 2016), ind)22    kinds = {r["kind"]: r for r in out}23    assert "yoy_drop" in kinds24    h = kinds["yoy_drop"]["headline"]25    assert h.startswith("Inflation fell 3.4 points to 3.4 % in 2025")26    assert "largest drop" in h27    assert 0 < kinds["yoy_drop"]["severity"] <= 1282930def test_floor_blocks_small_moves() -> None:31    ind = indicators_by_id()["inflation"]  # change_floor 2 points32    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 floor33    out = detect_changes(_series("inflation", vals), ind)34    assert not any(r["kind"] in ("yoy_jump", "yoy_drop") for r in out)353637def test_record_high_and_relative_headline() -> None:38    ind = indicators_by_id()["gdp-per-capita"]39    vals = [1000 * 1.03**k for k in range(15)]40    vals[-1] = vals[-2] * 1.2541    out = detect_changes(_series("gdp-per-capita", vals), ind)42    kinds = {r["kind"] for r in out}43    assert "record_high" in kinds and "yoy_jump" in kinds44    jump = next(r for r in out if r["kind"] == "yoy_jump")45    assert "25.0 %" in jump["headline"] and "US$" in jump["headline"]464748def test_sign_flip_and_events_history() -> None:49    ind = indicators_by_id()["gdp-growth"]50    vals = [3.0, 2.5, 3.1, 2.8, 3.0, 2.9, -4.5, 5.0, 2.0, -1.0]51    ch = detect_changes(_series("gdp-growth", vals), ind)52    assert any(r["kind"] == "sign_flip" for r in ch)53    ev = detect_events(_series("gdp-growth", vals), ind)54    flips = [r for r in ev if r["kind"] == "sign_flip"]55    assert len(flips) == 3  # 2006, 2007, 200956    assert any(r["kind"] == "yoy_drop" and r["year"] == 2006 for r in ev)575859def test_driver_over_frame() -> None:60    inds = indicators_by_id()61    rows = []62    for c in ("CAN", "FRA"):63        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]):64            rows.append({"country_id": c, "indicator_id": "inflation", "period": date(2016 + k, 1, 1), "year": 2016 + k, "value": v})65    df = pl.DataFrame(rows)66    ch, ev = compute_changes_and_events(df, inds)67    assert ch.filter(pl.col("kind") == "yoy_drop").height == 268    assert set(ch.columns) >= {"id", "country_id", "indicator_id", "kind", "headline", "severity", "detected_at"}69    assert ev.height >= 270