Pipeline: noise rules for events (index ≥10 %, annualised Q/M, caps); headline uses IMF general government debt
6 changed files +90 −6
modified
docs/ARCHITECTURE.md
+4 −1
@@ -231,7 +231,10 @@ No table or column of `schema.sql` was changed. The following precisions/deviati | ||
| 231 | 231 | reached after a ≥ 5-year gap since the previous record and sign flips, at most 30 per series; `changes` add N-year |
| 232 | 232 | highs/lows (N ∈ {10, 20, 30}) and acceleration/deceleration (3 consecutive increases/decreases of the difference) at the |
| 233 | 233 | latest period only. `id = sha1("changes|"+country+indicator+kind+period)[:16]` (`"events|"` for events). Headlines are |
| 234 | − English templates. | |
| 234 | + English templates. **Noise rules** (changes and events): Q/M series are annualised before detection (last value of the | |
| 235 | + calendar year for indices/shares/rates/ratios, calendar-year mean otherwise; `period` = Jan 1 of that year) so an indicator | |
| 236 | + yields at most one detection per year; indicators with `format: index` or `ranking_eligible: false` only report YoY moves | |
| 237 | + with |Δ%| ≥ 10 %; `events` drop rows with severity < 0.35 and keep at most the 3 most severe per (country, indicator, year). | |
| 235 | 238 | * **OWID freshness**: raw.githubusercontent.com sends no `Last-Modified`; `source_updated_at` for the co2/energy files is the |
| 236 | 239 | date of the last GitHub commit touching the file (`api.github.com/repos/owid/<repo>/commits?path=…`), grapher charts use |
| 237 | 240 | `lastUpdated` from their metadata. |
modified
docs/PIPELINE.md
+2 −1
@@ -106,7 +106,8 @@ country's differences, with a floor of 10 % (relative) or 2 % of the series rang | ||
| 106 | 106 | *"Inflation fell 3.4 points to 3.4 % in 2025 (largest drop since 2009)."* `changes` only keep detections at a series' |
| 107 | 107 | latest period when that period is recent (≤ 2 years behind the indicator's max year and ≤ 3 years behind today); |
| 108 | 108 | monotone series and `cumulative`-tagged indicators are silent; severity = detector score × importance weight |
| 109 | − (headline 1.0 / featured 0.9 / other 0.7). | |
| 109 | + (headline 1.0 / featured 0.9 / other 0.7). Q/M series are annualised first (one detection per year at most); index / | |
| 110 | + non-rankable series need a ≥ 10 % move; `events` keep severity ≥ 0.35 and ≤ 3 rows per (country, indicator, year). | |
| 110 | 111 | 7. `similarity` (5 modes, `registry/similarity.yaml`) and `country_dna` (9 percentile dimensions, `dna:` section of the |
| 111 | 112 | same file). 8. `insights` (`registry/insights.yaml`, 16 templates, all numbers computed). |
| 112 | 113 | 9. `meta` (schema_version, build_run_id, built_at, counts, connectors, duration) → `CHECKPOINT`. |
modified
registry/topics.yaml
+1 −1
@@ -9,7 +9,7 @@ headline: | ||
| 9 | 9 | - unemployment-rate |
| 10 | 10 | - life-expectancy |
| 11 | 11 | - median-age |
| 12 | − - government-debt-pct-gdp | |
| 12 | + - general-government-gross-debt-pct-gdp | |
| 13 | 13 | - co2-per-capita |
| 14 | 14 | - renewable-electricity-share |
| 15 | 15 | - internet-users |
modified
src/countryatlas/pipeline/build.py
+2 −1
@@ -427,7 +427,8 @@ def build(run_id: str | None = None, strict: bool = True, swap: bool = True) -> | ||
| 427 | 427 | |
| 428 | 428 | ind_by_id = registry.indicators_by_id() |
| 429 | 429 | series = con.execute( |
| 430 | − "SELECT country_id, indicator_id, period, year, value FROM obs_ok ORDER BY country_id, indicator_id, period" | |
| 430 | + "SELECT country_id, indicator_id, period, year, frequency, value FROM obs_ok " | |
| 431 | + "ORDER BY country_id, indicator_id, period" | |
| 431 | 432 | ).pl() |
| 432 | 433 | ch, ev = compute_changes_and_events(series, ind_by_id, headline_ids=set(registry.topics()["headline"])) |
| 433 | 434 | con.register("df_changes", ch) |
modified
src/countryatlas/pipeline/changes.py
+54 −1
@@ -38,6 +38,9 @@ MAX_EVENTS_PER_SERIES = 30 | ||
| 38 | 38 | MIN_SCALE_RATIO = 0.25 # lower bound of the robust scale, as a fraction of the floor (see _scale) |
| 39 | 39 | RECENT_YEARS_FROM_MAX = 2 # a change must be within 2 years of the indicator's latest year in the snapshot… |
| 40 | 40 | RECENT_YEARS_FROM_NOW = 3 # …and within 3 years of today |
| 41 | +INDEX_MIN_PCT = 10.0 # index / non-rankable series: a YoY move must be ≥ 10 % to be reported at all | |
| 42 | +EVENT_MIN_SEVERITY = 0.35 # events below this are noise and are dropped | |
| 43 | +EVENTS_PER_INDICATOR_YEAR = 3 # timeline cap per (country, indicator, year) | |
| 41 | 44 | RECORD_GAP_YEARS = 5 |
| 42 | 45 | SIGN_FLIP_HINTS = ("growth", "balance", "net-migration", "inflation", "change") |
| 43 | 46 | |
@@ -101,6 +104,42 @@ def _scale(mad: float, floor: float, floor_abs: bool, mode: str, ref: float) -> | ||
| 101 | 104 | return max(mad, MIN_SCALE_RATIO * floor_w) |
| 102 | 105 | |
| 103 | 106 | |
| 107 | +def _strict_pct(ind: Indicator) -> bool: | |
| 108 | + """Index-type or non-rankable series (house-price indices, IPI, exchange rates…): only large relative moves count.""" | |
| 109 | + return ind.format == "index" or not ind.ranking_eligible | |
| 110 | + | |
| 111 | + | |
| 112 | +def _clears_pct(ind: Indicator, cur: float, prev: float) -> bool: | |
| 113 | + if not _strict_pct(ind): | |
| 114 | + return True | |
| 115 | + p = _pct(cur, prev) | |
| 116 | + return p is not None and abs(p) >= INDEX_MIN_PCT | |
| 117 | + | |
| 118 | + | |
| 119 | +def annualise(df: pl.DataFrame, indicators: dict[str, Indicator]) -> pl.DataFrame: | |
| 120 | + """Collapse Q/M series to one value per calendar year so detectors fire at most once a year per indicator: | |
| 121 | + last value of the year for rates, shares, indices and ratios; calendar-year mean otherwise.""" | |
| 122 | + if df.is_empty() or "frequency" not in df.columns: | |
| 123 | + return df | |
| 124 | + sub = df.filter(pl.col("frequency") != "A") | |
| 125 | + if sub.is_empty(): | |
| 126 | + return df.drop("frequency") | |
| 127 | + last_ids = [s for s, ind in indicators.items() | |
| 128 | + if ind.format in ("index", "percent", "ratio") or (ind.unit or "").startswith("%")] | |
| 129 | + agg = ( | |
| 130 | + sub.sort("period") | |
| 131 | + .group_by(["country_id", "indicator_id", "year"], maintain_order=True) | |
| 132 | + .agg(pl.col("value").last().alias("_last"), pl.col("value").mean().alias("_mean")) | |
| 133 | + .with_columns( | |
| 134 | + pl.when(pl.col("indicator_id").is_in(last_ids)).then(pl.col("_last")).otherwise(pl.col("_mean")).alias("value"), | |
| 135 | + pl.date(pl.col("year"), 1, 1).alias("period"), | |
| 136 | + ) | |
| 137 | + .select("country_id", "indicator_id", "period", "year", "value") | |
| 138 | + ) | |
| 139 | + annual = df.filter(pl.col("frequency") == "A").drop("frequency") | |
| 140 | + return pl.concat([annual, agg.select(annual.columns)]).sort(["country_id", "indicator_id", "period"]) | |
| 141 | + | |
| 142 | + | |
| 104 | 143 | def _sign_flip_applicable(ind: Indicator) -> bool: |
| 105 | 144 | lo = (ind.bounds or [None, None])[0] |
| 106 | 145 | if lo is not None and float(lo) >= 0: |
@@ -204,7 +243,7 @@ def detect_changes(s: Series, ind: Indicator) -> list[dict[str, Any]]: | ||
| 204 | 243 | if dl is not None and n >= MIN_POINTS and scale > 0: |
| 205 | 244 | z = (dl - med) / scale |
| 206 | 245 | magnitude = abs(cur - prev) if floor_abs else abs(dl) |
| 207 | − if abs(z) > Z_THRESHOLD and magnitude >= floor > 0: | |
| 246 | + if abs(z) > Z_THRESHOLD and magnitude >= floor > 0 and _clears_pct(ind, cur, prev): | |
| 208 | 247 | kind = "yoy_jump" if dl > 0 else "yoy_drop" |
| 209 | 248 | verb = "rose" if dl > 0 else "fell" |
| 210 | 249 | # largest since: last earlier year whose move was at least as large in the same direction |
@@ -319,6 +358,8 @@ def detect_events(s: Series, ind: Indicator) -> list[dict[str, Any]]: | ||
| 319 | 358 | for j in hits: |
| 320 | 359 | i = int(j) + 1 |
| 321 | 360 | prev, cur = float(v[i - 1]), float(v[i]) |
| 361 | + if not _clears_pct(ind, cur, prev): | |
| 362 | + continue | |
| 322 | 363 | kind = "yoy_jump" if d[j] > 0 else "yoy_drop" |
| 323 | 364 | verb = "rose" if d[j] > 0 else "fell" |
| 324 | 365 | sev = _yoy_severity(float(z[j]), float(magnitude[j]), floor, floor_abs) |
@@ -397,6 +438,7 @@ def compute_changes_and_events( | ||
| 397 | 438 | """ |
| 398 | 439 | headline_ids = headline_ids or set() |
| 399 | 440 | now = now or datetime.now(UTC) |
| 441 | + df = annualise(df, indicators) # Q/M → one value per calendar year | |
| 400 | 442 | changes: list[dict[str, Any]] = [] |
| 401 | 443 | events: list[dict[str, Any]] = [] |
| 402 | 444 | n_series = n_skipped = 0 |
@@ -430,4 +472,15 @@ def compute_changes_and_events( | ||
| 430 | 472 | ch = pl.DataFrame(changes, schema=schema) if changes else pl.DataFrame(schema=schema) |
| 431 | 473 | ch = ch.with_columns(pl.lit(detected_at).cast(pl.Datetime("us")).alias("detected_at")) |
| 432 | 474 | ev = pl.DataFrame(events, schema=schema) if events else pl.DataFrame(schema=schema) |
| 475 | + if ev.height: | |
| 476 | + # timeline hygiene: drop noise and keep the 3 most severe events per (country, indicator, year) | |
| 477 | + ev = ( | |
| 478 | + ev.filter(pl.col("severity") >= EVENT_MIN_SEVERITY) | |
| 479 | + .sort(["country_id", "indicator_id", "year", "severity"], descending=[False, False, False, True]) | |
| 480 | + .with_columns(pl.int_range(pl.len()).over(["country_id", "indicator_id", "year"]).alias("_rk")) | |
| 481 | + .filter(pl.col("_rk") < EVENTS_PER_INDICATOR_YEAR) | |
| 482 | + .drop("_rk") | |
| 483 | + ) | |
| 484 | + log.info("events after hygiene (severity ≥ %.2f, ≤ %d per indicator-year): %d", EVENT_MIN_SEVERITY, | |
| 485 | + EVENTS_PER_INDICATOR_YEAR, ev.height) | |
| 433 | 486 | return ch, ev |
modified
tests/test_changes.py
+27 −1
@@ -1,6 +1,6 @@ | ||
| 1 | 1 | from __future__ import annotations |
| 2 | 2 | |
| 3 | −from datetime import date | |
| 3 | +from datetime import UTC, date, datetime | |
| 4 | 4 | |
| 5 | 5 | import numpy as np |
| 6 | 6 | import polars as pl |
@@ -90,6 +90,32 @@ def test_changes_are_recent_and_weighted() -> None: | ||
| 90 | 90 | assert ch_w["severity"][0] == pytest.approx(min(1.0, d["raw_severity"] * 0.9), rel=1e-3) |
| 91 | 91 | |
| 92 | 92 | |
| 93 | +def 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 false | |
| 96 | + assert ind.format == "index" and not ind.ranking_eligible | |
| 97 | + smooth = [100 + k for k in range(12)] | |
| 98 | + smooth[-1] = smooth[-2] * 1.05 # +5 %: big z, but below the 10 % rule | |
| 99 | + 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 year | |
| 104 | + 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.3 | |
| 108 | + if y == 2025: | |
| 109 | + v = 150.0 + q # +~25 % jump in 2025 | |
| 110 | + 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 == 1 | |
| 114 | + assert ev.group_by(["indicator_id", "year"]).len()["len"].max() <= 3 | |
| 115 | + assert ev["severity"].min() >= 0.35 | |
| 116 | + assert ch.filter(pl.col("kind") == "yoy_jump")["value"][0] == 153.0 # last quarter of 2025 | |
| 117 | + | |
| 118 | + | |
| 93 | 119 | def test_driver_over_frame() -> None: |
| 94 | 120 | inds = indicators_by_id() |
| 95 | 121 | rows = [] |
| 96 | 122 | |