diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 5baf54f3..6b9cf94b 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -37,11 +37,30 @@ jobs: - name: Run tests (shard ${{ matrix.shard }}/4) run: pytest -q --splits 4 --group ${{ matrix.shard }} --splitting-algorithm duration_based_chunks --durations-path .test_durations + graph-integration: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ['3.13', '3.14'] + steps: + - uses: actions/checkout@v6 + - uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python-version }} + cache: 'pip' + - name: Install the pinned optional graph integration + run: pip install -e ".[graph]" pytest + - name: Require typed graph capabilities + run: python -c "from populace_dynamics.graph._compat import require_graph; require_graph()" + - name: Run graph and existing mortality regressions + run: pytest -q tests/test_graph_mortality.py tests/test_m6_engine_refit.py tests/test_m6_engine_steps.py + # Fan-in jobs keeping the branch-protection context names # ("pytest (3.11)" / "pytest (3.13)") stable across the shard split. pytest: name: pytest (${{ matrix.python-version }}) - needs: pytest-shard + needs: [pytest-shard, graph-integration] if: always() runs-on: ubuntu-latest strategy: @@ -49,4 +68,6 @@ jobs: python-version: ['3.14'] steps: - name: Verify all shards passed - run: test "${{ needs.pytest-shard.result }}" = "success" + run: | + test "${{ needs.pytest-shard.result }}" = "success" + test "${{ needs.graph-integration.result }}" = "success" diff --git a/docs/population-graph.md b/docs/population-graph.md new file mode 100644 index 00000000..39e74771 --- /dev/null +++ b/docs/population-graph.md @@ -0,0 +1,128 @@ +# First population graph: mortality and ageing + +The optional `populace_dynamics.graph` package fits the existing M6 mortality +model on a historical synthetic panel, applies the fitted artifact to a +separate starting population, and adds surviving observations for the next +year. It produces an engineering report and a content-addressed execution +manifest. It does not change the existing projection loop, candidate +registries, scientific gates, or committed evidence. + +## Dependencies and execution + +The graph example requires Python **3.13 or 3.14**, NumPy 2+, pandas 2.3+, +and the Microcosm graph/frame revisions containing typed artifact edges and +`microcosm.graph.randomness.keyed_uniform`. The default Dynamics installation +continues to support Python 3.10–3.14 without importing Microcosm. The entry +point checks capabilities and gives installation guidance when they are absent. + +The `graph` extra pins both graph and Frame to core commit +`3ff92b0aea14407d09479bff5623dc7d1a92d008`. An older package merely sharing +the version number `0.1.0` is insufficient. In an isolated Python 3.13 or 3.14 +environment, install with `uv pip install '.[graph]'`. The core change must be +reviewed before this dependent integration is released; replace the Git pins +with a compatible published release when one exists. CI installs this exact +extra, refuses missing capabilities, and runs the integration on both supported +Python versions. Do not modify the existing scientific gate environment. No +rules engine, restricted microdata, or optional forest fitter is needed here. + +Run from that environment, choosing an output directory: + +```sh +python -m populace_dynamics.graph --synthetic --output-dir ./mortality-example +``` + +The command creates small synthetic inputs under `mortality-example/inputs`. +It preserves existing input files so that edits can test cache invalidation. +Repeated execution reuses the verified store under `mortality-example/store`. +The report, manifest, fitted JSON model, entity tables, and next-period slice +are written inside the chosen output directory. A failed engineering or +fixture verdict exits nonzero while retaining the diagnostics. + +Four explicit source paths can replace the generated inputs: + +```sh +python -m populace_dynamics.graph \ + --training ./inputs/training.json --rates ./inputs/rates.json \ + --initial ./inputs/initial.json --holdout ./inputs/holdout.json \ + --boundary-year 2014 --external-vintage-year 2014 \ + --experiment-id comparison-a --replicate 0 --base-seed 0 \ + --output-dir ./mortality-example +``` + +These inputs still exercise the synthetic engineering contract. The example +does not confer validity on a real-population projection. Source JSON rejects +duplicate members and nonfinite values. Each source is declared separately; +holdout bytes are available only to evaluation. Domain kernels read the +content-verified JSON directly: the registered source marker deliberately +does not pretend an external rate table or a holdout report is a population. + +## Executable ownership + +The graph has two CREATE roots, each carrying a `person_period` observation +entity and `person` and `period` groups. `person` retains stable identities; +`period.period` is the immutable mass-partition label. The training root +contains exposure records; the initial root contains recipients. Their only +connection is the explicitly typed mortality-model artifact. + +The fit node calls `prepare_mortality_refit_inputs` and +`fit_mortality_model`. Event year, required interview year, and declared +external vintage retain the existing cutoff checks. The JSON model contains +validated contiguous age bands, sex-specific probabilities, fit boundary, +external vintage, and retained row count. The manifest binds its producer +to source identities and implementation digests. The fitter's external-rate +factor cancels in its fitted-window level, so this is not evidence of +independent external calibration. + +Application calls `apply_mortality` with a graph-specific context. Every +uniform is keyed by the original person identity, process, year, and draw +index under the chosen experiment/replicate/seed. It does not use the legacy +ID-sorted ordinal registry. Reordering, splitting, or adding unrelated +people preserves the existing people's draws. Fit and application declare +platform-specific bitwise numeric behavior conservatively; cross-platform +equality is not claimed. + +EXPAND calls `advance_age` on survivors, adds their next-period observations +with lineage to the original observations, and attaches them to one newly +admitted period group. A same-version rewrite node claims the materialized +age values. Historical ages and memberships remain unchanged. The temporary +`year` returned by `advance_age` is never written over a carried observation +column. No new person, birth, or immigrant is implied by admission of the +period group. + +Typed person-period weights are the single authority. Every survivor carries +the same trajectory weight into the new period. The declared mass receipt +shows historical mass unchanged and new-period mass equal to surviving +weight. Total stored observation mass therefore grows by the additional +period. If everyone dies, the graph adds no observations and no orphan period +group; the report explicitly records next-period mass zero. + +## Evaluation and limits + +The report separates `engineering_verdict` (survivor/age parity and population +structure) from `fixture_verdict` (the independently sourced synthetic +death-rate and age expectations). It records weighted expected, observed, +and generated deaths, row counts, period mass, node/model identities, and +cache reuse. The fixture death-rate tolerance is an input named +`fixture_max_abs_death_rate_gap`; it is not a scientific acceptance threshold. +Changing all held-out outcomes to deaths fails that fixture check while +leaving fitting, application, draws, and accounting unchanged. + +Household accounting is explicitly unsupported and refused by the Python +entry point. Household weight sharing, marriage, births, immigration, +alignment replay, repeated years, and the full M6 loop remain later work. +No certified data release or scientific candidate is produced by this graph. + +## Tests + +```sh +python -m pytest -q tests/test_graph_mortality.py \ + tests/test_m6_engine_refit.py tests/test_m6_engine_steps.py +``` + +The integration tests cover direct execution with an independently injected +uniform vector, JSON validation, cutoff and holdout isolation, fitted-artifact +reuse, changed fitting weights, row/chunk/person invariance, cold/warm stores, +and zero/all-survivor expansion. They skip the optional runtime cases when +the required core capabilities are unavailable; the JSON and dependency +boundary tests still run. Importing `populace_dynamics.graph` remains safe +under Python 3.10–3.12. diff --git a/pyproject.toml b/pyproject.toml index 370d161a..3bdf706c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,6 +47,12 @@ dependencies = [ ] [project.optional-dependencies] +# Typed artifacts are not yet in the published 0.1.0 packages. Pin both +# graph and Frame to the same reviewed core revision; this extra needs 3.13+. +graph = [ + "microcosm-frame @ git+https://github.com/PolicyEngine/microcosm.git@3ff92b0aea14407d09479bff5623dc7d1a92d008#subdirectory=packages/microcosm-frame", + "microcosm-graph @ git+https://github.com/PolicyEngine/microcosm.git@3ff92b0aea14407d09479bff5623dc7d1a92d008#subdirectory=packages/microcosm-graph", +] dev = [ "pytest>=7.4.0", "black>=23.7.0", diff --git a/scripts/first_estimates_birth_evidence.py b/scripts/first_estimates_birth_evidence.py index efae0662..911ae21f 100644 --- a/scripts/first_estimates_birth_evidence.py +++ b/scripts/first_estimates_birth_evidence.py @@ -156,6 +156,15 @@ Path("src/populace_dynamics/estimates/anchor_context_registry.py"), Path("src/populace_dynamics/estimates/anchor_context_rehearsal.py"), Path("src/populace_dynamics/estimates/anchor_context_report.py"), + # The opt-in graph integration is outside the historical reducer and + # registered production call paths. Keep exact file exclusions, with + # import-reachability coverage, rather than changing any evidence pin. + Path("src/populace_dynamics/graph/__init__.py"), + Path("src/populace_dynamics/graph/__main__.py"), + Path("src/populace_dynamics/graph/_compat.py"), + Path("src/populace_dynamics/graph/model.py"), + Path("src/populace_dynamics/graph/runtime.py"), + Path("src/populace_dynamics/graph/synthetic.py"), ) POST_REVIEW_SHARED_SOURCE_BLOBS = { Path( diff --git a/src/populace_dynamics/graph/__init__.py b/src/populace_dynamics/graph/__init__.py new file mode 100644 index 00000000..03c8c7ab --- /dev/null +++ b/src/populace_dynamics/graph/__init__.py @@ -0,0 +1,18 @@ +"""Optional synthetic population-graph integration. + +Importing this package does not import Microcosm or change legacy execution. +The graph entry point checks Python and the installed graph capabilities. +""" + + +def run_mortality_graph(**kwargs): + """Run the existing mortality/ageing operations through Microcosm.""" + from ._compat import require_graph + + require_graph() + from .runtime import run_mortality_graph as run + + return run(**kwargs) + + +__all__ = ["run_mortality_graph"] diff --git a/src/populace_dynamics/graph/__main__.py b/src/populace_dynamics/graph/__main__.py new file mode 100644 index 00000000..fb502f35 --- /dev/null +++ b/src/populace_dynamics/graph/__main__.py @@ -0,0 +1,66 @@ +"""Run the mortality graph with explicit inputs and output placement.""" + +import argparse +from pathlib import Path + +from . import run_mortality_graph +from .synthetic import write_synthetic_inputs + + +def parser(): + result = argparse.ArgumentParser(description=__doc__) + result.add_argument("--output-dir", type=Path, required=True) + result.add_argument("--synthetic", action="store_true") + for name in ("training", "rates", "initial", "holdout"): + result.add_argument(f"--{name}", type=Path) + result.add_argument("--boundary-year", type=int, default=2014) + result.add_argument("--external-vintage-year", type=int, default=2014) + result.add_argument("--experiment-id", default="mortality") + result.add_argument("--replicate", type=int, default=0) + result.add_argument("--base-seed", type=int, default=0) + return result + + +def main(argv=None): + arg_parser = parser() + args = arg_parser.parse_args(argv) + sources = { + name: getattr(args, name) + for name in ("training", "rates", "initial", "holdout") + } + if args.synthetic: + if any(sources.values()): + arg_parser.error("--synthetic cannot be combined with input paths") + if args.boundary_year != 2014: + arg_parser.error( + "the supplied synthetic fixture has boundary year 2014" + ) + sources = write_synthetic_inputs(args.output_dir / "inputs") + elif not all(sources.values()): + arg_parser.error("supply all four input paths or --synthetic") + try: + run = run_mortality_graph( + **sources, + output_dir=args.output_dir, + boundary_year=args.boundary_year, + external_vintage_year=args.external_vintage_year, + experiment_id=args.experiment_id, + replicate=args.replicate, + base_seed=args.base_seed, + ) + except (ImportError, ValueError) as error: + arg_parser.exit(2, f"{error}\n") + print( + f"{args.output_dir / 'report.json'}: engineering={run.report['engineering_verdict']}, fixture={run.report['fixture_verdict']}" + ) + return ( + 0 + if run.report["engineering_verdict"] + == run.report["fixture_verdict"] + == "pass" + else 1 + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/populace_dynamics/graph/_compat.py b/src/populace_dynamics/graph/_compat.py new file mode 100644 index 00000000..86ffff07 --- /dev/null +++ b/src/populace_dynamics/graph/_compat.py @@ -0,0 +1,36 @@ +"""An explicit dependency boundary for the optional graph example.""" + +import importlib +import sys + +_GUIDANCE = ( + "The population graph needs Microcosm's typed model-artifact and keyed " + "randomness interfaces. Install the reviewed microcosm-graph and " + "microcosm-frame revisions together; see docs/population-graph.md. " + "Legacy Dynamics does not require these packages." +) + + +def _python_version(): + return sys.version_info[:2] + + +def require_graph(): + """Refuse unsupported Python or a graph lacking the required interfaces.""" + if _python_version() < (3, 13): + raise ImportError( + "The optional population graph requires Python >=3.13." + ) + try: + decl = importlib.import_module("microcosm.graph.decl") + kernel = importlib.import_module("microcosm.graph.kernel") + randomness = importlib.import_module("microcosm.graph.randomness") + except (ImportError, SyntaxError) as error: + raise ImportError(_GUIDANCE) from error + for name in ("ArtifactType", "ArtifactInput", "ArtifactOutput"): + if not hasattr(decl, name): + raise ImportError(_GUIDANCE) + if not hasattr(kernel.SeedSource, "KEYED") or not hasattr( + randomness, "keyed_uniform" + ): + raise ImportError(_GUIDANCE) diff --git a/src/populace_dynamics/graph/model.py b/src/populace_dynamics/graph/model.py new file mode 100644 index 00000000..0f2da4be --- /dev/null +++ b/src/populace_dynamics/graph/model.py @@ -0,0 +1,179 @@ +"""Validated JSON for the existing fitted M6 mortality law; no pickle.""" + +from __future__ import annotations + +import json +import math +from dataclasses import dataclass + +_FORMAT = "populace-dynamics.mortality" + + +def _unique_object(pairs): + result = {} + for key, value in pairs: + if key in result: + raise ValueError(f"duplicate JSON member {key!r}") + result[key] = value + return result + + +def read_json(path): + """Read JSON without accepting duplicate fields or nonfinite numbers.""" + return parse_json(path.read_bytes()) + + +def parse_json(payload): + def invalid(value): + raise ValueError(f"nonfinite JSON number {value}") + + try: + return json.loads( + payload, object_pairs_hook=_unique_object, parse_constant=invalid + ) + except (UnicodeError, json.JSONDecodeError) as error: + raise ValueError("invalid mortality JSON") from error + + +def json_bytes(value): + return json.dumps( + value, sort_keys=True, separators=(",", ":"), allow_nan=False + ).encode("utf-8") + + +def _integer(value, label): + if type(value) is not int: + raise ValueError(f"{label} must be an integer") + return value + + +@dataclass(frozen=True) +class MortalityArtifact: + """A compact model with an explicit fit boundary and source vintage.""" + + bands: tuple[tuple[int, int], ...] + probabilities: tuple[tuple[str, str, float], ...] + boundary_year: int + external_vintage_year: int + fit_rows: int + + def __post_init__(self): + _integer(self.boundary_year, "boundary_year") + _integer(self.external_vintage_year, "external_vintage_year") + if _integer(self.fit_rows, "fit_rows") <= 0: + raise ValueError("fit_rows must be positive") + if self.external_vintage_year > self.boundary_year: + raise ValueError("external vintage is later than the fit boundary") + seen = set() + for label, sex, probability in self.probabilities: + if not isinstance(label, str) or sex not in ("female", "male"): + raise ValueError("invalid mortality probability cell") + if (label, sex) in seen: + raise ValueError("duplicate mortality probability cell") + seen.add((label, sex)) + if isinstance(probability, bool) or not isinstance( + probability, (int, float) + ): + raise ValueError("mortality probability must be numeric") + if not math.isfinite(probability) or not 0 <= probability <= 1: + raise ValueError("mortality probability must lie in [0, 1]") + for band in self.bands: + if len(band) != 2 or any(type(age) is not int for age in band): + raise ValueError("mortality bands must contain integer bounds") + # The real model validates complete, contiguous bands and sex cells. + _ = self.model + + @property + def model(self): + from populace_dynamics.engine.steps import AgeSexMortalityModel + + return AgeSexMortalityModel( + self.bands, + {(band, sex): p for band, sex, p in self.probabilities}, + ) + + def to_bytes(self): + return json_bytes( + { + "format": _FORMAT, + "schema_version": 1, + "boundary_year": self.boundary_year, + "external_vintage_year": self.external_vintage_year, + "fit_rows": self.fit_rows, + "bands": self.bands, + "probabilities": [ + {"age_band": band, "sex": sex, "probability": p} + for band, sex, p in sorted(self.probabilities) + ], + } + ) + + @classmethod + def from_bytes(cls, payload): + raw = parse_json(payload) + expected = { + "format", + "schema_version", + "boundary_year", + "external_vintage_year", + "fit_rows", + "bands", + "probabilities", + } + if not isinstance(raw, dict) or set(raw) != expected: + raise ValueError("mortality model has an invalid field set") + if raw["format"] != _FORMAT or type(raw["schema_version"]) is not int: + raise ValueError("invalid mortality format or schema version") + if raw["schema_version"] != 1: + raise ValueError("unsupported mortality schema version") + if not isinstance(raw["bands"], list) or not all( + isinstance(band, list) for band in raw["bands"] + ): + raise ValueError("mortality bands must be arrays") + if not isinstance(raw["probabilities"], list): + raise ValueError("mortality probabilities must be an array") + for row in raw["probabilities"]: + if not isinstance(row, dict) or set(row) != { + "age_band", + "sex", + "probability", + }: + raise ValueError("invalid mortality probability fields") + return cls( + bands=tuple(tuple(band) for band in raw["bands"]), + probabilities=tuple( + (row["age_band"], row["sex"], row["probability"]) + for row in raw["probabilities"] + ), + boundary_year=raw["boundary_year"], + external_vintage_year=raw["external_vintage_year"], + fit_rows=raw["fit_rows"], + ) + + +def fit_mortality( + exposure, external_rates, *, boundary_year, external_vintage_year +): + """Use the existing cutoff-safe fitter and retain its compact results.""" + from populace_dynamics.engine.refit import ( + fit_mortality_model, + prepare_mortality_refit_inputs, + ) + + prepared = prepare_mortality_refit_inputs( + exposure, + external_rates, + boundary_year=boundary_year, + external_vintage_year=external_vintage_year, + ) + model = fit_mortality_model(prepared) + return MortalityArtifact( + bands=model.bands, + probabilities=tuple( + (band, sex, p) + for (band, sex), p in sorted(model.probability.items()) + ), + boundary_year=boundary_year, + external_vintage_year=external_vintage_year, + fit_rows=len(prepared.exposure), + ) diff --git a/src/populace_dynamics/graph/runtime.py b/src/populace_dynamics/graph/runtime.py new file mode 100644 index 00000000..4a6a332d --- /dev/null +++ b/src/populace_dynamics/graph/runtime.py @@ -0,0 +1,825 @@ +"""Fit the existing mortality law, transport it, and retain two periods. + +This module is imported only after the optional capability check. Its graph +is a synthetic engineering integration, separate from the locked M6 loop. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from types import SimpleNamespace + +import numpy as np +import pandas as pd +from microcosm.frame import EntitySchema, Frame, WeightKind, Weights +from microcosm.graph.canonical import canonical_json +from microcosm.graph.codecs import SOURCE_CODECS +from microcosm.graph.decl import ( + ArtifactInput, + ArtifactOutput, + ArtifactType, + Graph, + Node, + Owned, + Slice, + SourceRef, + StructuralDelta, + compile_graph, +) +from microcosm.graph.executor import run_graph +from microcosm.graph.kernel import ( + Capabilities, + Determinism, + KernelRegistry, + KernelResult, + KernelRole, + Numeric, + SeedSource, + source_hash, +) +from microcosm.graph.randomness import keyed_uniform +from microcosm.graph.store import ContentStore + +from populace_dynamics.engine.refit import ( + fit_mortality_model, + prepare_mortality_refit_inputs, +) +from populace_dynamics.engine.steps import advance_age, apply_mortality + +from . import model as model_module +from .model import ( + MortalityArtifact, + fit_mortality, + json_bytes, + parse_json, + read_json, +) + +OBS = "person_period" +PID = "person_period_person_id" +PERIOD_ID = "person_period_period_id" +OID = "person_period_id" +MODEL_TYPE = ArtifactType("populace-dynamics.mortality", 1) +CODEC = "dynamics-mortality-json-v1" +DEPENDENCIES = ("numpy", "pandas") +TRAIN_COLUMNS = ( + ("age_band", "string"), + ("sex", "string"), + ("required_interview_year", "int64"), + ("exposure", "float64"), + ("death", "float64"), +) + + +def _json_source_marker(path, *, store=None): + """Raw declared sources are read by domain kernels, never as Frames.""" + del path, store + raise ValueError("Dynamics JSON sources need their declared domain kernel") + + +def _series(values, ids, entity=OBS, dtype="int64"): + return pd.Series( + values, + index=pd.Index(ids, name=f"{entity}_id", dtype="int64"), + dtype=dtype, + ) + + +def _records(path, expected): + raw = read_json(path) + if not isinstance(raw, list) or not raw: + raise ValueError(f"{path.name} must contain nonempty record inputs") + if any( + not isinstance(row, dict) or set(row) != set(expected) for row in raw + ): + raise ValueError( + f"{path.name} has unsupported fields; expected {expected}" + ) + return pd.DataFrame(raw) + + +def _integer_column(frame, column): + if any(type(v) is not int for v in frame[column].tolist()): + raise ValueError(f"{column} must contain integer identifiers/years") + frame[column] = frame[column].astype("int64") + + +def _weights(values): + result = np.asarray(values, dtype=np.float64) + if not np.isfinite(result).all() or (result <= 0).any(): + raise ValueError("source weights must be positive and finite") + return Weights(result, WeightKind.DESIGN) + + +def _frame(records, periods, weights, columns): + """Give observations an explicit persistent-person and period partition.""" + observations = pd.DataFrame( + { + OID: np.arange(1, len(records) + 1, dtype=np.int64), + PID: records["person_id"].to_numpy(dtype=np.int64), + PERIOD_ID: np.asarray(periods, dtype=np.int64), + } + ) + for column, dtype in columns: + observations[column] = records[column].astype(dtype).array + person_ids = np.sort(observations[PID].unique()) + period_ids = np.sort(observations[PERIOD_ID].unique()) + return Frame( + { + OBS: observations, + "person": pd.DataFrame({"person_id": person_ids}), + "period": pd.DataFrame( + {"period_id": period_ids, "period": period_ids} + ), + }, + EntitySchema(person_entity=OBS, group_entities=("person", "period")), + {OBS: weights}, + pd.Series(["synthetic"] * len(observations), dtype=object), + ) + + +def _create_training(context): + fields = ["person_id", "event_year", "start_weight", *dict(TRAIN_COLUMNS)] + data = _records(context.sources["training"], fields) + for column in ("person_id", "event_year", "required_interview_year"): + _integer_column(data, column) + if data.duplicated(["person_id", "event_year"]).any(): + raise ValueError("training source repeats a person-period") + if not data.sex.isin(["female", "male"]).all(): + raise ValueError("training sex must be female or male") + for column in ("exposure", "death"): + numeric = data[column].to_numpy(dtype=np.float64) + if not np.isfinite(numeric).all() or (numeric < 0).any(): + raise ValueError( + f"training {column} must be finite and nonnegative" + ) + if (data.death > 1).any(): + raise ValueError("training death must lie in [0, 1]") + data = data.sort_values(["person_id", "event_year"]).reset_index(drop=True) + return KernelResult( + frame=_frame( + data, data.event_year, _weights(data.start_weight), TRAIN_COLUMNS + ) + ) + + +def _create_initial(context): + data = _records( + context.sources["initial"], ("person_id", "age", "sex", "weight") + ) + for column in ("person_id", "age"): + _integer_column(data, column) + if data.person_id.duplicated().any(): + raise ValueError("initial population repeats a person_id") + if not data.age.between(0, 120).all(): + raise ValueError("initial ages must lie in [0, 120]") + if not data.sex.isin(["female", "male"]).all(): + raise ValueError("initial sex must be female or male") + data = data.sort_values("person_id").reset_index(drop=True) + return KernelResult( + frame=_frame( + data, + [context.params["boundary_year"]] * len(data), + _weights(data.weight), + (("age", "int64"), ("sex", "string")), + ) + ) + + +def _periods(context, observations): + periods = context.tables["period"].set_index("period_id")["period"] + return observations[PERIOD_ID].map(periods).astype("int64") + + +def _fit(context): + observations = context.tables[OBS] + exposure = observations[[column for column, _ in TRAIN_COLUMNS]].copy() + exposure["person_id"] = observations[PID].to_numpy() + exposure["event_year"] = _periods(context, observations).to_numpy() + exposure["start_weight"] = context.weights[OBS].values + rates = pd.DataFrame(read_json(context.sources["rates"])) + artifact = fit_mortality( + exposure, + rates, + boundary_year=context.params["boundary_year"], + external_vintage_year=context.params["external_vintage_year"], + ) + return KernelResult( + artifacts={"model": artifact.to_bytes()}, + receipt={ + "fit_rows": artifact.fit_rows, + "boundary_year": artifact.boundary_year, + }, + ) + + +def mortality_uniforms( + person_ids, + *, + experiment_id="mortality", + replicate=0, + base_seed=0, + period=2015, + draw_index=0, +): + """Stable original-person draws, independent of observation row ordinals.""" + if any( + isinstance(pid, (bool, np.bool_)) + or not isinstance(pid, (int, np.integer)) + for pid in person_ids + ): + raise ValueError("person identities must be integers") + if ( + type(period) is not int + or type(draw_index) is not int + or draw_index < 0 + ): + raise ValueError("period and nonnegative draw index must be integers") + return keyed_uniform( + stream=("sha256-u53-v1", experiment_id, replicate, base_seed), + keys=[ + (int(pid), "mortality", int(period), int(draw_index)) + for pid in person_ids + ], + ) + + +class _PersonGenerator: + def __init__(self, person_id, module, period, stream): + self.person_id = person_id + self.module = getattr(module, "value", str(module)) + self.period = period + self.stream = stream + self.draw_index = 0 + + def random(self): + value = keyed_uniform( + stream=self.stream, + keys=[ + ( + int(self.person_id), + self.module, + self.period, + self.draw_index, + ) + ], + )[0] + self.draw_index += 1 + return float(value) + + +class _GraphPeriodContext: + """Adapter for existing steps, intentionally bypassing ordinal mapping.""" + + def __init__(self, params): + self.year = int(params["boundary_year"]) + 1 + self.metadata = {} + self.rng_registry = self # Existing apply_mortality tests for None. + self.stream = tuple(params["stream"]) + + def person_generator(self, module, person_id): + return _PersonGenerator(person_id, module, self.year, self.stream) + + +def _slice(context, observations=None): + observations = ( + context.tables[OBS] if observations is None else observations + ) + return pd.DataFrame( + { + "person_id": observations[PID].to_numpy(dtype=np.int64), + "age": observations.age.to_numpy(dtype=np.int64), + "sex": observations.sex.astype(str).to_numpy(), + "year": _periods(context, observations).to_numpy(), + } + ) + + +def _apply(context): + artifact = MortalityArtifact.from_bytes(context.artifacts["model"].payload) + if artifact.boundary_year != context.params["boundary_year"]: + raise ValueError("mortality fit and application boundary years differ") + model = artifact.model + initial = _slice(context) + survived = apply_mortality( + initial, _GraphPeriodContext(context.params), context.rng, model=model + ) + ids = context.tables[OBS][OID].tolist() + return KernelResult( + columns={ + (OBS, "death_probability"): _series( + model.probabilities(initial), ids, dtype="float64" + ), + (OBS, "survives"): _series( + initial.person_id.isin(survived.person_id).to_numpy(), + ids, + dtype="bool", + ), + } + ) + + +def _mass(weights, survivor_weights, strata, survivor_strata, boundary_year): + def totals(values, labels): + frame = pd.DataFrame({"weight": values, "stratum": list(labels)}) + return { + str(label): float(value) + for label, value in frame.groupby("stratum", observed=True) + .weight.sum() + .items() + } + + before = totals(weights, strata) + future = totals(survivor_weights, survivor_strata) + after = totals( + np.concatenate([weights, survivor_weights]), + [*strata, *survivor_strata], + ) + partition_after = {str(boundary_year): before} + if len(survivor_weights): + partition_after[str(boundary_year + 1)] = future + return { + "policy": "declared", + "before": float(np.sum(weights)), + "after": float(np.sum(np.concatenate([weights, survivor_weights]))), + "stratum_before": before, + "stratum_after": after, + "partition": { + "entity": "period", + "column": "period", + "stratum_before": {str(boundary_year): before}, + "stratum_after": partition_after, + }, + } + + +def _advance(context): + observations = context.tables[OBS] + mask = observations.survives.to_numpy(dtype=bool) + surviving = observations.loc[mask] + boundary = int(context.params["boundary_year"]) + # The existing adapter's year is used for the entrant period group only; + # it is never written back onto an incumbent observation. + aged = advance_age( + _slice(context, surviving), + SimpleNamespace(year=boundary + 1, metadata={}), + context.rng, + ) + old_ids = observations[OID].tolist() + new_ids = list(range(max(old_ids) + 1, max(old_ids) + 1 + len(surviving))) + target_ids = old_ids + new_ids + period_ids = context.tables["period"].period_id.tolist() + period_values = context.tables["period"].period.tolist() + next_period = [boundary + 1] if len(new_ids) else [] + weights = context.weights[OBS].values + survivor_weights = weights[mask] + return KernelResult( + expand={ + OBS: _series(surviving[OID].tolist(), new_ids), + "person": _series([], [], "person"), + "period": _series( + [pd.NA] * len(next_period), next_period, "period", "Int64" + ), + }, + columns={ + (OBS, PERIOD_ID): _series( + observations[PERIOD_ID].tolist() + + [boundary + 1] * len(new_ids), + target_ids, + ), + (OBS, "age"): _series( + observations.age.tolist() + aged.age.tolist(), target_ids + ), + ("period", "period"): _series( + period_values + next_period, period_ids + next_period, "period" + ), + }, + weights=Weights( + np.concatenate([weights, survivor_weights]), + context.weights[OBS].kind, + ), + receipt={ + "mass": _mass( + weights, + survivor_weights, + context.strata.tolist(), + context.strata.to_numpy()[mask], + boundary, + ) + }, + ) + + +def _age_claim(context): + observations = context.tables[OBS] + return KernelResult( + columns={ + (OBS, "age"): _series( + observations.age.tolist(), observations[OID].tolist() + ) + } + ) + + +def _evaluate(context): + observations = context.tables[OBS] + boundary = int(context.params["boundary_year"]) + periods = _periods(context, observations) + initial = observations.loc[periods == boundary].copy() + future = observations.loc[periods == boundary + 1].copy() + truth_document = read_json(context.sources["holdout"]) + if ( + not isinstance(truth_document, dict) + or set(truth_document) + != {"scope", "fixture_max_abs_death_rate_gap", "outcomes"} + or truth_document["scope"] != "synthetic_engineering" + ): + raise ValueError( + "holdout requires an explicit synthetic engineering contract" + ) + truth = pd.DataFrame(truth_document["outcomes"]) + if set(truth.columns) != {"person_id", "year", "age", "death"}: + raise ValueError("invalid held-out outcome columns") + for column in truth.columns: + _integer_column(truth, column) + if truth.person_id.duplicated().any() or set(truth.person_id) != set( + initial[PID] + ): + raise ValueError( + "held-out identities must match the initial population exactly" + ) + if ( + not (truth.year == boundary + 1).all() + or not truth.death.isin([0, 1]).all() + ): + raise ValueError( + "held-out outcomes must be binary deaths in the next period" + ) + truth = truth.set_index("person_id").loc[initial[PID]] + weights = context.weights[OBS].values + start_weights = weights[(periods == boundary).to_numpy()] + next_weights = weights[(periods == boundary + 1).to_numpy()] + mass = _mass( + start_weights, + next_weights, + context.strata[(periods == boundary).to_numpy()].tolist(), + context.strata[(periods == boundary + 1).to_numpy()].tolist(), + boundary, + ) + mass["next_period"] = float(next_weights.sum()) + probability = initial.death_probability.to_numpy() + expected_deaths = float(np.dot(start_weights, probability)) + observed_deaths = float(np.dot(start_weights, truth.death.to_numpy())) + generated_deaths = float( + np.dot(start_weights, ~initial.survives.to_numpy(dtype=bool)) + ) + discrepancy = abs(expected_deaths - observed_deaths) / float( + start_weights.sum() + ) + threshold = truth_document["fixture_max_abs_death_rate_gap"] + if ( + isinstance(threshold, bool) + or not isinstance(threshold, (int, float)) + or not np.isfinite(threshold) + or not 0 <= threshold <= 1 + ): + raise ValueError("fixture death-rate gap must be finite and in [0, 1]") + artifact = MortalityArtifact.from_bytes(context.artifacts["model"].payload) + original = _slice(context, initial).sort_values("person_id") + stream = tuple(context.params["stream"]) + uniforms = mortality_uniforms( + original.person_id.tolist(), + experiment_id=stream[1], + replicate=stream[2], + base_seed=stream[3], + period=boundary + 1, + ) + expected_ids = original.loc[ + uniforms >= artifact.model.probabilities(original), "person_id" + ].tolist() + actual_ids = sorted(future[PID].tolist()) + expected_ages = original.set_index("person_id").age + 1 + age_parity = all( + int(row.age) == int(expected_ages.loc[getattr(row, PID)]) + for row in future.itertuples(index=False) + ) + engineering_pass = ( + expected_ids == actual_ids + and age_parity + and not observations[OID].duplicated().any() + ) + heldout_age_error = float( + np.abs(truth.age.to_numpy() - (initial.age.to_numpy() + 1)).mean() + ) + fixture_pass = discrepancy <= threshold and heldout_age_error == 0 + report = { + "scope": "synthetic_engineering", + "boundary_year": boundary, + "next_period": boundary + 1, + "fit_rows": artifact.fit_rows, + "initial_records": len(initial), + "survivor_records": len(future), + "heldout_records": len(truth), + "expected_deaths": expected_deaths, + "observed_deaths": observed_deaths, + "generated_deaths": generated_deaths, + "absolute_death_rate_gap": discrepancy, + "fixture_max_abs_death_rate_gap": float(threshold), + "heldout_mean_absolute_age_error": heldout_age_error, + "fixture_verdict": "pass" if fixture_pass else "fail", + "engineering_verdict": "pass" if engineering_pass else "fail", + "mass": mass, + "limitations": [ + "Synthetic engineering fixture; no scientific or national-population certification.", + "The mortality fit's external-rate factor cancels in the fitted-window level.", + "Household accounting, births, immigration, and the full M6 loop are outside this slice.", + ], + } + return KernelResult( + artifacts={"report": json_bytes(report)}, + receipt={ + "outcome": "pass" if engineering_pass and fixture_pass else "fail", + "evidence": report, + }, + ) + + +class _Kernel: + def __init__( + self, + ref, + function, + *, + structural=StructuralDelta.NONE, + numeric=Numeric.PLATFORM_BITWISE, + seeded=False, + gate=False, + ): + self.ref = ref + self.function = function + self.capabilities = Capabilities( + Determinism.SEEDED if seeded else Determinism.DETERMINISTIC, + numeric=numeric, + structural=structural, + seed_source=SeedSource.KEYED if seeded else SeedSource.NONE, + role=KernelRole.GATE if gate else KernelRole.COMPUTE, + dependencies=DEPENDENCIES, + ) + + def implementation_hash(self): + return source_hash( + self.function, + model_module, + fit_mortality_model, + prepare_mortality_refit_inputs, + apply_mortality, + advance_age, + keyed_uniform, + canonical_json, + dependencies=self.capabilities.dependencies, + ) + + def run(self, context): + return self.function(context) + + +def build_graph( + *, + boundary_year=2014, + external_vintage_year=2014, + experiment_id="mortality", + replicate=0, + base_seed=0, +): + """Return the declared graph and registered existing-operation wrappers.""" + stream = ("sha256-u53-v1", experiment_id, replicate, base_seed) + keyed_uniform(stream=stream, keys=[]) # Validate even an empty population. + if ( + type(boundary_year) is not int + or type(external_vintage_year) is not int + ): + raise ValueError( + "fit boundary and external vintage must be integer years" + ) + params = {"boundary_year": boundary_year, "stream": stream} + binding = (ArtifactInput("model", "fit", "model", MODEL_TYPE),) + roots = ( + Node( + "training", + "dynamics.training@1", + sources=("training",), + structural=StructuralDelta.CREATE, + outputs=tuple( + Owned(OBS, column, dtype) for column, dtype in TRAIN_COLUMNS + ) + + (Owned("period", "period", "int64"),), + ), + Node( + "fit", + "dynamics.mortality.fit@1", + population="training", + sources=("rates",), + inputs=( + Slice(OBS, tuple(dict(TRAIN_COLUMNS))), + Slice("period", ("period",)), + ), + params={ + "boundary_year": boundary_year, + "external_vintage_year": external_vintage_year, + }, + artifact_outputs=(ArtifactOutput("model", MODEL_TYPE),), + ), + Node( + "initial", + "dynamics.initial@1", + sources=("initial",), + structural=StructuralDelta.CREATE, + params={"boundary_year": boundary_year}, + outputs=( + Owned(OBS, "age", "int64"), + Owned(OBS, "sex", "string"), + Owned("period", "period", "int64"), + ), + ), + Node( + "apply", + "dynamics.mortality.apply@1", + population="initial", + artifact_inputs=binding, + inputs=(Slice(OBS, ("age", "sex")), Slice("period", ("period",))), + params=params, + outputs=( + Owned(OBS, "death_probability", "float64"), + Owned(OBS, "survives", "bool"), + ), + ), + Node( + "advance", + "dynamics.advance@1", + base="initial", + structural=StructuralDelta.EXPAND, + entrants=True, + mass="declared", + inputs=( + Slice(OBS, ("age", "sex", "survives")), + Slice("period", ("period",)), + ), + params={ + "boundary_year": boundary_year, + "expand_cells": ( + (OBS, PERIOD_ID, "int64"), + (OBS, "age", "int64"), + ("period", "period", "int64"), + ), + "expand_weight_entity": OBS, + "expand_weight_kind": "design", + }, + ), + Node( + "age", + "dynamics.age-claim@1", + population="advance", + inputs=(Slice(OBS, ("age",)),), + outputs=(Owned(OBS, "age", "int64", rewrite=True),), + ), + Node( + "evaluate", + "dynamics.mortality.evaluate@1", + population="advance", + sources=("holdout",), + artifact_inputs=binding, + params=params, + inputs=( + Slice(OBS, ("age", "sex", "survives", "death_probability")), + Slice("period", ("period",)), + ), + ), + ) + registry = KernelRegistry() + for kernel in ( + _Kernel( + "dynamics.training@1", + _create_training, + structural=StructuralDelta.CREATE, + numeric=Numeric.BITWISE, + ), + _Kernel("dynamics.mortality.fit@1", _fit), + _Kernel( + "dynamics.initial@1", + _create_initial, + structural=StructuralDelta.CREATE, + numeric=Numeric.BITWISE, + ), + _Kernel("dynamics.mortality.apply@1", _apply, seeded=True), + _Kernel( + "dynamics.advance@1", _advance, structural=StructuralDelta.EXPAND + ), + _Kernel("dynamics.age-claim@1", _age_claim), + _Kernel( + "dynamics.mortality.evaluate@1", _evaluate, seeded=True, gate=True + ), + ): + registry.register(kernel) + SOURCE_CODECS.register(CODEC, _json_source_marker) + return ( + Graph( + "dynamics-mortality", + tuple( + SourceRef(name, CODEC) + for name in ("training", "rates", "initial", "holdout") + ), + roots, + mass_partition=("period", "period"), + ), + registry, + ) + + +@dataclass(frozen=True) +class MortalityGraphRun: + manifest: object + report: dict + model_payload: bytes + next_slice: pd.DataFrame + + +def run_mortality_graph( + *, + training, + rates, + initial, + holdout, + output_dir, + boundary_year=2014, + external_vintage_year=2014, + experiment_id="mortality", + replicate=0, + base_seed=0, + household_accounting=False, +): + """Run/reuse the graph and write artifacts only in the explicit directory.""" + if household_accounting: + raise ValueError( + "household accounting is unsupported by this person-period slice" + ) + output = Path(output_dir).resolve() + sources = { + name: Path(path).resolve() + for name, path in ( + ("training", training), + ("rates", rates), + ("initial", initial), + ("holdout", holdout), + ) + } + graph, registry = build_graph( + boundary_year=boundary_year, + external_vintage_year=external_vintage_year, + experiment_id=experiment_id, + replicate=replicate, + base_seed=base_seed, + ) + output.mkdir(parents=True, exist_ok=True) + store = ContentStore(output / "store") + manifest = run_graph( + compile_graph(graph), sources=sources, store=store, kernels=registry + ) + model_payload = store.load_bytes( + manifest.nodes["fit"].opaque_artifacts["model"] + ) + report = parse_json( + store.load_bytes(manifest.nodes["evaluate"].opaque_artifacts["report"]) + ) + report["node_keys"] = { + name: node.key for name, node in manifest.nodes.items() + } + report["cache_hits"] = { + name: node.hit for name, node in manifest.nodes.items() + } + report["model_artifact_key"] = manifest.nodes["fit"].opaque_artifacts[ + "model" + ] + population = manifest.population("advance") + observations = population.table(OBS) + next_rows = observations.loc[observations[PERIOD_ID] == boundary_year + 1] + next_slice = ( + pd.DataFrame( + { + "person_id": next_rows[PID].to_numpy(dtype=np.int64), + "age": next_rows.age.to_numpy(dtype=np.int64), + "year": np.full( + len(next_rows), boundary_year + 1, dtype=np.int64 + ), + } + ) + .sort_values("person_id") + .reset_index(drop=True) + ) + (output / "report.json").write_bytes(json_bytes(report)) + (output / "manifest.json").write_text(manifest.to_json(), encoding="utf-8") + (output / "model.json").write_bytes(model_payload) + for entity in (OBS, "person", "period"): + population.table(entity).to_csv(output / f"{entity}.csv", index=False) + next_slice.to_csv(output / "next_period.csv", index=False) + return MortalityGraphRun(manifest, report, model_payload, next_slice) diff --git a/src/populace_dynamics/graph/synthetic.py b/src/populace_dynamics/graph/synthetic.py new file mode 100644 index 00000000..24808993 --- /dev/null +++ b/src/populace_dynamics/graph/synthetic.py @@ -0,0 +1,78 @@ +"""Small hand-specified engineering inputs, independent of generated draws.""" + +from pathlib import Path + +from .model import json_bytes + + +def write_synthetic_inputs(directory): + """Write synthetic sources to an explicit directory; preserve edits.""" + directory = Path(directory) + directory.mkdir(parents=True, exist_ok=True) + training = [ + { + "person_id": 1001 + i, + "event_year": 2013, + "required_interview_year": 2013, + "age_band": "0+", + "sex": "female" if i < 4 else "male", + "start_weight": 1.0, + "exposure": 1.0, + "death": float(i % 4 == 0), + } + for i in range(8) + ] + training.extend( + [ + {**training[0], "person_id": 1009, "event_year": 2015}, + { + **training[0], + "person_id": 1010, + "required_interview_year": 2015, + }, + ] + ) + rates = [ + { + "lower_age": 0, + "upper_age": 120, + "age_band": "0+", + "sex": sex, + "central_rate": rate, + } + for sex, rate in (("female", 0.005), ("male", 0.006)) + ] + initial = [ + { + "person_id": 100 + i, + "age": 30 + 2 * i, + "sex": "female" if i % 2 == 0 else "male", + "weight": float(1 + i % 3), + } + for i in range(20) + ] + holdout = { + "scope": "synthetic_engineering", + "fixture_max_abs_death_rate_gap": 0.25, + "outcomes": [ + { + "person_id": row["person_id"], + "year": 2015, + "age": row["age"] + 1, + "death": int(i % 5 == 0), + } + for i, row in enumerate(initial) + ], + } + result = {} + for name, value in ( + ("training", training), + ("rates", rates), + ("initial", initial), + ("holdout", holdout), + ): + path = directory / f"{name}.json" + if not path.exists(): + path.write_bytes(json_bytes(value)) + result[name] = path + return result diff --git a/tests/README-tiers.md b/tests/README-tiers.md index a5ef9187..56e59849 100644 --- a/tests/README-tiers.md +++ b/tests/README-tiers.md @@ -38,9 +38,9 @@ pytest --collect-only -q -m oracle_policyengine | tail -1 | Tier | Tests at HEAD | |---|---:| -| `unit` | 1,563 | +| `unit` | 1,588 | | `artifact` | 2,668 | | `integration_psid` | 848 | | `reproduction_legacy` | 520 | | `oracle_policyengine` | 159 | -| **Total** | **5,758** | +| **Total** | **5,783** | diff --git a/tests/estimates/test_birth_evidence_artifact.py b/tests/estimates/test_birth_evidence_artifact.py index d4e838a1..2f4c176c 100644 --- a/tests/estimates/test_birth_evidence_artifact.py +++ b/tests/estimates/test_birth_evidence_artifact.py @@ -88,6 +88,12 @@ def test_post_review_sources_are_outside_historical_reducer_identity(): Path("src/populace_dynamics/estimates/anchor_context_registry.py"), Path("src/populace_dynamics/estimates/anchor_context_rehearsal.py"), Path("src/populace_dynamics/estimates/anchor_context_report.py"), + Path("src/populace_dynamics/graph/__init__.py"), + Path("src/populace_dynamics/graph/__main__.py"), + Path("src/populace_dynamics/graph/_compat.py"), + Path("src/populace_dynamics/graph/model.py"), + Path("src/populace_dynamics/graph/runtime.py"), + Path("src/populace_dynamics/graph/synthetic.py"), ) assert reducer.POST_REVIEW_SHARED_SOURCE_BLOBS == { Path( @@ -157,7 +163,7 @@ def _internal_imports( return imports -def test_psid_identity_exclusions_are_unreachable_from_birth_evidence(): +def test_psid_and_graph_exclusions_are_unreachable_from_birth_evidence(): module_paths = _repository_module_paths() root_module = "scripts.first_estimates_birth_evidence" psid_exclusions = { @@ -170,6 +176,13 @@ def test_psid_identity_exclusions_are_unreachable_from_birth_evidence(): } assert root_module in module_paths assert psid_exclusions.issubset(module_paths) + graph_exclusions = { + name + for name in module_paths + if name == "populace_dynamics.graph" + or name.startswith("populace_dynamics.graph.") + } + assert graph_exclusions module_by_path = { path.resolve(): module_name for module_name, path in module_paths.items() @@ -205,6 +218,10 @@ def test_psid_identity_exclusions_are_unreachable_from_birth_evidence(): "historically excluded PSID modules became reachable from the " f"birth-evidence reducer: {sorted(psid_exclusions & reachable)}" ) + assert graph_exclusions.isdisjoint(reachable), ( + "opt-in graph modules became reachable from the birth-evidence " + f"reducer: {sorted(graph_exclusions & reachable)}" + ) def test_reducer_accepts_explicit_unresolved_upstream_boundary(): diff --git a/tests/test_graph_mortality.py b/tests/test_graph_mortality.py new file mode 100644 index 00000000..6a300e6e --- /dev/null +++ b/tests/test_graph_mortality.py @@ -0,0 +1,324 @@ +"""Synthetic engineering tests for the optional population graph.""" + +import copy +import json +from pathlib import Path +from types import SimpleNamespace + +import numpy as np +import pandas as pd +import pytest + +from populace_dynamics.graph.model import MortalityArtifact, fit_mortality +from populace_dynamics.graph.synthetic import write_synthetic_inputs + + +@pytest.fixture +def inputs(tmp_path): + return write_synthetic_inputs(tmp_path / "inputs") + + +@pytest.fixture +def runtime(): + from populace_dynamics.graph._compat import require_graph + + try: + require_graph() + except ImportError as error: + pytest.skip(str(error)) + from populace_dynamics.graph import runtime + + return runtime + + +def _read(path): + return json.loads(path.read_text()) + + +def _write(path, value): + path.write_text(json.dumps(value)) + + +def _fit(inputs): + return fit_mortality( + pd.DataFrame(_read(inputs["training"])), + pd.DataFrame(_read(inputs["rates"])), + boundary_year=2014, + external_vintage_year=2014, + ) + + +def _run(runtime, inputs, tmp_path, **kwargs): + return runtime.run_mortality_graph( + **inputs, output_dir=tmp_path / "output", **kwargs + ) + + +def test_existing_mortality_fit_and_json_roundtrip(inputs): + artifact = _fit(inputs) + assert artifact.fit_rows == 8 + assert artifact.model.probability[("0+", "female")] == pytest.approx( + -np.expm1(-0.25) + ) + restored = MortalityArtifact.from_bytes(artifact.to_bytes()) + assert restored.to_bytes() == artifact.to_bytes() + assert restored.model == artifact.model + + +def test_future_exposure_and_interview_information_are_excluded(inputs): + baseline = _fit(inputs) + training = _read(inputs["training"]) + training[-1]["death"] = 0.0 + _write(inputs["training"], training) + assert _fit(inputs).to_bytes() == baseline.to_bytes() + with pytest.raises(ValueError, match="vintage"): + fit_mortality( + pd.DataFrame(training), + pd.DataFrame(_read(inputs["rates"])), + boundary_year=2013, + external_vintage_year=2014, + ) + + +@pytest.mark.parametrize("mutation", ["schema", "duplicate", "missing", "nan"]) +def test_model_payload_fails_closed(inputs, mutation): + raw = json.loads(_fit(inputs).to_bytes()) + if mutation == "schema": + raw["schema_version"] = 99 + elif mutation == "duplicate": + raw["probabilities"].append(copy.deepcopy(raw["probabilities"][0])) + elif mutation == "missing": + raw["probabilities"].pop() + else: + raw["probabilities"][0]["probability"] = float("nan") + with pytest.raises(ValueError): + MortalityArtifact.from_bytes(json.dumps(raw).encode()) + + +def test_model_rejects_duplicate_json_members(inputs): + payload = ( + _fit(inputs) + .to_bytes() + .replace( + b'"schema_version":1', b'"schema_version":1,"schema_version":1' + ) + ) + with pytest.raises(ValueError, match="duplicate"): + MortalityArtifact.from_bytes(payload) + + +def test_graph_reuses_fit_and_reports_period_mass(runtime, inputs, tmp_path): + cold = _run(runtime, inputs, tmp_path) + warm = _run(runtime, inputs, tmp_path) + assert cold.report["scope"] == "synthetic_engineering" + assert cold.report["fixture_verdict"] == "pass" + assert cold.report["engineering_verdict"] == "pass" + assert all(node.hit for node in warm.manifest.nodes.values()) + assert cold.model_payload == warm.model_payload + assert cold.report["mass"] == warm.report["mass"] + mass = cold.report["mass"] + assert mass["after"] == mass["before"] + mass["next_period"] + assert mass["partition"]["stratum_before"]["2014"] == ( + mass["partition"]["stratum_after"]["2014"] + ) + assert (tmp_path / "output" / "report.json").is_file() + assert (tmp_path / "output" / "manifest.json").is_file() + + +def test_graph_matches_existing_steps_with_explicit_uniforms( + runtime, inputs, tmp_path +): + from populace_dynamics.engine.steps import ( + advance_age, + apply_mortality, + ) + + result = _run(runtime, inputs, tmp_path) + initial = pd.DataFrame(_read(inputs["initial"])).sort_values("person_id") + initial["year"] = 2014 + uniforms = runtime.mortality_uniforms(initial.person_id.tolist()) + + class FixedUniforms: + def random(self, n): + assert n == len(uniforms) + return uniforms.copy() + + context = SimpleNamespace(rng_registry=None, year=2015, metadata={}) + model = MortalityArtifact.from_bytes(result.model_payload).model + survived = apply_mortality(initial, context, FixedUniforms(), model=model) + assert ( + survived.person_id.tolist() + == initial.loc[ + uniforms >= model.probabilities(initial), "person_id" + ].tolist() + ) + expected = advance_age(survived, context, np.random.default_rng(0)) + actual = result.next_slice.sort_values("person_id") + pd.testing.assert_frame_equal( + actual[["person_id", "age", "year"]].reset_index(drop=True), + expected[["person_id", "age", "year"]].reset_index(drop=True), + ) + + +def test_holdout_changes_only_evaluation_and_can_fail_fixture( + runtime, inputs, tmp_path +): + baseline = _run(runtime, inputs, tmp_path) + holdout = _read(inputs["holdout"]) + for row in holdout["outcomes"]: + row["death"] = 1 + _write(inputs["holdout"], holdout) + changed = _run(runtime, inputs, tmp_path) + assert changed.report["fixture_verdict"] == "fail" + assert changed.report["engineering_verdict"] == "pass" + assert changed.model_payload == baseline.model_payload + for node_id in ("training", "fit", "initial", "apply", "advance", "age"): + assert changed.manifest.nodes[node_id].hit + assert not changed.manifest.nodes["evaluate"].hit + pd.testing.assert_frame_equal(baseline.next_slice, changed.next_slice) + + +def test_recipient_edit_keeps_model_fit(runtime, inputs, tmp_path): + baseline = _run(runtime, inputs, tmp_path) + initial = _read(inputs["initial"]) + initial[0]["age"] += 1 + _write(inputs["initial"], initial) + changed = _run(runtime, inputs, tmp_path) + assert changed.manifest.nodes["fit"].hit + assert changed.model_payload == baseline.model_payload + assert not changed.manifest.nodes["apply"].hit + + +def test_training_weight_edit_invalidates_fit_and_application( + runtime, inputs, tmp_path +): + baseline = _run(runtime, inputs, tmp_path) + training = _read(inputs["training"]) + training[0]["start_weight"] = 8.0 + _write(inputs["training"], training) + changed = _run(runtime, inputs, tmp_path) + assert not changed.manifest.nodes["fit"].hit + assert not changed.manifest.nodes["apply"].hit + assert changed.model_payload != baseline.model_payload + + +def test_stable_draws_are_row_chunk_and_unrelated_person_invariant(runtime): + first = runtime.mortality_uniforms([11, 21, 31]) + np.testing.assert_array_equal( + runtime.mortality_uniforms([31, 11, 21]), first[[2, 0, 1]] + ) + np.testing.assert_array_equal( + runtime.mortality_uniforms([1, 11, 21, 31])[1:], first + ) + np.testing.assert_array_equal( + np.concatenate( + [ + runtime.mortality_uniforms([11]), + runtime.mortality_uniforms([21, 31]), + ] + ), + first, + ) + + +@pytest.mark.parametrize("identity", [None, 1.5, "1", True]) +def test_mortality_draws_reject_noninteger_person_ids(runtime, identity): + with pytest.raises(ValueError, match="identities"): + runtime.mortality_uniforms([identity]) + + +def test_population_reorder_and_unrelated_person_preserve_survivors( + runtime, inputs, tmp_path +): + baseline = _run(runtime, inputs, tmp_path) + initial = _read(inputs["initial"]) + _write(inputs["initial"], initial[::-1]) + reordered = _run(runtime, inputs, tmp_path) + pd.testing.assert_frame_equal(baseline.next_slice, reordered.next_slice) + initial.append({"person_id": 1, "age": 44, "sex": "male", "weight": 2.0}) + _write(inputs["initial"], initial) + holdout = _read(inputs["holdout"]) + holdout["outcomes"].append( + {"person_id": 1, "year": 2015, "age": 45, "death": 0} + ) + _write(inputs["holdout"], holdout) + extended = _run(runtime, inputs, tmp_path) + pd.testing.assert_frame_equal( + baseline.next_slice, + extended.next_slice.query("person_id != 1").reset_index(drop=True), + ) + + +@pytest.mark.parametrize("all_die", [False, True]) +def test_empty_or_complete_survivor_expansion( + runtime, inputs, tmp_path, all_die +): + training = _read(inputs["training"]) + for row in training: + row["death"] = 1.0 if all_die else 0.0 + row["exposure"] = 1e-9 if all_die else 1.0 + _write(inputs["training"], training) + result = _run(runtime, inputs, tmp_path) + warm = _run(runtime, inputs, tmp_path) + assert warm.manifest.nodes["advance"].hit + assert len(result.next_slice) == (0 if all_die else 20) + assert result.report["mass"]["next_period"] == ( + 0.0 if all_die else result.report["mass"]["before"] + ) + if all_die: + assert ( + "2015" not in result.report["mass"]["partition"]["stratum_after"] + ) + + +def test_household_accounting_is_explicitly_unsupported( + runtime, inputs, tmp_path +): + with pytest.raises(ValueError, match="household"): + _run(runtime, inputs, tmp_path, household_accounting=True) + + +def test_optional_entrypoint_has_actionable_missing_capability(monkeypatch): + from populace_dynamics.graph import _compat + + monkeypatch.setattr(_compat, "_python_version", lambda: (3, 12)) + with pytest.raises(ImportError, match="Python >=3.13"): + _compat.require_graph() + + +def test_optional_entrypoint_reports_old_core_capabilities(monkeypatch): + from populace_dynamics.graph import _compat + + monkeypatch.setattr(_compat, "_python_version", lambda: (3, 14)) + monkeypatch.setattr( + _compat.importlib, "import_module", lambda name: SimpleNamespace() + ) + with pytest.raises(ImportError, match="typed model-artifact"): + _compat.require_graph() + + +def test_cli_requires_output_directory(): + from populace_dynamics.graph.__main__ import parser + + with pytest.raises(SystemExit): + parser().parse_args(["--synthetic"]) + + +def test_keyed_kernel_hashes_random_coordinate_encoding(runtime, monkeypatch): + from microcosm.graph import canonical + + _, registry = runtime.build_graph() + kernel = registry.get("dynamics.mortality.apply@1") + assert kernel.capabilities.dependencies == ("numpy", "pandas") + before = kernel.implementation_hash() + encoding_source = Path(canonical.__file__).resolve() + original = Path.read_bytes + + def changed_source(path): + payload = original(path) + if path.resolve() == encoding_source: + payload += b"\n# coordinate encoding change\n" + return payload + + monkeypatch.setattr(Path, "read_bytes", changed_source) + assert kernel.implementation_hash() != before diff --git a/tests/tier_counts.json b/tests/tier_counts.json index 71769153..e184e443 100644 --- a/tests/tier_counts.json +++ b/tests/tier_counts.json @@ -1,7 +1,7 @@ { "schema_version": 1, "counts": { - "unit": 1563, + "unit": 1588, "artifact": 2668, "integration_psid": 848, "reproduction_legacy": 520,