diff --git a/src/language_reading_predictors/statistical_models/concurrent.py b/src/language_reading_predictors/statistical_models/concurrent.py new file mode 100644 index 00000000..b0f9dd12 --- /dev/null +++ b/src/language_reading_predictors/statistical_models/concurrent.py @@ -0,0 +1,280 @@ +# 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 concurrent-associations family (#394 pillar 4). + +Mirrors the ITT / gain-factor / level-factor / DiD run-plan pattern (:mod:`itt`, +:mod:`gain_factors`, :mod:`level_factors`, :mod:`did`) for the per-wave concurrent +conditional-associations (``kind="concurrent"``) 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 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 concurrent design fits, **at each wave separately**, a between-child +Beta-Binomial regression of the focal outcome's level on the standardised same-wave +logits of a predictor skill set (plus age and a group nuisance term), reported side +by side with matched bivariate refits. Every coefficient is an **adjusted +association**; the family makes no causal claim, so conditioning on contemporaneous +(post-treatment) skill levels is intentional and the Table-2 fallacy applies. + +Because the model is fit once per wave with a wave-specific usable-predictor subset +(and the bivariate refits vary ``include_age`` / ``include_group``), the factory is +called many times by the pipeline; this plan therefore owns the *settings*, the +single ``load_and_prepare`` call and the recorded metadata, while the per-wave +factory calls stay in ``fit_concurrent`` using the resolved plan attributes. +``predictor_slope_sigma`` defaults to ``None`` so the pipeline can fill the factory +default through ``_default_of`` reflection, keeping the anti-drift single source. +""" + +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 concurrent family +# understands. Anything else is a typo and must fail before a fit starts. +_LEGACY_KEYS = frozenset( + { + "predictor_symbols", + "covariates", + "include_age", + "include_group", + "predictor_slope_sigma", + } +) + +_DEFAULT_PREDICTORS = ("L", "B", "TR", "TE", "R", "E") + + +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 ConcurrentModelSettings: + """Immutable settings declared by a single concurrent-associations model module. + + Defaults encode the primary six-skill concurrent read with age and a group + nuisance term. ``predictor_slope_sigma`` is ``None`` by default so the pipeline + fills the ``build_concurrent_model`` default via ``_default_of`` reflection + rather than duplicating the numeric literal here. + """ + + predictor_symbols: tuple[str, ...] = _DEFAULT_PREDICTORS + covariates: tuple[str, ...] = () + include_age: bool = True + include_group: bool = True + predictor_slope_sigma: float | None = None + + def __post_init__(self) -> None: + object.__setattr__( + self, + "predictor_symbols", + _tuple_of_strings(self.predictor_symbols, name="predictor_symbols"), + ) + object.__setattr__( + self, "covariates", _tuple_of_strings(self.covariates, name="covariates") + ) + for flag in ("include_age", "include_group"): + if not isinstance(getattr(self, flag), bool): + raise TypeError(f"{flag} must be bool") + sigma = self.predictor_slope_sigma + if sigma is not None: + # bool is an int subclass but is never a valid slope scale. + if isinstance(sigma, bool) or not isinstance(sigma, (int, float)): + raise TypeError("predictor_slope_sigma must be a number or None") + if sigma <= 0: + raise ValueError("predictor_slope_sigma must be positive") + object.__setattr__(self, "predictor_slope_sigma", float(sigma)) + + @classmethod + def from_legacy_extra( + cls, extra: Mapping[str, Any], *, model_id: str + ) -> ConcurrentModelSettings: + """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 concurrent setting(s): {', '.join(unknown)}. " + "Declare ConcurrentModelSettings so misspellings fail fast." + ) + # Pass raw values through so __post_init__ is the single validation/coercion + # point. ``predictor_slope_sigma`` absent -> None (pipeline uses the factory + # default via _default_of), matching the former .get(key, _default_of(...)). + return cls( + predictor_symbols=extra.get("predictor_symbols", _DEFAULT_PREDICTORS), + covariates=extra.get("covariates", ()), + include_age=extra.get("include_age", True), + include_group=extra.get("include_group", True), + predictor_slope_sigma=extra.get("predictor_slope_sigma"), + ) + + +@dataclass(frozen=True, slots=True) +class ConcurrentRunPlan: + """Concrete, validated instructions consumed by preparation and modelling.""" + + model_id: str + outcome_symbol: str + settings_source: str + predictor_symbols: tuple[str, ...] + covariates: tuple[str, ...] + include_age: bool + include_group: bool + # ``None`` -> the pipeline fills the build_concurrent_model default via _default_of. + predictor_slope_sigma: float | None + # Recorded audit metadata (#394 pillar 4). + design: str + estimand: str + causal_status: str + analysis_population: str + missing_data_assumption: str + + @property + def measure_outcomes(self) -> tuple[str, ...]: + """The outcome plus its predictor skills, de-duplicated and order-preserving.""" + return tuple(dict.fromkeys((self.outcome_symbol, *self.predictor_symbols))) + + 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_and_prepare`` from the resolved plan. + + The concurrent family loads the per-wave ``levels`` panel with the outcome and + its predictor skills as bounded-count measures; the trait covariates are + t1-measured and enter as baseline covariates broadcast across the waves.""" + return { + "phase_mode": "levels", + "outcomes": self.measure_outcomes, + "baseline_covariates": self.covariates, + } + + def recipe_markdown(self, *, title: str) -> str: + """Undergraduate-friendly explanation generated from the resolved plan.""" + preds = ", ".join(self.predictor_symbols) if self.predictor_symbols else "none" + covs = ", ".join(self.covariates) if self.covariates else "none" + sigma = ( + "build default" + if self.predictor_slope_sigma is None + else f"{self.predictor_slope_sigma:g}" + ) + return ( + "Note: Generated from the validated concurrent-associations 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"Outcome: `{self.outcome_symbol}`. Predictor skills: {preds}. Trait " + f"covariates: {covs}. Age term: {self.include_age}. Group nuisance term: " + f"{self.include_group}. Predictor-slope prior sigma: {sigma}.\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_concurrent_settings( + spec: ModelSpec, +) -> tuple[ConcurrentModelSettings, 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}: concurrent settings cannot be split between " + "model_settings and extra" + ) + if not isinstance(settings, ConcurrentModelSettings): + raise TypeError( + f"{spec.model_id}: kind='concurrent' requires " + f"ConcurrentModelSettings, got {type(settings).__name__}" + ) + return settings, "typed" + return ( + ConcurrentModelSettings.from_legacy_extra(spec.extra, model_id=spec.model_id), + "legacy_extra", + ) + + +def resolve_concurrent_run_plan(spec: ModelSpec) -> ConcurrentRunPlan: + """Resolve and validate a concurrent-associations spec before any data are loaded.""" + if spec.kind != "concurrent": + raise ValueError( + f"{spec.model_id}: expected kind 'concurrent', got {spec.kind!r}" + ) + if not spec.outcome_symbol: + raise ValueError( + f"{spec.model_id}: outcome_symbol is required for a concurrent model" + ) + + settings, source = declared_concurrent_settings(spec) + own = spec.outcome_symbol + + design = ( + "Per-wave concurrent conditional associations: at each timepoint a " + "between-child Beta-Binomial regression of the outcome level on the " + "standardised same-wave logits of the predictor skills, plus optional age and " + "a group nuisance term. Four cross-sectional fits reported side by side, each " + "with a matched single-predictor (bivariate) refit." + ) + estimand = ( + "Adjusted per-wave conditional associations (per +1 SD of each predictor's " + "same-wave logit), reported alongside the unadjusted bivariate association. " + "No causal quantity is estimated: conditioning on contemporaneous " + "post-treatment skill levels is intentional and the Table-2 fallacy applies." + ) + causal_status = ( + "Associational only. The family makes no causal claim; every coefficient is a " + "latent-ability-confounded adjusted association at a wave, never a cause." + ) + analysis_population = ( + "Available-case children observed at each wave (about 53 per wave, " + "outcome-complete). The reported associations average over the fitted rows at " + "each wave -- a descriptive averaging population." + ) + missing_data_assumption = ( + "Rows missing the focal outcome are dropped (an outcome cannot be imputed); " + "missing predictor values are mean-imputed (a descriptive read -- imputation " + "can bias a conditional coefficient when missingness relates to the predictor, " + "outcome or other skills)." + ) + + return ConcurrentRunPlan( + model_id=spec.model_id, + outcome_symbol=own, + settings_source=source, + predictor_symbols=settings.predictor_symbols, + covariates=settings.covariates, + include_age=settings.include_age, + include_group=settings.include_group, + predictor_slope_sigma=settings.predictor_slope_sigma, + 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..a4ed9910 100644 --- a/src/language_reading_predictors/statistical_models/pipeline.py +++ b/src/language_reading_predictors/statistical_models/pipeline.py @@ -73,6 +73,9 @@ save_plotcollection, save_styled_figure, ) +from language_reading_predictors.statistical_models.concurrent import ( + resolve_concurrent_run_plan, +) from language_reading_predictors.statistical_models.context import ( ModelSpec, StatisticalFitContext, @@ -7385,35 +7388,39 @@ def fit_concurrent(spec: ModelSpec, config: str = "dev") -> StatisticalFitContex detailed probability/items marginals (wave × predictor × {+1 SD, +k items}). """ _require_spec(spec, "concurrent", outcome=True) - e = spec.extra - outcome = spec.outcome_symbol or "W" - predictor_symbols = list(e.get("predictor_symbols", ["L", "B", "TR", "TE", "R", "E"])) + # 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, the teaching recipe and config.json. + plan = resolve_concurrent_run_plan(spec) + outcome = plan.outcome_symbol + predictor_symbols = list(plan.predictor_symbols) # Trait covariates (non-verbal ability, hearing, speech, phonological memory), # aligned with the gains panel. They are t1-measured, so they enter as # baseline covariates broadcast across the waves (there is no per-wave value). - covariates = list(e.get("covariates", [])) - include_age = bool(e.get("include_age", True)) - include_group = bool(e.get("include_group", True)) - sigma0 = float( - e.get( - "predictor_slope_sigma", - _default_of(_factories.build_concurrent_model, "predictor_slope_sigma"), + covariates = list(plan.covariates) + include_age = plan.include_age + include_group = plan.include_group + # ``predictor_slope_sigma`` is None on the plan when a spec does not set it, so the + # build_concurrent_model default is filled via _default_of here — the anti-drift + # single source #394 retains until typed family defaults replace it. + sigma0 = ( + float(plan.predictor_slope_sigma) + if plan.predictor_slope_sigma is not None + else float( + _default_of(_factories.build_concurrent_model, "predictor_slope_sigma") ) ) from language_reading_predictors.statistical_models.measures import MEASURES ctx = make_context(spec, config) + ctx.resolved_plan = plan + _report.write_model_recipe(ctx) hdi = ctx.reporting.ci_prob N_focal = MEASURES[outcome].n_trials section_header("Prepare data") - measure_outcomes = tuple(dict.fromkeys([outcome, *predictor_symbols])) - prepared_all = load_and_prepare( - phase_mode="levels", - outcomes=measure_outcomes, - baseline_covariates=tuple(covariates), - ) + prepared_all = load_and_prepare(**plan.prepare_kwargs()) # Timepoints present; each wave's row count and its usable predictor set (a # predictor whose same-wave logit has positive variance on the wave's rows — diff --git a/src/language_reading_predictors/statistical_models/reporting.py b/src/language_reading_predictors/statistical_models/reporting.py index 8fac06fe..b4f529a6 100644 --- a/src/language_reading_predictors/statistical_models/reporting.py +++ b/src/language_reading_predictors/statistical_models/reporting.py @@ -27,6 +27,10 @@ from language_reading_predictors.statistical_models.context import ( StatisticalFitContext, ) +from language_reading_predictors.statistical_models.concurrent import ( + ConcurrentRunPlan, + resolve_concurrent_run_plan, +) from language_reading_predictors.statistical_models.gain_factors import ( GainFactorsRunPlan, resolve_gain_factors_run_plan, @@ -2273,12 +2277,22 @@ def _gain_factors_run_plan(context: StatisticalFitContext) -> GainFactorsRunPlan return resolve_gain_factors_run_plan(context.spec) +def _concurrent_run_plan(context: StatisticalFitContext) -> ConcurrentRunPlan: + """Return the concurrent plan resolved before loading, or reconstruct it.""" + resolved_plan = getattr(context, "resolved_plan", None) + if isinstance(resolved_plan, ConcurrentRunPlan): + return resolved_plan + return resolve_concurrent_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, concurrent), 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 == "concurrent": + return _concurrent_run_plan(context) return None diff --git a/tests/statistical_models/test_concurrent_run_plan.py b/tests/statistical_models/test_concurrent_run_plan.py new file mode 100644 index 00000000..82e45c22 --- /dev/null +++ b/tests/statistical_models/test_concurrent_run_plan.py @@ -0,0 +1,192 @@ +# Copyright (c) 2026 Down Syndrome Education International and contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +"""Tests for the typed concurrent-associations 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.concurrent import ( + ConcurrentModelSettings, + ConcurrentRunPlan, + resolve_concurrent_run_plan, +) +from language_reading_predictors.statistical_models.context import ModelSpec + +_META_FIELDS = ( + "design", + "estimand", + "causal_status", + "analysis_population", + "missing_data_assumption", +) + + +def _concurrent_specs() -> list[ModelSpec]: + """Every registered concurrent-associations model's SPEC.""" + root = os.path.dirname( + importlib.import_module( + "language_reading_predictors.statistical_models.concurrent" + ).__file__ + ) + specs: list[ModelSpec] = [] + for path in sorted(glob.glob(os.path.join(root, "lrp_rli_ca_*.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 == "concurrent": + specs.append(spec) + return specs + + +# --- settings validation ------------------------------------------------------ + + +def test_settings_reject_non_bool_include_age(): + with pytest.raises(TypeError, match="include_age"): + ConcurrentModelSettings(include_age=1) # type: ignore[arg-type] + + +def test_settings_reject_string_predictor_symbols(): + with pytest.raises(TypeError, match="predictor_symbols"): + ConcurrentModelSettings(predictor_symbols="L") # type: ignore[arg-type] + + +def test_settings_reject_string_covariates(): + with pytest.raises(TypeError, match="covariates"): + ConcurrentModelSettings(covariates="hs") # type: ignore[arg-type] + + +def test_settings_reject_non_positive_sigma(): + with pytest.raises(ValueError, match="predictor_slope_sigma must be positive"): + ConcurrentModelSettings(predictor_slope_sigma=0.0) + + +def test_settings_reject_bool_sigma(): + with pytest.raises(TypeError, match="predictor_slope_sigma"): + ConcurrentModelSettings(predictor_slope_sigma=True) # type: ignore[arg-type] + + +def test_from_legacy_extra_rejects_unknown_key(): + with pytest.raises(ValueError, match="unknown concurrent setting"): + ConcurrentModelSettings.from_legacy_extra( + {"predictor_symbols": ("L",), "predictor_symbol": ("B",)}, # typo + model_id="lrp-rli-ca-999", + ) + + +def test_from_legacy_extra_round_trips_and_sigma_defaults_none(): + settings = ConcurrentModelSettings.from_legacy_extra( + { + "predictor_symbols": ("L", "B"), + "covariates": ("hs", "blocks"), + "include_age": False, + "include_group": False, + }, + model_id="lrp-rli-ca-999", + ) + assert settings.predictor_symbols == ("L", "B") + assert settings.covariates == ("hs", "blocks") + assert settings.include_age is False + assert settings.include_group is False + # Absent -> None so the pipeline fills the factory default via _default_of. + assert settings.predictor_slope_sigma is None + + +# --- resolve ------------------------------------------------------------------ + + +def _spec(**extra) -> ModelSpec: + return ModelSpec( + model_id="lrp-rli-ca-000", + kind="concurrent", + title="test", + outcome_symbol="W", + 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 'concurrent'"): + resolve_concurrent_run_plan(spec) + + +def test_resolve_is_levels_frame_and_associational(): + plan = resolve_concurrent_run_plan(_spec(predictor_symbols=("L", "B"))) + assert plan.settings_source == "legacy_extra" + assert plan.predictor_slope_sigma is None # unset -> pipeline fills via _default_of + prep = plan.prepare_kwargs() + assert prep["phase_mode"] == "levels" + # outcome first, then predictors, de-duplicated. + assert prep["outcomes"] == ("W", "L", "B") + assert prep["baseline_covariates"] == () + assert plan.causal_status.startswith("Associational") + assert "Table-2 fallacy" in plan.estimand + + +def test_resolve_dedups_outcome_in_measure_outcomes(): + # A predictor equal to the outcome must not appear twice in the load list. + plan = resolve_concurrent_run_plan(_spec(predictor_symbols=("W", "L"))) + assert plan.prepare_kwargs()["outcomes"] == ("W", "L") + + +def test_resolve_keeps_covariates_and_explicit_sigma(): + plan = resolve_concurrent_run_plan( + _spec(covariates=("blocks", "hs"), predictor_slope_sigma=0.5) + ) + assert plan.prepare_kwargs()["baseline_covariates"] == ("blocks", "hs") + assert plan.predictor_slope_sigma == 0.5 + + +def test_typed_settings_are_accepted_and_sourced(): + spec = ModelSpec( + model_id="lrp-rli-ca-000", + kind="concurrent", + title="test", + outcome_symbol="W", + model_settings=ConcurrentModelSettings(predictor_symbols=("L",)), + ) + plan = resolve_concurrent_run_plan(spec) + assert plan.settings_source == "typed" + assert plan.predictor_symbols == ("L",) + + +def test_split_settings_between_typed_and_extra_is_rejected(): + spec = ModelSpec( + model_id="lrp-rli-ca-000", + kind="concurrent", + title="test", + outcome_symbol="W", + model_settings=ConcurrentModelSettings(), + extra={"include_age": False}, + ) + with pytest.raises(ValueError, match="cannot be split"): + resolve_concurrent_run_plan(spec) + + +# --- registered-specification coverage (acceptance criterion) ----------------- + + +def test_every_registered_concurrent_model_resolves_with_metadata(): + """Every registered concurrent model resolves to a validated plan that records the + design, estimand, causal status, analysis population and missing-data assumption + (#394 pillar 4).""" + specs = _concurrent_specs() + assert len(specs) >= 11, f"expected the full concurrent suite, found {len(specs)}" + for spec in specs: + plan = resolve_concurrent_run_plan(spec) + assert isinstance(plan, ConcurrentRunPlan) + recorded = plan.as_dict() + for field in _META_FIELDS: + assert isinstance(recorded[field], str) and recorded[field], ( + f"{spec.model_id}: {field} not recorded" + ) + # The outcome always loads as the first measure outcome. + assert plan.prepare_kwargs()["outcomes"][0] == spec.outcome_symbol