Build a Frozen 3PL Transform

Turn 3PL item curves into reusable FPCA scores with a frozen transform.

Introduction

30 Second Summary

A result that works during training can still change when new data arrives. Trust comes from knowing every score follows the same learned process.

In this project, you will build a tested statistical foundation that converts valid 3PL item parameters into smooth FPCA scores. You will freeze the fitted transform so unseen curves use the same coordinate system without refitting.

What You'll Build

When you run the finished foundation, you'll see 240 valid 3PL items become stable FPCA scores that a saved transform can reproduce for unseen curves.

By the end of this project, you'll have:

  • Deterministic synthetic data with 240 parameter rows across six configurable tiers. Tier labels remain in a separate file so fitting code cannot use them.
  • Aggregate validation for complete CSV files. Every failure identifies its physical row. It also identifies the affected column plus a clear message. Invalid data never enters the statistical pipeline.
  • Exact 201-point item characteristic curves processed through penalized B-spline smoothing before signed FPCA. A frozen transform projects unseen curves without fitting again. An empirical report records smoothing error, persistence checks, projection checks, and refit subspace angles.
  • Secret Mission: Project an extreme but valid item outside every configured tier range. Prove the frozen transform remains byte-for-byte unchanged.

Are there any prerequisites?

You should be comfortable reading Python. Familiarity with NumPy array shapes and basic PCA terminology also helps.

Before We Start

Before any code, this checkpoint defines the boundary between training and inference. The frozen transform accepts one or more valid exact 201-point 3PL ICC rows on its stored theta grid. It returns signed FPCA scores through stored smoothing and FPCA objects without fitting or mutation.

Set Up a Compatible WSL Environment

Your frozen transform depends on a numerical stack with compatible compiled packages. The existing Python 3.14 installation is newer than the environment selected for this project.

This project uses scikit-fda 0.10.1. That release supports scikit-learn 1.6.

A dedicated Python 3.12 environment in WSL gives the smoothing and FPCA code a reproducible foundation.

Why This Environment?

The selected scikit-fda release carries scikit-learn as a dependency. Pinning scikit-learn==1.6.1 keeps that transitive dependency inside the supported compatibility target.

Python 3.12 provides a conservative runtime for this compiled numerical stack. The isolated environment prevents your existing Python 3.14 packages from affecting the project.

In this step, get ready to:
  • Create an isolated Python 3.12 environment with pip 26.2.2.
  • Generate a platform-specific dependency lock and install its resolved packages.
  • Create the project directories and save the statistical defaults.
Create the Python environment

A virtual environment keeps this project's interpreter and packages separate from the rest of your WSL installation. First, confirm that the Python 3.12 executable is available.

Choose the tab that matches the result you see in your WSL terminal.

✔️ I See the Required Version

  • Confirm the available interpreter by running this command:
python3.12 --version

What Does This Check?

The version option prints the interpreter version without starting an interactive Python session. You should see Python 3.12 followed by its patch version.

ⓧ I See a Different Version

The installed interpreter does not match this project's compatibility target. Your existing Python 3.14 installation can remain available for other work.

  • Provision Python 3.12 in WSL with the package or version manager method you used for Python 3.14.
  • Confirm the newly provisioned interpreter by running this command:
python3.12 --version

What Should I See?

The command should now report Python 3.12 followed by its patch version. This confirms that the required interpreter is available alongside your newer installation.

ⓧ Command Not Found

WSL cannot currently find a Python 3.12 executable. Your ability to provision another Python version supplies the missing interpreter.

  • Provision Python 3.12 in your WSL distribution.
  • Confirm that WSL can find the interpreter by running this command:
python3.12 --version

What Should I See?

The command should report Python 3.12 followed by its patch version. That output proves WSL can use the selected interpreter.

  • Create the .venv environment and activate it by running these commands:
python3.12 -m venv .venv
source .venv/bin/activate

What Do These Commands Do?

  • The first command creates .venv with the Python 3.12 interpreter.
  • The second command makes that environment's Python and pip executables active in the current WSL terminal.

Your terminal prompt should now begin with (.venv). That prompt confirms the project environment is active.

Environment Creation Failed?

Confirm that the version check reports Python 3.12. A missing virtual-environment component can prevent the environment directory from being created.

After fixing the interpreter installation, remove any incomplete .venv directory before rerunning the creation commands.

Help me diagnose my Python virtual environment setup.

The active environment begins with its bundled pip version. Upgrade it before generating the dependency lock.

  • Upgrade the active environment to pip 26.2.2 by running:
python -m pip install --upgrade "pip==26.2.2"

What Does This Command Do?

The python -m pip form runs pip through the active Python interpreter. The quoted pin installs pip 26.2.2 inside .venv.

Pip Upgrade Failed?

Confirm that your prompt still begins with (.venv). Network or package-index access can also interrupt the download.

Help me troubleshoot the pip upgrade in WSL.

Lock and install the dependencies

A dependency lock records the exact packages selected for this Python version and WSL platform. It also records the source URLs and hashes needed to reproduce the installation.

The six direct packages power the project code. The additional scikit-learn pin constrains a dependency that scikit-fda uses for functional principal-component analysis.

  • Generate pylock.toml from the package pins by running:
python -m pip lock -o pylock.toml \
  "numpy==2.5.4" \
  "scipy==1.18.1" \
  "scikit-fda==0.10.1" \
  "pandas==3.1.0" \
  "PyYAML==6.0.3" \
  "pytest==9.1.1" \
  "scikit-learn==1.6.1"

What Goes Into the Lock?

  • NumPy and SciPy provide the array and numerical foundations.
  • pandas loads parameter tables. PyYAML loads the project configuration.
  • pytest runs the contract tests. The scikit-learn constraint keeps scikit-fda on its selected compatibility path.

The generated pylock.toml is specific to the current Python version and WSL platform. Regenerate it before using the project on another platform.

  • Confirm that the lock file exists by running:
ls pylock.toml

What Should I See?

The terminal should print pylock.toml. That filename confirms the resolver wrote the lock into the current project directory.

Lock File Missing?

Confirm that the pip upgrade completed inside .venv. A dependency-resolution failure prevents pylock.toml from being written.

Help me diagnose a failed pip lock command.

  • Install every resolved dependency from the generated lock by running:
python -m pip install -r pylock.toml

What Does This Command Do?

The -r pylock.toml option tells pip to install the resolved packages recorded in the lock. The active .venv receives those packages.

Dependency Installation Failed?

Confirm that pylock.toml was generated with Python 3.12 on this WSL platform. A lock created elsewhere may contain incompatible resolved artifacts.

Help me troubleshoot installation from the generated lock.

Create the project structure

The statistical foundation separates reusable modules from configuration, tests, scripts, and generated artifacts. This structure keeps training logic distinct from the frozen inference path you build later.

  • Create the five project directories and list them by running:
mkdir -p core config tests scripts artifacts
ls -d artifacts config core scripts tests

What Does This Create?

  • The core directory holds validation, I/O, generation, smoothing, FPCA, and transform modules.
  • The config directory holds the shared statistical defaults.
  • The tests and scripts directories hold contract tests and the end-to-end runner.
  • The artifacts directory receives generated CSVs, the saved transform, and the empirical report.

Directories Not Listed?

Confirm that you have permission to create files in the current WSL directory. Check the command for the exact directory names shown above.

Help me fix the WSL project directory creation.

The configuration fixes every shared grid, smoothing, FPCA, tolerance, and synthetic-data value. Later modules read this one file instead of duplicating statistical constants.

  • Create config/defaults.yaml with the complete configuration by running:
cat <<'YAML' > config/defaults.yaml
theta:
  start: -3.0
  stop: 3.0
  num_points: 201
smoothing:
  spline_order: 4
  num_basis: 40
  penalty_derivative_order: 2
  lambda_candidates:
    logspace: {start: -8, stop: 2, num: 7}
  max_seconds: 1200
fpca:
  max_components: 10
  variance_threshold: 0.95
  sign: largest-abs loading positive
tolerance:
  in_sample: 1.0e-6
  held_out: 1.0e-2
synthetic:
  seed: 42
  n_items: 240
  tiers:
    - {name: tier_1, a: [0.45, 0.80], b: [-2.50, -1.50], c: [0.00, 0.04]}
    - {name: tier_2, a: [0.80, 1.15], b: [-1.50, -0.50], c: [0.02, 0.08]}
    - {name: tier_3, a: [1.15, 1.50], b: [-0.50, 0.50], c: [0.04, 0.12]}
    - {name: tier_4, a: [1.50, 1.90], b: [0.50, 1.50], c: [0.06, 0.16]}
    - {name: tier_5, a: [1.90, 2.40], b: [1.50, 2.50], c: [0.10, 0.22]}
    - {name: tier_6, a: [0.60, 2.20], b: [-2.50, 2.50], c: [0.22, 0.34]}
YAML

What Does This Configuration Control?

  • The theta section defines 201 evaluation points from -3.0 through 3.0.
  • The smoothing section defines a cubic order-4 basis with 40 functions. It also defines the second-derivative penalty and seven lambda candidates.
  • The FPCA section caps the model at 10 components. It targets 0.95 cumulative explained variance.
  • The synthetic section fixes seed 42 for 240 items across six parameter tiers.
  • Print the saved configuration back to the terminal by running:
cat config/defaults.yaml

What Should I See?

You should see the complete YAML configuration in the terminal. Confirm that it starts with the theta grid and ends with tier_6.

Configuration Looks Incomplete?

Confirm that the closing YAML marker starts at the beginning of its line. Check that the tier lines retain their indentation.

Help me repair my defaults file.

Use these tabs to confirm the complete configuration before the environment check.

✔️ Awesome, I've got everything!

Great work. Your project now has one deterministic source for every statistical default.

ⓧ I'd like to double check the full code

theta:
  start: -3.0
  stop: 3.0
  num_points: 201
smoothing:
  spline_order: 4
  num_basis: 40
  penalty_derivative_order: 2
  lambda_candidates:
    logspace: {start: -8, stop: 2, num: 7}
  max_seconds: 1200
fpca:
  max_components: 10
  variance_threshold: 0.95
  sign: largest-abs loading positive
tolerance:
  in_sample: 1.0e-6
  held_out: 1.0e-2
synthetic:
  seed: 42
  n_items: 240
  tiers:
    - {name: tier_1, a: [0.45, 0.80], b: [-2.50, -1.50], c: [0.00, 0.04]}
    - {name: tier_2, a: [0.80, 1.15], b: [-1.50, -0.50], c: [0.02, 0.08]}
    - {name: tier_3, a: [1.15, 1.50], b: [-0.50, 0.50], c: [0.04, 0.12]}
    - {name: tier_4, a: [1.50, 1.90], b: [0.50, 1.50], c: [0.06, 0.16]}
    - {name: tier_5, a: [1.90, 2.40], b: [1.50, 2.50], c: [0.10, 0.22]}
    - {name: tier_6, a: [0.60, 2.20], b: [-2.50, 2.50], c: [0.22, 0.34]}

How to Compare This File

Compare each section with your saved config/defaults.yaml file. The indentation and numeric values must match exactly.

Before you run the final check, do you expect every import to come from the active environment? Hold your answer in mind.

  • Verify the interpreter and all required imports by running:
python --version && python -c "import numpy, scipy, skfda, pandas, yaml, pytest; print(skfda.__version__)"

What Should I See?

The first line should report Python 3.12 followed by its patch version. The final line should report 0.10.1.

The silent imports confirm that NumPy, SciPy, scikit-fda, pandas, PyYAML, and pytest all load successfully. That is the complete compatibility layer working.

Import Check Failed?

Confirm that your prompt begins with (.venv). Activate the environment again if the marker is missing.

Regenerate the lock inside Python 3.12 if the reported interpreter is different. Reinstall from pylock.toml if an import is missing.

Help me diagnose the final environment check.

Your Python 3.12 environment and project structure are ready. Next, you will generate deterministic 3PL parameters and turn them into exact ICC curves.

Generate Synthetic Items and Exact ICCs

Your pinned WSL environment is ready to turn the configured item tiers into a reproducible statistical baseline. This baseline needs parameter rows that stay separate from their synthetic tier labels.

The first curves also need an exact mathematical reference before smoothing begins. Raw CSV parsing has not yet proved that it can enforce the rules of the 3PL model.

In this step, get ready to:
  • Generate 240 deterministic item-parameter rows with separate tier labels.
  • Compute exact item characteristic curves across 201 theta points.
  • Test raw parsing with a deliberately invalid parameter row.
Generate deterministic items

A fixed random seed makes every run produce the same parameters. Separate labels prevent tier information from entering the parameter table used by the statistical pipeline.

The foundation runner needs a small configuration loader before it can read the existing config/defaults.yaml file.

  • Create io.py inside the existing core/ folder using the editor connected to your WSL project.
  • Add the configuration-loading helper to core/io.py by copying this code:
"""Configuration and validated CSV loading."""
from __future__ import annotations
from pathlib import Path
from typing import Any
import yaml

def load_config(path: str | Path) -> dict[str, Any]:
    with Path(path).open("r", encoding="utf-8") as handle:
        config = yaml.safe_load(handle)
    if not isinstance(config, dict):
        raise ValueError("Configuration root must be a mapping.")
    return config

What does this code do?

  • The module uses PyYAML to read the project configuration safely.
  • The load_config() function checks that the configuration root is a mapping.
  • The returned dictionary gives the runner access to the existing theta settings plus the synthetic tier settings.
  • Save core/io.py.
  • Confirm that the editor file explorer lists io.py inside core/.

Cannot save the configuration helper?

Check that io.py sits inside the existing core/ folder. Confirm that your editor is connected to the same WSL project shown in the terminal.

Help me place the configuration helper correctly.

✔️ Awesome, I've got everything!

Great. Save core/io.py before you continue.

ⓧ I'd like to double check the full code

"""Configuration and validated CSV loading."""
from __future__ import annotations
from pathlib import Path
from typing import Any
import yaml

def load_config(path: str | Path) -> dict[str, Any]:
    with Path(path).open("r", encoding="utf-8") as handle:
        config = yaml.safe_load(handle)
    if not isinstance(config, dict):
        raise ValueError("Configuration root must be a mapping.")
    return config

How to use this reference

Compare this reference with your saved file. The module docstring states its purpose.

