diff --git a/docs/CICs/README.md b/docs/CICs/README.md index 50c2d6f..177b1d6 100644 --- a/docs/CICs/README.md +++ b/docs/CICs/README.md @@ -52,7 +52,6 @@ Contracts must be clear enough that: - `UNFAOPostProcessorManager.md` — Pipeline orchestration manager (read → transform → validate → save) - `GaulLookupEnricher.md` — Precomputed GAUL lookup enrichment (ADR-011; replaced the runtime mapper) -- `ReconciliationModule.md` — Reconcile pgm forecasts to cm country totals (numpy-native) --- diff --git a/docs/CICs/ReconciliationModule.md b/docs/CICs/ReconciliationModule.md deleted file mode 100644 index 23b2b9f..0000000 --- a/docs/CICs/ReconciliationModule.md +++ /dev/null @@ -1,118 +0,0 @@ -# Class Intent Contract: ReconciliationModule - -**Status:** Active -**Owner:** PRIO MD&D Team -**Last reviewed:** 2026-06-24 -**Related ADRs:** ADR-003 (fail loud), views-frames ADR-014 (injected cross-level mapping); epic #31, migration `docs/reconciliation_migration.md`, origin #3 / views-reporting#72 - ---- - -## 1. Purpose - -> Make PRIO-GRID-month (pgm) forecasts consistent with country-month (cm) totals: within each `(time, country)`, scale the country's grid cells so their per-draw sum equals the country forecast, preserving each cell's relative share and its zeros. - -It is the frames-native, numpy-only home of the reconciliation that previously lived in views-reporting (`ForecastReconciler` + `ReconciliationModule`), ported parity-preserving. - ---- - -## 2. Non-Goals (Explicit Exclusions) - -- Does **not** embed or fetch geography. The `(time, priogrid_gid) -> country_id` mapping is **injected** at construction (views-frames ADR-014); the class never queries viewser, shapefiles, or the GAUL lookup. -- Does **not** change the algorithm. It is **top-down proportional scaling per posterior draw** (FPP3 forecast proportions), a faithful port — **not** principled joint probabilistic reconciliation (the upgrade is **C-37**, deferred). -- Does **not** load, save, or upload data; does **not** depend on pandas, torch, viewser, or wandb. -- Does **not** mutate its inputs. - ---- - -## 3. Responsibilities and Guarantees - -- Validates inputs fail-loud before any work (level, sample-count, time coverage, country coverage). -- For each `(time, country)`: scales grid cells so their per-draw sum **equals** the country forecast; preserves zero cells; clamps to non-negative. -- Returns a **new** pgm `PredictionFrame` with the same index/metadata as the input grid frame (de-mutation, C-184). -- Bit-for-bit reproduces the frozen views-reporting pipeline on the parity fixture. - ---- - -## 4. Inputs and Assumptions - -- Constructed with `map_keys` `(M, 2)` `(time, priogrid_gid)` and `map_vals` `(M,)` `country_id` covering every grid row. -- `reconcile(cm_frame, pgm_frame)`: a CM-level `PredictionFrame` (`country_id` units) and a PGM-level one (`priogrid_gid` units), **same** sample count `S` and **same** set of times. One target per call (multi-target = call per target). -- Time identifiers are opaque integers (`month_id`); country totals are authoritative. - ---- - -## 5. Outputs and Side Effects - -- Output: a new pgm `PredictionFrame` `(N, S)`, reconciled. **No** side effects (no I/O, no logging of data, no global state). -- **Memory ∝ frame size.** Grouping is `O(N log N)` (group-by-sort; register C-38), but the whole frame is held in memory at once — peak ≈ input + output ≈ `2·N·S·4` bytes. At global volume (`land` region) the **caller must chunk by time**: reconciliation is independent across months, so call `reconcile` per month-slice and write each result out rather than materialising the global frame. (C-38; verified on a global dry-run at S7, #39.) -- **Approximate where flagged:** for a draw in which *all* of a country's grid cells are zero, there are no proportions to distribute, so those cells stay zero and that draw's total is not conserved (the algorithm's documented edge case). Uncertainty is reconciled per-draw, which is a pragmatic approximation (C-37). - ---- - -## 6. Failure Modes and Loudness - -Raises `ValueError` (never silently degrades) when: -- `map_keys` is not `(M, 2)` or `map_vals` is not length `M` (constructor); -- a frame is at the wrong `SpatialLevel`; -- cm and pgm sample counts differ; -- cm and pgm cover different time steps; -- a grid row's `(time, priogrid_gid)` is absent from the mapping (raised by `cross_level_align`); -- a `(time, country)` group has no matching country forecast in `cm_frame`. - -Aligns with ADR-003: ambiguity fails loud, before computation. - ---- - -## 7. Boundaries and Interactions - -- **Trusts:** the leaf `reconcile_proportional` (the math), `grouping.reconcile_pgm_to_cm` (the cross-level grouping/scatter), `validation` (the guards), `frames` (array↔frame I/O), and `views_frames` (`PredictionFrame`, `SpatioTemporalIndex`, `cross_level_align`). -- **Must not depend on:** pandas, torch, viewser, wandb, the unfao delivery code, or any geography source. -- The injected mapping is treated as opaque, caller-owned truth. - ---- - -## 8. Examples of Correct Usage - -```python -from views_postprocessing.reconciliation import ReconciliationModule - -rm = ReconciliationModule(map_keys, map_vals) # injected (time, pgid) -> country_id -reconciled_pgm = rm.reconcile(cm_frame, pgm_frame) # one target; new frame -``` - -Multi-target: call `rm.reconcile(cm_t, pgm_t)` once per target. - ---- - -## 9. Examples of Incorrect Usage - -- Constructing it and expecting it to *derive* the country mapping (it never does — inject it). -- Passing a pgm frame where a cm frame is expected, or frames with different sample counts / times (raises, by design — do not pre-pad or coerce to silence it). -- Reusing it as a generic disaggregator for non-reconciliation tasks. - ---- - -## 10. Test Alignment - -- **Parity (gate):** `tests/test_reconciliation_e2e_parity.py` — the module reproduces the frozen oracle (`tests/fixtures/reconciliation_e2e_parity.npz`) bit-for-bit on every target. -- **Unit:** `tests/test_reconciliation_{frames,grouping,validation}.py` — adapters, grouping core, and each fail-loud guard. -- **Leaf parity:** `tests/test_reconciliation_parity.py` — `reconcile_proportional` vs the torch oracle. -- **Scale:** `tests/test_reconciliation_scale.py` — conservation holds across thousands of `(time, country)` groups (guards the group-by-sort logic; C-38). -- Regression-protected: bit-exact parity, zero-preservation, de-mutation, every `ValueError` guard, and grouping correctness at scale. - ---- - -## 11. Evolution Notes - -- **Stable:** the injected-mapping contract, the fail-loud guards, the de-mutated return. -- **Expected to change:** the *algorithm* — the principled probabilistic upgrade (**C-37**) will arrive as a sibling method behind this same interface (OCP); when it does, this contract's §2/§5 approximation notes must be revisited. -- The production mapping **source** (viewser-derived `country_id` vs the GAUL lookup) is decided at wiring time (S7, #39) and does not change this class's contract. - ---- - -## End of Contract - -This document defines the **intended meaning** of `ReconciliationModule`. - -Changes to behavior that violate this intent are bugs. -Changes to intent must update this contract. diff --git a/docs/reconciliation_migration.md b/docs/reconciliation_migration.md index f6df71a..2cc81cf 100644 --- a/docs/reconciliation_migration.md +++ b/docs/reconciliation_migration.md @@ -1,64 +1,17 @@ -# Reconciliation migration — notes & decisions - -Tracks the migration of forecast reconciliation from views-reporting into this -repo. Epic: **#31**; tracking checklist: **#41**; origin: **#3** / views-reporting#72. -Story S6 (#38) formalises this into the `ReconciliationModule` CIC. - -## Status - -- **Slice 1 (PR #30) — done.** Leaf algorithm `reconcile_proportional` - (`views_postprocessing/reconciliation/proportional.py`), pure numpy, **bit-exact** - parity vs the views-reporting torch oracle (`tests/test_reconciliation_parity.py`). -- **S0 (#32) — done.** End-to-end oracle fixture captured offline - (`tests/fixtures/reconciliation_e2e_parity.npz`, via - `scripts/gen_reconciliation_e2e_fixture.py`). Decisions below. -- **S1–S5 (#33–#37) — done.** The frames-native module is complete and - **end-to-end parity-proven**: `cm/pgm` adapters (`reconciliation/frames.py`), - the `cross_level_align` grouping core (`reconciliation/grouping.py`), fail-loud - validation (`reconciliation/validation.py`), and the public - `ReconciliationModule` (`reconciliation/module.py`) — which reproduces the - frozen views-reporting pipeline **bit-for-bit** on every target - (`tests/test_reconciliation_e2e_parity.py`). No torch / pandas / viewser / wandb. -- **S6 (#38) — done.** CIC at `docs/CICs/ReconciliationModule.md`. -- **C-38 (scale) — compute fixed.** The grouping is now `O(N log N)` (group-by-sort - in `reconciliation/grouping.py`); parity stays bit-exact and a scale guard - (`tests/test_reconciliation_scale.py`) protects it. Residual: global peak memory - is bounded by caller-side **chunk-by-time** (documented in the CIC), verified at S7. -- **In-repo migration complete.** Remaining: S7 (#39) pipeline-core repoint and - S8 (#40) views-reporting phase-out — both cross-repo, **blocked** on this - landing; and the principled-algorithm upgrade (**C-37**), a separate epic. - -## D-R1 — Group by injected VIEWS `country_id`, not GAUL (for parity) - -The frozen oracle groups grid cells by **VIEWS `country_id`** (from viewser's -`country_month` LOA). This repo's GAUL lookup (`data/gaul_lookup.parquet`) numbers -countries by **`admin1_gaul0_code`** — a *different* id system. The migration is -**parity-preserving**, so the frames-native module groups by the **same VIEWS -`country_id`**, **injected** by the caller (the leaf never embeds geography — -views-frames ADR-014). The fixture bypasses viewser by pre-setting -`pg_ds._country_to_grids_cache`. - -> **Deferred (S7, #39):** whether the *production* mapping should eventually come -> from our GAUL lookup instead of viewser is a separate decision, taken at wiring -> time. It does **not** affect parity and is out of scope until the migration is -> wired and proven. - -## D-R2 — The fixture deliberately includes the all-zero-country-draw edge case - -When *every* grid cell of a country is zero for a posterior draw, proportional -scaling has no proportions to distribute, so the oracle leaves those cells **zero** -(country total not conserved for that draw — the algorithm's documented edge case). -The S0 fixture's sparsity (~30% zeros, small countries) produces such draws -(178 across the battery), captured verbatim. The frames-native module must -**reproduce this behaviour** (parity, not "correctness"); improving it belongs to -the principled-reconciliation upgrade (**C-37**), not this migration. - -## Parity oracle (how the fixture is made) - -`scripts/gen_reconciliation_e2e_fixture.py` builds a realistic cm + pgm sample -(5 countries of varying size, 3 months, 2 targets, 100 samples), injects the -`country_id` mapping, runs the **untouched** views-reporting `ReconciliationModule` -on CPU with WandB patched out, and freezes `(cm, pg, pg_country, recon)` to npz. -It needs pipeline-core + views-reporting + torch **only at generation time** (the -`views_pipeline` conda env); the committed fixture is consumed offline (numpy only), -so CI needs none of them. +# Reconciliation — moved to views-frames + +Forecast reconciliation **no longer lives in views-postprocessing.** It moved to the +`views_frames_reconcile` sibling in the views-frames mono-wheel — a frame→frame +operation belongs in the frames foundation, not bolted onto FAO delivery (which does +not use reconciliation). + +- **Home:** `views_frames_reconcile` (views-frames ≥ 1.7.0) — Epic 11 / + views-platform/views-frames#131, ADR-023. Parity-proven bit-identical to the copy + that briefly lived here. +- **Consumer wiring:** views-models `reconciliation/reconciler_factory.py` (the ADR-014 + composition root) imports it directly — views-platform/views-models#191 (PR #202). +- **This repo:** vpp was only ever a way-station (epic #31); its copy was parity-proven + (PR #30) and then retired here (#62). The pipeline-core port collapse and + views-reporting retirement continue separately (pipeline-core#221, #40 / views-reporting#72). + +Migration history is preserved in git and in the risk register (C-37, C-38, C-42). diff --git a/scripts/gen_reconciliation_e2e_fixture.py b/scripts/gen_reconciliation_e2e_fixture.py deleted file mode 100644 index 6baccf1..0000000 --- a/scripts/gen_reconciliation_e2e_fixture.py +++ /dev/null @@ -1,140 +0,0 @@ -"""Capture the END-TO-END reconciliation oracle from the *untouched* views-reporting. - -Story S0 (#32) of the reconciliation epic (#31). Slice 1 froze the leaf math; -this freezes the **whole pipeline**: build a realistic cm + pgm sample, run -views-reporting's `ReconciliationModule` fully OFFLINE (inject the country->grids -mapping so no viewser; patch WandB; CPU), and save (cm, pgm, per-grid country, -reconciled pgm) to a committed npz. The frames-native module (S1-S4) must then -reproduce `recon__*` exactly — the parity gate (S5). - -The grouping is by VIEWS `country_id` (injected) — the only id system that lets -us match the oracle (the GAUL-vs-VIEWS choice is deferred to S7; see -`docs/reconciliation_migration.md`). - -Run once, in the views_pipeline env (needs pipeline-core + views-reporting + torch): - - PYTHONPATH=.:/home/simon/Documents/scripts/views_platform/views-reporting \\ - /home/simon/anaconda3/envs/views_pipeline/bin/python \\ - scripts/gen_reconciliation_e2e_fixture.py - -READS/RUNS the oracle; never edits views-reporting (parity). -""" - -from __future__ import annotations - -from pathlib import Path -from unittest.mock import patch - -import numpy as np -import pandas as pd -import torch - -from views_pipeline_core.data.handlers import CMDataset, PGMDataset -from views_reporting.reconciliation.reconciliation import ReconciliationModule - -_OUT = Path(__file__).resolve().parent.parent / "tests" / "fixtures" / "reconciliation_e2e_parity.npz" - -# Realistic-but-small hierarchy: 5 countries of varying size, 3 months, 2 targets. -_GRIDS = {1: [100, 101], 2: [102, 103, 104], 3: [105, 106], - 4: [107, 108, 109, 110], 5: [111, 112]} -_MONTHS = [528, 529, 530] -_TARGETS = ["pred_ged_sb", "pred_ged_ns"] -_SAMPLES = 100 - - -def _cm_df(rng): - rows = [(m, c) for m in _MONTHS for c in _GRIDS] - idx = pd.MultiIndex.from_tuples(rows, names=["month_id", "country_id"]) - # Country totals drawn INDEPENDENTLY of the grid (the realistic case). - data = {t: [rng.gamma(3.0, 20.0, _SAMPLES).astype(np.float64) for _ in rows] - for t in _TARGETS} - return pd.DataFrame(data, index=idx) - - -def _pg_df(rng): - rows = [(m, g) for m in _MONTHS for c in _GRIDS for g in _GRIDS[c]] - idx = pd.MultiIndex.from_tuples(rows, names=["month_id", "priogrid_id"]) - data = {} - for t in _TARGETS: - cells = [] - for _ in rows: - v = rng.gamma(2.0, 5.0, _SAMPLES).astype(np.float64) - v[rng.random(_SAMPLES) < 0.3] = 0.0 # ~30% zeros, per cell - cells.append(v) - data[t] = cells - return pd.DataFrame(data, index=idx) - - -def _stack(df, target): - """(N, S) float32 from an object-dtype column of per-cell sample arrays.""" - return np.stack([np.asarray(v, dtype=np.float32) for v in df[target].to_numpy()]) - - -def main() -> int: - rng = np.random.default_rng(20260624) - c_df, pg_df = _cm_df(rng), _pg_df(rng) - grid_to_country = {g: c for c, gs in _GRIDS.items() for g in gs} - - c_ds = CMDataset(source=c_df) - pg_ds = PGMDataset(source=pg_df) - # Inject the country<->grid mapping so build_country_to_grids_cache skips viewser. - pg_ds._country_to_grids_cache = {c: list(gs) for c, gs in _GRIDS.items()} - pg_ds._entity_metadata_cache = pd.DataFrame( - {"country_id": [grid_to_country[g] for (_, g) in pg_df.index]}, - index=pg_df.index, - ) - - with patch("views_reporting.reconciliation.reconciliation.WandBModule"): - rm = ReconciliationModule(c_ds, pg_ds, wandb_notifications=False) - rm._device = torch.device("cpu") - reconciled = rm.reconcile(max_workers=2) - - # --- assemble the fixture ------------------------------------------------ - cm_rows = list(c_df.index) - pg_rows = list(pg_df.index) - out: dict[str, np.ndarray] = { - "targets": np.array(_TARGETS), - "cm_time": np.array([m for (m, _) in cm_rows], dtype=np.int64), - "cm_unit": np.array([c for (_, c) in cm_rows], dtype=np.int64), - "pg_time": np.array([m for (m, _) in pg_rows], dtype=np.int64), - "pg_unit": np.array([g for (_, g) in pg_rows], dtype=np.int64), - "pg_country": np.array([grid_to_country[g] for (_, g) in pg_rows], dtype=np.int64), - } - for t in _TARGETS: - out[f"cm__{t}"] = _stack(c_df, t) - out[f"pg__{t}"] = _stack(pg_df, t) - out[f"recon__{t}"] = _stack(reconciled, t) - - # --- smoke check: the oracle conserves country totals per draw EXCEPT where a - # country's grid cells are all-zero for that draw (no proportions to distribute - # -> the cells stay zero; the algorithm's documented edge case). Verify exactly - # that, so the fixture is trusted before S1-S5 build against it. - worst_active = 0.0 - n_allzero = 0 - for t in _TARGETS: - recon = {row: out[f"recon__{t}"][i] for i, row in enumerate(pg_rows)} - pg_in = {row: out[f"pg__{t}"][i] for i, row in enumerate(pg_rows)} - cm = {row: out[f"cm__{t}"][i] for i, row in enumerate(cm_rows)} - for m in _MONTHS: - for c, gs in _GRIDS.items(): - inp = np.stack([pg_in[(m, g)] for g in gs]) - allzero = (inp == 0).all(axis=0) # per-draw: whole country zero - grid_sum = np.stack([recon[(m, g)] for g in gs]).sum(axis=0) - n_allzero += int(allzero.sum()) - assert (grid_sum[allzero] == 0).all(), "all-zero draw must stay zero" - if (~allzero).any(): - worst_active = max(worst_active, float( - np.max(np.abs(grid_sum[~allzero] - cm[(m, c)][~allzero])))) - print(f"smoke: worst |grid_sum - country| on active draws = {worst_active:.3e} " - f"(~0 expected); {n_allzero} all-zero country-draws left at zero " - f"(the oracle's edge case, captured for parity)") - - _OUT.parent.mkdir(parents=True, exist_ok=True) - np.savez_compressed(_OUT, **out) - print(f"wrote e2e parity fixture ({len(pg_rows)} grids x {_SAMPLES} samples x " - f"{len(_TARGETS)} targets) -> {_OUT}") - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/gen_reconciliation_parity_fixture.py b/scripts/gen_reconciliation_parity_fixture.py deleted file mode 100644 index 7bb3c88..0000000 --- a/scripts/gen_reconciliation_parity_fixture.py +++ /dev/null @@ -1,121 +0,0 @@ -"""Capture the reconciliation parity oracle from the *untouched* views-reporting. - -Runs views-reporting's torch `ForecastReconciler.reconcile_forecast` over a -battery of inputs and freezes ``(grid, country, expected)`` to a committed npz, -so the numpy port (`views_postprocessing.reconciliation.reconcile_proportional`) -can be proven to reproduce it in CI **without torch or views-reporting**. - -Run once, locally, from this repo's checkout (needs torch + views-reporting): - - PYTHONPATH=.:/home/simon/Documents/scripts/views_platform/views-reporting \\ - python scripts/gen_reconciliation_parity_fixture.py - -This script READS/RUNS the oracle; it never edits views-reporting (parity). -""" - -from __future__ import annotations - -from pathlib import Path - -import numpy as np -import torch - -from views_reporting.statistics import ForecastReconciler - -_OUT = Path(__file__).resolve().parent.parent / "tests" / "fixtures" / "reconciliation_parity.npz" - -# views-reporting/tests/test_statistics.py _PROB_CASES / _POINT_CASES. -_PROB_CASES = [ - (1000, 100, 0.3, 1.2, "prob-basic"), - (1000, 100, 1.0, 1.2, "prob-all-zeros"), - (1000, 100, 0.2, 10, "prob-extreme-skew"), - (1000, 100, 0.95, 1.2, "prob-sparse-95pct"), - (1000, 100, 0.3, 10, "prob-extreme-scaling"), - (1000, 100, 0.5, 1e-5, "prob-float-precision"), - (1000, 100, 0.7, 5, "prob-mixed-zeros-large"), - (500, 500, 0.5, 1.1, "prob-large-ish"), -] -_POINT_CASES = [ - (100, 0.3, 1.2, "point-basic"), - (100, 1.0, 1.2, "point-all-zeros"), - (100, 0.2, 10, "point-extreme-skew"), - (100, 0.95, 1.2, "point-sparse-95pct"), - (100, 0.3, 10, "point-extreme-scaling"), - (100, 0.5, 1e-5, "point-float-precision"), - (100, 0.7, 5, "point-mixed-zeros-large"), -] - - -def _prob_grid_country(num_samples, num_grid_cells, zero_fraction, scaling_factor): - """Reproduce the test fixture's exact construction (country = grid.sum * k).""" - torch.manual_seed(42) - zero_mask = torch.rand((num_samples, num_grid_cells)) < zero_fraction - grid = torch.randint(1, 100, (num_samples, num_grid_cells), dtype=torch.float32) - grid[zero_mask] = 0 - country = grid.sum(dim=1) * scaling_factor - return grid, country - - -def _point_grid_country(num_grid_cells, zero_fraction, scaling_factor): - torch.manual_seed(42) - zero_mask = torch.rand(num_grid_cells) < zero_fraction - grid = torch.randint(1, 100, (num_grid_cells,), dtype=torch.float32) - grid[zero_mask] = 0 - country = grid.sum().item() * scaling_factor - return grid, country - - -def main() -> int: - rec = ForecastReconciler(device="cpu") - cases: list[tuple[str, np.ndarray, np.ndarray, np.ndarray]] = [] - - for num_samples, n_cells, zf, sf, label in _PROB_CASES: - grid, country = _prob_grid_country(num_samples, n_cells, zf, sf) - adjusted = rec.reconcile_forecast(grid, country) - cases.append((label, grid.numpy(), country.numpy(), adjusted.cpu().numpy())) - - for n_cells, zf, sf, label in _POINT_CASES: - grid, country = _point_grid_country(n_cells, zf, sf) - adjusted = rec.reconcile_forecast(grid, country) - cases.append( - (label, grid.numpy(), np.float32(country), adjusted.cpu().numpy()) - ) - - # Extra: the realistic case the test suite never probes — country drawn - # INDEPENDENTLY of the grid (separate model), so index-pairing matters. - torch.manual_seed(7) - grid = torch.randint(0, 80, (300, 50), dtype=torch.float32) - country = torch.rand(300) * 5000.0 # independent of grid.sum - adjusted = rec.reconcile_forecast(grid, country) - cases.append( - ("prob-independent-country", grid.numpy(), country.numpy(), adjusted.cpu().numpy()) - ) - - # Extra: negatives present (the >0 mask drops them) and an all-zero draw. - grid = torch.tensor( - [[-5.0, 10.0, 0.0, 30.0], [0.0, 0.0, 0.0, 0.0], [2.0, -1.0, 4.0, 0.0]], - dtype=torch.float32, - ) - country = torch.tensor([100.0, 50.0, 12.0], dtype=torch.float32) - adjusted = rec.reconcile_forecast(grid, country) - cases.append( - ("prob-negatives-and-zero-draw", grid.numpy(), country.numpy(), adjusted.cpu().numpy()) - ) - - out: dict[str, np.ndarray] = {"n_cases": np.int64(len(cases))} - labels = [] - for i, (label, g, c, e) in enumerate(cases): - out[f"grid_{i}"] = g.astype(np.float32) - out[f"country_{i}"] = np.asarray(c, dtype=np.float32) - out[f"expected_{i}"] = e.astype(np.float32) - labels.append(label) - out["labels"] = np.array(labels) - - _OUT.parent.mkdir(parents=True, exist_ok=True) - np.savez_compressed(_OUT, **out) - print(f"wrote {len(cases)} parity cases -> {_OUT}") - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/tests/fixtures/reconciliation_e2e_parity.npz b/tests/fixtures/reconciliation_e2e_parity.npz deleted file mode 100644 index 8c82a21..0000000 Binary files a/tests/fixtures/reconciliation_e2e_parity.npz and /dev/null differ diff --git a/tests/fixtures/reconciliation_parity.npz b/tests/fixtures/reconciliation_parity.npz deleted file mode 100644 index 24770d0..0000000 Binary files a/tests/fixtures/reconciliation_parity.npz and /dev/null differ diff --git a/tests/test_reconciliation_e2e_parity.py b/tests/test_reconciliation_e2e_parity.py deleted file mode 100644 index 1c45c67..0000000 --- a/tests/test_reconciliation_e2e_parity.py +++ /dev/null @@ -1,93 +0,0 @@ -"""S4/S5 (#36/#37): the public `ReconciliationModule`, end-to-end parity gate. - -The gate that turns "leaf ported" into "pipeline migrated": the whole -frames-native module reproduces the frozen views-reporting pipeline on the S0 -fixture. Offline — numpy + views-frames + the committed fixture only. -""" - -from pathlib import Path - -import numpy as np -import pytest - -from views_frames import SpatialLevel - -from views_postprocessing.reconciliation import ReconciliationModule -from views_postprocessing.reconciliation.frames import prediction_frame_from_arrays - -_FIX = Path(__file__).resolve().parent / "fixtures" / "reconciliation_e2e_parity.npz" -_TARGETS = ["pred_ged_sb", "pred_ged_ns"] -_GRIDS = {1: [100, 101], 2: [102, 103, 104], 3: [105, 106], - 4: [107, 108, 109, 110], 5: [111, 112]} -_MONTHS = [528, 529, 530] - - -@pytest.fixture(scope="module") -def fix(): - return np.load(_FIX) - - -@pytest.fixture(scope="module") -def module(fix): - return ReconciliationModule( - np.stack([fix["pg_time"], fix["pg_unit"]], axis=1), fix["pg_country"] - ) - - -def _frames(fix, target): - pgm = prediction_frame_from_arrays( - fix["pg_time"], fix["pg_unit"], fix[f"pg__{target}"], level=SpatialLevel.PGM - ) - cm = prediction_frame_from_arrays( - fix["cm_time"], fix["cm_unit"], fix[f"cm__{target}"], level=SpatialLevel.CM - ) - return cm, pgm - - -class TestEndToEndParity: - @pytest.mark.parametrize("target", _TARGETS) - def test_module_reproduces_oracle(self, fix, module, target): - cm, pgm = _frames(fix, target) - out = module.reconcile(cm, pgm) - np.testing.assert_allclose( - out.values, fix[f"recon__{target}"], rtol=1e-5, atol=1e-6, - err_msg=f"module output drifts from the frozen oracle on {target}", - ) - - -class TestModuleProperties: - def test_de_mutated(self, fix, module): - cm, pgm = _frames(fix, "pred_ged_sb") - before = pgm.values.copy() - out = module.reconcile(cm, pgm) - assert out is not pgm - np.testing.assert_array_equal(pgm.values, before) - - def test_sum_constraint_on_active_draws(self, fix, module): - cm, pgm = _frames(fix, "pred_ged_sb") - out = module.reconcile(cm, pgm) - recon = {(int(t), int(u)): out.values[i] - for i, (t, u) in enumerate(zip(fix["pg_time"], fix["pg_unit"]))} - pin = {(int(t), int(u)): fix["pg__pred_ged_sb"][i] - for i, (t, u) in enumerate(zip(fix["pg_time"], fix["pg_unit"]))} - cmv = {(int(t), int(u)): fix["cm__pred_ged_sb"][i] - for i, (t, u) in enumerate(zip(fix["cm_time"], fix["cm_unit"]))} - for m in _MONTHS: - for c, gs in _GRIDS.items(): - allzero = np.stack([pin[(m, g)] for g in gs]).sum(axis=0) == 0 - grid_sum = np.stack([recon[(m, g)] for g in gs]).sum(axis=0) - active = ~allzero - np.testing.assert_allclose( - grid_sum[active], cmv[(m, c)][active], rtol=1e-4, atol=1e-3 - ) - assert (grid_sum[allzero] == 0).all() # all-zero draws stay zero - - def test_bad_mapping_shape_raises(self, fix): - with pytest.raises(ValueError, match="map_keys must be"): - ReconciliationModule(fix["pg_time"], fix["pg_country"]) # 1-D keys - - def test_validates_before_reconciling(self, fix, module): - # cm frame at the wrong level -> validation raises (not a mid-compute crash) - cm_wrong, pgm = _frames(fix, "pred_ged_sb") - with pytest.raises(ValueError, match="SpatialLevel.CM"): - module.reconcile(pgm, pgm) # pass pgm where cm expected diff --git a/tests/test_reconciliation_frames.py b/tests/test_reconciliation_frames.py deleted file mode 100644 index 16475af..0000000 --- a/tests/test_reconciliation_frames.py +++ /dev/null @@ -1,70 +0,0 @@ -"""S1 (#33): cm/pgm `PredictionFrame` adapters build contract-valid frames. - -Offline — consumes the committed S0 fixture; needs only numpy + views-frames. -""" - -from pathlib import Path - -import numpy as np -import pytest - -from views_frames import PredictionFrame, SpatialLevel -from views_frames.conformance import assert_frame_contract - -from views_postprocessing.reconciliation.frames import prediction_frame_from_arrays - -_FIX = Path(__file__).resolve().parent / "fixtures" / "reconciliation_e2e_parity.npz" - - -@pytest.fixture(scope="module") -def fix(): - return np.load(_FIX) - - -class TestPgmAdapter: - def test_satisfies_contract_and_identity(self, fix): - pf = prediction_frame_from_arrays( - fix["pg_time"], fix["pg_unit"], fix["pg__pred_ged_sb"], level=SpatialLevel.PGM - ) - assert_frame_contract(pf) - assert isinstance(pf, PredictionFrame) - assert pf.index.level is SpatialLevel.PGM - assert pf.n_rows == len(fix["pg_time"]) - assert pf.sample_count == fix["pg__pred_ged_sb"].shape[1] - np.testing.assert_array_equal(pf.index.time, fix["pg_time"]) - np.testing.assert_array_equal(pf.index.unit, fix["pg_unit"]) # priogrid_gid - - -class TestCmAdapter: - def test_satisfies_contract_and_country_units(self, fix): - cf = prediction_frame_from_arrays( - fix["cm_time"], fix["cm_unit"], fix["cm__pred_ged_sb"], level=SpatialLevel.CM - ) - assert_frame_contract(cf) - assert cf.index.level is SpatialLevel.CM - assert cf.n_rows == len(fix["cm_time"]) - np.testing.assert_array_equal(cf.index.unit, fix["cm_unit"]) # country_id - - -class TestBoundaryBehaviour: - def test_values_preserved_and_input_not_mutated(self, fix): - vals = fix["pg__pred_ged_ns"].copy() - before = vals.copy() - pf = prediction_frame_from_arrays( - fix["pg_time"], fix["pg_unit"], vals, level=SpatialLevel.PGM - ) - np.testing.assert_array_equal(pf.values, vals.astype(np.float32)) - np.testing.assert_array_equal(vals, before) # adapter did not mutate input - - def test_non_2d_values_raise(self, fix): - with pytest.raises(ValueError, match="2-D"): - prediction_frame_from_arrays( - fix["pg_time"], fix["pg_unit"], fix["pg_unit"], level=SpatialLevel.PGM - ) - - def test_length_mismatch_raises(self, fix): - with pytest.raises(ValueError, match="length N"): - prediction_frame_from_arrays( - fix["pg_time"][:-1], fix["pg_unit"], fix["pg__pred_ged_sb"], - level=SpatialLevel.PGM, - ) diff --git a/tests/test_reconciliation_grouping.py b/tests/test_reconciliation_grouping.py deleted file mode 100644 index 166a2be..0000000 --- a/tests/test_reconciliation_grouping.py +++ /dev/null @@ -1,70 +0,0 @@ -"""S2 (#34): the grouping core reproduces the oracle's reconciled grid (parity). - -Offline — builds cm/pgm frames from the committed S0 fixture, runs -`reconcile_pgm_to_cm`, and asserts it reproduces the oracle's `recon__*` (the -frozen views-reporting output captured in S0). Needs only numpy + views-frames. -""" - -from pathlib import Path - -import numpy as np -import pytest - -from views_frames import SpatialLevel - -from views_postprocessing.reconciliation.frames import prediction_frame_from_arrays -from views_postprocessing.reconciliation.grouping import reconcile_pgm_to_cm - -_FIX = Path(__file__).resolve().parent / "fixtures" / "reconciliation_e2e_parity.npz" -_TARGETS = ["pred_ged_sb", "pred_ged_ns"] - - -@pytest.fixture(scope="module") -def fix(): - return np.load(_FIX) - - -def _frames(fix, target): - pgm = prediction_frame_from_arrays( - fix["pg_time"], fix["pg_unit"], fix[f"pg__{target}"], level=SpatialLevel.PGM - ) - cm = prediction_frame_from_arrays( - fix["cm_time"], fix["cm_unit"], fix[f"cm__{target}"], level=SpatialLevel.CM - ) - map_keys = np.stack([fix["pg_time"], fix["pg_unit"]], axis=1) - map_vals = fix["pg_country"] - return pgm, cm, map_keys, map_vals - - -class TestGroupingParity: - @pytest.mark.parametrize("target", _TARGETS) - def test_reproduces_oracle(self, fix, target): - pgm, cm, mk, mv = _frames(fix, target) - out = reconcile_pgm_to_cm(pgm, cm, mk, mv) - np.testing.assert_allclose( - out.values, fix[f"recon__{target}"], rtol=1e-5, atol=1e-6, - err_msg=f"grouping core drifts from the oracle on {target}", - ) - - def test_de_mutated(self, fix): - pgm, cm, mk, mv = _frames(fix, "pred_ged_sb") - before = pgm.values.copy() - out = reconcile_pgm_to_cm(pgm, cm, mk, mv) - assert out is not pgm - np.testing.assert_array_equal(pgm.values, before) # input untouched - - -class TestGuards: - def test_missing_country_forecast_raises(self, fix): - # Drop a country from the cm frame -> its grid group has no total. - keep = fix["cm_unit"] != fix["cm_unit"][0] - cm = prediction_frame_from_arrays( - fix["cm_time"][keep], fix["cm_unit"][keep], - fix["cm__pred_ged_sb"][keep], level=SpatialLevel.CM, - ) - pgm = prediction_frame_from_arrays( - fix["pg_time"], fix["pg_unit"], fix["pg__pred_ged_sb"], level=SpatialLevel.PGM - ) - mk = np.stack([fix["pg_time"], fix["pg_unit"]], axis=1) - with pytest.raises(ValueError, match="no country forecast"): - reconcile_pgm_to_cm(pgm, cm, mk, fix["pg_country"]) diff --git a/tests/test_reconciliation_parity.py b/tests/test_reconciliation_parity.py deleted file mode 100644 index 63cd791..0000000 --- a/tests/test_reconciliation_parity.py +++ /dev/null @@ -1,80 +0,0 @@ -"""Parity: the numpy reconciler reproduces the views-reporting torch oracle. - -Phase 2, slice 1. The fixture (``tests/fixtures/reconciliation_parity.npz``) was -captured from the *untouched* views-reporting ``ForecastReconciler`` by -``scripts/gen_reconciliation_parity_fixture.py``. This test needs neither torch -nor views-reporting — it proves the pure-numpy port matches the frozen oracle, -which is the gate for migrating reconciliation here (views-reporting issue #72). - -Parity tolerance is relative (``rtol=1e-5``): the only expected difference is -float32 summation-order noise between torch and numpy; a genuinely different -algorithm diverges by O(value), not O(1e-5). -""" - -from pathlib import Path - -import numpy as np -import pytest - -from views_postprocessing.reconciliation import reconcile_proportional - -_FIXTURE = Path(__file__).resolve().parent / "fixtures" / "reconciliation_parity.npz" - - -@pytest.fixture(scope="module") -def parity_cases(): - data = np.load(_FIXTURE) - labels = data["labels"] - return [ - (str(labels[i]), data[f"grid_{i}"], data[f"country_{i}"], data[f"expected_{i}"]) - for i in range(int(data["n_cases"])) - ] - - -class TestParityWithOracle: - def test_reproduces_oracle_every_case(self, parity_cases): - assert parity_cases, "fixture is empty — regenerate it" - for label, grid, country, expected in parity_cases: - arg_country = float(country) if grid.ndim == 1 else country - got = reconcile_proportional(grid, arg_country) - assert got.shape == expected.shape, f"shape drift in {label}" - np.testing.assert_allclose( - got, expected, rtol=1e-5, atol=1e-6, - err_msg=f"parity drift in case {label}", - ) - - -class TestProperties: - def test_sum_constraint_per_draw(self): - rng = np.random.default_rng(0) - grid = rng.integers(1, 100, (1000, 100)).astype(np.float32) - country = grid.sum(axis=1) * np.float32(1.2) - adjusted = reconcile_proportional(grid, country) - np.testing.assert_allclose(adjusted.sum(axis=1), country, rtol=1e-4) - - def test_zeros_preserved(self): - rng = np.random.default_rng(1) - grid = rng.integers(1, 100, (200, 50)).astype(np.float32) - grid[rng.random((200, 50)) < 0.4] = 0 - country = grid.sum(axis=1) * np.float32(1.5) - adjusted = reconcile_proportional(grid, country) - assert np.all(adjusted[grid == 0] == 0) - - def test_non_negative(self): - rng = np.random.default_rng(2) - grid = rng.standard_normal((100, 50)).astype(np.float32) # has negatives - country = np.abs(grid).sum(axis=1).astype(np.float32) - assert reconcile_proportional(grid, country).min() >= 0 - - def test_point_shape_sum_and_zero(self): - grid = np.array([10.0, 20.0, 30.0, 0.0, 15.0], dtype=np.float32) - adjusted = reconcile_proportional(grid, 100.0) - assert adjusted.shape == grid.shape - assert abs(float(adjusted.sum()) - 100.0) < 1e-2 - assert adjusted[3] == 0.0 - - def test_sample_count_mismatch_raises(self): - grid = np.zeros((100, 50), dtype=np.float32) - country = np.zeros(200, dtype=np.float32) - with pytest.raises(ValueError, match="Mismatch in sample count"): - reconcile_proportional(grid, country) diff --git a/tests/test_reconciliation_scale.py b/tests/test_reconciliation_scale.py deleted file mode 100644 index 4c65ce2..0000000 --- a/tests/test_reconciliation_scale.py +++ /dev/null @@ -1,72 +0,0 @@ -"""C-38 regression guard: reconciliation grouping scales (not O(groups x N)). - -Reconciles a frame with **many** (time, country) groups and asserts it stays -correct and fast. The primary guard is correctness-at-scale — the group-by-sort -`argsort`/`bounds` logic (`grouping.py`) must hold conservation across thousands -of groups; the time budget is a loose smoke. The real global-volume verification -is the S7 dry-run (register C-38). Offline; numpy + views-frames only. -""" - -import time - -import numpy as np - -from views_frames import SpatialLevel - -from views_postprocessing.reconciliation import ReconciliationModule -from views_postprocessing.reconciliation.frames import prediction_frame_from_arrays - - -def _many_group_frames(n_countries=2000, cells_per=8, month=528, samples=20): - """A single-month frame with `n_countries` groups of `cells_per` grid cells.""" - rng = np.random.default_rng(0) - times, grids, gcountry, cm_units = [], [], [], [] - g = 1000 - for c in range(1, n_countries + 1): - for _ in range(cells_per): - times.append(month) - grids.append(g) - gcountry.append(c) - g += 1 - cm_units.append(c) - times = np.asarray(times) - grids = np.asarray(grids) - gcountry = np.asarray(gcountry) - - pg = rng.gamma(2.0, 5.0, (len(grids), samples)).astype(np.float32) - pg[rng.random(pg.shape) < 0.3] = 0.0 # sparsity -> some all-zero groups - cm_vals = rng.gamma(3.0, 20.0, (n_countries, samples)).astype(np.float32) - - pgm = prediction_frame_from_arrays(times, grids, pg, level=SpatialLevel.PGM) - cm = prediction_frame_from_arrays( - np.full(n_countries, month), np.asarray(cm_units), cm_vals, level=SpatialLevel.CM - ) - map_keys = np.stack([times, grids], axis=1) - return cm, pgm, map_keys, gcountry, month - - -class TestScale: - def test_many_groups_correct_and_fast(self): - cm, pgm, map_keys, gcountry, month = _many_group_frames() - rm = ReconciliationModule(map_keys, gcountry) - - t0 = time.perf_counter() - out = rm.reconcile(cm, pgm) - elapsed = time.perf_counter() - t0 - - assert out.values.shape == pgm.values.shape - # Loose smoke: a reintroduced O(groups x N) grouping would balloon this. - assert elapsed < 10.0, f"reconcile unexpectedly slow ({elapsed:.1f}s)" - - # Correctness across many groups: conservation holds per draw, except - # all-zero-input draws (which stay zero) — checked on a spread of countries. - cmv = {int(u): cm.values[i] for i, u in enumerate(cm.index.unit)} - for c in (1, 137, 1000, 2000): - sel = np.nonzero(gcountry == c)[0] - inp_allzero = (pgm.values[sel] == 0).all(axis=0) - grid_sum = out.values[sel].sum(axis=0) - active = ~inp_allzero - np.testing.assert_allclose( - grid_sum[active], cmv[c][active], rtol=1e-3, atol=1e-2 - ) - assert (grid_sum[inp_allzero] == 0).all() diff --git a/tests/test_reconciliation_validation.py b/tests/test_reconciliation_validation.py deleted file mode 100644 index 0bb3a8d..0000000 --- a/tests/test_reconciliation_validation.py +++ /dev/null @@ -1,76 +0,0 @@ -"""S3 (#35): fail-loud validation guards for reconciliation inputs. - -Offline — builds frames from the committed S0 fixture; numpy + views-frames only. -""" - -from pathlib import Path - -import numpy as np -import pytest - -from views_frames import SpatialLevel - -from views_postprocessing.reconciliation.frames import prediction_frame_from_arrays -from views_postprocessing.reconciliation.validation import validate_reconciliation_inputs - -_FIX = Path(__file__).resolve().parent / "fixtures" / "reconciliation_e2e_parity.npz" - - -@pytest.fixture(scope="module") -def fix(): - return np.load(_FIX) - - -def _cm(fix, time=None, unit=None, vals=None): - return prediction_frame_from_arrays( - fix["cm_time"] if time is None else time, - fix["cm_unit"] if unit is None else unit, - fix["cm__pred_ged_sb"] if vals is None else vals, - level=SpatialLevel.CM, - ) - - -def _pgm(fix): - return prediction_frame_from_arrays( - fix["pg_time"], fix["pg_unit"], fix["pg__pred_ged_sb"], level=SpatialLevel.PGM - ) - - -def _mk(fix): - return np.stack([fix["pg_time"], fix["pg_unit"]], axis=1), fix["pg_country"] - - -class TestValidInputsPass: - def test_no_raise_on_consistent_inputs(self, fix): - mk, mv = _mk(fix) - validate_reconciliation_inputs(_cm(fix), _pgm(fix), mk, mv) # must not raise - - -class TestGuards: - def test_wrong_level_raises(self, fix): - mk, mv = _mk(fix) - # pass the pgm frame where a cm frame is expected - with pytest.raises(ValueError, match="SpatialLevel.CM"): - validate_reconciliation_inputs(_pgm(fix), _pgm(fix), mk, mv) - - def test_sample_count_mismatch_raises(self, fix): - mk, mv = _mk(fix) - cm = _cm(fix, vals=fix["cm__pred_ged_sb"][:, :50]) # half the samples - with pytest.raises(ValueError, match="sample-count mismatch"): - validate_reconciliation_inputs(cm, _pgm(fix), mk, mv) - - def test_time_mismatch_raises(self, fix): - mk, mv = _mk(fix) - keep = fix["cm_time"] != 530 # drop a month from cm - cm = _cm(fix, time=fix["cm_time"][keep], unit=fix["cm_unit"][keep], - vals=fix["cm__pred_ged_sb"][keep]) - with pytest.raises(ValueError, match="different time steps"): - validate_reconciliation_inputs(cm, _pgm(fix), mk, mv) - - def test_missing_country_raises(self, fix): - mk, mv = _mk(fix) - keep = fix["cm_unit"] != fix["cm_unit"][0] # drop a country from cm - cm = _cm(fix, time=fix["cm_time"][keep], unit=fix["cm_unit"][keep], - vals=fix["cm__pred_ged_sb"][keep]) - with pytest.raises(ValueError, match="no country forecast"): - validate_reconciliation_inputs(cm, _pgm(fix), mk, mv) diff --git a/views_postprocessing/reconciliation/__init__.py b/views_postprocessing/reconciliation/__init__.py deleted file mode 100644 index 19f26ad..0000000 --- a/views_postprocessing/reconciliation/__init__.py +++ /dev/null @@ -1,11 +0,0 @@ -"""Forecast reconciliation (pgm forecasts reconciled to cm totals). - -Slice 1 ports the top-down proportional method from views-reporting as a pure -numpy function. New methods (e.g. principled probabilistic reconciliation, C-37) -should be added as sibling modules, not by modifying ``proportional``. -""" - -from views_postprocessing.reconciliation.module import ReconciliationModule -from views_postprocessing.reconciliation.proportional import reconcile_proportional - -__all__ = ["ReconciliationModule", "reconcile_proportional"] diff --git a/views_postprocessing/reconciliation/frames.py b/views_postprocessing/reconciliation/frames.py deleted file mode 100644 index 9db2500..0000000 --- a/views_postprocessing/reconciliation/frames.py +++ /dev/null @@ -1,62 +0,0 @@ -"""Array → `PredictionFrame` adapters for reconciliation (epic #31, story #33). - -Reconciliation works on views-frames `PredictionFrame`s at two spatial levels: -country (`cm`) and PRIO-GRID (`pgm`). Both are built the same way from -`(time, unit, values)` arrays — only the `SpatialLevel` and the unit identifier -(`country_id` vs `priogrid_gid`) differ. Predictions here carry a real posterior -sample axis, so values are `(N, S)` with `S >= 1`. - -This is the reconciliation package's **own** I/O: it differs from the unfao -delivery adapters (`unfao/frames.py`), which wrap pandas *scalar* columns as -`(N, 1)` point frames — so per CRP they are not forced together. numpy + -views-frames only; no torch, no pandas. -""" - -from __future__ import annotations - -import numpy as np -from numpy.typing import NDArray - -from views_frames import FrameMetadata, PredictionFrame, SpatialLevel, SpatioTemporalIndex - - -def prediction_frame_from_arrays( - time: NDArray[np.integer] | object, - unit: NDArray[np.integer] | object, - values: NDArray[np.floating] | object, - *, - level: SpatialLevel, - metadata: FrameMetadata | None = None, -) -> PredictionFrame: - """Build a `PredictionFrame` from `(time, unit, values)` at ``level``. - - Args: - time: 1-D integer array, length ``N`` (``month_id``). - unit: 1-D integer array, length ``N`` — ``country_id`` for - ``SpatialLevel.CM``, ``priogrid_gid`` for ``SpatialLevel.PGM``. - values: ``(N, S)`` float32-coercible array of posterior samples. - level: the frame's spatial level. - - Returns: - A `PredictionFrame` of shape ``(N, S)`` at ``level``. The values buffer - is reused without copy when already float32 (views-frames C-07); the - input arrays are never mutated. - - Raises: - ValueError: ``values`` is not 2-D, or ``time``/``unit`` are not 1-D of - length ``N``. - """ - time_arr = np.asarray(time, dtype=np.int64) - unit_arr = np.asarray(unit, dtype=np.int64) - vals = np.asarray(values, dtype=np.float32) - - if vals.ndim != 2: - raise ValueError(f"values must be 2-D (N, S); got ndim={vals.ndim}") - if time_arr.shape != (vals.shape[0],) or unit_arr.shape != (vals.shape[0],): - raise ValueError( - f"time {time_arr.shape} and unit {unit_arr.shape} must both be 1-D " - f"of length N={vals.shape[0]}" - ) - - index = SpatioTemporalIndex(time=time_arr, unit=unit_arr, level=level) - return PredictionFrame(vals, index, metadata) diff --git a/views_postprocessing/reconciliation/grouping.py b/views_postprocessing/reconciliation/grouping.py deleted file mode 100644 index f31dcde..0000000 --- a/views_postprocessing/reconciliation/grouping.py +++ /dev/null @@ -1,87 +0,0 @@ -"""Reconcile a pgm `PredictionFrame` to cm country totals (epic #31, story #34). - -The heart of the migration: for each `(time, country)`, scale that country's grid -cells so their per-draw sum matches the country forecast, using the parity-proven -leaf `reconcile_proportional` (PR #30). Grid rows are labelled by country with -views-frames `cross_level_align` — the sanctioned cm↔pgm primitive, which fails -loud if any grid row lacks a country (mirrors the original's "valid countries" -guard). De-mutated: returns a **new** pgm frame (C-184); the input is untouched. - -numpy + views-frames only. The loop is over `(time, country)` groups (a small -number), not over rows; each group call is fully vectorised over cells × samples. -""" - -from __future__ import annotations - -import numpy as np -from numpy.typing import NDArray - -from views_frames import PredictionFrame, SpatialLevel - -from views_postprocessing.reconciliation.proportional import reconcile_proportional - - -def reconcile_pgm_to_cm( - pgm_frame: PredictionFrame, - cm_frame: PredictionFrame, - map_keys: NDArray[np.integer] | object, - map_vals: NDArray[np.integer] | object, -) -> PredictionFrame: - """Return a new pgm `PredictionFrame` reconciled to ``cm_frame``'s totals. - - Args: - pgm_frame: grid forecasts at PGM level, values ``(N_pg, S)``. - cm_frame: country forecasts at CM level, values ``(N_cm, S)``. - map_keys: ``(M, 2)`` int ``(time, priogrid_gid)`` covering every pgm row. - map_vals: ``(M,)`` int ``country_id`` for each key (injected; geography is - never embedded here — views-frames ADR-014). - - Returns: - A new pgm `PredictionFrame` (same index/metadata as ``pgm_frame``) whose - cells sum, per draw, to the country forecast — except all-zero country - draws, which stay zero (the leaf's documented edge case). - - Raises: - ValueError: a grid row has no country mapping (raised by - ``cross_level_align``), or a ``(time, country)`` group has no matching - country forecast in ``cm_frame``. - """ - # 1. Label every grid row with its country (cm-level units); fails loud if a - # row's (time, priogrid) is absent from the injected mapping. - cm_units = pgm_frame.index.cross_level_align_arrays( - np.asarray(map_keys), np.asarray(map_vals), SpatialLevel.CM - ).unit # (N_pg,) - pg_time = pgm_frame.index.time - - # 2. (time, country) -> row position in the country frame. - cm_time, cm_unit, cm_vals = cm_frame.index.time, cm_frame.index.unit, cm_frame.values - cm_pos = {(int(cm_time[j]), int(cm_unit[j])): j for j in range(cm_frame.n_rows)} - - # 3. Group grid rows by (time, country) and reconcile each group with the leaf. - # Group-by-sort: a single argsort lays the rows out contiguously per group - # (O(N log N), one index array) — no per-group full-array scan — so it - # scales to the full grid (register C-38). Group iteration order is - # irrelevant: each group writes only its own rows. - pg_vals = pgm_frame.values - out = np.empty_like(pg_vals) - group_key = np.stack([pg_time, cm_units], axis=1) # (N_pg, 2) - unique_groups, inverse, counts = np.unique( - group_key, axis=0, return_inverse=True, return_counts=True - ) - inverse = np.asarray(inverse).reshape(-1) - order = np.argsort(inverse, kind="stable") # rows grouped contiguously by group - bounds = np.concatenate(([0], np.cumsum(counts))) # group gi -> order[bounds[gi]:bounds[gi+1]] - - for gi in range(unique_groups.shape[0]): - t, c = int(unique_groups[gi, 0]), int(unique_groups[gi, 1]) - if (t, c) not in cm_pos: - raise ValueError( - f"grid group (time={t}, country={c}) has no country forecast in cm_frame" - ) - rows = order[bounds[gi]:bounds[gi + 1]] - country_total = cm_vals[cm_pos[(t, c)]] # (S,) - # leaf convention: grid is (samples, cells); our frame slice is (cells, samples) - scaled = reconcile_proportional(pg_vals[rows].T, country_total) # (S, n_cells) - out[rows] = scaled.T # back to (n_cells, S) - - return PredictionFrame(out, pgm_frame.index, pgm_frame.metadata) diff --git a/views_postprocessing/reconciliation/module.py b/views_postprocessing/reconciliation/module.py deleted file mode 100644 index 4ba306f..0000000 --- a/views_postprocessing/reconciliation/module.py +++ /dev/null @@ -1,67 +0,0 @@ -"""Frames-native reconciliation orchestration (epic #31, story #36). - -`ReconciliationModule` holds the injected `(time, priogrid_gid) -> country_id` -mapping (geography is injected, never embedded — views-frames ADR-014) and -applies it: `reconcile(cm_frame, pgm_frame)` validates the inputs and scales the -grid forecasts to the country totals, returning a **new** pgm frame (de-mutated, -C-184). - -**SRP:** orchestration only — the scaling math is the leaf (`proportional`), the -grouping is `grouping`, the guards are `validation`, the I/O is `frames`. No -torch, no pandas, no viewser, no wandb: the original's `ProcessPoolExecutor` and -WandB alerting are dropped (numpy is fast; there is no GPU). If scale ever needs -parallelism, add it behind this same interface (OCP). Multi-target inputs are -reconciled by calling `reconcile` once per target. -""" - -from __future__ import annotations - -import numpy as np -from numpy.typing import NDArray - -from views_frames import PredictionFrame - -from views_postprocessing.reconciliation.grouping import reconcile_pgm_to_cm -from views_postprocessing.reconciliation.validation import validate_reconciliation_inputs - - -class ReconciliationModule: - """Reconcile pgm forecasts to cm country totals (one target per call).""" - - def __init__( - self, - map_keys: NDArray[np.integer] | object, - map_vals: NDArray[np.integer] | object, - ) -> None: - """Inject the `(time, priogrid_gid) -> country_id` mapping. - - Args: - map_keys: ``(M, 2)`` int array of ``(time, priogrid_gid)`` pairs. - map_vals: ``(M,)`` int ``country_id`` for each key. - - Raises: - ValueError: ``map_keys`` is not ``(M, 2)`` or ``map_vals`` is not - length ``M``. - """ - keys = np.asarray(map_keys) - vals = np.asarray(map_vals) - if keys.ndim != 2 or keys.shape[1] != 2: - raise ValueError("map_keys must be an (M, 2) array of (time, priogrid_gid)") - if vals.shape != (keys.shape[0],): - raise ValueError("map_vals must be a length-M array aligned to map_keys") - self._map_keys = keys - self._map_vals = vals - - def reconcile( - self, cm_frame: PredictionFrame, pgm_frame: PredictionFrame - ) -> PredictionFrame: - """Validate the inputs, then return a new pgm frame reconciled to cm totals. - - Raises: - ValueError: the inputs fail validation (level / sample-count / time - coverage / missing country forecast). - """ - validate_reconciliation_inputs( - cm_frame, pgm_frame, self._map_keys, self._map_vals - ) - return reconcile_pgm_to_cm(pgm_frame, cm_frame, self._map_keys, self._map_vals) diff --git a/views_postprocessing/reconciliation/proportional.py b/views_postprocessing/reconciliation/proportional.py deleted file mode 100644 index 2c21167..0000000 --- a/views_postprocessing/reconciliation/proportional.py +++ /dev/null @@ -1,72 +0,0 @@ -"""Top-down proportional reconciliation (numpy port — phase 2, slice 1). - -Makes PRIO-GRID-month (pgm) forecasts sum to their country-month (cm) total by -**top-down disaggregation using forecast proportions** (FPP3 terminology), -applied **per posterior draw**: within a draw, each grid cell keeps its relative -share and the cells are rescaled so they sum to that draw's country total. Zeros -stay zero; country totals are authoritative; the result is non-negative. - -This is a *faithful, numpy-only* port of views-reporting's -``ForecastReconciler.reconcile_forecast`` (torch), migrated here because the -algorithm belongs in post-processing, not reporting (views-reporting issue #72). -It is intentionally the **same** method — a pragmatic per-draw approximation, not -principled joint probabilistic reconciliation. The upgrade to the latter is -tracked as **C-37** and is deliberately deferred until this port's parity with -the original is proven and the move is wired. - -No torch, no pandas — numpy only. -""" - -from __future__ import annotations - -import numpy as np -from numpy.typing import NDArray - -_EPS = np.float32(1e-8) - - -def reconcile_proportional( - grid: NDArray[np.floating] | object, - country: NDArray[np.floating] | float | object, -) -> NDArray[np.float32]: - """Rescale grid forecasts so each draw sums to its country total. - - Args: - grid: Grid-level forecasts, float32-coercible. Either - ``(num_samples, num_grid_cells)`` (probabilistic) or - ``(num_grid_cells,)`` (point). - country: Country-level total. Either ``(num_samples,)`` (probabilistic) - or a scalar (point). Must align with ``grid``'s sample axis. - - Returns: - Adjusted grid forecasts, float32, same shape as ``grid``. ``sum`` over - grid cells equals ``country`` per sample; zero cells stay zero; values - are clamped to be non-negative. - - Raises: - ValueError: the grid and country sample counts disagree. - """ - grid_arr = np.asarray(grid, dtype=np.float32) - is_point = grid_arr.ndim == 1 - - if is_point: - grid_arr = grid_arr[np.newaxis, :] # (1, N) - country_arr = np.asarray([country], dtype=np.float32) - else: - country_arr = np.asarray(country, dtype=np.float32).reshape(-1) - - if grid_arr.shape[0] != country_arr.shape[0]: - raise ValueError( - f"Mismatch in sample count: grid has {grid_arr.shape[0]}, " - f"country has {country_arr.shape[0]}" - ) - - # Preserve zeros: only strictly-positive cells carry probability mass. - nonzero = np.where(grid_arr > 0, grid_arr, np.float32(0.0)) - - # Per-draw proportional scaling to the (authoritative) country total. - sum_nonzero = nonzero.sum(axis=1, keepdims=True) # (S, 1) - scaling = country_arr.reshape(-1, 1) / (sum_nonzero + _EPS) # (S, 1) - adjusted = np.clip(nonzero * scaling, 0.0, None).astype(np.float32) - - return adjusted[0] if is_point else adjusted diff --git a/views_postprocessing/reconciliation/validation.py b/views_postprocessing/reconciliation/validation.py deleted file mode 100644 index 082bf79..0000000 --- a/views_postprocessing/reconciliation/validation.py +++ /dev/null @@ -1,77 +0,0 @@ -"""Fail-loud validation for reconciliation inputs (epic #31, story #35). - -Ports views-reporting `ReconciliationModule.__init__`'s guards to frames-native -checks, run before any work. SRP: small, independently testable helpers — not -buried in the orchestrator. The original's intents map as: - - - dataset type checks -> spatial-level guard (cm@CM, pgm@PGM) - - same time steps + exact overlap -> identical set of time values - - (per-draw scaling needs it) -> equal `sample_count` - - valid countries -> every (time, country) the grid maps to has - a country forecast in the cm frame - -The original's "different time units" (e.g. month_id vs year_id) check is -subsumed by the level guard; per-target intersection is handled by the -orchestrator, since frames here are single-target. -""" - -from __future__ import annotations - -import numpy as np -from numpy.typing import NDArray - -from views_frames import PredictionFrame, SpatialLevel - - -def validate_reconciliation_inputs( - cm_frame: PredictionFrame, - pgm_frame: PredictionFrame, - map_keys: NDArray[np.integer] | object, - map_vals: NDArray[np.integer] | object, -) -> None: - """Raise ``ValueError`` if the reconciliation inputs are inconsistent. - - Checks spatial levels, sample-count alignment, identical time coverage, and - that every country the grid maps to has a forecast in ``cm_frame``. - """ - if cm_frame.index.level is not SpatialLevel.CM: - raise ValueError( - f"country frame must be at SpatialLevel.CM, got {cm_frame.index.level}" - ) - if pgm_frame.index.level is not SpatialLevel.PGM: - raise ValueError( - f"grid frame must be at SpatialLevel.PGM, got {pgm_frame.index.level}" - ) - if cm_frame.sample_count != pgm_frame.sample_count: - raise ValueError( - f"sample-count mismatch: cm has {cm_frame.sample_count}, " - f"pgm has {pgm_frame.sample_count}" - ) - - cm_times = {int(t) for t in np.unique(cm_frame.index.time)} - pg_times = {int(t) for t in np.unique(pgm_frame.index.time)} - if cm_times != pg_times: - raise ValueError( - "cm and pgm cover different time steps: " - f"cm-only={sorted(cm_times - pg_times)}, " - f"pgm-only={sorted(pg_times - cm_times)}" - ) - - # Valid countries: every (time, country) the grid maps to must have a forecast. - cm_units = pgm_frame.index.cross_level_align_arrays( - np.asarray(map_keys), np.asarray(map_vals), SpatialLevel.CM - ).unit - needed = { - (int(t), int(c)) - for t, c in zip(pgm_frame.index.time, cm_units, strict=True) - } - have = { - (int(t), int(c)) - for t, c in zip(cm_frame.index.time, cm_frame.index.unit, strict=True) - } - missing = needed - have - if missing: - raise ValueError( - f"{len(missing)} grid group(s) have no country forecast in cm_frame, " - f"e.g. {sorted(missing)[:5]}" - )