diff --git a/src/language_reading_predictors/statistical_models/growth.py b/src/language_reading_predictors/statistical_models/growth.py new file mode 100644 index 00000000..8d489393 --- /dev/null +++ b/src/language_reading_predictors/statistical_models/growth.py @@ -0,0 +1,233 @@ +# Copyright (c) 2026 Down Syndrome Education International and contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +"""Typed settings and a resolved run plan for the latent growth-curve family (#394 pillar 4). + +Mirrors the ITT / gain-factor / level-factor / DiD / concurrent / aligned run-plan +pattern for the joint multivariate latent growth-curve (``kind="growth"``) models. A +model module declares its settings; the plan is resolved and **validated before any +data are loaded or an output directory is reset**, then drives data preparation, +factory construction and the ``config.json`` / ``model_recipe.md`` audit trail. This +removes the untyped ``spec.extra`` boundary (where a misspelled key silently +defaulted) and records the resolved design, estimand, causal status, analysis +population and missing-data assumption alongside every fit. + +The growth design characterises each measure's within-child trajectory across the +four RLI waves (linear in standardised age) and asks whether a **baseline** covariate +(non-verbal ability) predicts trajectory shape: ``gamma`` on the growth *rate* (the +headline Q5 estimand) and ``delta`` on the baseline *level*. It is a multi-outcome +family (no single ``outcome_symbol``); every non-randomised term is an **adjusted, +latent-general-ability-confounded association**, never causal. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import asdict, dataclass +from typing import Any + +from language_reading_predictors.statistical_models.context import ModelSpec + +# The complete, closed set of legacy ``spec.extra`` keys the growth family +# understands. Anything else is a typo and must fail before a fit starts. +_LEGACY_KEYS = frozenset( + { + "outcomes", + "baseline_covariate", + "use_shared_factor", + "age_ability_interaction", + } +) + +_DEFAULT_OUTCOMES = ("R", "E", "T", "W", "L") + + +def _tuple_of_strings(value: Any, *, name: str) -> tuple[str, ...]: + if isinstance(value, str) or not hasattr(value, "__iter__"): + raise TypeError(f"{name} must be a sequence of strings, got {value!r}") + out = tuple(value) + for item in out: + if not isinstance(item, str) or not item: + raise TypeError(f"{name} must contain non-empty strings, got {item!r}") + return out + + +@dataclass(frozen=True, slots=True) +class GrowthModelSettings: + """Immutable settings declared by a single latent growth-curve model module. + + Defaults encode the primary five-measure growth-curve read on non-verbal ability + (``blocks``), without the shared growth-tempo factor or the age x ability term. + """ + + outcomes: tuple[str, ...] = _DEFAULT_OUTCOMES + baseline_covariate: str = "blocks" + use_shared_factor: bool = False + age_ability_interaction: bool = False + + def __post_init__(self) -> None: + object.__setattr__( + self, "outcomes", _tuple_of_strings(self.outcomes, name="outcomes") + ) + if not self.outcomes: + raise ValueError("outcomes must list at least one measure") + if not isinstance(self.baseline_covariate, str) or not self.baseline_covariate: + raise TypeError("baseline_covariate must be a non-empty string") + for flag in ("use_shared_factor", "age_ability_interaction"): + if not isinstance(getattr(self, flag), bool): + raise TypeError(f"{flag} must be bool") + + @classmethod + def from_legacy_extra( + cls, extra: Mapping[str, Any], *, model_id: str + ) -> GrowthModelSettings: + """Strictly translate the former ``spec.extra`` dictionary boundary. + + Rejects unknown keys so a misspelling fails before data loading rather than + silently taking a default.""" + unknown = sorted(set(extra) - _LEGACY_KEYS) + if unknown: + raise ValueError( + f"{model_id}: unknown growth setting(s): {', '.join(unknown)}. " + "Declare GrowthModelSettings so misspellings fail fast." + ) + # Pass raw values through so __post_init__ is the single validation/coercion + # point; pre-coercing here would silently reshape misshaped legacy settings. + return cls( + outcomes=extra.get("outcomes", _DEFAULT_OUTCOMES), + baseline_covariate=extra.get("baseline_covariate", "blocks"), + use_shared_factor=extra.get("use_shared_factor", False), + age_ability_interaction=extra.get("age_ability_interaction", False), + ) + + +@dataclass(frozen=True, slots=True) +class GrowthRunPlan: + """Concrete, validated instructions consumed by preparation and modelling.""" + + model_id: str + settings_source: str + outcomes: tuple[str, ...] + baseline_covariate: str + use_shared_factor: bool + age_ability_interaction: bool + # Recorded audit metadata (#394 pillar 4). + design: str + estimand: str + causal_status: str + analysis_population: str + missing_data_assumption: str + + def as_dict(self) -> dict[str, Any]: + """Return the JSON-ready run-plan contract for ``config.json``.""" + return asdict(self) + + def prepare_kwargs(self) -> dict[str, Any]: + """Arguments for ``load_wave_panel`` from the resolved plan. + + The growth family loads the wave panel of its measures with the single + t1-only baseline covariate (non-verbal ability).""" + return { + "outcomes": self.outcomes, + "baseline_covariates": (self.baseline_covariate,), + } + + def factory_kwargs(self) -> dict[str, Any]: + """Arguments for ``build_growth_model`` for this plan.""" + return { + "baseline_covariate": self.baseline_covariate, + "use_shared_factor": self.use_shared_factor, + "age_ability_interaction": self.age_ability_interaction, + } + + def recipe_markdown(self, *, title: str) -> str: + """Undergraduate-friendly explanation generated from the resolved plan.""" + outcomes = ", ".join(self.outcomes) + return ( + "Note: Generated from the validated latent growth-curve run plan; " + "template drafted by an LLM-based AI tool (Claude Code/Opus 4.8).\n\n" + f"# Model recipe: {title}\n\n" + f"Model ID: `{self.model_id}`.\n\n" + f"## Design\n\n{self.design}\n\n" + f"## Estimand\n\n{self.estimand}\n\n" + f"## Causal status\n\n{self.causal_status}\n\n" + f"## Analysis population\n\n{self.analysis_population}\n\n" + f"## Missing data\n\n{self.missing_data_assumption}\n\n" + "## Terms\n\n" + f"Measures: {outcomes}. Baseline covariate: `{self.baseline_covariate}`. " + f"Shared growth-tempo factor: {self.use_shared_factor}. Age x ability " + f"interaction: {self.age_ability_interaction}.\n\n" + "## Uncertainty and checks\n\n" + "The fit reports a posterior distribution; interpret it only after the " + "convergence gate and posterior-predictive checks pass. The saved " + "`config.json` contains the same resolved run plan in machine-readable " + "form.\n" + ) + + +def declared_growth_settings(spec: ModelSpec) -> tuple[GrowthModelSettings, str]: + """Return typed settings and their source, rejecting mixed declarations.""" + settings = spec.model_settings + if settings is not None: + if spec.extra: + raise ValueError( + f"{spec.model_id}: growth settings cannot be split between " + "model_settings and extra" + ) + if not isinstance(settings, GrowthModelSettings): + raise TypeError( + f"{spec.model_id}: kind='growth' requires GrowthModelSettings, got " + f"{type(settings).__name__}" + ) + return settings, "typed" + return ( + GrowthModelSettings.from_legacy_extra(spec.extra, model_id=spec.model_id), + "legacy_extra", + ) + + +def resolve_growth_run_plan(spec: ModelSpec) -> GrowthRunPlan: + """Resolve and validate a latent growth-curve spec before any data are loaded.""" + if spec.kind != "growth": + raise ValueError(f"{spec.model_id}: expected kind 'growth', got {spec.kind!r}") + + settings, source = declared_growth_settings(spec) + + design = ( + "Joint multivariate latent growth-curve model on the logit scale: each " + "measure's within-child trajectory across the four waves (linear in " + "standardised age), with a per-measure child-level random intercept and slope; " + "optionally a rank-1 shared growth-tempo factor coupling the slopes." + ) + estimand = ( + "gamma (baseline non-verbal ability -> growth RATE) is the headline Q5 " + "association; delta is the association with baseline LEVEL. Both are adjusted, " + "latent-general-ability-confounded associations, never causal (block design is " + "an off-DAG ability proxy)." + ) + causal_status = ( + "Associational only: every non-randomised term is a latent-general-ability-" + "confounded adjusted association under the locked DAG, never a causal effect." + ) + analysis_population = ( + "Available-case children with wave-panel trajectories (about 54). The " + "associations describe this observed cohort, not a randomised contrast." + ) + missing_data_assumption = ( + "Available-case analysis under ignorable missingness: missing wave scores are " + "assumed ignorable given the modelled trajectory and baseline covariate." + ) + + return GrowthRunPlan( + model_id=spec.model_id, + settings_source=source, + outcomes=settings.outcomes, + baseline_covariate=settings.baseline_covariate, + use_shared_factor=settings.use_shared_factor, + age_ability_interaction=settings.age_ability_interaction, + design=design, + estimand=estimand, + causal_status=causal_status, + analysis_population=analysis_population, + missing_data_assumption=missing_data_assumption, + ) diff --git a/src/language_reading_predictors/statistical_models/pipeline.py b/src/language_reading_predictors/statistical_models/pipeline.py index aab60fe5..61ec01ce 100644 --- a/src/language_reading_predictors/statistical_models/pipeline.py +++ b/src/language_reading_predictors/statistical_models/pipeline.py @@ -82,6 +82,9 @@ from language_reading_predictors.statistical_models.gain_factors import ( resolve_gain_factors_run_plan, ) +from language_reading_predictors.statistical_models.growth import ( + resolve_growth_run_plan, +) from language_reading_predictors.statistical_models.itt import ( IttRunPlan, build_itt_from_plan, @@ -8021,25 +8024,27 @@ def fit_growth(spec: ModelSpec, config: str = "dev") -> StatisticalFitContext: """ _require_spec(spec, "growth") + # Resolve and validate the family contract before the context resets an output + # directory or the loader reads any data (#394 pillar 4). One plan drives + # preparation, factory arguments, the teaching recipe and config.json. + plan = resolve_growth_run_plan(spec) ctx = make_context(spec, config) + ctx.resolved_plan = plan + _report.write_model_recipe(ctx) + + outcomes = plan.outcomes + baseline_cov = plan.baseline_covariate + use_factor = plan.use_shared_factor + age_ability = plan.age_ability_interaction section_header("Prepare data") - outcomes = tuple(spec.extra.get("outcomes", ("R", "E", "T", "W", "L"))) - baseline_cov = spec.extra.get("baseline_covariate", "blocks") - use_factor = bool(spec.extra.get("use_shared_factor", False)) - age_ability = bool(spec.extra.get("age_ability_interaction", False)) - panel = load_wave_panel(outcomes=outcomes, baseline_covariates=(baseline_cov,)) + panel = load_wave_panel(**plan.prepare_kwargs()) ctx.prepared = panel _print_header(ctx) section_header("Build model") - built = _factories.build_growth_model( - panel, - baseline_covariate=baseline_cov, - use_shared_factor=use_factor, - age_ability_interaction=age_ability, - ) + built = _factories.build_growth_model(panel, **plan.factory_kwargs()) _attach_built(ctx, built) _render_model_graph(ctx) diff --git a/src/language_reading_predictors/statistical_models/reporting.py b/src/language_reading_predictors/statistical_models/reporting.py index 8fac06fe..48d081be 100644 --- a/src/language_reading_predictors/statistical_models/reporting.py +++ b/src/language_reading_predictors/statistical_models/reporting.py @@ -31,6 +31,10 @@ GainFactorsRunPlan, resolve_gain_factors_run_plan, ) +from language_reading_predictors.statistical_models.growth import ( + GrowthRunPlan, + resolve_growth_run_plan, +) from language_reading_predictors.statistical_models.itt import ( IttRunPlan, declared_settings_dict, @@ -2273,12 +2277,22 @@ def _gain_factors_run_plan(context: StatisticalFitContext) -> GainFactorsRunPlan return resolve_gain_factors_run_plan(context.spec) +def _growth_run_plan(context: StatisticalFitContext) -> GrowthRunPlan: + """Return the growth plan resolved before loading, or reconstruct it.""" + resolved_plan = getattr(context, "resolved_plan", None) + if isinstance(resolved_plan, GrowthRunPlan): + return resolved_plan + return resolve_growth_run_plan(context.spec) + + def _resolved_run_plan(context: StatisticalFitContext): - """The typed run plan for families that have one (ITT, gain factors), else None.""" + """The typed run plan for families that have one (ITT, gain, growth), else None.""" if context.spec.kind == "itt": return _itt_run_plan(context) if context.spec.kind == "gain_factors": return _gain_factors_run_plan(context) + if context.spec.kind == "growth": + return _growth_run_plan(context) return None diff --git a/tests/statistical_models/test_growth_run_plan.py b/tests/statistical_models/test_growth_run_plan.py new file mode 100644 index 00000000..fb2acab3 --- /dev/null +++ b/tests/statistical_models/test_growth_run_plan.py @@ -0,0 +1,179 @@ +# Copyright (c) 2026 Down Syndrome Education International and contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +"""Tests for the typed latent growth-curve settings and run plan (#394 pillar 4).""" + +from __future__ import annotations + +import glob +import importlib +import os + +import pytest + +from language_reading_predictors.statistical_models.context import ModelSpec +from language_reading_predictors.statistical_models.growth import ( + GrowthModelSettings, + GrowthRunPlan, + resolve_growth_run_plan, +) + +_META_FIELDS = ( + "design", + "estimand", + "causal_status", + "analysis_population", + "missing_data_assumption", +) + + +def _growth_specs() -> list[ModelSpec]: + """Every registered latent growth-curve model's SPEC.""" + root = os.path.dirname( + importlib.import_module( + "language_reading_predictors.statistical_models.growth" + ).__file__ + ) + specs: list[ModelSpec] = [] + for path in sorted(glob.glob(os.path.join(root, "lrp_rli_gc_*.py"))): + mod = importlib.import_module( + "language_reading_predictors.statistical_models." + os.path.basename(path)[:-3] + ) + spec = getattr(mod, "SPEC", None) + if spec is not None and spec.kind == "growth": + specs.append(spec) + return specs + + +# --- settings validation ------------------------------------------------------ + + +def test_settings_reject_string_outcomes(): + with pytest.raises(TypeError, match="outcomes"): + GrowthModelSettings(outcomes="W") # type: ignore[arg-type] + + +def test_settings_reject_empty_outcomes(): + with pytest.raises(ValueError, match="at least one measure"): + GrowthModelSettings(outcomes=()) + + +def test_settings_reject_empty_baseline_covariate(): + with pytest.raises(TypeError, match="baseline_covariate"): + GrowthModelSettings(baseline_covariate="") + + +def test_settings_reject_non_bool_use_shared_factor(): + with pytest.raises(TypeError, match="use_shared_factor"): + GrowthModelSettings(use_shared_factor=1) # type: ignore[arg-type] + + +def test_from_legacy_extra_rejects_unknown_key(): + with pytest.raises(ValueError, match="unknown growth setting"): + GrowthModelSettings.from_legacy_extra( + {"use_shared_factor": True, "use_shared_factr": False}, # typo + model_id="lrp-rli-gc-999", + ) + + +def test_from_legacy_extra_round_trips_known_keys(): + settings = GrowthModelSettings.from_legacy_extra( + { + "outcomes": ("W", "L"), + "baseline_covariate": "blocks2", + "use_shared_factor": True, + "age_ability_interaction": True, + }, + model_id="lrp-rli-gc-999", + ) + assert settings.outcomes == ("W", "L") + assert settings.baseline_covariate == "blocks2" + assert settings.use_shared_factor is True + assert settings.age_ability_interaction is True + + +# --- resolve ------------------------------------------------------------------ + + +def _spec(**extra) -> ModelSpec: + # Growth is a multi-outcome family with no single outcome_symbol. + return ModelSpec(model_id="lrp-rli-gc-000", kind="growth", title="test", extra=extra) + + +def test_resolve_rejects_wrong_kind(): + spec = ModelSpec(model_id="x", kind="itt", title="t", outcome_symbol="W") + with pytest.raises(ValueError, match="expected kind 'growth'"): + resolve_growth_run_plan(spec) + + +def test_resolve_defaults_five_measures_on_blocks(): + plan = resolve_growth_run_plan(_spec()) + assert plan.settings_source == "legacy_extra" + prep = plan.prepare_kwargs() + assert prep["outcomes"] == ("R", "E", "T", "W", "L") + assert prep["baseline_covariates"] == ("blocks",) + fac = plan.factory_kwargs() + assert fac == { + "baseline_covariate": "blocks", + "use_shared_factor": False, + "age_ability_interaction": False, + } + assert "associational" in plan.causal_status.lower() + assert "growth rate" in plan.estimand.lower() + + +def test_resolve_keeps_shared_factor_and_interaction(): + plan = resolve_growth_run_plan( + _spec(use_shared_factor=True, age_ability_interaction=True) + ) + assert plan.factory_kwargs()["use_shared_factor"] is True + assert plan.factory_kwargs()["age_ability_interaction"] is True + + +def test_typed_settings_are_accepted_and_sourced(): + spec = ModelSpec( + model_id="lrp-rli-gc-000", + kind="growth", + title="test", + model_settings=GrowthModelSettings(use_shared_factor=True), + ) + plan = resolve_growth_run_plan(spec) + assert plan.settings_source == "typed" + assert plan.use_shared_factor is True + + +def test_split_settings_between_typed_and_extra_is_rejected(): + spec = ModelSpec( + model_id="lrp-rli-gc-000", + kind="growth", + title="test", + model_settings=GrowthModelSettings(), + extra={"use_shared_factor": True}, + ) + with pytest.raises(ValueError, match="cannot be split"): + resolve_growth_run_plan(spec) + + +# --- registered-specification coverage (acceptance criterion) ----------------- + + +def test_every_registered_growth_model_resolves_with_metadata(): + """Every registered latent growth-curve model resolves to a validated plan that + records the design, estimand, causal status, analysis population and missing-data + assumption (#394 pillar 4).""" + specs = _growth_specs() + assert len(specs) >= 3, f"expected the full growth suite, found {len(specs)}" + saw_factor = saw_interaction = False + for spec in specs: + plan = resolve_growth_run_plan(spec) + assert isinstance(plan, GrowthRunPlan) + recorded = plan.as_dict() + for field in _META_FIELDS: + assert isinstance(recorded[field], str) and recorded[field], ( + f"{spec.model_id}: {field} not recorded" + ) + assert plan.prepare_kwargs()["baseline_covariates"] == (plan.baseline_covariate,) + saw_factor |= plan.use_shared_factor + saw_interaction |= plan.age_ability_interaction + assert saw_factor, "no shared-factor growth model found (gc-070)" + assert saw_interaction, "no age x ability growth model found (gc-085)"