The generator uses NumPy for seeded sampling. It returns a pandas DataFrame containing only the four model fields.

  • Create synthetic.py inside the existing core/ folder using your editor.
  • Add the generator dependencies to core/synthetic.py by copying this code:
"""Deterministic synthetic 3PL item generation."""
from __future__ import annotations
from collections.abc import Sequence
from typing import Any
import numpy as np
import pandas as pd

Why these imports?

  • The module docstring records that this file owns deterministic item generation.
  • The typing imports describe the configurable sequence of tier dictionaries.
  • NumPy handles seeded sampling. pandas builds the parameter table.
  • Save core/synthetic.py.
  • Confirm that the editor file explorer lists synthetic.py inside core/.

Seeing unresolved import indicators?

Confirm that your editor uses the active Python 3.12 .venv. The pinned environment already contains NumPy plus pandas.

Help me check the imports in core/synthetic.py.

  • Add generate_synthetic() below the imports in core/synthetic.py by copying this code:
def generate_synthetic(
    n_items: int,
    seed: int,
    tiers: Sequence[dict[str, Any]],
) -> tuple[pd.DataFrame, np.ndarray]:
    """Return parameters and tier labels as separate objects."""
    if n_items <= 0 or not tiers:
        raise ValueError("n_items and tiers must be non-empty.")
    rng = np.random.default_rng(seed)
    counts = np.full(len(tiers), n_items // len(tiers), dtype=int)
    counts[: n_items % len(tiers)] += 1
    a_parts: list[np.ndarray] = []
    b_parts: list[np.ndarray] = []
    c_parts: list[np.ndarray] = []
    labels: list[str] = []
    for tier, count in zip(tiers, counts, strict=True):
        a_parts.append(rng.uniform(*tier["a"], size=int(count)))
        b_parts.append(rng.uniform(*tier["b"], size=int(count)))
        c_parts.append(rng.uniform(*tier["c"], size=int(count)))
        labels.extend([str(tier["name"])] * int(count))
    params = pd.DataFrame({
        "item_id": [f"ITEM-{index + 1:04d}" for index in range(n_items)],
        "a": np.concatenate(a_parts),
        "b": np.concatenate(b_parts),
        "c": np.concatenate(c_parts),
    })
    return params, np.asarray(labels, dtype=str)

What does this generator do?

  • The fixed seed initializes the same random-number sequence on every run.
  • The counts array distributes the requested item total across every configured tier.
  • The parameter DataFrame contains item_id, a, b, and c.
  • The tier names leave the function as a separate NumPy string array.
  • Save core/synthetic.py.
  • Confirm that generate_synthetic() ends by returning params plus the separate label array.

Generator structure looks different?

Check that params has exactly four keys. Keep labels outside the DataFrame constructor.

Help me compare my synthetic generator with the required structure.

✔️ Awesome, I've got everything!

Your deterministic generator is ready. Double check that core/synthetic.py is saved.

ⓧ I'd like to double check the full code

"""Deterministic synthetic 3PL item generation."""
from __future__ import annotations
from collections.abc import Sequence
from typing import Any
import numpy as np
import pandas as pd

def generate_synthetic(
    n_items: int,
    seed: int,
    tiers: Sequence[dict[str, Any]],
) -> tuple[pd.DataFrame, np.ndarray]:
    """Return parameters and tier labels as separate objects."""
    if n_items <= 0 or not tiers:
        raise ValueError("n_items and tiers must be non-empty.")
    rng = np.random.default_rng(seed)
    counts = np.full(len(tiers), n_items // len(tiers), dtype=int)
    counts[: n_items % len(tiers)] += 1
    a_parts: list[np.ndarray] = []
    b_parts: list[np.ndarray] = []
    c_parts: list[np.ndarray] = []
    labels: list[str] = []
    for tier, count in zip(tiers, counts, strict=True):
        a_parts.append(rng.uniform(*tier["a"], size=int(count)))
        b_parts.append(rng.uniform(*tier["b"], size=int(count)))
        c_parts.append(rng.uniform(*tier["c"], size=int(count)))
        labels.extend([str(tier["name"])] * int(count))
    params = pd.DataFrame({
        "item_id": [f"ITEM-{index + 1:04d}" for index in range(n_items)],
        "a": np.concatenate(a_parts),
        "b": np.concatenate(b_parts),
        "c": np.concatenate(c_parts),
    })
    return params, np.asarray(labels, dtype=str)

How to use this reference

Compare the imports plus the complete function with your saved file. The function docstring highlights the separation between parameters and labels.

Compute exact ICCs

An item characteristic curve maps each theta value to the probability of a correct response. Broadcasting evaluates every item against the shared theta grid without a Python loop over rows.

  • Create icc.py inside the existing core/ folder using your editor.
  • Add the exact 3PL equation plus the grid builder to core/icc.py by copying this code:
"""Exact vectorized three-parameter logistic ICCs."""
from __future__ import annotations
import numpy as np
import pandas as pd

def three_pl_probability(a, b, c, theta) -> np.ndarray:
    """Evaluate the 3PL equation with NumPy broadcasting."""
    return c + (1.0 - c) / (1.0 + np.exp(-a * (theta - b)))

def compute_iccs(
    params: pd.DataFrame,
    start: float,
    stop: float,
    num_points: int,
) -> tuple[np.ndarray, np.ndarray]:
    theta = np.linspace(start, stop, num=num_points)
    a = params["a"].to_numpy(dtype=float)[:, None]
    b = params["b"].to_numpy(dtype=float)[:, None]
    c = params["c"].to_numpy(dtype=float)[:, None]
    return three_pl_probability(a, b, c, theta[None, :]), theta

What does this code do?

  • The three_pl_probability() function implements the exact three-parameter logistic equation.
  • The theta array contains the configured points from -3.0 through 3.0.
  • The extra axis on each parameter array gives every item one row.
  • NumPy broadcasting combines 240 parameter rows with 201 theta columns.
  • Save core/icc.py.
  • Confirm that the editor file explorer lists icc.py inside core/.

ICC dimensions look unclear?

Check that each parameter conversion ends with [:, None]. Check that the theta argument uses theta[None, :].

Help me understand the broadcasting shapes in core/icc.py.

✔️ Awesome, I've got everything!

The exact ICC functions are in place. Save core/icc.py before building the runner.

ⓧ I'd like to double check the full code

"""Exact vectorized three-parameter logistic ICCs."""
from __future__ import annotations
import numpy as np
import pandas as pd

def three_pl_probability(a, b, c, theta) -> np.ndarray:
    """Evaluate the 3PL equation with NumPy broadcasting."""
    return c + (1.0 - c) / (1.0 + np.exp(-a * (theta - b)))

def compute_iccs(
    params: pd.DataFrame,
    start: float,
    stop: float,
    num_points: int,
) -> tuple[np.ndarray, np.ndarray]:
    theta = np.linspace(start, stop, num=num_points)
    a = params["a"].to_numpy(dtype=float)[:, None]
    b = params["b"].to_numpy(dtype=float)[:, None]
    c = params["c"].to_numpy(dtype=float)[:, None]
    return three_pl_probability(a, b, c, theta[None, :]), theta

How to use this reference

Compare this complete file with your saved version. The function docstring records why broadcasting is used.

Run the deliberately corrupt preview

The runner turns the generator plus the ICC functions into one visible workflow. It also creates a two-row preview with a nonpositive discrimination parameter for the raw parser to read.

  • Create run_foundation.py inside the existing scripts/ folder using your editor.
  • Add the runner imports plus its project path to scripts/run_foundation.py by copying this code:
"""Run the complete foundation and write reproducible artifacts."""
from __future__ import annotations
from pathlib import Path
import pandas as pd
from core.icc import compute_iccs
from core.io import load_config
from core.synthetic import generate_synthetic

ROOT = Path(__file__).resolve().parents[1]

Why these runner imports?

  • The ROOT path anchors configuration plus artifact locations to the project folder.
  • The imports connect configuration loading to synthetic generation.
  • The ICC import gives the runner access to exact curve evaluation.
  • Save scripts/run_foundation.py.
  • Confirm that the editor file explorer lists run_foundation.py inside scripts/.

Runner imports cannot find core?

Confirm that scripts/ sits beside core/ in the same project folder. Run the module from that project folder during the final check.

Help me check the module layout for scripts.run_foundation.

  • Add the initial main() workflow below the ROOT line in scripts/run_foundation.py by copying this code:
def main() -> None:
    config = load_config(ROOT / "config/defaults.yaml")
    output = ROOT / "artifacts"; output.mkdir(exist_ok=True)
    lines: list[str] = []
    def emit(message: str) -> None:
        print(message); lines.append(message)

    synthetic = config["synthetic"]
    params, labels = generate_synthetic(
        synthetic["n_items"], synthetic["seed"], synthetic["tiers"],
    )
    params_path, labels_path = output / "params.csv", output / "labels.csv"
    params.to_csv(params_path, index=False)
    pd.DataFrame({"item_id": params["item_id"], "tier": labels}).to_csv(
        labels_path, index=False,
    )

    corrupt = params.iloc[:2].copy()
    corrupt.loc[corrupt.index[0], "a"] = -1.0
    corrupt_path = output / "corrupt_preview.csv"
    corrupt.to_csv(corrupt_path, index=False)
    raw_corrupt = pd.read_csv(corrupt_path)
    emit(f"raw parser accepted corrupt preview a={raw_corrupt.loc[0, 'a']}")

    curves, grid = compute_iccs(params, **config["theta"])
    emit(f"icc shape: {curves.shape}, grid shape: {grid.shape}")

if __name__ == "__main__":
    main()

What does this workflow do?

  • The runner reads the synthetic settings from config/defaults.yaml.
  • It writes parameters to artifacts/params.csv.
  • It writes the separate tier labels to artifacts/labels.csv.
  • It writes the invalid two-row preview to artifacts/corrupt_preview.csv.
  • It prints the observed raw value plus both computed array shapes.
  • Save scripts/run_foundation.py.
  • Confirm that the file ends with the call to main().

Runner structure looks incomplete?

Check that every statement after def main() is indented by four spaces. Keep the final module guard outside the function.

Help me check the indentation in scripts/run_foundation.py.

✔️ Awesome, I've got everything!

The initial runner is ready. Save scripts/run_foundation.py before running it.

ⓧ I'd like to double check the full code

"""Run the complete foundation and write reproducible artifacts."""
from __future__ import annotations
from pathlib import Path
import pandas as pd
from core.icc import compute_iccs
from core.io import load_config
from core.synthetic import generate_synthetic

ROOT = Path(__file__).resolve().parents[1]

def main() -> None:
    config = load_config(ROOT / "config/defaults.yaml")
    output = ROOT / "artifacts"; output.mkdir(exist_ok=True)
    lines: list[str] = []
    def emit(message: str) -> None:
        print(message); lines.append(message)

    synthetic = config["synthetic"]
    params, labels = generate_synthetic(
        synthetic["n_items"], synthetic["seed"], synthetic["tiers"],
    )
    params_path, labels_path = output / "params.csv", output / "labels.csv"
    params.to_csv(params_path, index=False)
    pd.DataFrame({"item_id": params["item_id"], "tier": labels}).to_csv(
        labels_path, index=False,
    )

    corrupt = params.iloc[:2].copy()
    corrupt.loc[corrupt.index[0], "a"] = -1.0
    corrupt_path = output / "corrupt_preview.csv"
    corrupt.to_csv(corrupt_path, index=False)
    raw_corrupt = pd.read_csv(corrupt_path)
    emit(f"raw parser accepted corrupt preview a={raw_corrupt.loc[0, 'a']}")

    curves, grid = compute_iccs(params, **config["theta"])
    emit(f"icc shape: {curves.shape}, grid shape: {grid.shape}")

if __name__ == "__main__":
    main()

How to use this reference

Compare this complete runner with your saved version. Its module docstring identifies the workflow as the reproducible artifact entry point.

Before you run the workflow, do you think raw parsing will reject the nonpositive a value or continue with it?

  • Run the complete foundation workflow from your project folder with this command:
python -m scripts.run_foundation

What did the run prove?

You will see raw parser accepted corrupt preview a=-1.0. That result proves raw parsing accepts the deliberately invalid value.

You will also see the ICC shape (240, 201) plus the grid shape (201,). The first dimension represents items. The second dimension represents theta points.

You have a working exact-curve baseline. The parameters remain isolated from their tier labels while the runner exposes the validation gap.

Runner did not produce the expected files?

Confirm that the active terminal is inside the project folder containing config/, core/, and scripts/.

Check the traceback for the first filename it mentions. Compare that file with the corresponding full-code reference above.

Help me diagnose my foundation runner output.

Your project now generates reproducible 3PL items plus exact curves. Next, you will replace the exposed parsing gap with aggregate validation that reports every invalid cell.

Validate Every CSV Value

Your exact ICC baseline revealed a dangerous gap. CSV files can pass pandas parsing while still containing an invalid discrimination value such as a=-1.0.

Aggregate validation inspects the complete parameter table before statistical processing begins. You will collect every bad cell with its physical CSV row so one validation run reveals the full repair list.

In this step, get ready to:
  • Collect every parameter error as a sorted row-column-message tuple.
  • Load cleaned parameter data only after the complete CSV passes validation.
  • Prove every validation rule with focused automated tests.
Build aggregate validation

A validation result needs to preserve all failures from one pass. Its is_valid property gives later code one clear gate without discarding the detailed errors.

  • Create core/validation.py in the existing core folder by copying this first section:
"""Aggregate validation for 3PL parameter tables."""
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
import pandas as pd

ErrorTuple = tuple[int, str, str]
REQUIRED_COLUMNS = ("item_id", "a", "b", "c")

@dataclass(frozen=True)
class ValidationResult:
    """Every validation error found in a table."""
    errors: tuple[ErrorTuple, ...]

    @property
    def is_valid(self) -> bool:
        return not self.errors

What Does This Code Do?

  • ErrorTuple fixes the error shape as a physical row number, column name, and message.
  • REQUIRED_COLUMNS defines the fields every parameter table must contain.
  • ValidationResult is frozen so validation evidence cannot change after creation.
  • is_valid becomes true only when the error tuple is empty.
  • Save core/validation.py.
  • Confirm core/validation.py now appears beside core/icc.py and core/synthetic.py in your file tree.

Seeing an Import Warning?

Confirm your editor is using the active Python 3.12 environment. The existing environment already contains NumPy and pandas.

Check that the filename is exactly validation.py inside core. Help me diagnose imports in core/validation.py

The validator first checks table structure. It then checks every ID while preserving the original DataFrame positions as physical CSV row numbers.

  • Add validate_params below ValidationResult by copying this code:
def validate_params(frame: pd.DataFrame) -> ValidationResult:
    """Validate all rows and return every error without short-circuiting."""
    errors: list[ErrorTuple] = []
    for column in REQUIRED_COLUMNS:
        if column not in frame.columns:
            errors.append((1, column, "required column is missing"))

    if "item_id" in frame.columns:
        ids = frame["item_id"].fillna("").astype(str).str.strip()
        empty = ids.eq("")
        duplicate = ids.duplicated(keep=False) & ~empty
        for position in range(len(frame)):
            row = position + 2
            if bool(empty.iloc[position]):
                errors.append((row, "item_id", "must be non-empty"))
            elif bool(duplicate.iloc[position]):
                errors.append((row, "item_id", "must be unique"))
    return ValidationResult(tuple(sorted(errors)))

How Are Rows Tracked?

  • Missing columns use row 1 because that is the CSV header row.
  • position + 2 converts a zero-based DataFrame position into a physical CSV data row.
  • duplicated(keep=False) reports every occurrence of a repeated ID.
  • sorted(errors) makes the final error order deterministic.
  • Save core/validation.py.
  • Confirm the function ends with a ValidationResult containing sorted errors.

Are CSV Row Numbers Off by One?

Keep row = position + 2 inside the row loop. The extra two positions account for zero-based indexing plus the header row.

Make sure duplicate excludes empty IDs. Help me check physical CSV row numbering

Numeric fields need two separate checks. Blank cells receive a missing-value error while text or nonfinite values receive a numeric error.

  • Find the final return ValidationResult(tuple(sorted(errors))) line in validate_params.
  • Insert this numeric validation loop immediately above that return line:
    for column in ("a", "b", "c"):
        if column not in frame.columns:
            continue
        raw = frame[column]
        text = raw.fillna("").astype(str).str.strip()
        missing = text.eq("")
        numeric = pd.to_numeric(raw.where(~missing), errors="coerce")
        for position in range(len(frame)):
            row = position + 2
            if bool(missing.iloc[position]):
                errors.append((row, column, "must not be missing"))
                continue
            value = numeric.iloc[position]
            if pd.isna(value) or not np.isfinite(float(value)):
                errors.append((row, column, "must be numeric and finite"))
                continue
            number = float(value)
            if column == "a" and number <= 0:
                errors.append((row, column, "must be greater than 0"))
            if column == "c" and not 0 <= number < 1:
                errors.append((row, column, "must satisfy 0 <= c < 1"))

How Does Numeric Validation Work?

  • pd.to_numeric converts valid text into numbers. Invalid text becomes a missing numeric value for inspection.
  • np.isfinite rejects infinite values that numeric conversion can otherwise accept.
  • a must remain greater than zero for a valid 3PL discrimination parameter.
  • c must stay inside the half-open interval from zero to one.
  • Save core/validation.py.
  • Confirm the numeric loop appears before the final sorted-error return.

Seeing an Indentation Error?

Place the numeric loop at the same indentation level as the earlier required-column loop. Keep its final return at the function indentation level.

Check that each continue remains inside the inner row loop. Help me fix validate_params indentation

✔️ Awesome, I've got everything!

Great. Your validator now gathers structural, ID, and numeric errors in one deterministic result.

ⓧ I'd like to double check the full code

"""Aggregate validation for 3PL parameter tables."""
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
import pandas as pd

ErrorTuple = tuple[int, str, str]
REQUIRED_COLUMNS = ("item_id", "a", "b", "c")

@dataclass(frozen=True)
class ValidationResult:
    """Every validation error found in a table."""
    errors: tuple[ErrorTuple, ...]

    @property
    def is_valid(self) -> bool:
        return not self.errors

def validate_params(frame: pd.DataFrame) -> ValidationResult:
    """Validate all rows and return every error without short-circuiting."""
    errors: list[ErrorTuple] = []
    for column in REQUIRED_COLUMNS:
        if column not in frame.columns:
            errors.append((1, column, "required column is missing"))

    if "item_id" in frame.columns:
        ids = frame["item_id"].fillna("").astype(str).str.strip()
        empty = ids.eq("")
        duplicate = ids.duplicated(keep=False) & ~empty
        for position in range(len(frame)):
            row = position + 2
            if bool(empty.iloc[position]):
                errors.append((row, "item_id", "must be non-empty"))
            elif bool(duplicate.iloc[position]):
                errors.append((row, "item_id", "must be unique"))

    for column in ("a", "b", "c"):
        if column not in frame.columns:
            continue
        raw = frame[column]
        text = raw.fillna("").astype(str).str.strip()
        missing = text.eq("")
        numeric = pd.to_numeric(raw.where(~missing), errors="coerce")
        for position in range(len(frame)):
            row = position + 2
            if bool(missing.iloc[position]):
                errors.append((row, column, "must not be missing"))
                continue
            value = numeric.iloc[position]
            if pd.isna(value) or not np.isfinite(float(value)):
                errors.append((row, column, "must be numeric and finite"))
                continue
            number = float(value)
            if column == "a" and number <= 0:
                errors.append((row, column, "must be greater than 0"))
            if column == "c" and not 0 <= number < 1:
                errors.append((row, column, "must satisfy 0 <= c < 1"))
    return ValidationResult(tuple(sorted(errors)))

What Should Match?

Compare the complete file with yours from top to bottom. Pay close attention to loop indentation and the final sorted-error return.

Gate CSV loading on validation

The loader is the boundary between raw CSV text and statistical code. It preserves extra columns for downstream use while withholding the complete DataFrame whenever one validation error exists.

  • Create core/io.py in the existing core folder by copying this configuration and result section:
"""Configuration and validated CSV loading."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import pandas as pd
import yaml
from core.validation import ValidationResult, validate_params

@dataclass(frozen=True)
class LoadResult:
    data: pd.DataFrame | None
    validation: ValidationResult

def load_config(path: str | Path) -> dict[str, Any]:
    with Path(path).open("r", encoding="utf-8") as handle:
        config = yaml.safe_load(handle)
    if not isinstance(config, dict):
        raise ValueError("Configuration root must be a mapping.")
    return config

Why Keep One I/O Module?

LoadResult keeps cleaned data beside the evidence that allowed or blocked it. load_config remains available to the existing foundation runner.

  • Save core/io.py.
  • Confirm LoadResult and load_config appear in the file.

Is Configuration Loading Broken?

Keep the existing load_config function exactly as shown. The runner still depends on it for config/defaults.yaml.

Check that yaml.safe_load remains inside the file context. Help me restore core/io.py

  • Add load_params_csv below load_config by copying this code:
def load_params_csv(path: str | Path) -> LoadResult:
    """Return normalized data only when every rule passes."""
    frame = pd.read_csv(path, dtype=str, keep_default_na=False)
    validation = validate_params(frame)
    if not validation.is_valid:
        return LoadResult(None, validation)
    clean = frame.copy()
    clean["item_id"] = clean["item_id"].astype(str).str.strip()
    for column in ("a", "b", "c"):
        clean[column] = pd.to_numeric(clean[column], errors="raise")
    return LoadResult(clean, validation)

How Does the Loader Protect the Pipeline?

  • dtype=str preserves the original cell text for validation.
  • keep_default_na=False keeps blank CSV cells available for the explicit missing-value rule.
  • LoadResult(None, validation) prevents partial or invalid data from reaching ICC computation.
  • frame.copy() preserves every extra column while normalizing the required fields.
  • Save core/io.py.
  • Confirm the invalid path returns data=None before any numeric conversion occurs.

Are Extra Columns Disappearing?

Create clean with frame.copy(). Selecting only the required columns would discard permitted metadata.

Convert only a, b, and c after validation succeeds. Help me preserve extra CSV columns

✔️ Awesome, I've got everything!

Your loader now exposes cleaned parameter data only after every rule passes.

ⓧ I'd like to double check the full code

"""Configuration and validated CSV loading."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import pandas as pd
import yaml
from core.validation import ValidationResult, validate_params

@dataclass(frozen=True)
class LoadResult:
    data: pd.DataFrame | None
    validation: ValidationResult

def load_config(path: str | Path) -> dict[str, Any]:
    with Path(path).open("r", encoding="utf-8") as handle:
        config = yaml.safe_load(handle)
    if not isinstance(config, dict):
        raise ValueError("Configuration root must be a mapping.")
    return config

def load_params_csv(path: str | Path) -> LoadResult:
    """Return normalized data only when every rule passes."""
    frame = pd.read_csv(path, dtype=str, keep_default_na=False)
    validation = validate_params(frame)
    if not validation.is_valid:
        return LoadResult(None, validation)
    clean = frame.copy()
    clean["item_id"] = clean["item_id"].astype(str).str.strip()
    for column in ("a", "b", "c"):
        clean[column] = pd.to_numeric(clean[column], errors="raise")
    return LoadResult(clean, validation)

What Should Match?

The invalid branch must return before normalization. The successful branch must copy the complete frame so extra columns survive.

The existing runner already imports load_config from this module. A baseline run now confirms that the replacement preserved the earlier generation and ICC workflow.

  • Run the existing foundation workflow to check the updated I/O module:
python -m scripts.run_foundation

What Does This Check Prove?

The command imports the updated core/io.py through the existing configuration path. A complete run proves the new module did not break deterministic generation or exact ICC computation.

You should still see the corrupt preview accepted by raw parsing. You should also see the ICC matrix shape (240, 201).

Did the Baseline Run Stop?

Check the first traceback path for core/io.py. A missing import or indentation error there prevents the runner from reaching generation.

Keep the Python 3.12 virtual environment active. Help me debug the updated I/O module

The runner now needs to use the validated loader before computing curves. This creates one enforced entry point for parameter CSV files.

  • In scripts/run_foundation.py, find this import line:
from core.io import load_config

What Are You Locating?

This is the runner's current configuration import. The next edit adds the validated parameter loader from the same module.

  • Replace that import with this updated line:
from core.io import load_config, load_params_csv

What Changed?

The runner can now load artifacts/params.csv through the validation gate before ICC computation.

  • Find the raw corrupt-preview message in scripts/run_foundation.py.
  • Insert this validation gate immediately below that message:
    loaded = load_params_csv(params_path)
    emit(
        f"validation: valid={loaded.validation.is_valid}, "
        f"errors={len(loaded.validation.errors)}",
    )
    if loaded.data is None:
        for error in loaded.validation.errors:
            emit(f"validation error: {error}")
        raise SystemExit("Validation failed before ICC computation.")

How Does the Gate Work?

  • load_params_csv validates the generated parameter file as raw CSV input.
  • emit records whether validation passed plus the complete error count.
  • loaded.data is None blocks ICC computation when any error exists.
  • validation error output preserves every row-column-message tuple for repair.
  • Replace the existing ICC call with this validated-data version:
    curves, grid = compute_iccs(loaded.data, **config["theta"])

Why Use Loaded Data?

ICC computation now receives normalized parameters from the successful loader branch. Invalid raw data has no path into the curve matrix.

  • Save scripts/run_foundation.py.

Before you run the workflow, do you expect the valid generated parameter file to reach ICC computation?

  • Run the updated foundation workflow to test the validation gate:
python -m scripts.run_foundation

What Should the Runner Prove?

The runner validates artifacts/params.csv before using it. The corrupt preview remains a separate demonstration of raw parsing behavior.

You should see validation: valid=True, errors=0 before the ICC shape. The later icc shape: (240, 201), grid shape: (201,) message proves validated rows reached the exact curve calculation.

Did Validation Block the Generated File?

Read every emitted error tuple before changing code. Its row number points to the physical row in artifacts/params.csv.

Confirm the ICC call uses loaded.data after the invalid-data branch. Help me diagnose the runner validation gate

✔️ Awesome, I've got everything!

That is the boundary working. Only a complete valid parameter table now reaches ICC computation.

ⓧ I'd like to double check the full code

"""Run the complete foundation and write reproducible artifacts."""
from __future__ import annotations
from pathlib import Path
import pandas as pd
from core.icc import compute_iccs
from core.io import load_config, load_params_csv
from core.synthetic import generate_synthetic

ROOT = Path(__file__).resolve().parents[1]

def main() -> None:
    config = load_config(ROOT / "config/defaults.yaml")
    output = ROOT / "artifacts"; output.mkdir(exist_ok=True)
    lines: list[str] = []
    def emit(message: str) -> None:
        print(message); lines.append(message)

    synthetic = config["synthetic"]
    params, labels = generate_synthetic(
        synthetic["n_items"], synthetic["seed"], synthetic["tiers"],
    )
    params_path, labels_path = output / "params.csv", output / "labels.csv"
    params.to_csv(params_path, index=False)
    pd.DataFrame({"item_id": params["item_id"], "tier": labels}).to_csv(
        labels_path, index=False,
    )

    corrupt = params.iloc[:2].copy()
    corrupt.loc[corrupt.index[0], "a"] = -1.0
    corrupt_path = output / "corrupt_preview.csv"
    corrupt.to_csv(corrupt_path, index=False)
    raw_corrupt = pd.read_csv(corrupt_path)
    emit(f"raw parser accepted corrupt preview a={raw_corrupt.loc[0, 'a']}")

    loaded = load_params_csv(params_path)
    emit(
        f"validation: valid={loaded.validation.is_valid}, "
        f"errors={len(loaded.validation.errors)}",
    )
    if loaded.data is None:
        for error in loaded.validation.errors:
            emit(f"validation error: {error}")
        raise SystemExit("Validation failed before ICC computation.")

    curves, grid = compute_iccs(loaded.data, **config["theta"])
    emit(f"icc shape: {curves.shape}, grid shape: {grid.shape}")

if __name__ == "__main__":
    main()

What Should Match?

The loader gate must sit after CSV generation. The ICC call must use loaded.data instead of the unvalidated DataFrame.

Prove every validation rule

pytest turns each validation promise into an executable contract. The tests cover isolated failures first so a broken rule points to one precise behavior.

  • Create tests/test_foundation.py in the existing tests folder by copying this first test group:
"""Contract tests for the statistical foundation."""
from __future__ import annotations
from pathlib import Path
import pandas as pd
from core.io import load_params_csv
from core.validation import validate_params

ROOT = Path(__file__).resolve().parents[1]

def valid_frame() -> pd.DataFrame:
    return pd.DataFrame({
        "item_id": ["A", "B"],
        "a": ["1.0", "1.2"],
        "b": ["0.0", "0.5"],
        "c": ["0.1", "0.2"],
    })

def test_validation_missing_required_column():
    assert (1, "b", "required column is missing") in validate_params(
        valid_frame().drop(columns="b"),
    ).errors

def test_validation_missing_numeric_value():
    frame = valid_frame(); frame.loc[0, "a"] = ""
    assert (2, "a", "must not be missing") in validate_params(frame).errors

def test_validation_nonnumeric_value():
    frame = valid_frame(); frame.loc[0, "b"] = "wrong"
    assert (2, "b", "must be numeric and finite") in validate_params(frame).errors

What Does This Test Group Cover?

  • valid_frame supplies a small valid baseline for one-change tests.
  • The first test assigns missing-column errors to physical CSV row 1.
  • The next tests separate blank numeric cells from nonnumeric text.
  • Save tests/test_foundation.py.
  • Run the validation-focused tests:
pytest -q -s tests/test_foundation.py -k validation

What Should Pass?

The missing-column, missing-value, and nonnumeric-value tests should pass. This confirms the first three contracts independently.

Did One Error Tuple Differ?

Compare the complete tuple in the failed assertion. The row number, column, and message must all match.

Use row 1 only for required-column failures. Help me diagnose a validation test failure

  • Add the range and ID tests below the existing tests:
def test_validation_nonpositive_a():
    frame = valid_frame(); frame.loc[0, "a"] = "0"
    assert (2, "a", "must be greater than 0") in validate_params(frame).errors

def test_validation_c_range():
    frame = valid_frame(); frame.loc[0, "c"] = "1"
    assert (2, "c", "must satisfy 0 <= c < 1") in validate_params(frame).errors

def test_validation_empty_item_id():
    frame = valid_frame(); frame.loc[0, "item_id"] = "  "
    assert (2, "item_id", "must be non-empty") in validate_params(frame).errors

def test_validation_duplicate_item_id():
    frame = valid_frame(); frame.loc[1, "item_id"] = "A"
    errors = validate_params(frame).errors
    assert (2, "item_id", "must be unique") in errors
    assert (3, "item_id", "must be unique") in errors

What Does This Test Group Cover?

  • The range tests enforce positive a plus the allowed interval for c.
  • Whitespace-only IDs become empty after normalization.
  • Duplicate detection reports both physical CSV rows that share the same ID.
  • Save tests/test_foundation.py.
  • Run the expanded validation-focused tests:
pytest -q -s tests/test_foundation.py -k validation

What Should Pass Now?

The range and ID cases should join the earlier passing tests. Both duplicate rows must appear in the collected errors.

Is Only One Duplicate Reported?

Check that duplicate detection uses keep=False. Keeping the first or last occurrence would hide one bad cell.

Exclude empty IDs from duplicate detection so they receive the more precise empty-ID message. Help me fix duplicate ID validation

The final tests prove aggregate behavior and the successful loader path. A clean file must preserve an unrelated column after normalization.

  • Add the multi-error and clean-loader tests below the existing tests:
def test_validation_reports_all_errors():
    frame = pd.DataFrame({
        "item_id": ["", "X", "X"],
        "a": ["0", "bad", ""],
        "b": ["bad", "", "0"],
        "c": ["1", "-0.1", "bad"],
    })
    assert len(validate_params(frame).errors) >= 9

def test_clean_loader_passes_and_keeps_extra_column(tmp_path):
    path = tmp_path / "clean.csv"
    valid_frame().assign(extra=["keep", "me"]).to_csv(path, index=False)
    result = load_params_csv(path)
    assert result.validation.is_valid and result.data is not None
    assert "extra" in result.data.columns

What Do the Final Tests Prove?

  • test_validation_reports_all_errors proves validation continues after the first failure.
  • tmp_path provides an isolated location for the clean CSV fixture.
  • The extra-column assertion proves cleaning preserves fields outside the required parameter set.
  • Save tests/test_foundation.py.
  • Run the complete current test file:
pytest -q -s

What Should the Full Run Show?

Every current test should pass. The successful run includes the clean loader case because the full command does not filter test names.

Did the Clean Loader Test Fail?

Confirm the successful branch begins with clean = frame.copy(). The copied frame is what preserves extra.

Make sure numeric conversion happens after the validation gate. Help me debug the clean loader test

✔️ Awesome, I've got everything!

Your test suite now covers every failure category plus the successful loader path.

ⓧ I'd like to double check the full code

"""Contract tests for the statistical foundation."""
from __future__ import annotations
from pathlib import Path
import pandas as pd
from core.io import load_params_csv
from core.validation import validate_params

ROOT = Path(__file__).resolve().parents[1]

def valid_frame() -> pd.DataFrame:
    return pd.DataFrame({
        "item_id": ["A", "B"],
        "a": ["1.0", "1.2"],
        "b": ["0.0", "0.5"],
        "c": ["0.1", "0.2"],
    })

def test_validation_missing_required_column():
    assert (1, "b", "required column is missing") in validate_params(
        valid_frame().drop(columns="b"),
    ).errors

def test_validation_missing_numeric_value():
    frame = valid_frame(); frame.loc[0, "a"] = ""
    assert (2, "a", "must not be missing") in validate_params(frame).errors

def test_validation_nonnumeric_value():
    frame = valid_frame(); frame.loc[0, "b"] = "wrong"
    assert (2, "b", "must be numeric and finite") in validate_params(frame).errors

def test_validation_nonpositive_a():
    frame = valid_frame(); frame.loc[0, "a"] = "0"
    assert (2, "a", "must be greater than 0") in validate_params(frame).errors

def test_validation_c_range():
    frame = valid_frame(); frame.loc[0, "c"] = "1"
    assert (2, "c", "must satisfy 0 <= c < 1") in validate_params(frame).errors

def test_validation_empty_item_id():
    frame = valid_frame(); frame.loc[0, "item_id"] = "  "
    assert (2, "item_id", "must be non-empty") in validate_params(frame).errors

def test_validation_duplicate_item_id():
    frame = valid_frame(); frame.loc[1, "item_id"] = "A"
    errors = validate_params(frame).errors
    assert (2, "item_id", "must be unique") in errors
    assert (3, "item_id", "must be unique") in errors

def test_validation_reports_all_errors():
    frame = pd.DataFrame({
        "item_id": ["", "X", "X"],
        "a": ["0", "bad", ""],
        "b": ["bad", "", "0"],
        "c": ["1", "-0.1", "bad"],
    })
    assert len(validate_params(frame).errors) >= 9

def test_clean_loader_passes_and_keeps_extra_column(tmp_path):
    path = tmp_path / "clean.csv"
    valid_frame().assign(extra=["keep", "me"]).to_csv(path, index=False)
    result = load_params_csv(path)
    assert result.validation.is_valid and result.data is not None
    assert "extra" in result.data.columns

What Should Match?

Compare each expected tuple character for character. The clean-loader test should remain last because it verifies the complete successful path.

Before the final focused run, which validation rule would reveal a duplicated ID in both affected CSV rows?

  • Run the required validation-focused check:
pytest -q -s tests/test_foundation.py -k validation

What Does the Final Check Confirm?

The focused run executes every test whose name contains validation. It covers missing columns, missing values, nonnumeric values, invalid ranges, empty IDs, duplicate IDs, and aggregate multi-error reporting.

The clean-loader test may appear as deselected because its name does not contain the filter keyword. Its successful path was proven by the complete test run immediately above.

Is the Focused Run Failing?

Start with the first failed assertion because later failures may share the same loop or indentation mistake. Keep the configuration unchanged while you repair the implementation.

Compare the actual tuple list with the expected row-column-message values. Help me debug the focused validation suite

That closes the parsing gap. Next, you will turn each validated 201-point ICC row into shared B-spline coordinates with an explicitly selected smoothing penalty.

Select the B-Spline Smoother

Your validation layer now keeps malformed parameters out of the statistical pipeline. Every accepted item can reach the exact ICC calculation with a complete set of valid values.

Each exact ICC still contains 201 values. This step converts those values into a shared cubic B-spline representation. The shared representation creates functional coordinates for FPCA while exposing the approximation error for every item.

In this step, get ready to:
  • Define the smoothing result plus its grid and error helpers.
  • Fit every configured penalty candidate with pooled GCV selection.
  • Verify the selected lambda, edge flag, and maximum smoothing RMSE.
Define the smoothing contract

A common basis gives every curve the same functional coordinate system. scikit-fda supplies the basis, regularization, and smoothing primitives used here.

Why use cubic B-splines?

An order-4 B-spline is cubic because its degree is one lower than its order. Forty basis functions preserve local curve shape while reducing each ICC from 201 grid values to a common basis representation.

The second-derivative penalty discourages excessive bending. The selected penalty balances fidelity against smoothness.

  • Create `core/smoothing.py` inside the existing `core` folder in your project editor.
  • Add the imports and smoothing result container by pasting this code into `core/smoothing.py`:
"""Penalized B-spline smoothing selected by pooled GCV."""
from __future__ import annotations
from dataclasses import dataclass
from time import perf_counter
from typing import Any
import numpy as np
from skfda import FDataGrid
from skfda.misc.operators import LinearDifferentialOperator
from skfda.misc.regularization import L2Regularization
from skfda.preprocessing.smoothing import BasisSmoother
from skfda.representation.basis import BSplineBasis, FDataBasis

@dataclass(frozen=True)
class SmoothingResult:
    fdata: FDataBasis
    smoother: BasisSmoother
    selected_lambda: float
    edge_flag: bool
    candidates: np.ndarray
    gcv_values: np.ndarray
    smoothed_values: np.ndarray
    rmse: np.ndarray

What does this code do?

  • `FDataGrid` holds the exact ICC rows on their shared theta grid.
  • `BSplineBasis` supplies the common functional representation.
  • `BasisSmoother` stores the fitted mapping that later projects curves into basis coefficients.
  • `SmoothingResult` keeps the fitted objects and every diagnostic needed by tests or reports.
  • Save `core/smoothing.py`.
  • Check that the existing test suite still passes by running this command:
pytest

What does this check prove?

Pytest reruns the existing contracts after the new module is added. A passing suite confirms that the new file has not disturbed parameter validation or exact ICC behavior.

You should see the existing tests pass. The smoothing result can now carry the fitted model and its diagnostics.

Seeing an import failure?

Confirm that your Python 3.12 virtual environment is still active. Check that `core/smoothing.py` sits beside `core/icc.py` and `core/validation.py`.

Compare each scikit-fda import with the code block above. A changed import path prevents the module from loading.

Help me diagnose the smoothing module import failure.

The helper functions keep candidate creation, grid evaluation, and error measurement separate from model fitting. Each helper has one numerical responsibility.

  • Append the smoothing helper functions to `core/smoothing.py`:
def lambda_candidates_from_config(config: dict[str, Any]) -> np.ndarray:
    spec = config["lambda_candidates"]["logspace"]
    return np.logspace(spec["start"], spec["stop"], num=spec["num"])

def evaluate_on_grid(fdata: FDataBasis, grid: np.ndarray) -> np.ndarray:
    return np.asarray(fdata(grid), dtype=float)[..., 0]

def per_item_rmse(exact: np.ndarray, fitted: np.ndarray) -> np.ndarray:
    return np.sqrt(np.mean((exact - fitted) ** 2, axis=1))

What do these helpers calculate?

  • `lambda_candidates_from_config` creates the seven configured powers-of-ten candidates.
  • `evaluate_on_grid` converts basis functions back into sample-first grid values.
  • `per_item_rmse` measures the approximation error for each ICC row.
  • Save `core/smoothing.py`.
  • Confirm that the helper additions preserve the existing contracts by running:
pytest

What should pass?

The existing validation and ICC tests should remain green. This confirms that the helper definitions load successfully in the pinned environment.

Your smoothing module now has a stable output contract plus focused numerical helpers.

Fit every lambda candidate

The smoother uses forty order-4 basis functions across the stored theta domain. `LinearDifferentialOperator(2)` supplies the second-derivative operator used by the L2 penalty.

Generalized cross-validation, or GCV, compares candidates using pooled residual error and the hat-matrix trace. The minimum value selects the stored smoother.

  • Append the `fit_smoothing` signature and shared setup to `core/smoothing.py`:
def fit_smoothing(
    curves: np.ndarray,
    grid: np.ndarray,
    spline_order: int,
    num_basis: int,
    penalty_derivative_order: int,
    lambda_candidates: np.ndarray,
    max_seconds: float = 1200.0,
) -> SmoothingResult:
    """Fit every candidate and retain the minimum-GCV model."""
    started = perf_counter()
    fd_grid = FDataGrid(data_matrix=curves, grid_points=grid)
    basis = BSplineBasis(
        domain_range=(float(grid[0]), float(grid[-1])),
        n_basis=num_basis,
        order=spline_order,
    )
    regularization = L2Regularization(
        LinearDifferentialOperator(penalty_derivative_order),
    )
    smoothers: list[BasisSmoother] = []
    basis_rows: list[FDataBasis] = []
    evaluated_rows: list[np.ndarray] = []
    gcv_values: list[float] = []
    n_rows, n_points = curves.shape

How is the shared model prepared?

  • `FDataGrid` packages all exact curves against the same grid.
  • `BSplineBasis` creates the configured basis across the grid endpoints.
  • `L2Regularization` applies the configured differential operator.
  • The result lists retain each fitted candidate until GCV identifies the winner.
  • Save `core/smoothing.py`.
  • Append the candidate-fitting loop below `n_rows, n_points = curves.shape`:
    for candidate in np.asarray(lambda_candidates, dtype=float):
        smoother = BasisSmoother(
            basis,
            smoothing_parameter=float(candidate),
            regularization=regularization,
            method="svd",
            return_basis=True,
        )
        fitted = smoother.fit_transform(fd_grid)
        evaluated = evaluate_on_grid(fitted, grid)
        hat = np.asarray(smoother.hat_matrix(), dtype=float)
        effective_df = float(hat.diagonal().sum())
        rss = float(np.sum((curves - evaluated) ** 2)) / (n_rows * n_points)
        gcv_values.append(rss / (1.0 - effective_df / n_points) ** 2)
        smoothers.append(smoother)
        basis_rows.append(fitted)
        evaluated_rows.append(evaluated)
        if perf_counter() - started > max_seconds:
            raise TimeoutError("Smoothing/GCV exceeded the 20-minute stop rule.")

How is each candidate scored?

  • Each lambda receives its own fitted `BasisSmoother` with the same basis and regularization operator.
  • The residual sum of squares measures the pooled grid approximation error.
  • The hat-matrix trace estimates the effective degrees of freedom used by the fit.
  • The elapsed-time check enforces the configured 1200-second guard.
  • Save `core/smoothing.py`.
  • Check that the incomplete return path remains syntactically valid by running:
pytest

What does this intermediate check prove?

The test collection imports project modules before running the existing contracts. A passing run confirms that the candidate loop is valid Python before the selection logic is added.

You should see the existing tests pass. The candidate loop is now ready for its minimum-GCV selection.

  • Append the selection and result construction to the end of `fit_smoothing`:
    gcv = np.asarray(gcv_values)
    selected = int(np.argmin(gcv))
    values = evaluated_rows[selected]
    return SmoothingResult(
        basis_rows[selected],
        smoothers[selected],
        float(lambda_candidates[selected]),
        selected in (0, len(lambda_candidates) - 1),
        np.asarray(lambda_candidates, dtype=float),
        gcv,
        values,
        per_item_rmse(curves, values),
    )

How is the winner retained?

  • `np.argmin` identifies the candidate with the lowest pooled GCV.
  • `edge_flag` records whether the winner is the first or last configured candidate.
  • The returned RMSE array preserves one approximation error per item.
  • The fitted smoother remains available for frozen projection in the next step.
  • Save `core/smoothing.py`.

✔️ Awesome, I've got everything!

Great. Double-check that `core/smoothing.py` is saved before wiring it into the runner.

ⓧ I'd like to double check the full code

"""Penalized B-spline smoothing selected by pooled GCV."""
from __future__ import annotations
from dataclasses import dataclass
from time import perf_counter
from typing import Any
import numpy as np
from skfda import FDataGrid
from skfda.misc.operators import LinearDifferentialOperator
from skfda.misc.regularization import L2Regularization
from skfda.preprocessing.smoothing import BasisSmoother
from skfda.representation.basis import BSplineBasis, FDataBasis

@dataclass(frozen=True)
class SmoothingResult:
    fdata: FDataBasis
    smoother: BasisSmoother
    selected_lambda: float
    edge_flag: bool
    candidates: np.ndarray
    gcv_values: np.ndarray
    smoothed_values: np.ndarray
    rmse: np.ndarray

def lambda_candidates_from_config(config: dict[str, Any]) -> np.ndarray:
    spec = config["lambda_candidates"]["logspace"]
    return np.logspace(spec["start"], spec["stop"], num=spec["num"])

def evaluate_on_grid(fdata: FDataBasis, grid: np.ndarray) -> np.ndarray:
    return np.asarray(fdata(grid), dtype=float)[..., 0]

def per_item_rmse(exact: np.ndarray, fitted: np.ndarray) -> np.ndarray:
    return np.sqrt(np.mean((exact - fitted) ** 2, axis=1))

def fit_smoothing(
    curves: np.ndarray,
    grid: np.ndarray,
    spline_order: int,
    num_basis: int,
    penalty_derivative_order: int,
    lambda_candidates: np.ndarray,
    max_seconds: float = 1200.0,
) -> SmoothingResult:
    """Fit every candidate and retain the minimum-GCV model."""
    started = perf_counter()
    fd_grid = FDataGrid(data_matrix=curves, grid_points=grid)
    basis = BSplineBasis(
        domain_range=(float(grid[0]), float(grid[-1])),
        n_basis=num_basis,
        order=spline_order,
    )
    regularization = L2Regularization(
        LinearDifferentialOperator(penalty_derivative_order),
    )
    smoothers: list[BasisSmoother] = []
    basis_rows: list[FDataBasis] = []
    evaluated_rows: list[np.ndarray] = []
    gcv_values: list[float] = []
    n_rows, n_points = curves.shape
    for candidate in np.asarray(lambda_candidates, dtype=float):
        smoother = BasisSmoother(
            basis,
            smoothing_parameter=float(candidate),
            regularization=regularization,
            method="svd",
            return_basis=True,
        )
        fitted = smoother.fit_transform(fd_grid)
        evaluated = evaluate_on_grid(fitted, grid)
        hat = np.asarray(smoother.hat_matrix(), dtype=float)
        effective_df = float(hat.diagonal().sum())
        rss = float(np.sum((curves - evaluated) ** 2)) / (n_rows * n_points)
        gcv_values.append(rss / (1.0 - effective_df / n_points) ** 2)
        smoothers.append(smoother)
        basis_rows.append(fitted)
        evaluated_rows.append(evaluated)
        if perf_counter() - started > max_seconds:
            raise TimeoutError("Smoothing/GCV exceeded the 20-minute stop rule.")
    gcv = np.asarray(gcv_values)
    selected = int(np.argmin(gcv))
    values = evaluated_rows[selected]
    return SmoothingResult(
        basis_rows[selected],
        smoothers[selected],
        float(lambda_candidates[selected]),
        selected in (0, len(lambda_candidates) - 1),
        np.asarray(lambda_candidates, dtype=float),
        gcv,
        values,
        per_item_rmse(curves, values),
    )

The runner already validates `artifacts/params.csv` before computing exact ICCs. The next addition passes those accepted curves through the configured candidate search.

  • Add the smoothing import below the existing `core.io` import in `scripts/run_foundation.py`:
from core.smoothing import fit_smoothing, lambda_candidates_from_config

Why import both functions?

`lambda_candidates_from_config` converts the YAML logspace specification into concrete candidates. `fit_smoothing` evaluates those candidates against the validated exact curves.

  • Save `scripts/run_foundation.py`.
  • Confirm that the completed smoothing module imports through the runner by running:
python -m scripts.run_foundation

What should the runner show?

The runner still follows its existing path because the smoothing call has not been added yet. You should see valid parameter loading followed by the ICC and grid shapes.

The successful import proves that the complete smoothing module loads inside the real project entry point.

  • Append the smoothing run below the existing ICC shape output in `scripts/run_foundation.py`:
    smooth_cfg = config["smoothing"]
    smoothing = fit_smoothing(
        curves, grid,
        smooth_cfg["spline_order"], smooth_cfg["num_basis"],
        smooth_cfg["penalty_derivative_order"],
        lambda_candidates_from_config(smooth_cfg), smooth_cfg["max_seconds"],
    )
    emit(
        f"smoothing: lambda={smoothing.selected_lambda:.12g}, "
        f"edge_flag={smoothing.edge_flag}",
    )
    emit(
        "rmse min/median/max: "
        f"{np.min(smoothing.rmse):.12g} / "
        f"{np.median(smoothing.rmse):.12g} / "
        f"{np.max(smoothing.rmse):.12g}",
    )

What do these diagnostics reveal?

  • The selected lambda identifies the minimum-GCV candidate.
  • The edge flag shows whether the chosen value sits at either end of the search range.
  • The minimum, median, and maximum RMSE values summarize approximation quality across all 240 curves.
  • Save `scripts/run_foundation.py`.

This candidate search is the slowest calculation so far. The 1200-second guard stops the loop if it exceeds the configured limit.

Before you run the foundation, which candidate do you think pooled GCV will select?

  • Run the complete foundation with smoothing enabled:
python -m scripts.run_foundation

What should the completed run report?

You should see the ICC shape followed by the selected lambda and edge flag. The next line shows the observed minimum, median, and maximum RMSE values.

These are empirical results from your pinned environment. Preserve the observed values instead of replacing them with assumed results.

That is the candidate search working across all 240 curves. Your runner now exposes the chosen penalty and the approximation error distribution.

Did the smoothing run stop?

Confirm that `curves` has shape `(240, 201)` before the smoothing call. A different grid shape can prevent the shared basis fit from completing.

If the time guard fires, keep `config/defaults.yaml` unchanged. Record the timeout as the observed result.

Help me diagnose the smoothing run.

✔️ Awesome, I've got everything!

Your runner now selects the smoother and prints its observed diagnostics.

ⓧ I'd like to double check the full code

"""Run the complete foundation and write reproducible artifacts."""
from __future__ import annotations
from pathlib import Path
import numpy as np
import pandas as pd
from core.icc import compute_iccs
from core.io import load_config, load_params_csv
from core.smoothing import fit_smoothing, lambda_candidates_from_config
from core.synthetic import generate_synthetic

ROOT = Path(__file__).resolve().parents[1]

def main() -> None:
    config = load_config(ROOT / "config/defaults.yaml")
    output = ROOT / "artifacts"; output.mkdir(exist_ok=True)
    lines: list[str] = []
    def emit(message: str) -> None:
        print(message); lines.append(message)

    synthetic = config["synthetic"]
    params, labels = generate_synthetic(
        synthetic["n_items"], synthetic["seed"], synthetic["tiers"],
    )
    params_path, labels_path = output / "params.csv", output / "labels.csv"
    params.to_csv(params_path, index=False)
    pd.DataFrame({"item_id": params["item_id"], "tier": labels}).to_csv(
        labels_path, index=False,
    )

    corrupt = params.iloc[:2].copy()
    corrupt.loc[corrupt.index[0], "a"] = -1.0
    corrupt_path = output / "corrupt_preview.csv"
    corrupt.to_csv(corrupt_path, index=False)
    raw_corrupt = pd.read_csv(corrupt_path)
    emit(f"raw parser accepted corrupt preview a={raw_corrupt.loc[0, 'a']}")

    loaded = load_params_csv(params_path)
    emit(
        f"validation: valid={loaded.validation.is_valid}, "
        f"errors={len(loaded.validation.errors)}",
    )
    if loaded.data is None:
        for error in loaded.validation.errors:
            emit(f"validation error: {error}")
        raise SystemExit("Validation failed before ICC computation.")

    curves, grid = compute_iccs(loaded.data, **config["theta"])
    emit(f"icc shape: {curves.shape}, grid shape: {grid.shape}")
    smooth_cfg = config["smoothing"]
    smoothing = fit_smoothing(
        curves, grid,
        smooth_cfg["spline_order"], smooth_cfg["num_basis"],
        smooth_cfg["penalty_derivative_order"],
        lambda_candidates_from_config(smooth_cfg), smooth_cfg["max_seconds"],
    )
    emit(
        f"smoothing: lambda={smoothing.selected_lambda:.12g}, "
        f"edge_flag={smoothing.edge_flag}",
    )
    emit(
        "rmse min/median/max: "
        f"{np.min(smoothing.rmse):.12g} / "
        f"{np.median(smoothing.rmse):.12g} / "
        f"{np.max(smoothing.rmse):.12g}",
    )

if __name__ == "__main__":
    main()
Prove the smoothing contract

The runner makes the diagnostics visible. The contract test turns the selected-candidate rule and the RMSE threshold into repeatable checks.

  • Update the import section at the top of `tests/test_foundation.py` to include the smoothing dependencies shown below:
"""Contract tests for the statistical foundation."""
from __future__ import annotations
from pathlib import Path
import numpy as np
import pandas as pd
from core.icc import compute_iccs, three_pl_probability
from core.io import load_config, load_params_csv
from core.smoothing import fit_smoothing, lambda_candidates_from_config
from core.synthetic import generate_synthetic
from core.validation import validate_params

ROOT = Path(__file__).resolve().parents[1]

Why does the test need these imports?

The test recreates the deterministic parameters and exact ICCs before fitting the smoother. This keeps the contract tied to the same configuration used by the runner.

  • Save `tests/test_foundation.py`.
  • Add the exact ICC contract below the existing validation tests:
def test_icc_closed_form_and_shape():
    params = pd.DataFrame({
        "item_id": ["c0", "cpos"],
        "a": [1.0, 1.5], "b": [0.0, -0.2], "c": [0.0, 0.25],
    })
    curves, grid = compute_iccs(params, -3.0, 3.0, 201)
    theta = np.asarray([-1.0, 0.0, 1.0])
    expected = 0.25 + 0.75 / (1.0 + np.exp(-1.5 * (theta + 0.2)))
    np.testing.assert_allclose(
        three_pl_probability(1.5, -0.2, 0.25, theta),
        expected,
        atol=0.0,
        rtol=0.0,
    )
    assert curves.shape == (2, 201) and grid.shape == (201,)

What does the ICC contract protect?

The closed-form comparison checks the vectorized probability calculation against an independently written expression. The final assertion protects the 201-point shape required by the smoother.

  • Save `tests/test_foundation.py`.
  • Run the ICC-focused test before adding the slower smoothing contract:
pytest -q -s tests/test_foundation.py -k icc

What should this focused test confirm?

The ICC test should pass with a `(2, 201)` curve matrix and a `(201,)` grid. This protects the input shape expected by every smoothing candidate.

The exact-curve baseline is still intact. You can now test the approximation layer against it.

  • Append the smoothing contract to `tests/test_foundation.py`:
def test_smoothing_contract():
    config = load_config(ROOT / "config/defaults.yaml")
    synthetic = config["synthetic"]
    params, labels = generate_synthetic(
        synthetic["n_items"], synthetic["seed"], synthetic["tiers"],
    )
    curves, grid = compute_iccs(params, **config["theta"])
    smooth_cfg = config["smoothing"]
    smoothing = fit_smoothing(
        curves,
        grid,
        smooth_cfg["spline_order"],
        smooth_cfg["num_basis"],
        smooth_cfg["penalty_derivative_order"],
        lambda_candidates_from_config(smooth_cfg),
        smooth_cfg["max_seconds"],
    )
    assert smoothing.selected_lambda in smoothing.candidates
    selected = int(np.argmin(smoothing.gcv_values))
    assert smoothing.edge_flag == (selected in (0, len(smoothing.candidates) - 1))
    observed = float(np.max(smoothing.rmse))
    print(f"observed max smoothing RMSE={observed:.12g}")
    assert observed < 0.01

What does the smoothing contract prove?

  • The selected lambda must belong to the configured candidate array.
  • The edge flag must match the selected candidate's position.
  • The observed maximum per-item RMSE must remain below `0.01`.
  • The printed value preserves the empirical result from this run.
  • Save `tests/test_foundation.py`.

✔️ Awesome, I've got everything!

Your test file now protects aggregate validation, exact ICC behavior, and smoothing selection.

ⓧ I'd like to double check the full code

"""Contract tests for the statistical foundation."""
from __future__ import annotations
from pathlib import Path
import numpy as np
import pandas as pd
from core.icc import compute_iccs, three_pl_probability
from core.io import load_config, load_params_csv
from core.smoothing import fit_smoothing, lambda_candidates_from_config
from core.synthetic import generate_synthetic
from core.validation import validate_params

ROOT = Path(__file__).resolve().parents[1]

def valid_frame() -> pd.DataFrame:
    return pd.DataFrame({
        "item_id": ["A", "B"],
        "a": ["1.0", "1.2"],
        "b": ["0.0", "0.5"],
        "c": ["0.1", "0.2"],
    })

def test_validation_missing_required_column():
    assert (1, "b", "required column is missing") in validate_params(
        valid_frame().drop(columns="b"),
    ).errors

def test_validation_missing_numeric_value():
    frame = valid_frame(); frame.loc[0, "a"] = ""
    assert (2, "a", "must not be missing") in validate_params(frame).errors

def test_validation_nonnumeric_value():
    frame = valid_frame(); frame.loc[0, "b"] = "wrong"
    assert (2, "b", "must be numeric and finite") in validate_params(frame).errors

def test_validation_nonpositive_a():
    frame = valid_frame(); frame.loc[0, "a"] = "0"
    assert (2, "a", "must be greater than 0") in validate_params(frame).errors

def test_validation_c_range():
    frame = valid_frame(); frame.loc[0, "c"] = "1"
    assert (2, "c", "must satisfy 0 <= c < 1") in validate_params(frame).errors

def test_validation_empty_item_id():
    frame = valid_frame(); frame.loc[0, "item_id"] = "  "
    assert (2, "item_id", "must be non-empty") in validate_params(frame).errors

def test_validation_duplicate_item_id():
    frame = valid_frame(); frame.loc[1, "item_id"] = "A"
    errors = validate_params(frame).errors
    assert (2, "item_id", "must be unique") in errors
    assert (3, "item_id", "must be unique") in errors

def test_validation_reports_all_errors():
    frame = pd.DataFrame({
        "item_id": ["", "X", "X"],
        "a": ["0", "bad", ""],
        "b": ["bad", "", "0"],
        "c": ["1", "-0.1", "bad"],
    })
    assert len(validate_params(frame).errors) >= 9

def test_clean_loader_passes_and_keeps_extra_column(tmp_path):
    path = tmp_path / "clean.csv"
    valid_frame().assign(extra=["keep", "me"]).to_csv(path, index=False)
    result = load_params_csv(path)
    assert result.validation.is_valid and result.data is not None
    assert "extra" in result.data.columns

def test_icc_closed_form_and_shape():
    params = pd.DataFrame({
        "item_id": ["c0", "cpos"],
        "a": [1.0, 1.5], "b": [0.0, -0.2], "c": [0.0, 0.25],
    })
    curves, grid = compute_iccs(params, -3.0, 3.0, 201)
    theta = np.asarray([-1.0, 0.0, 1.0])
    expected = 0.25 + 0.75 / (1.0 + np.exp(-1.5 * (theta + 0.2)))
    np.testing.assert_allclose(
        three_pl_probability(1.5, -0.2, 0.25, theta),
        expected,
        atol=0.0,
        rtol=0.0,
    )
    assert curves.shape == (2, 201) and grid.shape == (201,)

def test_smoothing_contract():
    config = load_config(ROOT / "config/defaults.yaml")
    synthetic = config["synthetic"]
    params, labels = generate_synthetic(
        synthetic["n_items"], synthetic["seed"], synthetic["tiers"],
    )
    curves, grid = compute_iccs(params, **config["theta"])
    smooth_cfg = config["smoothing"]
    smoothing = fit_smoothing(
        curves,
        grid,
        smooth_cfg["spline_order"],
        smooth_cfg["num_basis"],
        smooth_cfg["penalty_derivative_order"],
        lambda_candidates_from_config(smooth_cfg),
        smooth_cfg["max_seconds"],
    )
    assert smoothing.selected_lambda in smoothing.candidates
    selected = int(np.argmin(smoothing.gcv_values))
    assert smoothing.edge_flag == (selected in (0, len(smoothing.candidates) - 1))
    observed = float(np.max(smoothing.rmse))
    print(f"observed max smoothing RMSE={observed:.12g}")
    assert observed < 0.01

Before you run the smoothing test, do you expect the selected candidate and its edge flag to agree?

  • Run the focused smoothing contract with printed output enabled:
pytest -q -s tests/test_foundation.py -k smoothing

What should the final check show?

You should see the printed maximum smoothing RMSE followed by a passing smoothing test. A pass proves that the selected lambda belongs to the candidate array and that the edge flag matches its position.

The maximum observed RMSE should be below `0.01`. That threshold confirms that the shared basis closely approximates every exact ICC.

Strong result. Your exact 201-point ICC rows now have a selected cubic B-spline representation with explicit approximation diagnostics.

Did the smoothing contract fail?

  • Check that `lambda_candidates_from_config` reads the nested `logspace` values from `config/defaults.yaml`.
  • Check that `selected` comes from `np.argmin(gcv)`.
  • Preserve the configured candidate range if the observed RMSE exceeds `0.01`.

Help me debug the smoothing contract.

Your smoother now turns validated ICC rows into shared functional coordinates. Next up, you will fit FPCA and freeze the fitted projection path for unseen curves.

Freeze and Verify FPCA

Your exact ICCs now share a smooth B-spline representation. The final statistical layer converts those curves into reusable coordinates with functional PCA.

Separate FPCA fits can choose opposite component signs. A fitted coordinate system can also leak training work into inference unless projection uses stored objects only. This step fixes both problems before proving the complete foundation.

In this step, get ready to:
  • Fit FPCA with variance-based component retention plus deterministic component signs.
  • Freeze the fitted smoother plus FPCA objects for projection without refitting.
  • Prove persistence plus held-out behavior with tests and an empirical report.
Fit signed FPCA components

FPCA compresses the smoothed curves into component scores. The retained prefix reaches the configured variance threshold whenever the ten-component cap permits it.

  • In your code editor's file explorer, select the existing core folder.
  • Create a file named fpca.py inside the core folder.
  • Use the complete reference below to populate core/fpca.py.

✔️ Awesome, I've got everything!

Great. Save core/fpca.py before connecting it to the runner.

ⓧ I'd like to double check the full code

"""Functional PCA with deterministic component signs."""
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
from skfda.preprocessing.dim_reduction import FPCA
from core.smoothing import SmoothingResult, evaluate_on_grid

SIGN_CONVENTION = "largest-abs loading positive"

@dataclass(frozen=True)
class FPCAResult:
    fpca: FPCA
    k: int
    signs: np.ndarray
    training_scores: np.ndarray
    explained_variance_ratio: np.ndarray
    cumulative_variance: float
    components_on_grid: np.ndarray

def fit_fpca(
    smoothing: SmoothingResult,
    grid: np.ndarray,
    max_components: int,
    variance_threshold: float,
) -> FPCAResult:
    """Fit the cap, choose k, then align component and score signs."""
    n_components = min(
        max_components,
        smoothing.fdata.n_samples,
        smoothing.fdata.basis.n_basis,
    )
    model = FPCA(n_components=n_components)
    raw_scores = model.fit_transform(smoothing.fdata)
    ratios = np.asarray(model.explained_variance_ratio_, dtype=float)
    cumulative = np.cumsum(ratios)
    k = (
        int(np.searchsorted(cumulative, variance_threshold, side="left") + 1)
        if cumulative[-1] >= variance_threshold
        else n_components
    )
    raw_components = evaluate_on_grid(model.components_, grid)
    pivot = np.argmax(np.abs(raw_components), axis=1)
    signs = np.where(
        raw_components[np.arange(n_components), pivot] < 0,
        -1.0,
        1.0,
    )
    return FPCAResult(
        model,
        k,
        signs[:k],
        np.asarray(raw_scores[:, :k]) * signs[:k],
        ratios[:k],
        float(cumulative[k - 1]),
        raw_components[:k] * signs[:k, None],
    )

How does signed FPCA work?

  • The FPCA model fits no more components than the configured cap or the available functional dimensions allow.
  • The cumulative variance determines the smallest retained prefix k.
  • The largest-absolute loading becomes the sign anchor for each retained component.
  • The same signs multiply the component evaluations plus the training scores. This creates one deterministic orientation.
  • Save core/fpca.py.
  • Add the fit_fpca import beside the existing core imports in scripts/run_foundation.py.
  • In scripts/run_foundation.py, place the following block after the smoothing RMSE output:
    fpca_cfg = config["fpca"]
    fpca_result = fit_fpca(
        smoothing, grid,
        fpca_cfg["max_components"], fpca_cfg["variance_threshold"],
    )
    emit(
        f"fpca: k={fpca_result.k}, "
        f"cumulative_variance={fpca_result.cumulative_variance:.12g}",
    )

What does this runner block prove?

The runner passes the selected smoothed basis rows into fit_fpca. Its output exposes the retained component count plus the cumulative variance reached by that prefix.

  • Save scripts/run_foundation.py.
  • Run the foundation script to exercise the new FPCA layer:
python -m scripts.run_foundation

What should you see?

You should see an fpca: k= line after the smoothing diagnostics. The line also reports the observed cumulative variance.

FPCA output missing?

Confirm that core/fpca.py is saved. Check that the runner imports fit_fpca from the new module.

Compare the arguments passed to fit_fpca with the complete reference. A grid or smoothing variable typo prevents the FPCA block from running.

Help me debug the FPCA runner output

Freeze projection and persistence

A frozen transform owns the fitted smoother plus the fitted FPCA coordinate system. Inference calls their stored transformation methods without exposing any fitting path.

  • In your code editor's file explorer, select the existing core folder.
  • Create a file named transform.py inside the core folder.
  • Use the complete reference below to populate core/transform.py.

✔️ Awesome, I've got everything!

Your frozen inference module is ready. Save it before adding persistence checks to the runner.

ⓧ I'd like to double check the full code

"""Frozen smoothing and FPCA projection."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
import pickle
import numpy as np
from skfda import FDataGrid
from skfda.preprocessing.dim_reduction import FPCA
from skfda.preprocessing.smoothing import BasisSmoother
from skfda.representation.basis import Basis, FDataBasis
from core.fpca import FPCAResult, SIGN_CONVENTION
from core.smoothing import SmoothingResult

@dataclass
class FrozenTransform:
    basis: Basis
    selected_lambda: float
    fpca_mean: FDataBasis
    fpca_components: FDataBasis
    fpca: FPCA
    smoother: BasisSmoother
    k: int
    signs: np.ndarray
    sign_convention: str
    grid: np.ndarray

    def project(self, exact_icc_rows: np.ndarray) -> np.ndarray:
        """Use stored objects only. This method has no fit path."""
        rows = np.asarray(exact_icc_rows, dtype=float)
        if rows.ndim == 1:
            rows = rows[None, :]
        if rows.ndim != 2 or rows.shape[1] != len(self.grid):
            raise ValueError("ICC rows must have shape (n_items, len(grid)).")
        fd_grid = FDataGrid(data_matrix=rows, grid_points=self.grid)
        basis_rows = self.smoother.transform(fd_grid)
        return np.asarray(self.fpca.transform(basis_rows)[:, : self.k]) * self.signs

    def save(self, path: str | Path) -> None:
        destination = Path(path)
        destination.parent.mkdir(parents=True, exist_ok=True)
        with destination.open("wb") as handle:
            pickle.dump(self, handle, protocol=pickle.HIGHEST_PROTOCOL)

    @classmethod
    def load(cls, path: str | Path) -> "FrozenTransform":
        with Path(path).open("rb") as handle:
            loaded = pickle.load(handle)
        if not isinstance(loaded, cls):
            raise TypeError("Pickle does not contain a FrozenTransform.")
        return loaded

def freeze_transform(
    smoothing: SmoothingResult,
    fpca_result: FPCAResult,
    grid: np.ndarray,
) -> FrozenTransform:
    return FrozenTransform(
        smoothing.fdata.basis,
        smoothing.selected_lambda,
        fpca_result.fpca.mean_,
        fpca_result.fpca.components_,
        fpca_result.fpca,
        smoothing.smoother,
        fpca_result.k,
        fpca_result.signs.copy(),
        SIGN_CONVENTION,
        np.asarray(grid, dtype=float).copy(),
    )

What makes projection frozen?

  • The transform stores the fitted basis smoother plus the fitted FPCA model.
  • The project method accepts one row or a matrix whose width matches the stored grid.
  • The smoother converts exact ICC rows into the stored B-spline basis with transform.
  • The FPCA model converts those basis rows into signed scores with its stored coordinate system.

Load Trusted Pickles Only

The transform pickle belongs to this local project. Keep that trust boundary intact because loading an untrusted pickle can execute malicious code.

  • Save core/transform.py.
  • Add the FrozenTransform plus freeze_transform import beside the existing core imports in scripts/run_foundation.py.
  • In scripts/run_foundation.py, place the following persistence block after the FPCA output:
    frozen = freeze_transform(smoothing, fpca_result, grid)
    before_save = frozen.project(curves)
    transform_path = output / "transform.pkl"
    frozen.save(transform_path)
    reloaded = FrozenTransform.load(transform_path)
    projected = reloaded.project(curves)
    in_sample = float(np.max(np.abs(projected - fpca_result.training_scores)))
    emit(f"max in-sample difference: {in_sample:.12g}")
    emit(f"persistence identical: {np.array_equal(before_save, projected)}")

What does this persistence check do?

The runner projects the training curves before saving artifacts/transform.pkl. It reloads the transform before checking that the scores remain exactly identical.

  • Save scripts/run_foundation.py.
  • Run the foundation script to create the persisted transform:
python -m scripts.run_foundation

What should you see?

You should see a maximum in-sample difference plus a persistence result. The artifacts folder should now contain transform.pkl.

Persistence check failing?

Confirm that freeze_transform copies the signs plus the grid. Check that FrozenTransform.load returns the loaded transform object.

If projection rejects the curves, compare their second dimension with the stored grid length. The project expects 201 values per ICC row.

Help me debug frozen transform persistence

Complete the diagnostic proof

Matching training scores proves internal consistency. Held-out projection adds a stronger check because the wrapper must agree with direct use of the same stored objects on unseen rows.

Independent FPCA refits define different coordinate systems. Principal subspace angles compare their component spaces without pretending that their score columns are directly interchangeable.

  • Add the subspace_angles plus FDataGrid imports to scripts/run_foundation.py.
  • Place this held-out projection block after the persistence output:
    indices = np.random.default_rng(synthetic["seed"]).choice(
        len(curves), size=10, replace=False,
    )
    mask = np.ones(len(curves), dtype=bool); mask[indices] = False
    held_smoothing = fit_smoothing(
        curves[mask], grid,
        smooth_cfg["spline_order"], smooth_cfg["num_basis"],
        smooth_cfg["penalty_derivative_order"],
        lambda_candidates_from_config(smooth_cfg), smooth_cfg["max_seconds"],
    )
    held_result = fit_fpca(
        held_smoothing, grid,
        fpca_cfg["max_components"], fpca_cfg["variance_threshold"],
    )
    held_transform = freeze_transform(held_smoothing, held_result, grid)
    held_scores = held_transform.project(curves[indices])
    held_basis = held_smoothing.smoother.transform(
        FDataGrid(data_matrix=curves[indices], grid_points=grid),
    )
    direct = (
        held_result.fpca.transform(held_basis)[:, : held_result.k]
        * held_result.signs
    )
    held_differences = np.max(np.abs(held_scores - direct), axis=0)
    emit(f"held-out frozen differences: {held_differences.tolist()}")

Why compare wrapper and direct projection?

Both paths use one held-out model plus one coordinate system. Their difference isolates mistakes in the frozen wrapper without confusing them with legitimate variation between refitted FPCA models.

  • Save scripts/run_foundation.py.
  • Run the foundation script to observe the held-out differences:
python -m scripts.run_foundation

What should you see?

You should see a held-out frozen differences list. Each value measures wrapper-versus-direct agreement inside the same fitted coordinate system.

  • In scripts/run_foundation.py, place this leave-one-out diagnostics block after the held-out output:
    emit("leave-one-out refit subspace angles in degrees:")
    for index in indices:
        loo_smoothing = fit_smoothing(
            np.delete(curves, index, axis=0), grid,
            smooth_cfg["spline_order"], smooth_cfg["num_basis"],
            smooth_cfg["penalty_derivative_order"],
            lambda_candidates_from_config(smooth_cfg), smooth_cfg["max_seconds"],
        )
        loo = fit_fpca(
            loo_smoothing, grid,
            fpca_cfg["max_components"], fpca_cfg["variance_threshold"],
        )
        common = min(fpca_result.k, loo.k)
        angles = np.rad2deg(subspace_angles(
            fpca_result.components_on_grid[:common].T,
            loo.components_on_grid[:common].T,
        ))
        emit(f"{params.loc[index, 'item_id']}: {angles.tolist()}")

What do the angles represent?

Each vector compares the all-data FPCA component space with one leave-one-out refit. The runner reports every finite angle without turning refit variation into a pass threshold.

  • In scripts/run_foundation.py, place this final status plus report-writing block after the angle loop:
    emit(
        "rmse threshold status: "
        f"{'PASS' if np.max(smoothing.rmse) < 0.01 else 'FAIL'}",
    )
    emit(
        "in-sample threshold status: "
        f"{'PASS' if in_sample <= config['tolerance']['in_sample'] else 'FAIL'}",
    )
    emit(
        "held-out threshold status: "
        f"{'PASS' if np.all(held_differences < config['tolerance']['held_out']) else 'FAIL'}",
    )
    (output / "report.txt").write_text(
        "\n".join(lines) + "\n",
        encoding="utf-8",
    )

What does the report capture?

The final block records observed threshold statuses after every diagnostic has run. It writes the complete output history to artifacts/report.txt.

  • Save scripts/run_foundation.py.
  • Compare the completed runner with this full-file reference:

✔️ Awesome, I've got everything!

Your runner now covers generation through frozen diagnostics. Keep the saved version in scripts/run_foundation.py.

ⓧ I'd like to double check the full code

"""Run the complete foundation and write reproducible artifacts."""
from __future__ import annotations
from pathlib import Path
import numpy as np
import pandas as pd
from scipy.linalg import subspace_angles
from skfda import FDataGrid
from core.fpca import fit_fpca
from core.icc import compute_iccs
from core.io import load_config, load_params_csv
from core.smoothing import fit_smoothing, lambda_candidates_from_config
from core.synthetic import generate_synthetic
from core.transform import FrozenTransform, freeze_transform

ROOT = Path(__file__).resolve().parents[1]

def main() -> None:
    config = load_config(ROOT / "config/defaults.yaml")
    output = ROOT / "artifacts"; output.mkdir(exist_ok=True)
    lines: list[str] = []
    def emit(message: str) -> None:
        print(message); lines.append(message)

    synthetic = config["synthetic"]
    params, labels = generate_synthetic(
        synthetic["n_items"], synthetic["seed"], synthetic["tiers"],
    )
    params_path, labels_path = output / "params.csv", output / "labels.csv"
    params.to_csv(params_path, index=False)
    pd.DataFrame({"item_id": params["item_id"], "tier": labels}).to_csv(
        labels_path, index=False,
    )

    corrupt = params.iloc[:2].copy()
    corrupt.loc[corrupt.index[0], "a"] = -1.0
    corrupt_path = output / "corrupt_preview.csv"
    corrupt.to_csv(corrupt_path, index=False)
    raw_corrupt = pd.read_csv(corrupt_path)
    emit(f"raw parser accepted corrupt preview a={raw_corrupt.loc[0, 'a']}")

    loaded = load_params_csv(params_path)
    emit(
        f"validation: valid={loaded.validation.is_valid}, "
        f"errors={len(loaded.validation.errors)}",
    )
    if loaded.data is None:
        for error in loaded.validation.errors:
            emit(f"validation error: {error}")
        raise SystemExit("Validation failed before ICC computation.")

    curves, grid = compute_iccs(loaded.data, **config["theta"])
    emit(f"icc shape: {curves.shape}, grid shape: {grid.shape}")
    smooth_cfg = config["smoothing"]
    smoothing = fit_smoothing(
        curves, grid,
        smooth_cfg["spline_order"], smooth_cfg["num_basis"],
        smooth_cfg["penalty_derivative_order"],
        lambda_candidates_from_config(smooth_cfg), smooth_cfg["max_seconds"],
    )
    emit(
        f"smoothing: lambda={smoothing.selected_lambda:.12g}, "
        f"edge_flag={smoothing.edge_flag}",
    )
    emit(
        "rmse min/median/max: "
        f"{np.min(smoothing.rmse):.12g} / "
        f"{np.median(smoothing.rmse):.12g} / "
        f"{np.max(smoothing.rmse):.12g}",
    )

    fpca_cfg = config["fpca"]
    fpca_result = fit_fpca(
        smoothing, grid,
        fpca_cfg["max_components"], fpca_cfg["variance_threshold"],
    )
    emit(
        f"fpca: k={fpca_result.k}, "
        f"cumulative_variance={fpca_result.cumulative_variance:.12g}",
    )

    frozen = freeze_transform(smoothing, fpca_result, grid)
    before_save = frozen.project(curves)
    transform_path = output / "transform.pkl"
    frozen.save(transform_path)
    reloaded = FrozenTransform.load(transform_path)
    projected = reloaded.project(curves)
    in_sample = float(np.max(np.abs(projected - fpca_result.training_scores)))
    emit(f"max in-sample difference: {in_sample:.12g}")
    emit(f"persistence identical: {np.array_equal(before_save, projected)}")

    indices = np.random.default_rng(synthetic["seed"]).choice(
        len(curves), size=10, replace=False,
    )
    mask = np.ones(len(curves), dtype=bool); mask[indices] = False
    held_smoothing = fit_smoothing(
        curves[mask], grid,
        smooth_cfg["spline_order"], smooth_cfg["num_basis"],
        smooth_cfg["penalty_derivative_order"],
        lambda_candidates_from_config(smooth_cfg), smooth_cfg["max_seconds"],
    )
    held_result = fit_fpca(
        held_smoothing, grid,
        fpca_cfg["max_components"], fpca_cfg["variance_threshold"],
    )
    held_transform = freeze_transform(held_smoothing, held_result, grid)
    held_scores = held_transform.project(curves[indices])
    held_basis = held_smoothing.smoother.transform(
        FDataGrid(data_matrix=curves[indices], grid_points=grid),
    )
    direct = (
        held_result.fpca.transform(held_basis)[:, : held_result.k]
        * held_result.signs
    )
    held_differences = np.max(np.abs(held_scores - direct), axis=0)
    emit(f"held-out frozen differences: {held_differences.tolist()}")

    emit("leave-one-out refit subspace angles in degrees:")
    for index in indices:
        loo_smoothing = fit_smoothing(
            np.delete(curves, index, axis=0), grid,
            smooth_cfg["spline_order"], smooth_cfg["num_basis"],
            smooth_cfg["penalty_derivative_order"],
            lambda_candidates_from_config(smooth_cfg), smooth_cfg["max_seconds"],
        )
        loo = fit_fpca(
            loo_smoothing, grid,
            fpca_cfg["max_components"], fpca_cfg["variance_threshold"],
        )
        common = min(fpca_result.k, loo.k)
        angles = np.rad2deg(subspace_angles(
            fpca_result.components_on_grid[:common].T,
            loo.components_on_grid[:common].T,
        ))
        emit(f"{params.loc[index, 'item_id']}: {angles.tolist()}")

    emit(
        "rmse threshold status: "
        f"{'PASS' if np.max(smoothing.rmse) < 0.01 else 'FAIL'}",
    )
    emit(
        "in-sample threshold status: "
        f"{'PASS' if in_sample <= config['tolerance']['in_sample'] else 'FAIL'}",
    )
    emit(
        "held-out threshold status: "
        f"{'PASS' if np.all(held_differences < config['tolerance']['held_out']) else 'FAIL'}",
    )
    (output / "report.txt").write_text(
        "\n".join(lines) + "\n",
        encoding="utf-8",
    )

if __name__ == "__main__":
    main()

How does the complete runner flow?

The runner keeps labels separate before validating parameters. It computes exact curves before smoothing plus signed FPCA.

The frozen checks reuse stored objects. The refit diagnostics report subspace angles as separate observations.

  • Replace tests/test_foundation.py with the completed contract suite shown below.

✔️ Awesome, I've got everything!

Your test suite now covers the full training plus frozen-inference contract. Save tests/test_foundation.py.

ⓧ I'd like to double check the full code

"""Contract tests for the statistical foundation."""
from __future__ import annotations
from functools import lru_cache
import inspect
from pathlib import Path
import numpy as np
import pandas as pd
from scipy.linalg import subspace_angles
from skfda import FDataGrid
from core.fpca import fit_fpca
from core.icc import compute_iccs, three_pl_probability
from core.io import load_config, load_params_csv
from core.smoothing import fit_smoothing, lambda_candidates_from_config
from core.synthetic import generate_synthetic
from core.transform import FrozenTransform, freeze_transform
from core.validation import validate_params

ROOT = Path(__file__).resolve().parents[1]

def valid_frame() -> pd.DataFrame:
    return pd.DataFrame({
        "item_id": ["A", "B"],
        "a": ["1.0", "1.2"],
        "b": ["0.0", "0.5"],
        "c": ["0.1", "0.2"],
    })

@lru_cache(maxsize=1)
def bundle():
    config = load_config(ROOT / "config/defaults.yaml")
    synthetic = config["synthetic"]
    params, labels = generate_synthetic(
        synthetic["n_items"], synthetic["seed"], synthetic["tiers"],
    )
    curves, grid = compute_iccs(params, **config["theta"])
    smooth_cfg = config["smoothing"]
    smoothing = fit_smoothing(
        curves,
        grid,
        smooth_cfg["spline_order"],
        smooth_cfg["num_basis"],
        smooth_cfg["penalty_derivative_order"],
        lambda_candidates_from_config(smooth_cfg),
        smooth_cfg["max_seconds"],
    )
    fpca_cfg = config["fpca"]
    fpca_result = fit_fpca(
        smoothing,
        grid,
        fpca_cfg["max_components"],
        fpca_cfg["variance_threshold"],
    )
    return (
        config, params, labels, curves, grid, smoothing, fpca_result,
        freeze_transform(smoothing, fpca_result, grid),
    )

def test_validation_missing_required_column():
    assert (1, "b", "required column is missing") in validate_params(
        valid_frame().drop(columns="b"),
    ).errors

def test_validation_missing_numeric_value():
    frame = valid_frame(); frame.loc[0, "a"] = ""
    assert (2, "a", "must not be missing") in validate_params(frame).errors

def test_validation_nonnumeric_value():
    frame = valid_frame(); frame.loc[0, "b"] = "wrong"
    assert (2, "b", "must be numeric and finite") in validate_params(frame).errors

def test_validation_nonpositive_a():
    frame = valid_frame(); frame.loc[0, "a"] = "0"
    assert (2, "a", "must be greater than 0") in validate_params(frame).errors

def test_validation_c_range():
    frame = valid_frame(); frame.loc[0, "c"] = "1"
    assert (2, "c", "must satisfy 0 <= c < 1") in validate_params(frame).errors

def test_validation_empty_item_id():
    frame = valid_frame(); frame.loc[0, "item_id"] = "  "
    assert (2, "item_id", "must be non-empty") in validate_params(frame).errors

def test_validation_duplicate_item_id():
    frame = valid_frame(); frame.loc[1, "item_id"] = "A"
    errors = validate_params(frame).errors
    assert (2, "item_id", "must be unique") in errors
    assert (3, "item_id", "must be unique") in errors

def test_validation_reports_all_errors():
    frame = pd.DataFrame({
        "item_id": ["", "X", "X"],
        "a": ["0", "bad", ""],
        "b": ["bad", "", "0"],
        "c": ["1", "-0.1", "bad"],
    })
    assert len(validate_params(frame).errors) >= 9

def test_clean_loader_passes_and_keeps_extra_column(tmp_path):
    path = tmp_path / "clean.csv"
    valid_frame().assign(extra=["keep", "me"]).to_csv(path, index=False)
    result = load_params_csv(path)
    assert result.validation.is_valid and result.data is not None
    assert "extra" in result.data.columns

def test_icc_closed_form_and_shape():
    params = pd.DataFrame({
        "item_id": ["c0", "cpos"],
        "a": [1.0, 1.5], "b": [0.0, -0.2], "c": [0.0, 0.25],
    })
    curves, grid = compute_iccs(params, -3.0, 3.0, 201)
    theta = np.asarray([-1.0, 0.0, 1.0])
    expected = 0.25 + 0.75 / (1.0 + np.exp(-1.5 * (theta + 0.2)))
    np.testing.assert_allclose(
        three_pl_probability(1.5, -0.2, 0.25, theta),
        expected,
        atol=0.0,
        rtol=0.0,
    )
    assert curves.shape == (2, 201) and grid.shape == (201,)

def test_smoothing_contract():
    config, _, _, _, _, smoothing, _, _ = bundle()
    assert smoothing.selected_lambda in smoothing.candidates
    selected = int(np.argmin(smoothing.gcv_values))
    assert smoothing.edge_flag == (selected in (0, len(smoothing.candidates) - 1))
    observed = float(np.max(smoothing.rmse))
    print(f"observed max smoothing RMSE={observed:.12g}")
    assert observed < 0.01

def test_fpca_contract_and_signs():
    config, _, _, _, _, _, result, _ = bundle()
    assert result.k <= config["fpca"]["max_components"]
    if result.cumulative_variance < config["fpca"]["variance_threshold"]:
        assert result.k == config["fpca"]["max_components"]
    pivot = np.argmax(np.abs(result.components_on_grid), axis=1)
    assert np.all(result.components_on_grid[np.arange(result.k), pivot] > 0)

def test_in_sample_projection():
    config, _, _, curves, _, _, result, frozen = bundle()
    difference = np.max(np.abs(frozen.project(curves) - result.training_scores))
    assert difference <= config["tolerance"]["in_sample"]

def test_frozen_held_out_projection_matches_stored_objects():
    config, _, _, curves, grid, _, _, _ = bundle()
    indices = np.random.default_rng(config["synthetic"]["seed"]).choice(
        len(curves), size=10, replace=False,
    )
    mask = np.ones(len(curves), dtype=bool); mask[indices] = False
    smooth_cfg = config["smoothing"]
    smoothing = fit_smoothing(
        curves[mask], grid,
        smooth_cfg["spline_order"], smooth_cfg["num_basis"],
        smooth_cfg["penalty_derivative_order"],
        lambda_candidates_from_config(smooth_cfg), smooth_cfg["max_seconds"],
    )
    fpca_cfg = config["fpca"]
    result = fit_fpca(
        smoothing, grid,
        fpca_cfg["max_components"], fpca_cfg["variance_threshold"],
    )
    frozen = freeze_transform(smoothing, result, grid)
    wrapped = frozen.project(curves[indices])
    basis_rows = smoothing.smoother.transform(
        FDataGrid(data_matrix=curves[indices], grid_points=grid),
    )
    direct = result.fpca.transform(basis_rows)[:, : result.k] * result.signs
    differences = np.max(np.abs(wrapped - direct), axis=0)
    print("held-out frozen differences", differences.tolist())
    assert np.all(differences < config["tolerance"]["held_out"])

def test_refit_subspace_angles_are_reported_not_gated():
    config, params, _, curves, grid, _, all_fpca, _ = bundle()
    indices = np.random.default_rng(config["synthetic"]["seed"]).choice(
        len(curves), size=10, replace=False,
    )
    smooth_cfg, fpca_cfg = config["smoothing"], config["fpca"]
    for index in indices:
        loo_smoothing = fit_smoothing(
            np.delete(curves, index, axis=0), grid,
            smooth_cfg["spline_order"], smooth_cfg["num_basis"],
            smooth_cfg["penalty_derivative_order"],
            lambda_candidates_from_config(smooth_cfg), smooth_cfg["max_seconds"],
        )
        loo = fit_fpca(
            loo_smoothing, grid,
            fpca_cfg["max_components"], fpca_cfg["variance_threshold"],
        )
        common = min(all_fpca.k, loo.k)
        angles = np.rad2deg(subspace_angles(
            all_fpca.components_on_grid[:common].T,
            loo.components_on_grid[:common].T,
        ))
        print(params.loc[index, "item_id"], angles.tolist())
        assert np.all(np.isfinite(angles))

def test_persistence_scores_are_identical(tmp_path):
    _, _, _, curves, _, _, _, frozen = bundle()
    before = frozen.project(curves[:10])
    path = tmp_path / "transform.pkl"; frozen.save(path)
    after = FrozenTransform.load(path).project(curves[:10])
    assert np.array_equal(before, after)

def test_determinism():
    config, params, _, curves, grid, smoothing, result, _ = bundle()
    synthetic = config["synthetic"]
    params_2, _ = generate_synthetic(
        synthetic["n_items"], synthetic["seed"], synthetic["tiers"],
    )
    curves_2, grid_2 = compute_iccs(params_2, **config["theta"])
    smooth_cfg = config["smoothing"]
    smoothing_2 = fit_smoothing(
        curves_2, grid_2,
        smooth_cfg["spline_order"], smooth_cfg["num_basis"],
        smooth_cfg["penalty_derivative_order"],
        lambda_candidates_from_config(smooth_cfg), smooth_cfg["max_seconds"],
    )
    fpca_cfg = config["fpca"]
    result_2 = fit_fpca(
        smoothing_2, grid_2,
        fpca_cfg["max_components"], fpca_cfg["variance_threshold"],
    )
    assert params.equals(params_2)
    assert np.array_equal(curves, curves_2) and np.array_equal(grid, grid_2)
    assert smoothing.selected_lambda == smoothing_2.selected_lambda
    assert np.array_equal(result.training_scores, result_2.training_scores)

def test_label_leakage_and_no_streamlit():
    _, params, labels, _, _, _, _, _ = bundle()
    assert "label" not in params.columns and len(labels) == len(params)
    from core import fpca, icc, smoothing, synthetic, transform
    functions = [
        synthetic.generate_synthetic, icc.compute_iccs,
        smoothing.fit_smoothing, fpca.fit_fpca, transform.freeze_transform,
    ]
    for function in functions:
        assert all(
            "label" not in name.lower()
            for name in inspect.signature(function).parameters
        )
    for path in (ROOT / "core").glob("*.py"):
        assert "streamlit" not in path.read_text(encoding="utf-8").lower()

What does the suite protect?

The suite keeps every earlier validation plus ICC contract. It adds signed FPCA checks plus frozen in-sample, held-out, persistence, determinism, leakage, and subspace diagnostics.

  • Create README.md beside the existing core folder.
  • Use the complete reference below to document the final project.

✔️ Awesome, I've got everything!

Your project now documents the reproducible environment plus the statistical rationale. Save README.md.

ⓧ I'd like to double check the full code

# Frozen 3PL Functional Transform

This repository contains a headless statistical core. It has no assignment,
feedback, versioning, UI, or Streamlit dependency.

## Environment

The reproducible path uses Python 3.12 in WSL. scikit-fda 0.10.1 includes the
0.10.0 support for scikit-learn 1.6. `scikit-learn==1.6.1` is a transitive
compatibility constraint even though application code uses only the six
requested direct dependencies.

```bash
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade "pip==26.2.2"
python -m pip lock -o pylock.toml \
  "numpy==2.5.4" \
  "scipy==1.18.1" \
  "scikit-fda==0.10.1" \
  "pandas==3.1.0" \
  "PyYAML==6.0.3" \
  "pytest==9.1.1" \
  "scikit-learn==1.6.1"
python -m pip install -r pylock.toml
```

`pylock.toml` is generated for the current Python version and WSL platform. Do
not hand-edit it or reuse it on another platform without regenerating it.

## Run

```bash
pytest -q -s
python -m scripts.run_foundation
```

The script writes separate parameter and label CSVs, a deliberately corrupt
preview, `transform.pkl`, and the observed `report.txt` under `artifacts/`.

## Smoothing path

The project uses scikit-fda 0.10.1 with a 40-function order-4 B-spline basis,
`L2Regularization(LinearDifferentialOperator(2))`, and
`BasisSmoother(..., method="svd", return_basis=True)`.

GCV is pooled mean squared residual error divided by
`(1 - trace(H) / n_grid)^2`. The smallest candidate wins. The edge flag is true
when that candidate is the first or last. No setting changes after a test result.

## FPCA and held-out rationale

FPCA keeps the smallest prefix reaching 95% cumulative variance, capped at ten.
Each component is signed so its largest-absolute evaluated loading is positive.

Frozen-model held-out projection is gated because the curves share one stored
coordinate system. Leave-one-out and all-data fits define different FPCA
coordinates, so their component scores are not compared directly. The script
reports principal subspace angles for those refits without a pass threshold.
The configured held-out tolerance applies only to wrapper-versus-direct
projection through the same stored objects.

## Results and failure policy

Run the defaults and use `artifacts/report.txt` for selected lambda, edge flag,
component count, cumulative variance, RMSE summary, frozen held-out differences,
subspace angles, persistence, and maximum in-sample difference. Do not invent or
paste unobserved values into this README.

If the penalty API check or GCV loop exceeds 20 minutes, stop and report it. If
a fixed threshold fails, preserve the settings and report the observed value.
Only unpickle a transform created by this trusted local project.

Why record observed results separately?

The README documents the method plus failure policy. The generated report holds the observed values from your own environment, so the documentation never invents a selected lambda or diagnostic result.

The final checks perform repeated smoothing fits for the leave-one-out diagnostics. This can take several minutes, so sustained computation during these commands is expected.

Before you run the full suite, do you expect every frozen projection path to agree without calling a fitting method?

  • Run the complete contract suite:
pytest -q -s

What should you see?

You should see the complete suite pass. The output includes the observed smoothing RMSE, held-out frozen differences, and ten finite leave-one-out subspace-angle vectors.

A contract test failed?

Keep the configured thresholds unchanged. Use the failing test name to identify whether the mismatch comes from signs, projection shape, persistence, or refit diagnostics.

Confirm that project calls only the stored smoother plus FPCA transformation methods. Any new fit call breaks the frozen-inference contract.

Help me debug the failing foundation test

Before you run the final report, which diagnostics do you expect to appear after the persistence result?

  • Run the end-to-end foundation script:
python -m scripts.run_foundation

What should you see?

You should see valid parameter loading plus ICC shapes. The output continues through smoothing, FPCA, in-sample projection, persistence, held-out differences, subspace angles, and three observed threshold statuses.

You should also see artifacts/transform.pkl plus artifacts/report.txt in the existing artifacts folder.

Report or transform missing?

Check the terminal for the first traceback above the missing artifact. The runner writes the report only after all diagnostic loops complete.

If the run exceeds the configured guard, preserve the current settings. Report the timeout instead of changing the lambda grid or smoothing basis.

Help me diagnose the incomplete foundation run

That is the full foundation proven. Your stored transform now converts unseen exact ICC rows into signed FPCA scores through one reusable coordinate system.

Secret mission

Stress an Extrapolative Item Without Refitting

Held-out training rows still resemble the curves used to fit your model. Project an extreme valid item beyond every configured tier. Then prove the frozen transform remains byte-for-byte unchanged.

Clean Up Your Resources

Clean Up Your Resources

Your foundation runs entirely inside WSL with no ongoing service costs. Choose the cleanup option that matches what you want to do next.

Resources you used:

  • Python 3.12 virtual environment in .venv with the pinned project dependencies.
  • Application source modules in core/.
  • Project configuration in config/.
  • Complete test suite in tests/.
  • End-to-end runner in scripts/.
  • Generated CSV files, empirical report, and frozen transform in artifacts/.
  • Generated dependency lock in pylock.toml.
  • Project documentation in README.md.

Keep everything running

No action needed. Choose this if you want the tested foundation ready for more experiments.

  • Leave .venv plus every project file in place.
  • Reuse scripts/run_foundation.py whenever you want to regenerate the empirical artifacts.
  • Load only the trusted local artifacts/transform.pkl artifact.

Pause - I'll come back to this later

Deactivate the active environment to pause your session. Every source file, test, lock entry, and generated artifact stays available.

  • Stop the active Python environment by running this command:
deactivate

What does deactivation change?

The command removes the virtual environment from your current shell session. It preserves .venv and all installed dependencies on disk.

  • Confirm that the virtual environment name disappears from your shell prompt.

Your session is paused. The complete frozen-transform foundation remains ready for your next WSL session.

Delete - I don't want to use this again

Remove all project resources to clear the local foundation from WSL. Deletion is permanent, but the cleanup command targets only the paths listed above.

  • Confirm that the terminal is still inside the directory containing README.md by running this command:
ls

What does this check show?

The ls command lists the visible entries in your current directory. This check keeps the cleanup scoped to the foundation you built.

You should see README.md, core, config, tests, scripts, artifacts, and pylock.toml in the output.

  • Remove the active environment plus every listed project resource by running these commands:
deactivate
rm -rf .venv core config tests scripts artifacts pylock.toml README.md

What do these commands remove?

  • The deactivate command stops using the project environment in the current shell.
  • The rm -rf command permanently removes the named environment, folders, files, generated transform, and empirical artifacts.
  • Check the remaining visible contents of the same WSL directory by running this command:
ls

What confirms cleanup?

The command lists any visible entries that remain. None of the named foundation resources should appear.

The local foundation is now removed. WSL and the Python installations that existed before this project remain available.

Still see a project resource?

A remaining named path may be protected by local file permissions. Confirm that you are checking the same directory that previously contained README.md.

Ask for help removing the remaining local files safely.

Nice Work!

Nice Work!

Outstanding work! You built a headless functional-data foundation that converts validated 3PL item parameters into stable FPCA scores. The saved transform now projects unseen ICC curves with stored objects only.

You've learned how to:

  • Built a deterministic synthetic-data pipeline that generates 240 parameter rows with separate tier labels. Evaluated exact ICCs across a 201-point theta grid using vectorized calculations.
  • Added aggregate CSV validation that returns every invalid cell with its physical row number. Applied cubic B-spline smoothing with explicit GCV selection plus per-item RMSE diagnostics.
  • Created deterministically signed FPCA components with a retained variance prefix. Persisted a frozen inference transform that reuses the fitted smoother plus coordinate system. Proved persistence with in-sample checks plus held-out checks. Reported leave-one-out subspace-angle vectors for independently refitted models.
  • Secret Mission: Stress-tested the frozen transform with the out-of-tier EXTREME item. Proved that projection leaves the serialized transform byte-for-byte unchanged.

Ready to quiz yourself?