Build Leakage-Resistant Features

Introduction

30 Second Summary

A prediction can look convincing when its evaluation quietly includes clues from the answer. The score rises, but the evidence no longer reflects what would happen with new runners.

In this project, you will extend your running-readiness repository with a feature experiment controlled by a prediction-time schema. The experiment compares raw predictors with engineered predictors through group-safe cross-validation after the inherited evidence passes its audit.

What You'll Build

You will see a development-only report reveal whether three schema-supported engineered predictors improve grouped validation without reopening the consumed holdout.

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

• A terminal fail-closed audit result proving every protected fingerprint still matches the recorded evidence boundary.
• A JSON experiment report comparing raw features with engineered features across the same four runner-safe folds.
• A portfolio-ready evidence package linking observed results to a fingerprinted Level 5 handoff.
• Secret Mission: attack the schema gate with a target-derived leakage proxy that produces perfect but invalid validation before the repaired path rejects it.

Are there any prerequisites?

This project assumes you completed Levels 1 through 3 in the existing level-1-data repository. Your local copy must still contain data/v1/processed/train.csv, artifacts/v1/holdout_consumed.json, and reports/v1/level3_evaluation.json.

You also need the repository's existing .venv environment.

Before We Start

Level 4 builds on evidence already established in the local level-1-data repository. Changing feature behavior before confirming that starting point would weaken every result that follows.

The consumed v1 holdout is already committed evidence. Its contract permanently forbids feature selection, model comparison, model selection, and threshold tuning.

In this step, get ready to:

• Open the existing level-1-data repository in VS Code.
• Confirm the existing virtual environment and immutable evidence files are present.
• Define the Level 4 claim within the consumed-holdout boundary.

Open the existing repository

The existing local copy contains private evidence that the public repository intentionally excludes. Open that copy in VS Code so every later audit uses the learner-attested files already on your Mac.

• Press Cmd+Space on macOS or the Windows key on Windows to open system search.
• Type VS Code and press Enter to open it.
• Click File in the top menu bar.
• Click Open Folder.
• Select your existing level-1-data folder.
• Click Open.

You should see level-1-data at the top of the Explorer sidebar. This confirms that VS Code is using the existing local repository.

Can't find the repository?

• Search Finder for level-1-data.
• Avoid cloning a fresh copy because the public repository excludes the required local evidence.

Still stuck? Help me locate my existing level-1-data repository without replacing its private evidence.

Confirm the immutable starting evidence

The next step treats these files as immutable evidence. A missing file or changed byte must stop the experiment before feature work begins.

• Use the disclosure arrows in the Explorer sidebar to expand the folders named below.
• Confirm .venv appears at the top level without activating it.
• Confirm data/v1/processed/train.csv is present.
• Confirm contracts/v1/schema_contract.json is present.
• Confirm artifacts/v1 contains preprocessing.json, level3_evaluation_contract.json, level3_selection_lock.json, level4_features_contract.json, and holdout_consumed.json.
• Confirm reports/v1/level3_evaluation.json is present.

All eight evidence files should already exist. This check leaves every file unchanged.

Why must these files stay unchanged?

A SHA-256 fingerprint binds each recorded contract to the exact bytes it describes. Editing or regenerating an input would break that evidence chain.

The audit in the next step fails closed when evidence is missing or altered. That breakpoint prevents experimentation from continuing on an unverified starting state.

Define the evidence-safe claim

A predeclared claim prevents later results from reshaping the question. The claim must identify the candidate change and the evidence allowed to judge it.

What is the Level 4 claim?

Level 4 tests whether schema-supported engineered predictors change grouped development validation behavior relative to the five raw features.

Only grouped development evidence can answer that question. The consumed holdout is permanently excluded from feature selection, model comparison, model selection, and threshold tuning.

• State the Level 4 claim in the text checkpoint below.
• Explain the holdout exclusion using its four forbidden uses.

Before you make the final visual check, do you expect Level 4 to have created any new file?

• Inspect the Explorer sidebar without opening or editing any evidence file.

You should see the existing .venv plus all eight immutable inputs. The repository contains no new Level 4 output at this point.

That secures the most important starting condition. Level 4 can begin without weakening the evidence inherited from earlier levels.

Your evidence boundary is explicit. Next, you'll activate the existing environment and make the audit fail closed before any feature work begins.

Audit the Immutable Level 3 Handoff

Your Level 4 claim compares schema-supported engineered predictors with the five raw features on grouped development validation.

The consumed v1 holdout is permanently excluded from that decision. This audit checks the immutable evidence boundary before any feature experiment can run.

In this step, get ready to:

• Check the existing virtual environment without reinstalling its packages.
• Build the foundation of the Level 4 audit module.
• Run a fail-closed audit against every protected Level 3 handoff artifact.

Check the existing environment

The existing virtual environment isolates the repository's pinned Python packages. Those packages include pandas, scikit-learn, NumPy, and joblib.

Version drift could change experimental behavior. You will inspect the installed versions without reinstalling anything.

• Click Terminal in the top menu of VS Code.
• Select New Terminal.
• Activate the existing .venv by running this command:

source .venv/bin/activate

What does activation change?

Activation puts the environment's executable directory at the front of your shell path. The python command now uses the interpreter and packages already stored in .venv.

You should see (.venv) at the start of the terminal prompt.

Environment name missing from the prompt?

Confirm the terminal is positioned inside the open level-1-data repository. The activation script must exist at .venv/bin/activate.

Still stuck? Help me diagnose why my existing Python virtual environment does not activate on macOS.

• Inspect the active Python version and installed package versions by running these commands:

python --version
python -m pip list

What do these checks prove?

• The first command reports the interpreter selected by the active environment.
• The second command lists distributions installed in that same environment.

• Confirm the Python output reports version 3.15.
• Select requirements.txt in the VS Code Explorer.
• Confirm the installed NumPy version matches 2.5.4.
• Confirm the installed pandas version matches 3.1.0.
• Confirm the installed scikit-learn version matches 1.9.1.
• Confirm the installed joblib version matches 1.6.0.

The environment matches the repository pins. No dependency files or installed packages have changed.

See a missing or different version?

Stop before editing the environment. Confirm the terminal prompt still begins with (.venv).

A mismatch means the existing environment has drifted from requirements.txt. Help me compare the active environment with the repository pins without reinstalling packages.

Establish the immutable audit boundary

A SHA-256 fingerprint converts file bytes into a fixed digest. Any changed byte produces a different value.

The audit treats mismatches as hard failures. It never repairs or regenerates the evidence it is checking.

• Select the src folder in the VS Code Explorer.
• Click the New File icon in the Explorer toolbar.
• Enter feature_experiment.py as the file name.
• Add the module imports to src/feature_experiment.py by pasting this code:

from __future__ import annotations

import json
import sys
from collections import Counter
from hashlib import sha256
from itertools import combinations
from pathlib import Path
from statistics import fmean, pstdev
from typing import Any

import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin, clone
from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.impute import SimpleImputer
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score
from sklearn.pipeline import Pipeline
from sklearn.tree import DecisionTreeClassifier

from src.train_models import make_group_folds

What do these imports prepare?

• The standard-library imports support JSON contracts, paths, hashes, statistics, and command routing.
• The pandas import supports the development table used by the later experiment.
• The scikit-learn imports prepare fold-local transformations, selection, metrics, and the locked decision tree.
• The existing make_group_folds() import preserves the repository's established runner-safe split logic.

• Save src/feature_experiment.py.
• Check the VS Code Problems panel for red Python syntax errors.

Seeing an import error?

Confirm the file is inside the existing src folder. Confirm train_models.py remains in that folder.

Need help? Help me trace an import error in the Level 4 module without changing the existing repository structure.

• Add the version labels and protected model contract below the imports by pasting this code:

SOURCE_DATASET_VERSION = "v1"
FEATURE_DATASET_VERSION = "v2"
RANDOM_STATE = 42
SELECT_K = 4
RAW_FEATURES = (
    "weekly_km_to_date",
    "sleep_hours_last_night",
    "resting_hr_morning",
    "soreness_score_evening",
    "prior_injury",
)
TARGET_COLUMN = "needs_recovery_next_day"
GROUP_COLUMN = "runner_id"
METRIC_NAMES = ("accuracy", "precision", "recall", "f1")
FORBIDDEN_HOLDOUT_USES = {
    "feature_selection",
    "threshold_tuning",
    "model_comparison",
    "model_selection",
}

What does this contract protect?

• The dataset labels distinguish the inherited v1 evidence from the planned v2 feature contract.
• The raw feature tuple preserves the prediction-time column order recorded by the schema.
• The target and group constants prevent silent changes to the prediction task or runner boundary.
• The forbidden-use set encodes the decisions that can never consult the consumed holdout.

• Save src/feature_experiment.py.
• Confirm the Problems panel shows no red syntax errors for the new constants.

Seeing a syntax error in the constants?

Check the commas after every entry in RAW_FEATURES. Check the closing brace after FORBIDDEN_HOLDOUT_USES.

Still blocked? Help me compare my Level 4 constants with the expected Python tuple and set syntax.

• Add the immutable input paths and planned output paths below the constants by pasting this code:

TRAIN_PATH = Path("data/v1/processed/train.csv")
SCHEMA_PATH = Path("contracts/v1/schema_contract.json")
PREPROCESSING_PATH = Path("artifacts/v1/preprocessing.json")
LEVEL3_CONTRACT_PATH = Path("artifacts/v1/level3_evaluation_contract.json")
SELECTION_LOCK_PATH = Path("artifacts/v1/level3_selection_lock.json")
LEVEL4_CONTRACT_PATH = Path("artifacts/v1/level4_features_contract.json")
HOLDOUT_MARKER_PATH = Path("artifacts/v1/holdout_consumed.json")
EVALUATION_REPORT_PATH = Path("reports/v1/level3_evaluation.json")
V2_SCHEMA_PATH = Path("contracts/v2/feature_schema_contract.json")
V2_REPORT_PATH = Path("reports/v2/feature_experiment.json")
LEVEL5_HANDOFF_PATH = Path("artifacts/v2/level5_features_handoff.json")
DOC_PATH = Path("docs/level-4-features.md")

Why centralize these paths?

Each path names one evidence boundary explicitly. The audit reads protected v1 inputs through these constants.

The planned v2 paths reserve separate locations for later outputs. This separation keeps new evidence away from immutable files.

• Save src/feature_experiment.py.
• Confirm the Problems panel shows no red syntax errors for the path declarations.

Path declaration underlined?

Confirm every value uses Path with matching parentheses. Keep each repository-relative path exactly as shown.

Need a second pair of eyes? Help me find a typo in the immutable Level 4 path declarations.

• Add the fingerprint and JSON helpers below the path constants by pasting this code:

def fingerprint(path: Path) -> str:
    return sha256(path.read_bytes()).hexdigest()


def read_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def write_json_exclusive(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("x", encoding="utf-8") as handle:
        json.dump(payload, handle, indent=2, sort_keys=True)
        handle.write("\n")

What do these helpers enforce?

• The fingerprint() helper hashes the exact bytes of a protected file.
• The read_json() helper reads contract evidence without mutating it.
• The write_json_exclusive() helper uses exclusive creation for future evidence outputs. Existing evidence causes the write to fail.

• Save src/feature_experiment.py.
• Confirm VS Code recognizes three function definitions without red syntax errors.

Helper function indentation broken?

Use four spaces inside each function. Keep the two write operations inside the with block.

Still seeing an error? Help me repair the indentation of the fingerprint and JSON helpers.

Implement the fail-closed checks

The audit starts by proving that every required handoff file exists. It then loads only the contracts needed to compare recorded evidence with current bytes.

• Add the file-presence checks and contract loading below the helper functions by pasting this code:

def audit_handoff(root: Path) -> dict[str, Any]:
    paths = {
        "train": root / TRAIN_PATH,
        "schema": root / SCHEMA_PATH,
        "preprocessing": root / PREPROCESSING_PATH,
        "level3_contract": root / LEVEL3_CONTRACT_PATH,
        "selection_lock": root / SELECTION_LOCK_PATH,
        "level4_contract": root / LEVEL4_CONTRACT_PATH,
        "holdout_marker": root / HOLDOUT_MARKER_PATH,
        "evaluation_report": root / EVALUATION_REPORT_PATH,
    }
    missing = [path.relative_to(root).as_posix() for path in paths.values() if not path.exists()]
    if missing:
        raise FileNotFoundError(f"Missing immutable handoff files: {missing}")

    schema = read_json(paths["schema"])
    preprocessing = read_json(paths["preprocessing"])
    level3 = read_json(paths["level3_contract"])
    lock = read_json(paths["selection_lock"])
    level4 = read_json(paths["level4_contract"])
    marker = read_json(paths["holdout_marker"])

Why check presence first?

The missing-file check runs before any fingerprint comparison. A partial handoff therefore stops with a list of absent files.

The audit does not create replacements. Missing evidence remains visible as a failed boundary check.

• Save src/feature_experiment.py.
• Confirm the Problems panel shows no red syntax errors in audit_handoff().

Audit function ends too early?

Keep every line after the function declaration indented by four spaces. The next chunks continue inside this same function.

Need help with the structure? Help me verify the indentation of the first audit_handoff chunk.

• Continue audit_handoff() below the marker-loading line by pasting this code:

    expected_inputs = level3["input_fingerprints"]
    actual_inputs = {
        "train": fingerprint(paths["train"]),
        "schema": fingerprint(paths["schema"]),
        "preprocessing": fingerprint(paths["preprocessing"]),
    }
    for name, actual in actual_inputs.items():
        if expected_inputs[name]["sha256"] != actual:
            raise ValueError(f"Immutable v1 fingerprint mismatch: {name}")

How does fingerprint rejection work?

The contract supplies the expected hashes. The audit independently hashes the current training CSV, schema, and preprocessing artifact.

Any mismatch raises a ValueError before the audit can report success.

• Save src/feature_experiment.py.
• Confirm the Problems panel shows the fingerprint loop nested inside audit_handoff().

Fingerprint loop outside the function?

The expected_inputs line needs four leading spaces. The comparison inside the nested if needs eight leading spaces.

Still unsure? Help me align the nested fingerprint comparison inside audit_handoff.

Input hashes protect the inherited dataset foundation. The next checks protect the Level 3 decision evidence and consumed-holdout policy.

• Continue audit_handoff() below the fingerprint loop by pasting this code:

    selection_lock_sha = fingerprint(paths["selection_lock"])
    if level4["locked_selection"]["selection_lock_sha256"] != selection_lock_sha:
        raise ValueError("Level 3 selection lock fingerprint mismatch")
    marker_sha = fingerprint(paths["holdout_marker"])
    if level4["holdout_policy"]["marker_sha256"] != marker_sha:
        raise ValueError("Consumed-holdout marker fingerprint mismatch")
    report_sha = fingerprint(paths["evaluation_report"])
    expected_report_sha = level4["evaluation_evidence"]["json_report"]["sha256"]
    if expected_report_sha != report_sha:
        raise ValueError("Level 3 evaluation report fingerprint mismatch")
    if not marker["holdout_consumed"] or not level4["holdout_policy"]["holdout_consumed"]:
        raise ValueError("The v1 holdout must remain marked as consumed")
    if set(marker["forbidden_future_uses"]) != FORBIDDEN_HOLDOUT_USES:
        raise ValueError("Consumed-holdout restrictions changed")
    if set(level4["holdout_policy"]["forbidden_future_uses"]) != FORBIDDEN_HOLDOUT_USES:
        raise ValueError("Level 4 holdout restrictions changed")

What evidence does this protect?

• The selection-lock hash proves the locked model evidence still has the recorded bytes.
• The marker hash proves the consumed-holdout declaration still matches its contract.
• The evaluation-report hash proves the final Level 3 evidence has not drifted.
• The policy comparisons preserve every forbidden future use as a set.

• Save src/feature_experiment.py.
• Confirm the Problems panel shows no red syntax errors in the evidence checks.

Evidence check is outside audit_handoff?

Keep selection_lock_sha aligned with expected_inputs from the previous chunk. Both belong at the function's four-space indentation level.

Need help comparing the nesting? Help me place the Level 3 evidence checks inside audit_handoff.

Matching file hashes prove byte-level integrity. Semantic checks also confirm that the contracts still describe the intended features, target, group, and locked model.

• Finish audit_handoff() below the holdout-policy checks by pasting this code:

    if tuple(schema["feature_columns"]) != RAW_FEATURES:
        raise ValueError("The repository schema no longer matches the verified Level 4 design")
    if tuple(preprocessing["feature_columns"]) != RAW_FEATURES:
        raise ValueError("The preprocessing feature order changed")
    if level4["target_column"] != TARGET_COLUMN or level4["group_column"] != GROUP_COLUMN:
        raise ValueError("Target or group column changed")
    if lock["selected_model"] != level4["locked_selection"]["model"]:
        raise ValueError("Locked model identity changed")

    return {
        "audit_status": "passed",
        "development_only": True,
        "holdout_consumed": True,
        "holdout_read": False,
        "training_data": {"path": TRAIN_PATH.as_posix(), "sha256": actual_inputs["train"]},
        "schema": {"path": SCHEMA_PATH.as_posix(), "sha256": actual_inputs["schema"]},
        "preprocessing": {
            "path": PREPROCESSING_PATH.as_posix(),
            "sha256": actual_inputs["preprocessing"],
            "inherited_fit_scope": preprocessing["fit_scope"],
        },
        "selection_lock": {"path": SELECTION_LOCK_PATH.as_posix(), "sha256": selection_lock_sha},
        "holdout_marker": {"path": HOLDOUT_MARKER_PATH.as_posix(), "sha256": marker_sha},
        "evaluation_report": {"path": EVALUATION_REPORT_PATH.as_posix(), "sha256": report_sha},
        "forbidden_holdout_uses": sorted(FORBIDDEN_HOLDOUT_USES),
    }

When can the audit pass?

Success is reachable only after every existence check, fingerprint check, schema check, identity check, and policy check completes.

The returned evidence states that the run is development-only. It also records that the consumed holdout was not read.

• Save src/feature_experiment.py.
• Confirm the Problems panel shows the return dictionary inside audit_handoff().

Return dictionary underlined?

Confirm the return line has four leading spaces. Check each nested dictionary for matching braces.

Still seeing a problem? Help me locate a brace or indentation error in the audit result dictionary.

• Add the audit command entry point below audit_handoff() by pasting this code:

def main() -> None:
    if len(sys.argv) != 2 or sys.argv[1] not in {"audit", "plan", "run", "document"}:
        raise SystemExit("Usage: python -m src.feature_experiment [audit|plan|run|document]")
    root = Path(__file__).resolve().parents[1]
    if sys.argv[1] == "audit":
        print(json.dumps(audit_handoff(root), indent=2, sort_keys=True))


if __name__ == "__main__":
    main()

How does the command reach the audit?

The entry point resolves the repository directory from the module's location. The audit branch passes that directory into audit_handoff().

The result is printed as sorted JSON. Later steps extend the same command router with the remaining operations.

• Save src/feature_experiment.py.
• Confirm the Problems panel shows no red syntax errors in the command entry point.

Module entry point not recognized?

Confirm the double underscores remain around __name__ and __main__.

Need help? Help me compare my Python module entry point with the provided Level 4 code.

✔️ Awesome, I've got everything!

Great. Save src/feature_experiment.py before running the audit.

ⓧ I'd like to double check the full code

from __future__ import annotations

import json
import sys
from collections import Counter
from hashlib import sha256
from itertools import combinations
from pathlib import Path
from statistics import fmean, pstdev
from typing import Any

import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin, clone
from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.impute import SimpleImputer
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score
from sklearn.pipeline import Pipeline
from sklearn.tree import DecisionTreeClassifier

from src.train_models import make_group_folds

SOURCE_DATASET_VERSION = "v1"
FEATURE_DATASET_VERSION = "v2"
RANDOM_STATE = 42
SELECT_K = 4
RAW_FEATURES = (
    "weekly_km_to_date",
    "sleep_hours_last_night",
    "resting_hr_morning",
    "soreness_score_evening",
    "prior_injury",
)
TARGET_COLUMN = "needs_recovery_next_day"
GROUP_COLUMN = "runner_id"
METRIC_NAMES = ("accuracy", "precision", "recall", "f1")
FORBIDDEN_HOLDOUT_USES = {
    "feature_selection",
    "threshold_tuning",
    "model_comparison",
    "model_selection",
}

TRAIN_PATH = Path("data/v1/processed/train.csv")
SCHEMA_PATH = Path("contracts/v1/schema_contract.json")
PREPROCESSING_PATH = Path("artifacts/v1/preprocessing.json")
LEVEL3_CONTRACT_PATH = Path("artifacts/v1/level3_evaluation_contract.json")
SELECTION_LOCK_PATH = Path("artifacts/v1/level3_selection_lock.json")
LEVEL4_CONTRACT_PATH = Path("artifacts/v1/level4_features_contract.json")
HOLDOUT_MARKER_PATH = Path("artifacts/v1/holdout_consumed.json")
EVALUATION_REPORT_PATH = Path("reports/v1/level3_evaluation.json")
V2_SCHEMA_PATH = Path("contracts/v2/feature_schema_contract.json")
V2_REPORT_PATH = Path("reports/v2/feature_experiment.json")
LEVEL5_HANDOFF_PATH = Path("artifacts/v2/level5_features_handoff.json")
DOC_PATH = Path("docs/level-4-features.md")


def fingerprint(path: Path) -> str:
    return sha256(path.read_bytes()).hexdigest()


def read_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def write_json_exclusive(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("x", encoding="utf-8") as handle:
        json.dump(payload, handle, indent=2, sort_keys=True)
        handle.write("\n")


def audit_handoff(root: Path) -> dict[str, Any]:
    paths = {
        "train": root / TRAIN_PATH,
        "schema": root / SCHEMA_PATH,
        "preprocessing": root / PREPROCESSING_PATH,
        "level3_contract": root / LEVEL3_CONTRACT_PATH,
        "selection_lock": root / SELECTION_LOCK_PATH,
        "level4_contract": root / LEVEL4_CONTRACT_PATH,
        "holdout_marker": root / HOLDOUT_MARKER_PATH,
        "evaluation_report": root / EVALUATION_REPORT_PATH,
    }
    missing = [path.relative_to(root).as_posix() for path in paths.values() if not path.exists()]
    if missing:
        raise FileNotFoundError(f"Missing immutable handoff files: {missing}")

    schema = read_json(paths["schema"])
    preprocessing = read_json(paths["preprocessing"])
    level3 = read_json(paths["level3_contract"])
    lock = read_json(paths["selection_lock"])
    level4 = read_json(paths["level4_contract"])
    marker = read_json(paths["holdout_marker"])

    expected_inputs = level3["input_fingerprints"]
    actual_inputs = {
        "train": fingerprint(paths["train"]),
        "schema": fingerprint(paths["schema"]),
        "preprocessing": fingerprint(paths["preprocessing"]),
    }
    for name, actual in actual_inputs.items():
        if expected_inputs[name]["sha256"] != actual:
            raise ValueError(f"Immutable v1 fingerprint mismatch: {name}")

    selection_lock_sha = fingerprint(paths["selection_lock"])
    if level4["locked_selection"]["selection_lock_sha256"] != selection_lock_sha:
        raise ValueError("Level 3 selection lock fingerprint mismatch")
    marker_sha = fingerprint(paths["holdout_marker"])
    if level4["holdout_policy"]["marker_sha256"] != marker_sha:
        raise ValueError("Consumed-holdout marker fingerprint mismatch")
    report_sha = fingerprint(paths["evaluation_report"])
    expected_report_sha = level4["evaluation_evidence"]["json_report"]["sha256"]
    if expected_report_sha != report_sha:
        raise ValueError("Level 3 evaluation report fingerprint mismatch")
    if not marker["holdout_consumed"] or not level4["holdout_policy"]["holdout_consumed"]:
        raise ValueError("The v1 holdout must remain marked as consumed")
    if set(marker["forbidden_future_uses"]) != FORBIDDEN_HOLDOUT_USES:
        raise ValueError("Consumed-holdout restrictions changed")
    if set(level4["holdout_policy"]["forbidden_future_uses"]) != FORBIDDEN_HOLDOUT_USES:
        raise ValueError("Level 4 holdout restrictions changed")
    if tuple(schema["feature_columns"]) != RAW_FEATURES:
        raise ValueError("The repository schema no longer matches the verified Level 4 design")
    if tuple(preprocessing["feature_columns"]) != RAW_FEATURES:
        raise ValueError("The preprocessing feature order changed")
    if level4["target_column"] != TARGET_COLUMN or level4["group_column"] != GROUP_COLUMN:
        raise ValueError("Target or group column changed")
    if lock["selected_model"] != level4["locked_selection"]["model"]:
        raise ValueError("Locked model identity changed")

    return {
        "audit_status": "passed",
        "development_only": True,
        "holdout_consumed": True,
        "holdout_read": False,
        "training_data": {"path": TRAIN_PATH.as_posix(), "sha256": actual_inputs["train"]},
        "schema": {"path": SCHEMA_PATH.as_posix(), "sha256": actual_inputs["schema"]},
        "preprocessing": {
            "path": PREPROCESSING_PATH.as_posix(),
            "sha256": actual_inputs["preprocessing"],
            "inherited_fit_scope": preprocessing["fit_scope"],
        },
        "selection_lock": {"path": SELECTION_LOCK_PATH.as_posix(), "sha256": selection_lock_sha},
        "holdout_marker": {"path": HOLDOUT_MARKER_PATH.as_posix(), "sha256": marker_sha},
        "evaluation_report": {"path": EVALUATION_REPORT_PATH.as_posix(), "sha256": report_sha},
        "forbidden_holdout_uses": sorted(FORBIDDEN_HOLDOUT_USES),
    }


def main() -> None:
    if len(sys.argv) != 2 or sys.argv[1] not in {"audit", "plan", "run", "document"}:
        raise SystemExit("Usage: python -m src.feature_experiment [audit|plan|run|document]")
    root = Path(__file__).resolve().parents[1]
    if sys.argv[1] == "audit":
        print(json.dumps(audit_handoff(root), indent=2, sort_keys=True))


if __name__ == "__main__":
    main()

Before you run the audit, predict whether the current local bytes still match every recorded Level 3 contract.

• Run the immutable handoff audit from the repository terminal with this command:

python -m src.feature_experiment audit

What should the audit report?

The JSON should include audit_status set to passed. It should include development_only set to true.

It should also include holdout_consumed set to true. The holdout_read field should be false.

That is the boundary proven. Your experiment can use fingerprinted development evidence without reopening the consumed holdout.

Audit stopped on a mismatch?

A missing file or fingerprint mismatch is an intended fail-closed result. Do not regenerate, edit, delete, or replace any v1 evidence.

Use the reported path or evidence name to investigate the repository state. Help me interpret the Level 4 audit failure without repairing immutable evidence.

Your immutable Level 3 handoff is now verified. Next, you will build the schema gate that accepts three supported transformations while recording the unsupported prior-week ratio.

Build the Schema-Gated Feature Transformer

The immutable audit proved that your development data still matches the Level 3 handoff. It also preserved the consumed-holdout boundary before any feature work began.

A plausible engineered feature can still depend on information that is unavailable when a prediction is made. This step uses the prediction-time schema to decide which proposals may reach the model.

In this step, get ready to:

• Resolve each proposed transformation against the prediction-time schema.
• Implement a cloneable transformer for the supported synthetic predictors.
• Build a pipeline that keeps later preprocessing and selection inside each training fold.

Define the prediction-time feature plan

A feature proposal names the columns needed to calculate a candidate predictor. The resolver checks those sources before any transformation can run.

• In the existing src/feature_experiment.py tab, add the proposal registry below the path constants by pasting this code:

FEATURE_PROPOSALS = (
    {
        "name": "training_load_to_sleep_ratio",
        "sources": ("weekly_km_to_date", "sleep_hours_last_night"),
        "formula": "weekly_km_to_date / (sleep_hours_last_night + 1.0)",
    },
    {
        "name": "training_load_to_resting_hr_ratio",
        "sources": ("weekly_km_to_date", "resting_hr_morning"),
        "formula": "weekly_km_to_date / (resting_hr_morning + 1.0)",
    },
    {
        "name": "injury_soreness_interaction",
        "sources": ("prior_injury", "soreness_score_evening"),
        "formula": "prior_injury * soreness_score_evening",
    },
    {
        "name": "prior_week_load_ratio",
        "sources": ("weekly_km_to_date", "prior_week_km"),
        "formula": "weekly_km_to_date / (prior_week_km + 1.0)",
    },
)

What does this registry record?

• Each record names one candidate predictor.
• The sources tuple identifies the columns required by its calculation.
• The formula value documents the proposed calculation without executing it.
• The registry includes unsupported ideas so the gate can record why they were excluded.

• Add the schema allowlist helper and proposal resolver below audit_handoff() by pasting this code:

def prediction_time_features(schema: dict[str, Any]) -> set[str]:
    return {
        column["name"]
        for column in schema["columns"]
        if column["role"] == "feature" and column["available_at_prediction"] is True
    }


def resolve_feature_plan(schema: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
    allowed = prediction_time_features(schema)
    supported: list[dict[str, Any]] = []
    skipped: list[dict[str, Any]] = []
    for proposal in FEATURE_PROPOSALS:
        missing_sources = sorted(set(proposal["sources"]) - allowed)
        record = {
            "name": proposal["name"],
            "sources": list(proposal["sources"]),
            "formula": proposal["formula"],
        }
        if missing_sources:
            record["reason"] = "unsupported source columns"
            record["missing_sources"] = missing_sources
            skipped.append(record)
        else:
            record["status"] = "supported by prediction-time schema"
            supported.append(record)
    return {"supported": supported, "skipped": skipped}

How does the resolver fail closed?

• The allowlist includes columns whose schema role is feature.
• The allowlist also requires available_at_prediction to be true.
• The resolver records missing sources under skipped.
• A proposal reaches supported only when every source passes both checks.

• Save src/feature_experiment.py.

Before you run the plan, do you expect every proposed transformation to pass the prediction-time schema?

• Resolve the proposed transformations by running this command:

python -m src.feature_experiment plan

What did the gate discover?

The command runs the immutable audit before resolving the feature plan. A failed audit stops the command before feature eligibility is considered.

The output places three transformations under supported. It places prior_week_load_ratio under skipped because prior_week_km appears in missing_sources.

This is the intended shortfall. The schema has stopped an attractive proposal from becoming a fabricated input.

That is the gate doing its job. You now have a reviewable record of three supported transformations and one unsupported proposal.

Plan output does not match?

If all four proposals appear under supported, confirm that the resolver subtracts the allowed prediction-time columns from every proposal's sources.

If the module does not import, check the closing braces in FEATURE_PROPOSALS.

Still blocked? Help me trace the feature plan without adding prior_week_km.

Implement the schema-gated transformer

The plan decides which proposals are eligible. The transformer provides executable paths only for the three supported row-local calculations.

• Add the reusable model-input guard below resolve_feature_plan() by pasting this code:

def validate_model_inputs(candidate_columns: list[str], schema: dict[str, Any]) -> None:
    allowed = prediction_time_features(schema)
    rejected = sorted(set(candidate_columns) - allowed)
    if rejected:
        raise ValueError(f"Rejected non-prediction-time model inputs: {rejected}")

What does input validation protect?

The guard compares candidate model columns with the prediction-time allowlist. Any undeclared column raises a ValueError before fitting begins.

Later experiments can use this guard to reject development-only or target-derived proxy columns.

• Add the transformer constructor and fitting checks immediately below validate_model_inputs() by pasting this class definition:

class SchemaFeatureTransformer(TransformerMixin, BaseEstimator):
    def __init__(self, *, mode: str, engineered_features: tuple[str, ...] = ()) -> None:
        self.mode = mode
        self.engineered_features = engineered_features

    def fit(self, X: pd.DataFrame, y: Any = None) -> SchemaFeatureTransformer:
        if self.mode not in {"raw", "engineered"}:
            raise ValueError(f"Unknown feature mode: {self.mode}")
        missing = sorted(set(RAW_FEATURES) - set(X.columns))
        if missing:
            raise ValueError(f"Missing raw feature columns: {missing}")
        self.n_features_in_ = len(RAW_FEATURES)
        return self

What does fitting verify?

• The constructor stores the configuration needed by estimator cloning.
• The mode check permits only raw or engineered.
• The column check requires all five verified raw inputs.
• The fitted feature count records the expected raw input width.

• Save src/feature_experiment.py.
• Confirm that the new guard and class import without changing the feature decision by running this command:

python -m src.feature_experiment plan

What should remain unchanged?

You should still see three supported transformations and one skipped transformation. The guard and fitting checks add enforcement without changing schema eligibility.

Transformer class fails to import?

Confirm that SchemaFeatureTransformer starts at the file's left indentation level.

Check that TransformerMixin and BaseEstimator remain imported from sklearn.base.

Need help isolating the issue? Help me debug the cloneable transformer constructor and fit method.

• Inside SchemaFeatureTransformer, add transform() immediately below fit() by pasting this method:

    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
        missing = sorted(set(RAW_FEATURES) - set(X.columns))
        if missing:
            raise ValueError(f"Missing raw feature columns: {missing}")
        output = X[list(RAW_FEATURES)].copy()
        if self.mode == "raw":
            return output
        for name in self.engineered_features:
            if name == "training_load_to_sleep_ratio":
                output[name] = output["weekly_km_to_date"] / (
                    output["sleep_hours_last_night"] + 1.0
                )
            elif name == "training_load_to_resting_hr_ratio":
                output[name] = output["weekly_km_to_date"] / (
                    output["resting_hr_morning"] + 1.0
                )
            elif name == "injury_soreness_interaction":
                output[name] = (
                    output["prior_injury"] * output["soreness_score_evening"]
                )
            else:
                raise ValueError(f"No verified transformation implementation: {name}")
        return output

How are transformations constrained?

• The method starts with a copy of the five verified raw features.
• Raw mode returns those columns without adding predictors.
• Engineered mode implements only the three schema-supported calculations.
• An unknown engineered feature raises an error instead of triggering a guessed calculation.

How should you interpret the ratios?

The ratios are experimental synthetic predictors for this educational comparison. They are not validated coaching rules or medical evidence.

• Save src/feature_experiment.py.
• Confirm that the executable transformer leaves the unsupported proposal inactive by running this command:

python -m src.feature_experiment plan

What does this check confirm?

The output still lists prior_week_load_ratio under skipped. The transformer contains no calculation that fabricates prior_week_km.

The schema decision and executable code now enforce the same boundary.

Unexpected transformation path?

Confirm that transform() contains branches for exactly three engineered feature names.

Check that the method aligns with fit() inside the class.

Need help reviewing the implementation? Help me verify that the transformer implements only the supported features.

Assemble the fold-local pipeline

A Pipeline packages feature transformation, imputation, selection, and classification into one cloneable scikit-learn estimator. A later fold fit can therefore learn every new preprocessing value from training runners only.

• Add the feature-name helper and pipeline builder below SchemaFeatureTransformer by pasting this code:

def feature_names(mode: str, engineered_features: tuple[str, ...]) -> list[str]:
    names = list(RAW_FEATURES)
    if mode == "engineered":
        names.extend(engineered_features)
    return names


def build_pipeline(mode: str, engineered_features: tuple[str, ...]) -> Pipeline:
    return Pipeline(
        steps=[
            (
                "features",
                SchemaFeatureTransformer(
                    mode=mode,
                    engineered_features=engineered_features,
                ),
            ),
            ("imputer", SimpleImputer(strategy="mean")),
            ("selector", SelectKBest(score_func=f_classif, k=SELECT_K)),
            ("classifier", DecisionTreeClassifier(random_state=RANDOM_STATE)),
        ]
    )

What does the pipeline control?

• The features step applies the selected schema-gated mode.
• The imputer step learns mean values during each future fold fit.
• The selector step scores candidates from each fold's training rows before keeping four features.
• The classifier step fixes the decision-tree seed at 42 for the paired comparison.

• Save src/feature_experiment.py.

Before the final check, which proposal do you expect the completed module to exclude?

• Run the completed feature plan by using this command:

python -m src.feature_experiment plan

What confirms success?

You should see training_load_to_sleep_ratio, training_load_to_resting_hr_ratio, and injury_soreness_interaction under supported.

You should see prior_week_load_ratio under skipped. Its missing_sources list should contain prior_week_km.

The successful output also proves that the immutable handoff audit still passes before feature planning begins.

You have closed the unsupported-feature gap. The model pipeline can now receive only schema-supported row-local predictors.

Final feature plan does not match?

Check that FEATURE_PROPOSALS contains four records. Confirm that the prediction-time schema supports every source for only three records.

If the module fails during import, compare the pipeline step names with features, imputer, selector, and classifier.

Still blocked? Help me debug the completed feature plan and pipeline.

✔️ Awesome, I've got everything!

Your saved module now contains the feature gate, three supported transformation implementations, and a cloneable fold-local pipeline.

ⓧ I'd like to double check the full code

from __future__ import annotations

import json
import sys
from collections import Counter
from hashlib import sha256
from itertools import combinations
from pathlib import Path
from statistics import fmean, pstdev
from typing import Any

import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin, clone
from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.impute import SimpleImputer
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score
from sklearn.pipeline import Pipeline
from sklearn.tree import DecisionTreeClassifier

from src.train_models import make_group_folds

SOURCE_DATASET_VERSION = "v1"
FEATURE_DATASET_VERSION = "v2"
RANDOM_STATE = 42
SELECT_K = 4
RAW_FEATURES = (
    "weekly_km_to_date",
    "sleep_hours_last_night",
    "resting_hr_morning",
    "soreness_score_evening",
    "prior_injury",
)
TARGET_COLUMN = "needs_recovery_next_day"
GROUP_COLUMN = "runner_id"
METRIC_NAMES = ("accuracy", "precision", "recall", "f1")
FORBIDDEN_HOLDOUT_USES = {
    "feature_selection",
    "threshold_tuning",
    "model_comparison",
    "model_selection",
}

TRAIN_PATH = Path("data/v1/processed/train.csv")
SCHEMA_PATH = Path("contracts/v1/schema_contract.json")
PREPROCESSING_PATH = Path("artifacts/v1/preprocessing.json")
LEVEL3_CONTRACT_PATH = Path("artifacts/v1/level3_evaluation_contract.json")
SELECTION_LOCK_PATH = Path("artifacts/v1/level3_selection_lock.json")
LEVEL4_CONTRACT_PATH = Path("artifacts/v1/level4_features_contract.json")
HOLDOUT_MARKER_PATH = Path("artifacts/v1/holdout_consumed.json")
EVALUATION_REPORT_PATH = Path("reports/v1/level3_evaluation.json")
V2_SCHEMA_PATH = Path("contracts/v2/feature_schema_contract.json")
V2_REPORT_PATH = Path("reports/v2/feature_experiment.json")
LEVEL5_HANDOFF_PATH = Path("artifacts/v2/level5_features_handoff.json")
DOC_PATH = Path("docs/level-4-features.md")

FEATURE_PROPOSALS = (
    {
        "name": "training_load_to_sleep_ratio",
        "sources": ("weekly_km_to_date", "sleep_hours_last_night"),
        "formula": "weekly_km_to_date / (sleep_hours_last_night + 1.0)",
    },
    {
        "name": "training_load_to_resting_hr_ratio",
        "sources": ("weekly_km_to_date", "resting_hr_morning"),
        "formula": "weekly_km_to_date / (resting_hr_morning + 1.0)",
    },
    {
        "name": "injury_soreness_interaction",
        "sources": ("prior_injury", "soreness_score_evening"),
        "formula": "prior_injury * soreness_score_evening",
    },
    {
        "name": "prior_week_load_ratio",
        "sources": ("weekly_km_to_date", "prior_week_km"),
        "formula": "weekly_km_to_date / (prior_week_km + 1.0)",
    },
)


def fingerprint(path: Path) -> str:
    return sha256(path.read_bytes()).hexdigest()


def read_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def write_json_exclusive(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("x", encoding="utf-8") as handle:
        json.dump(payload, handle, indent=2, sort_keys=True)
        handle.write("\n")


def audit_handoff(root: Path) -> dict[str, Any]:
    paths = {
        "train": root / TRAIN_PATH,
        "schema": root / SCHEMA_PATH,
        "preprocessing": root / PREPROCESSING_PATH,
        "level3_contract": root / LEVEL3_CONTRACT_PATH,
        "selection_lock": root / SELECTION_LOCK_PATH,
        "level4_contract": root / LEVEL4_CONTRACT_PATH,
        "holdout_marker": root / HOLDOUT_MARKER_PATH,
        "evaluation_report": root / EVALUATION_REPORT_PATH,
    }
    missing = [path.relative_to(root).as_posix() for path in paths.values() if not path.exists()]
    if missing:
        raise FileNotFoundError(f"Missing immutable handoff files: {missing}")

    schema = read_json(paths["schema"])
    preprocessing = read_json(paths["preprocessing"])
    level3 = read_json(paths["level3_contract"])
    lock = read_json(paths["selection_lock"])
    level4 = read_json(paths["level4_contract"])
    marker = read_json(paths["holdout_marker"])

    expected_inputs = level3["input_fingerprints"]
    actual_inputs = {
        "train": fingerprint(paths["train"]),
        "schema": fingerprint(paths["schema"]),
        "preprocessing": fingerprint(paths["preprocessing"]),
    }
    for name, actual in actual_inputs.items():
        if expected_inputs[name]["sha256"] != actual:
            raise ValueError(f"Immutable v1 fingerprint mismatch: {name}")

    selection_lock_sha = fingerprint(paths["selection_lock"])
    if level4["locked_selection"]["selection_lock_sha256"] != selection_lock_sha:
        raise ValueError("Level 3 selection lock fingerprint mismatch")
    marker_sha = fingerprint(paths["holdout_marker"])
    if level4["holdout_policy"]["marker_sha256"] != marker_sha:
        raise ValueError("Consumed-holdout marker fingerprint mismatch")
    report_sha = fingerprint(paths["evaluation_report"])
    expected_report_sha = level4["evaluation_evidence"]["json_report"]["sha256"]
    if expected_report_sha != report_sha:
        raise ValueError("Level 3 evaluation report fingerprint mismatch")
    if not marker["holdout_consumed"] or not level4["holdout_policy"]["holdout_consumed"]:
        raise ValueError("The v1 holdout must remain marked as consumed")
    if set(marker["forbidden_future_uses"]) != FORBIDDEN_HOLDOUT_USES:
        raise ValueError("Consumed-holdout restrictions changed")
    if set(level4["holdout_policy"]["forbidden_future_uses"]) != FORBIDDEN_HOLDOUT_USES:
        raise ValueError("Level 4 holdout restrictions changed")
    if tuple(schema["feature_columns"]) != RAW_FEATURES:
        raise ValueError("The repository schema no longer matches the verified Level 4 design")
    if tuple(preprocessing["feature_columns"]) != RAW_FEATURES:
        raise ValueError("The preprocessing feature order changed")
    if level4["target_column"] != TARGET_COLUMN or level4["group_column"] != GROUP_COLUMN:
        raise ValueError("Target or group column changed")
    if lock["selected_model"] != level4["locked_selection"]["model"]:
        raise ValueError("Locked model identity changed")

    return {
        "audit_status": "passed",
        "development_only": True,
        "holdout_consumed": True,
        "holdout_read": False,
        "training_data": {"path": TRAIN_PATH.as_posix(), "sha256": actual_inputs["train"]},
        "schema": {"path": SCHEMA_PATH.as_posix(), "sha256": actual_inputs["schema"]},
        "preprocessing": {
            "path": PREPROCESSING_PATH.as_posix(),
            "sha256": actual_inputs["preprocessing"],
            "inherited_fit_scope": preprocessing["fit_scope"],
        },
        "selection_lock": {"path": SELECTION_LOCK_PATH.as_posix(), "sha256": selection_lock_sha},
        "holdout_marker": {"path": HOLDOUT_MARKER_PATH.as_posix(), "sha256": marker_sha},
        "evaluation_report": {"path": EVALUATION_REPORT_PATH.as_posix(), "sha256": report_sha},
        "forbidden_holdout_uses": sorted(FORBIDDEN_HOLDOUT_USES),
    }


def prediction_time_features(schema: dict[str, Any]) -> set[str]:
    return {
        column["name"]
        for column in schema["columns"]
        if column["role"] == "feature" and column["available_at_prediction"] is True
    }


def resolve_feature_plan(schema: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
    allowed = prediction_time_features(schema)
    supported: list[dict[str, Any]] = []
    skipped: list[dict[str, Any]] = []
    for proposal in FEATURE_PROPOSALS:
        missing_sources = sorted(set(proposal["sources"]) - allowed)
        record = {
            "name": proposal["name"],
            "sources": list(proposal["sources"]),
            "formula": proposal["formula"],
        }
        if missing_sources:
            record["reason"] = "unsupported source columns"
            record["missing_sources"] = missing_sources
            skipped.append(record)
        else:
            record["status"] = "supported by prediction-time schema"
            supported.append(record)
    return {"supported": supported, "skipped": skipped}


def validate_model_inputs(candidate_columns: list[str], schema: dict[str, Any]) -> None:
    allowed = prediction_time_features(schema)
    rejected = sorted(set(candidate_columns) - allowed)
    if rejected:
        raise ValueError(f"Rejected non-prediction-time model inputs: {rejected}")


class SchemaFeatureTransformer(TransformerMixin, BaseEstimator):
    def __init__(self, *, mode: str, engineered_features: tuple[str, ...] = ()) -> None:
        self.mode = mode
        self.engineered_features = engineered_features

    def fit(self, X: pd.DataFrame, y: Any = None) -> SchemaFeatureTransformer:
        if self.mode not in {"raw", "engineered"}:
            raise ValueError(f"Unknown feature mode: {self.mode}")
        missing = sorted(set(RAW_FEATURES) - set(X.columns))
        if missing:
            raise ValueError(f"Missing raw feature columns: {missing}")
        self.n_features_in_ = len(RAW_FEATURES)
        return self

    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
        missing = sorted(set(RAW_FEATURES) - set(X.columns))
        if missing:
            raise ValueError(f"Missing raw feature columns: {missing}")
        output = X[list(RAW_FEATURES)].copy()
        if self.mode == "raw":
            return output
        for name in self.engineered_features:
            if name == "training_load_to_sleep_ratio":
                output[name] = output["weekly_km_to_date"] / (
                    output["sleep_hours_last_night"] + 1.0
                )
            elif name == "training_load_to_resting_hr_ratio":
                output[name] = output["weekly_km_to_date"] / (
                    output["resting_hr_morning"] + 1.0
                )
            elif name == "injury_soreness_interaction":
                output[name] = (
                    output["prior_injury"] * output["soreness_score_evening"]
                )
            else:
                raise ValueError(f"No verified transformation implementation: {name}")
        return output


def feature_names(mode: str, engineered_features: tuple[str, ...]) -> list[str]:
    names = list(RAW_FEATURES)
    if mode == "engineered":
        names.extend(engineered_features)
    return names


def build_pipeline(mode: str, engineered_features: tuple[str, ...]) -> Pipeline:
    return Pipeline(
        steps=[
            (
                "features",
                SchemaFeatureTransformer(
                    mode=mode,
                    engineered_features=engineered_features,
                ),
            ),
            ("imputer", SimpleImputer(strategy="mean")),
            ("selector", SelectKBest(score_func=f_classif, k=SELECT_K)),
            ("classifier", DecisionTreeClassifier(random_state=RANDOM_STATE)),
        ]
    )


def main() -> None:
    if len(sys.argv) != 2 or sys.argv[1] not in {"audit", "plan", "run", "document"}:
        raise SystemExit("Usage: python -m src.feature_experiment [audit|plan|run|document]")
    root = Path(__file__).resolve().parents[1]
    if sys.argv[1] == "audit":
        print(json.dumps(audit_handoff(root), indent=2, sort_keys=True))
    elif sys.argv[1] == "plan":
        audit_handoff(root)
        print(
            json.dumps(
                resolve_feature_plan(read_json(root / SCHEMA_PATH)),
                indent=2,
                sort_keys=True,
            )
        )


if __name__ == "__main__":
    main()

Use this reference to compare ordering, indentation, identifiers, and literal values. Keep the previously audited code unchanged wherever it already matches.

Your schema-gated transformer is ready. Next, you will fit fresh pipeline clones on identical runner-safe folds and measure the raw-versus-engineered difference.

Run the Paired Group-Safe Experiment

Your schema gate now permits three row-local synthetic predictors. It also records the unsupported prior-week ratio without fabricating its missing source.

The next risk comes from evaluation design. A paired GroupKFold experiment must hold runners, selector policy, and estimator settings constant so the feature variant remains the only changed condition.

In this step, get ready to:

• Measure raw and engineered variants on the same four grouped folds.
• Record fold-local metrics, generalization gaps, and feature stability.
• Write the v2 feature contract, experiment report, and Level 5 handoff.

Measure each grouped fold

A paired comparison reuses identical validation rows for both variants. Each fold receives a fresh cloned Pipeline so its imputer, selector, and classifier learn only from that fold's fit rows.

• Switch back to src/feature_experiment.py in VS Code.
• Find build_pipeline() in the editor.
• Add metric_values() below build_pipeline() by copying this code:

def metric_values(y_true: Any, y_pred: Any) -> dict[str, float]:
    return {
        "accuracy": round(float(accuracy_score(y_true, y_pred)), 6),
        "precision": round(float(precision_score(y_true, y_pred, zero_division=0.0)), 6),
        "recall": round(float(recall_score(y_true, y_pred, zero_division=0.0)), 6),
        "f1": round(float(f1_score(y_true, y_pred, zero_division=0.0)), 6),
    }

What Does This Function Measure?

• The function calculates accuracy plus positive-class precision, recall, and F1.
• The six-decimal rounding keeps generated evidence deterministic across repeated runs.
• The zero-division policy gives folds with no positive predictions a numeric score instead of a warning-dependent result.

• Add fold_signature() below metric_values() by copying this code:

def fold_signature(fit_indices: Any, validation_indices: Any) -> str:
    payload = {
        "fit": [int(value) for value in fit_indices.tolist()],
        "validation": [int(value) for value in validation_indices.tolist()],
    }
    return sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()

Why Fingerprint Each Fold?

• The payload captures the exact fit and validation row indices.
• The SHA-256 digest creates a compact signature for that split.
• Matching signatures later prove that raw and engineered variants used identical evidence.

• Save src/feature_experiment.py.
• Check that the file still parses through the existing feature-plan path by running:

python -m src.feature_experiment plan

You should see the same three supported transformations and the skipped prior_week_load_ratio. This confirms the new metric helpers did not disturb the audited plan path.

Does the Plan Command Stop?

Check the closing braces in metric_values() and the indentation inside fold_signature().

Need a second pair of eyes? Help me debug the new metric and fold-signature helpers in feature_experiment.py.

Per-fold results become useful evidence when you summarize their central behavior and spread. The accuracy gap also shows how far training performance sits above grouped validation performance.

• Add aggregate_folds() below fold_signature() by copying this code:

def aggregate_folds(folds: list[dict[str, Any]]) -> dict[str, Any]:
    mean_training = {
        metric: round(fmean(fold["training_metrics"][metric] for fold in folds), 6)
        for metric in METRIC_NAMES
    }
    mean_validation = {
        metric: round(fmean(fold["validation_metrics"][metric] for fold in folds), 6)
        for metric in METRIC_NAMES
    }
    validation_std = {
        metric: round(pstdev(fold["validation_metrics"][metric] for fold in folds), 6)
        for metric in METRIC_NAMES
    }
    gaps = {
        metric: round(mean_training[metric] - mean_validation[metric], 6)
        for metric in METRIC_NAMES
    }
    return {
        "mean_training_metrics": mean_training,
        "mean_validation_metrics": mean_validation,
        "validation_metric_std": validation_std,
        "generalization_gaps": gaps,
    }

What Does the Aggregate Preserve?

• The training and validation dictionaries retain each metric's observed fold mean.
• The validation standard deviation records how much each metric varies across the four grouped folds.
• Each generalization gap subtracts validation performance from training performance.

• Add feature_stability() below aggregate_folds() by copying this code:

def feature_stability(folds: list[dict[str, Any]]) -> dict[str, Any]:
    selected_sets = [set(fold["selected_features"]) for fold in folds]
    counts = Counter(feature for selected in selected_sets for feature in selected)
    pairs = list(combinations(selected_sets, 2))
    jaccards = [
        len(left.intersection(right)) / len(left.union(right))
        for left, right in pairs
    ]
    return {
        "selection_frequency": {
            name: round(count / len(folds), 6) for name, count in sorted(counts.items())
        },
        "mean_pairwise_jaccard": round(fmean(jaccards), 6) if jaccards else 1.0,
    }

What Is Feature Stability?

Selection frequency measures how often each candidate survives the fold-local selector. Mean pairwise Jaccard measures how much the selected sets overlap across folds.

A validation mean can look promising while selected features change sharply between folds. Stability exposes that hidden sensitivity.

• Save src/feature_experiment.py.
• Confirm the existing audited plan still runs by using:

python -m src.feature_experiment plan

You should again see three supported transformations and one skipped transformation. The successful output confirms that both summary helpers load correctly.

Seeing a Helper Error?

Check that aggregate_folds() and feature_stability() sit at the file's top indentation level.

Still blocked? Help me find the syntax problem in my aggregate_folds and feature_stability functions.

The evaluator now combines these controls inside each fold. It rejects runner overlap before fitting a fresh clone on the fit rows.

• Add evaluate_variant() below feature_stability() by copying this function:

def evaluate_variant(
    train: pd.DataFrame,
    folds: list[tuple[Any, Any]],
    mode: str,
    engineered_features: tuple[str, ...],
) -> dict[str, Any]:
    pipeline = build_pipeline(mode, engineered_features)
    candidate_names = feature_names(mode, engineered_features)
    fold_results: list[dict[str, Any]] = []
    for number, (fit_indices, validation_indices) in enumerate(folds, start=1):
        fit = train.iloc[fit_indices]
        validation = train.iloc[validation_indices]
        fit_groups = {str(value) for value in fit[GROUP_COLUMN].unique()}
        validation_groups = {str(value) for value in validation[GROUP_COLUMN].unique()}
        overlap = sorted(fit_groups.intersection(validation_groups))
        if overlap:
            raise ValueError(f"Runner overlap in fold {number}: {overlap}")
        model = clone(pipeline)
        model.fit(fit[list(RAW_FEATURES)], fit[TARGET_COLUMN])
        training_predictions = model.predict(fit[list(RAW_FEATURES)])
        validation_predictions = model.predict(validation[list(RAW_FEATURES)])
        selected_indices = model.named_steps["selector"].get_support(indices=True)
        selected = [candidate_names[int(index)] for index in selected_indices.tolist()]
        fold_results.append(
            {
                "fold": number,
                "fold_signature": fold_signature(fit_indices, validation_indices),
                "fit_runner_count": len(fit_groups),
                "validation_runner_count": len(validation_groups),
                "runner_overlap": overlap,
                "training_metrics": metric_values(fit[TARGET_COLUMN], training_predictions),
                "validation_metrics": metric_values(
                    validation[TARGET_COLUMN], validation_predictions
                ),
                "selected_features": selected,
                "imputer_statistics": [
                    round(float(value), 6)
                    for value in model.named_steps["imputer"].statistics_.tolist()
                ],
            }
        )
    return {
        "variant": mode,
        "candidate_features": candidate_names,
        "pipeline_steps": ["features", "imputer", "selector", "classifier"],
        "folds": fold_results,
        "aggregate": aggregate_folds(fold_results),
        "feature_stability": feature_stability(fold_results),
    }

How Does the Fold Boundary Hold?

• The function receives precomputed folds instead of generating a different set for each variant.
• The fit and validation runner sets must have an empty intersection before training starts.
• The cloned pipeline fits its transformer, imputer, selector, and classifier on fit rows only.
• The fitted selector exposes the feature names chosen inside that fold.

• Save src/feature_experiment.py.
• Check the evaluator's syntax through the existing plan command by running:

python -m src.feature_experiment plan

You should see the unchanged feature plan. The evaluator is now ready to receive one shared set of grouped folds.

Does the Evaluator Fail to Load?

Check that the final return dictionary stays inside evaluate_variant() while sitting outside its fold loop.

Need help tracing the nesting? Help me inspect the indentation in evaluate_variant without changing its fold-local behavior.

Write the paired experiment evidence

The experiment produces three linked JSON artifacts. Their fingerprints connect the feature policy, observed comparison, and Level 5 handoff without materializing a replacement dataset.

• Add build_v2_schema() below evaluate_variant() by copying this code:

def build_v2_schema(plan: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
    engineered = [record["name"] for record in plan["supported"]]
    return {
        "schema_version": "2.0.0",
        "dataset_version": FEATURE_DATASET_VERSION,
        "source_dataset_version": SOURCE_DATASET_VERSION,
        "prediction_timestamp": "20:00 local time on observation day D",
        "raw_feature_columns": list(RAW_FEATURES),
        "engineered_feature_columns": engineered,
        "candidate_feature_columns": [*RAW_FEATURES, *engineered],
        "target_column": TARGET_COLUMN,
        "group_column": GROUP_COLUMN,
        "supported_transformations": plan["supported"],
        "skipped_transformations": plan["skipped"],
        "feature_policy": "Only row-local information marked available at prediction time may enter a transformation.",
        "scientific_status": "Training-load ratios are experimental synthetic predictors, not validated coaching rules.",
        "materialized_dataset_created": False,
    }

Why Create a Schema Without a Dataset?

The contract records every valid candidate plus the skipped proposal. The transformer can reproduce those candidates inside each fold when needed.

The explicit scientific status keeps the synthetic ratios from being presented as validated coaching rules.

• Add run_experiment() below build_v2_schema() by copying this function:

def run_experiment(root: Path) -> dict[str, Any]:
    for relative in (V2_SCHEMA_PATH, V2_REPORT_PATH, LEVEL5_HANDOFF_PATH):
        if (root / relative).exists():
            raise FileExistsError(f"Refusing to overwrite Level 4 evidence: {relative}")
    audit = audit_handoff(root)
    schema = read_json(root / SCHEMA_PATH)
    plan = resolve_feature_plan(schema)
    engineered_features = tuple(record["name"] for record in plan["supported"])
    train = pd.read_csv(root / TRAIN_PATH)
    required = {*RAW_FEATURES, TARGET_COLUMN, GROUP_COLUMN}
    missing = sorted(required - set(train.columns))
    if missing:
        raise ValueError(f"Development data is missing columns: {missing}")

    folds = make_group_folds(train, schema)
    raw = evaluate_variant(train, folds, "raw", engineered_features)
    engineered = evaluate_variant(train, folds, "engineered", engineered_features)
    raw_signatures = [fold["fold_signature"] for fold in raw["folds"]]
    engineered_signatures = [fold["fold_signature"] for fold in engineered["folds"]]
    if raw_signatures != engineered_signatures:
        raise ValueError("Raw and engineered variants did not use identical folds")

    results = {"raw": raw, "engineered": engineered}
    selected_variant = max(
        results,
        key=lambda name: (
            results[name]["aggregate"]["mean_validation_metrics"]["accuracy"],
            -abs(results[name]["aggregate"]["generalization_gaps"]["accuracy"]),
            1 if name == "raw" else 0,
        ),
    )
    deltas = {
        metric: round(
            engineered["aggregate"]["mean_validation_metrics"][metric]
            - raw["aggregate"]["mean_validation_metrics"][metric],
            6,
        )
        for metric in METRIC_NAMES
    }

    v2_schema = build_v2_schema(plan)
    write_json_exclusive(root / V2_SCHEMA_PATH, v2_schema)
    report = {
        "report_version": "2.0.0",
        "experiment_id": "level-4-features-v2",
        "development_only": True,
        "consumed_v1_holdout_read": False,
        "audit": audit,
        "feature_plan": plan,
        "cross_validation": {
            "method": "GroupKFold",
            "n_splits": len(folds),
            "identical_fold_signatures": raw_signatures,
            "selection_data": "v1 development rows only",
        },
        "model_control": {
            "estimator": "DecisionTreeClassifier",
            "random_state": RANDOM_STATE,
            "select_k": SELECT_K,
            "selection_score": "f_classif",
            "fold_local_steps": ["imputer", "selector", "classifier"],
        },
        "results": results,
        "engineered_minus_raw_validation_deltas": deltas,
        "selected_variant": selected_variant,
        "selection_basis": "highest mean grouped validation accuracy; smaller absolute accuracy gap; raw on an exact tie",
        "v2_schema": {
            "path": V2_SCHEMA_PATH.as_posix(),
            "sha256": fingerprint(root / V2_SCHEMA_PATH),
        },
        "limitations": [
            "The v1 processed training data is an immutable inherited input whose Level 1 mean imputation was fitted on training runners before Level 2 cross-validation.",
            "Every new Level 4 imputer and selector is fitted inside each fold, but Level 4 does not reconstruct pre-imputation v1 rows.",
            "The data, target, and engineered ratios are synthetic educational constructs and are not medical or coaching evidence.",
            "No consumed v1 holdout label participates in this comparison or selection.",
        ],
    }
    write_json_exclusive(root / V2_REPORT_PATH, report)
    handoff = {
        "handoff_version": "2.0.0",
        "next_level": "Level 5",
        "source_dataset_version": SOURCE_DATASET_VERSION,
        "feature_dataset_version": FEATURE_DATASET_VERSION,
        "development_data": audit["training_data"],
        "feature_schema": {
            "path": V2_SCHEMA_PATH.as_posix(),
            "sha256": fingerprint(root / V2_SCHEMA_PATH),
        },
        "feature_experiment": {
            "path": V2_REPORT_PATH.as_posix(),
            "sha256": fingerprint(root / V2_REPORT_PATH),
        },
        "selected_variant": selected_variant,
        "selection_basis": report["selection_basis"],
        "selected_candidate_features": results[selected_variant]["candidate_features"],
        "holdout_policy": {
            "v1_holdout_consumed": True,
            "marker": audit["holdout_marker"],
            "forbidden_future_uses": sorted(FORBIDDEN_HOLDOUT_USES),
            "required_level5_evidence": "grouped development validation or a new untouched holdout",
        },
        "scientific_status": "Synthetic educational feature experiment, not a validated coaching system.",
    }
    write_json_exclusive(root / LEVEL5_HANDOFF_PATH, handoff)
    return report

How Does the Orchestrator Protect the Evidence?

• Exclusive output checks stop the run before any existing Level 4 evidence can be overwritten.
• The immutable handoff audit passes before the development CSV is loaded.
• The code calls make_group_folds() once before passing the same index pairs to both variants.
• The report fingerprints the v2 schema before the Level 5 handoff fingerprints both generated evidence files.

How Is the Variant Selected?

Mean grouped validation accuracy is the primary selection criterion. A smaller absolute accuracy gap breaks the next tie.

An exact remaining tie selects the raw variant. That final rule avoids claiming extra feature complexity without observed validation benefit.

• Save src/feature_experiment.py.

Does the File Stop Loading?

Check that both output dictionaries close before their corresponding exclusive-write calls.

For targeted help, ask me to compare your run_experiment function with the supplied Complete Code.

Run and verify the paired comparison

This run is the evidence-producing checkpoint. Before you run it, predict whether the engineered variant will improve every validation metric or produce a mixed result.

• Run the audited paired experiment with this command:

python -m src.feature_experiment run

You should see a selected variant, observed validation deltas, the report path, and the Level 5 handoff path. The result comes only from grouped development folds.

That is the core experiment complete. Both variants now have comparable fold signatures, fold-local fitted steps, and observed stability evidence.

Did the Run Stop Safely?

A missing or changed v1 artifact means the immutable audit blocked the experiment. Preserve the evidence boundary instead of regenerating or editing the protected artifact.

Existing Level 4 output files also stop exclusive creation. Help me determine why the paired experiment stopped without deleting or overwriting evidence.

• Select contracts/v2/feature_schema_contract.json in the VS Code Explorer sidebar.
• Confirm materialized_dataset_created is false.
• Select reports/v2/feature_experiment.json in the Explorer sidebar.
• Confirm the raw and engineered results each contain four folds.
• Confirm every runner_overlap list is empty.
• Confirm both variants correspond to the same four entries under identical_fold_signatures.
• Select artifacts/v2/level5_features_handoff.json in the Explorer sidebar.
• Confirm the handoff preserves the consumed-holdout restrictions.

The three files now form a reproducible chain from prediction-time schema policy to observed development evidence to the Level 5 contract.

✔️ Awesome, I've got everything!

Great. Double-check that src/feature_experiment.py is saved and keep the generated v2 evidence unchanged for the next step.

ⓧ I'd like to double check the full code

Compare the experiment section below with the functions you added after build_pipeline().

def metric_values(y_true: Any, y_pred: Any) -> dict[str, float]:
    return {
        "accuracy": round(float(accuracy_score(y_true, y_pred)), 6),
        "precision": round(float(precision_score(y_true, y_pred, zero_division=0.0)), 6),
        "recall": round(float(recall_score(y_true, y_pred, zero_division=0.0)), 6),
        "f1": round(float(f1_score(y_true, y_pred, zero_division=0.0)), 6),
    }


def fold_signature(fit_indices: Any, validation_indices: Any) -> str:
    payload = {
        "fit": [int(value) for value in fit_indices.tolist()],
        "validation": [int(value) for value in validation_indices.tolist()],
    }
    return sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()


def aggregate_folds(folds: list[dict[str, Any]]) -> dict[str, Any]:
    mean_training = {
        metric: round(fmean(fold["training_metrics"][metric] for fold in folds), 6)
        for metric in METRIC_NAMES
    }
    mean_validation = {
        metric: round(fmean(fold["validation_metrics"][metric] for fold in folds), 6)
        for metric in METRIC_NAMES
    }
    validation_std = {
        metric: round(pstdev(fold["validation_metrics"][metric] for fold in folds), 6)
        for metric in METRIC_NAMES
    }
    gaps = {
        metric: round(mean_training[metric] - mean_validation[metric], 6)
        for metric in METRIC_NAMES
    }
    return {
        "mean_training_metrics": mean_training,
        "mean_validation_metrics": mean_validation,
        "validation_metric_std": validation_std,
        "generalization_gaps": gaps,
    }


def feature_stability(folds: list[dict[str, Any]]) -> dict[str, Any]:
    selected_sets = [set(fold["selected_features"]) for fold in folds]
    counts = Counter(feature for selected in selected_sets for feature in selected)
    pairs = list(combinations(selected_sets, 2))
    jaccards = [
        len(left.intersection(right)) / len(left.union(right))
        for left, right in pairs
    ]
    return {
        "selection_frequency": {
            name: round(count / len(folds), 6) for name, count in sorted(counts.items())
        },
        "mean_pairwise_jaccard": round(fmean(jaccards), 6) if jaccards else 1.0,
    }


def evaluate_variant(
    train: pd.DataFrame,
    folds: list[tuple[Any, Any]],
    mode: str,
    engineered_features: tuple[str, ...],
) -> dict[str, Any]:
    pipeline = build_pipeline(mode, engineered_features)
    candidate_names = feature_names(mode, engineered_features)
    fold_results: list[dict[str, Any]] = []
    for number, (fit_indices, validation_indices) in enumerate(folds, start=1):
        fit = train.iloc[fit_indices]
        validation = train.iloc[validation_indices]
        fit_groups = {str(value) for value in fit[GROUP_COLUMN].unique()}
        validation_groups = {str(value) for value in validation[GROUP_COLUMN].unique()}
        overlap = sorted(fit_groups.intersection(validation_groups))
        if overlap:
            raise ValueError(f"Runner overlap in fold {number}: {overlap}")
        model = clone(pipeline)
        model.fit(fit[list(RAW_FEATURES)], fit[TARGET_COLUMN])
        training_predictions = model.predict(fit[list(RAW_FEATURES)])
        validation_predictions = model.predict(validation[list(RAW_FEATURES)])
        selected_indices = model.named_steps["selector"].get_support(indices=True)
        selected = [candidate_names[int(index)] for index in selected_indices.tolist()]
        fold_results.append(
            {
                "fold": number,
                "fold_signature": fold_signature(fit_indices, validation_indices),
                "fit_runner_count": len(fit_groups),
                "validation_runner_count": len(validation_groups),
                "runner_overlap": overlap,
                "training_metrics": metric_values(fit[TARGET_COLUMN], training_predictions),
                "validation_metrics": metric_values(
                    validation[TARGET_COLUMN], validation_predictions
                ),
                "selected_features": selected,
                "imputer_statistics": [
                    round(float(value), 6)
                    for value in model.named_steps["imputer"].statistics_.tolist()
                ],
            }
        )
    return {
        "variant": mode,
        "candidate_features": candidate_names,
        "pipeline_steps": ["features", "imputer", "selector", "classifier"],
        "folds": fold_results,
        "aggregate": aggregate_folds(fold_results),
        "feature_stability": feature_stability(fold_results),
    }


def build_v2_schema(plan: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
    engineered = [record["name"] for record in plan["supported"]]
    return {
        "schema_version": "2.0.0",
        "dataset_version": FEATURE_DATASET_VERSION,
        "source_dataset_version": SOURCE_DATASET_VERSION,
        "prediction_timestamp": "20:00 local time on observation day D",
        "raw_feature_columns": list(RAW_FEATURES),
        "engineered_feature_columns": engineered,
        "candidate_feature_columns": [*RAW_FEATURES, *engineered],
        "target_column": TARGET_COLUMN,
        "group_column": GROUP_COLUMN,
        "supported_transformations": plan["supported"],
        "skipped_transformations": plan["skipped"],
        "feature_policy": "Only row-local information marked available at prediction time may enter a transformation.",
        "scientific_status": "Training-load ratios are experimental synthetic predictors, not validated coaching rules.",
        "materialized_dataset_created": False,
    }


def run_experiment(root: Path) -> dict[str, Any]:
    for relative in (V2_SCHEMA_PATH, V2_REPORT_PATH, LEVEL5_HANDOFF_PATH):
        if (root / relative).exists():
            raise FileExistsError(f"Refusing to overwrite Level 4 evidence: {relative}")
    audit = audit_handoff(root)
    schema = read_json(root / SCHEMA_PATH)
    plan = resolve_feature_plan(schema)
    engineered_features = tuple(record["name"] for record in plan["supported"])
    train = pd.read_csv(root / TRAIN_PATH)
    required = {*RAW_FEATURES, TARGET_COLUMN, GROUP_COLUMN}
    missing = sorted(required - set(train.columns))
    if missing:
        raise ValueError(f"Development data is missing columns: {missing}")

    folds = make_group_folds(train, schema)
    raw = evaluate_variant(train, folds, "raw", engineered_features)
    engineered = evaluate_variant(train, folds, "engineered", engineered_features)
    raw_signatures = [fold["fold_signature"] for fold in raw["folds"]]
    engineered_signatures = [fold["fold_signature"] for fold in engineered["folds"]]
    if raw_signatures != engineered_signatures:
        raise ValueError("Raw and engineered variants did not use identical folds")

    results = {"raw": raw, "engineered": engineered}
    selected_variant = max(
        results,
        key=lambda name: (
            results[name]["aggregate"]["mean_validation_metrics"]["accuracy"],
            -abs(results[name]["aggregate"]["generalization_gaps"]["accuracy"]),
            1 if name == "raw" else 0,
        ),
    )
    deltas = {
        metric: round(
            engineered["aggregate"]["mean_validation_metrics"][metric]
            - raw["aggregate"]["mean_validation_metrics"][metric],
            6,
        )
        for metric in METRIC_NAMES
    }

    v2_schema = build_v2_schema(plan)
    write_json_exclusive(root / V2_SCHEMA_PATH, v2_schema)
    report = {
        "report_version": "2.0.0",
        "experiment_id": "level-4-features-v2",
        "development_only": True,
        "consumed_v1_holdout_read": False,
        "audit": audit,
        "feature_plan": plan,
        "cross_validation": {
            "method": "GroupKFold",
            "n_splits": len(folds),
            "identical_fold_signatures": raw_signatures,
            "selection_data": "v1 development rows only",
        },
        "model_control": {
            "estimator": "DecisionTreeClassifier",
            "random_state": RANDOM_STATE,
            "select_k": SELECT_K,
            "selection_score": "f_classif",
            "fold_local_steps": ["imputer", "selector", "classifier"],
        },
        "results": results,
        "engineered_minus_raw_validation_deltas": deltas,
        "selected_variant": selected_variant,
        "selection_basis": "highest mean grouped validation accuracy; smaller absolute accuracy gap; raw on an exact tie",
        "v2_schema": {
            "path": V2_SCHEMA_PATH.as_posix(),
            "sha256": fingerprint(root / V2_SCHEMA_PATH),
        },
        "limitations": [
            "The v1 processed training data is an immutable inherited input whose Level 1 mean imputation was fitted on training runners before Level 2 cross-validation.",
            "Every new Level 4 imputer and selector is fitted inside each fold, but Level 4 does not reconstruct pre-imputation v1 rows.",
            "The data, target, and engineered ratios are synthetic educational constructs and are not medical or coaching evidence.",
            "No consumed v1 holdout label participates in this comparison or selection.",
        ],
    }
    write_json_exclusive(root / V2_REPORT_PATH, report)
    handoff = {
        "handoff_version": "2.0.0",
        "next_level": "Level 5",
        "source_dataset_version": SOURCE_DATASET_VERSION,
        "feature_dataset_version": FEATURE_DATASET_VERSION,
        "development_data": audit["training_data"],
        "feature_schema": {
            "path": V2_SCHEMA_PATH.as_posix(),
            "sha256": fingerprint(root / V2_SCHEMA_PATH),
        },
        "feature_experiment": {
            "path": V2_REPORT_PATH.as_posix(),
            "sha256": fingerprint(root / V2_REPORT_PATH),
        },
        "selected_variant": selected_variant,
        "selection_basis": report["selection_basis"],
        "selected_candidate_features": results[selected_variant]["candidate_features"],
        "holdout_policy": {
            "v1_holdout_consumed": True,
            "marker": audit["holdout_marker"],
            "forbidden_future_uses": sorted(FORBIDDEN_HOLDOUT_USES),
            "required_level5_evidence": "grouped development validation or a new untouched holdout",
        },
        "scientific_status": "Synthetic educational feature experiment, not a validated coaching system.",
    }
    write_json_exclusive(root / LEVEL5_HANDOFF_PATH, handoff)
    return report

Your development-only comparison is now fingerprinted for review. Next, you will test its controls and turn the observed result into documentation for Level 5.

Test, Document, and Hand Off

Your paired experiment now compares both feature variants on the same four runner-safe folds. Its v2 report and Level 5 handoff preserve the consumed-holdout boundary.

A single successful experiment is weak evidence unless deterministic tests reproduce its controls. Reviewers also need a clear path from the observed results to the next versioned contract.

In this step, get ready to:

• Build a deterministic test suite for the immutable audit and paired experiment.
• Recheck the existing test suites without rerunning the consumed held-out evaluation.
• Generate the Level 4 documentation and prepare the Level 5 evidence for commit.

Build the deterministic test suite

The tests use temporary repository fixtures, so they can mutate fingerprints without touching your real evidence. They also rerun the experiment in isolated directories to check reproducibility.

• Switch back to the repository in VS Code from the previous step.
• Create tests/test_features.py inside the existing tests folder.
• Use the full-file reference below to add the deterministic Level 4 tests.

✔️ I added the test suite

Great. Save tests/test_features.py before running the suite.

ⓧ I'd like to double check the full code

import inspect
import json
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory

import pandas as pd

import src.feature_experiment as feature_experiment
from src.feature_experiment import (
    FORBIDDEN_HOLDOUT_USES,
    LEVEL5_HANDOFF_PATH,
    RAW_FEATURES,
    SCHEMA_PATH,
    V2_REPORT_PATH,
    audit_handoff,
    build_pipeline,
    fingerprint,
    read_json,
    resolve_feature_plan,
    run_experiment,
)


class FeatureExperimentTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temporary_directory = TemporaryDirectory()
        self.addCleanup(self.temporary_directory.cleanup)
        self.root = Path(self.temporary_directory.name)
        self.build_fixture(self.root)

    def write_json(self, path: Path, payload: dict) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(payload, indent=2), encoding="utf-8")

    def build_fixture(self, root: Path) -> None:
        rows = []
        for number in range(12):
            rows.append(
                {
                    "runner_id": f"runner-{number:02d}",
                    "weekly_km_to_date": 20.0 + number * 2.0,
                    "sleep_hours_last_night": 8.0 - (number % 4) * 0.5,
                    "resting_hr_morning": 45.0 + number % 5,
                    "soreness_score_evening": float(number % 10),
                    "prior_injury": int(number % 3 == 0),
                    "needs_recovery_next_day": int(number % 3 != 0),
                }
            )
        train_path = root / "data/v1/processed/train.csv"
        train_path.parent.mkdir(parents=True, exist_ok=True)
        pd.DataFrame(rows).to_csv(train_path, index=False)

        columns = [
            {"name": name, "role": "feature", "available_at_prediction": True}
            for name in RAW_FEATURES
        ]
        columns.extend(
            [
                {"name": "runner_id", "role": "group_identifier", "available_at_prediction": True},
                {"name": "needs_recovery_next_day", "role": "target", "available_at_prediction": False},
            ]
        )
        schema = {
            "schema_version": "1.0.0",
            "dataset_version": "v1",
            "feature_columns": list(RAW_FEATURES),
            "target_column": "needs_recovery_next_day",
            "group_column": "runner_id",
            "columns": columns,
        }
        schema_path = root / "contracts/v1/schema_contract.json"
        preprocessing_path = root / "artifacts/v1/preprocessing.json"
        self.write_json(schema_path, schema)
        self.write_json(
            preprocessing_path,
            {
                "dataset_version": "v1",
                "fit_scope": "training runners only",
                "feature_columns": list(RAW_FEATURES),
            },
        )
        level3_path = root / "artifacts/v1/level3_evaluation_contract.json"
        self.write_json(
            level3_path,
            {
                "input_fingerprints": {
                    "train": {"path": "data/v1/processed/train.csv", "sha256": fingerprint(train_path)},
                    "schema": {"path": "contracts/v1/schema_contract.json", "sha256": fingerprint(schema_path)},
                    "preprocessing": {"path": "artifacts/v1/preprocessing.json", "sha256": fingerprint(preprocessing_path)},
                }
            },
        )
        lock_path = root / "artifacts/v1/level3_selection_lock.json"
        self.write_json(lock_path, {"selected_model": "decision_tree"})
        marker_path = root / "artifacts/v1/holdout_consumed.json"
        self.write_json(
            marker_path,
            {
                "holdout_consumed": True,
                "forbidden_future_uses": sorted(FORBIDDEN_HOLDOUT_USES),
            },
        )
        evaluation_path = root / "reports/v1/level3_evaluation.json"
        self.write_json(evaluation_path, {"holdout_consumed": True})
        self.write_json(
            root / "artifacts/v1/level4_features_contract.json",
            {
                "locked_selection": {
                    "model": "decision_tree",
                    "selection_lock_sha256": fingerprint(lock_path),
                },
                "holdout_policy": {
                    "holdout_consumed": True,
                    "marker_sha256": fingerprint(marker_path),
                    "forbidden_future_uses": sorted(FORBIDDEN_HOLDOUT_USES),
                },
                "evaluation_evidence": {
                    "json_report": {"sha256": fingerprint(evaluation_path)}
                },
                "target_column": "needs_recovery_next_day",
                "group_column": "runner_id",
            },
        )

    def test_immutable_audit_passes_and_rejects_drift(self) -> None:
        self.assertEqual(audit_handoff(self.root)["audit_status"], "passed")
        train_path = self.root / "data/v1/processed/train.csv"
        train_path.write_text(train_path.read_text(encoding="utf-8") + "\n", encoding="utf-8")
        with self.assertRaises(ValueError):
            audit_handoff(self.root)

    def test_schema_gate_supports_three_and_skips_missing_source(self) -> None:
        plan = resolve_feature_plan(read_json(self.root / SCHEMA_PATH))
        self.assertEqual(len(plan["supported"]), 3)
        self.assertEqual(plan["skipped"][0]["name"], "prior_week_load_ratio")
        self.assertEqual(plan["skipped"][0]["missing_sources"], ["prior_week_km"])

    def test_pipeline_owns_imputation_selection_and_classifier(self) -> None:
        plan = resolve_feature_plan(read_json(self.root / SCHEMA_PATH))
        engineered = tuple(record["name"] for record in plan["supported"])
        pipeline = build_pipeline("engineered", engineered)
        self.assertEqual(
            list(pipeline.named_steps),
            ["features", "imputer", "selector", "classifier"],
        )

    def test_variants_reuse_identical_group_folds(self) -> None:
        report = run_experiment(self.root)
        raw = [fold["fold_signature"] for fold in report["results"]["raw"]["folds"]]
        engineered = [
            fold["fold_signature"]
            for fold in report["results"]["engineered"]["folds"]
        ]
        self.assertEqual(raw, engineered)
        self.assertTrue(
            all(
                fold["runner_overlap"] == []
                for variant in report["results"].values()
                for fold in variant["folds"]
            )
        )

    def test_report_and_level5_handoff_are_deterministic(self) -> None:
        first_report = run_experiment(self.root)
        with TemporaryDirectory() as second_directory:
            second_root = Path(second_directory)
            self.build_fixture(second_root)
            second_report = run_experiment(second_root)
            self.assertEqual(
                json.dumps(first_report, sort_keys=True),
                json.dumps(second_report, sort_keys=True),
            )
            self.assertEqual(
                (self.root / LEVEL5_HANDOFF_PATH).read_text(encoding="utf-8"),
                (second_root / LEVEL5_HANDOFF_PATH).read_text(encoding="utf-8"),
            )

    def test_source_and_handoff_preserve_consumed_holdout_boundary(self) -> None:
        source = inspect.getsource(feature_experiment)
        self.assertNotIn("test.csv", source)
        run_experiment(self.root)
        report = read_json(self.root / V2_REPORT_PATH)
        handoff = read_json(self.root / LEVEL5_HANDOFF_PATH)
        self.assertFalse(report["consumed_v1_holdout_read"])
        self.assertEqual(
            set(handoff["holdout_policy"]["forbidden_future_uses"]),
            FORBIDDEN_HOLDOUT_USES,
        )


if __name__ == "__main__":
    unittest.main()

What does this test suite prove?

• The temporary fixture recreates the minimum immutable handoff needed by audit_handoff().
• The drift test changes the training bytes after recording their SHA-256 fingerprint. The audit must reject that mismatch.
• The pipeline test proves the order remains features, imputer, selector, and classifier.
• The determinism test runs the experiment in two independent fixtures. It requires identical reports and Level 5 handoffs.
• The boundary test rejects any source reference to test.csv while preserving every forbidden future holdout use.

• Save tests/test_features.py.

Before you run the suite, do you expect the temporary fingerprint mutation to be rejected without changing the real v1 files?

• Run the Level 4 tests from the repository folder by running:

python -m unittest -v tests/test_features.py

What should you see?

You should see six named tests complete. The final status should be OK.

That result proves the controls survive clean temporary runs. Your private v1 evidence remains untouched.

Tests not finishing with OK?

Check that tests/test_features.py matches the full-file reference. A missing fixture artifact can cause the audit test to stop early.

Make sure you run the command from the repository folder. Python needs that location to import src.feature_experiment.

Still stuck? Help me diagnose the failing Level 4 feature tests.

Recheck the complete test boundary

The new suite covers Level 4 in isolation. The repository also needs its earlier unit tests to keep passing after the feature module changes.

The Level 3 held-out evaluation remains consumed evidence. Its evaluation command stays closed during this regression check.

Before you rerun the earlier suites, do you expect the Level 4 additions to change any existing Level 1 through Level 3 behavior?

• Return to the terminal history that contains your existing Level 1 through Level 3 unit-test commands.
• Rerun each existing unit-test suite command.
• Skip the Level 3 held-out evaluation command.
• Confirm every existing suite finishes with OK.
• Rerun the Level 4 suite after the existing suites finish.

Why keep the evaluation closed?

Regression tests inspect code behavior with controlled fixtures. The consumed Level 3 holdout contains final evaluation evidence that cannot participate in feature selection or model comparison.

Keeping that command out of the loop preserves the evidence boundary you audited in Step 1.

Did an earlier suite fail?

Read the first failing test before changing code. A failure in an earlier suite can reveal an accidental change to an inherited contract.

Do not repair a failure by regenerating v1 data or consumed-holdout evidence.

Need help isolating the regression? Help me compare the failing earlier test with the Level 4 changes.

Generate the evidence-ready documentation

The experiment report is machine-readable evidence. The documentation layer turns its observed metrics and limitations into a reviewable project narrative.

The README update uses marker comments to protect the rest of the repository. Repeated documentation runs replace only the Level 4 marker block.

• In src/feature_experiment.py, add markdown_document() below run_experiment().
• Add update_readme() below markdown_document().
• Add document_experiment() below update_readme().
• Update main() with the final CLI routing shown in the full-file reference.

✔️ I added the documentation path

Good. Save src/feature_experiment.py before generating the documentation.

ⓧ I'd like to double check the new code

def markdown_document(report: dict[str, Any], handoff: dict[str, Any]) -> str:
    raw = report["results"]["raw"]["aggregate"]
    engineered = report["results"]["engineered"]["aggregate"]
    rows = []
    for name, aggregate in (("raw", raw), ("engineered", engineered)):
        validation = aggregate["mean_validation_metrics"]
        gaps = aggregate["generalization_gaps"]
        rows.append(
            f"| {name} | {validation['accuracy']:.3f} | {validation['precision']:.3f} | "
            f"{validation['recall']:.3f} | {validation['f1']:.3f} | {gaps['accuracy']:.3f} |"
        )
    table = "\n".join(rows)
    skipped = ", ".join(
        record["name"] for record in report["feature_plan"]["skipped"]
    )
    return f"""# Level 4: Features

This experiment compares raw and schema-gated engineered features using identical grouped development folds. The consumed v1 holdout was not read.

| Variant | Validation accuracy | Precision 1 | Recall 1 | F1 1 | Accuracy gap |
| --- | ---: | ---: | ---: | ---: | ---: |
{table}

Selected variant: `{report['selected_variant']}`

Mean pairwise feature stability:

- Raw: `{report['results']['raw']['feature_stability']['mean_pairwise_jaccard']}`
- Engineered: `{report['results']['engineered']['feature_stability']['mean_pairwise_jaccard']}`

Skipped unsupported transformations: `{skipped}`

The training-load ratios are experimental synthetic predictors. They are not validated coaching rules. The inherited v1 processed input remains subject to its documented Level 1 preprocessing history.

Level 5 handoff: `{LEVEL5_HANDOFF_PATH.as_posix()}`
Selected Level 5 variant: `{handoff['selected_variant']}`
"""


def update_readme(root: Path, document: str) -> Path:
    readme_path = root / "README.md"
    existing = readme_path.read_text(encoding="utf-8") if readme_path.exists() else ""
    start = "<!-- LEVEL-4-FEATURES:START -->"
    end = "<!-- LEVEL-4-FEATURES:END -->"
    section = f"{start}\n## Level 4: Features\n\n[Feature experiment](docs/level-4-features.md)\n\n{end}"
    if start in existing:
        if end not in existing:
            raise ValueError("Incomplete Level 4 README marker block")
        before = existing.split(start, 1)[0].rstrip()
        after = existing.split(end, 1)[1].lstrip()
        updated = f"{before}\n\n{section}\n" + (f"\n{after.rstrip()}\n" if after else "")
    else:
        prefix = existing.rstrip()
        updated = f"{prefix}\n\n{section}\n" if prefix else f"{section}\n"
    readme_path.write_text(updated, encoding="utf-8")
    return readme_path


def document_experiment(root: Path) -> list[str]:
    report = read_json(root / V2_REPORT_PATH)
    handoff = read_json(root / LEVEL5_HANDOFF_PATH)
    document = markdown_document(report, handoff)
    doc_path = root / DOC_PATH
    doc_path.parent.mkdir(parents=True, exist_ok=True)
    doc_path.write_text(document, encoding="utf-8")
    readme = update_readme(root, document)
    return [doc_path.relative_to(root).as_posix(), readme.relative_to(root).as_posix()]


def main() -> None:
    if len(sys.argv) != 2 or sys.argv[1] not in {"audit", "plan", "run", "document"}:
        raise SystemExit("Usage: python -m src.feature_experiment [audit|plan|run|document]")
    root = Path(__file__).resolve().parents[1]
    if sys.argv[1] == "audit":
        print(json.dumps(audit_handoff(root), indent=2, sort_keys=True))
    elif sys.argv[1] == "plan":
        audit_handoff(root)
        print(
            json.dumps(
                resolve_feature_plan(read_json(root / SCHEMA_PATH)),
                indent=2,
                sort_keys=True,
            )
        )
    elif sys.argv[1] == "run":
        report = run_experiment(root)
        print(
            json.dumps(
                {
                    "selected_variant": report["selected_variant"],
                    "validation_deltas": report[
                        "engineered_minus_raw_validation_deltas"
                    ],
                    "report": V2_REPORT_PATH.as_posix(),
                    "level5_handoff": LEVEL5_HANDOFF_PATH.as_posix(),
                },
                indent=2,
                sort_keys=True,
            )
        )
    else:
        print(json.dumps(document_experiment(root), indent=2))


if __name__ == "__main__":
    main()

What does this documentation path do?

• The markdown_document() function reads observed raw and engineered aggregates from the existing report.
• The generated table includes validation accuracy, positive-class precision, recall, F1, and the accuracy generalization gap.
• The update_readme() function creates or replaces exactly one marker-delimited Level 4 section.
• The document_experiment() function writes docs/level-4-features.md from existing evidence. It does not rerun the experiment.
• The final CLI branch exposes the document command without changing the audit, plan, or run paths.

• Save src/feature_experiment.py.

Before you generate the files, which two paths do you expect the command to print?

• Generate the documentation from the existing v2 evidence by running:

python -m src.feature_experiment document

What should you see?

You should see a JSON list containing docs/level-4-features.md and README.md.

The markdown file now records the observed comparison. The README contains one Level 4 marker block linking to it.

Documentation command failed?

Confirm reports/v2/feature_experiment.json and artifacts/v2/level5_features_handoff.json still exist. The command reads both files before writing documentation.

If the README marker block is incomplete, restore both marker lines before rerunning the command.

Need help tracing the generator? Help me debug the Level 4 documentation command.

• Open docs/level-4-features.md in the repository editor.
• Confirm the document shows one row for raw and one row for engineered.
• Confirm the document records both mean pairwise feature-stability values.
• Confirm the document lists prior_week_load_ratio as skipped.
• Open README.md in the repository editor.
• Confirm exactly one LEVEL-4-FEATURES marker block links to docs/level-4-features.md.
• Prepare the Level 4 source, tests, v2 contracts, report, documentation, README update, and Level 5 handoff for commit with your established Git workflow.
• Confirm the existing ignore policy still excludes private datasets and consumed-holdout evidence.

That completes the evidence chain. Your tests reproduce the controls while the generated documentation points Level 5 to development-only results.

Secret mission

Attack the Gate with a Development-Only Leakage Proxy

A target-derived feature can make grouped validation look perfect while destroying the experiment's credibility. Inject that proxy into an in-memory development copy, observe the naive failure, then prove the prediction-time schema gate rejects it before fitting.

Clean Up Your Resources

Clean Up Your Resources

Everything here stays on your computer at $0 ongoing cost. Choose whether to retain the Level 4 evidence, pause the existing virtual environment, or remove only the Level 4 additions.

Resources you used:

• Level 4 implementation files: src/feature_experiment.py plus tests/test_features.py.
• Versioned feature evidence: contracts/v2/feature_schema_contract.json, reports/v2/feature_experiment.json, plus artifacts/v2/level5_features_handoff.json.
• Level 4 documentation: docs/level-4-features.md plus the marker-delimited section in README.md.
• Secret Mission evidence: src/leakage_proxy_experiment.py, tests/test_leakage_proxy.py, plus reports/v2/leakage_proxy_experiment.json.

Keep everything running

No action needed. Choose this if you want to retain the evidence for Level 5 or future review.

• Your fingerprinted v2 schema remains available for the next level.
• Your raw-versus-engineered report remains available with its grouped validation metrics.
• Your Level 5 handoff continues to preserve the consumed-holdout restrictions.
• Your leakage proxy report remains clearly labeled as an intentionally invalid demonstration.
• Your local files remain on your computer without ongoing charges.

Pause - I'll come back to this later

Pausing leaves every repository file intact while ending the active virtual environment in your current terminal.

• Exit the active virtual environment by running this command:

deactivate

What does this command do?

The deactivate command exits the active virtual environment for the current terminal session. It leaves the existing .venv directory plus every repository file unchanged.

Virtual Environment Already Inactive?

If your shell cannot run the command, the existing environment is probably already inactive. You can leave the repository files as they are.

Help me confirm whether my Python virtual environment is active.

Delete - I don't want to use this again

Deleting creates a clean rollback to the pre-Level 4 repository state. This is permanent for uncommitted files, so the instructions isolate the Level 4 paths.

Protect the v1 Evidence

This cleanup removes only the Level 4 additions. Keep every existing v1 dataset, artifact, contract, plus report unchanged.

Main Level 4 files:

• Return to the Level 4 repository in VS Code.
• Delete src/feature_experiment.py from the repository file tree.
• Delete tests/test_features.py from the repository file tree.
• Delete contracts/v2/feature_schema_contract.json from the repository file tree.
• Delete reports/v2/feature_experiment.json from the repository file tree.
• Delete artifacts/v2/level5_features_handoff.json from the repository file tree.
• Delete docs/level-4-features.md from the repository file tree.

Secret Mission files:

• Delete src/leakage_proxy_experiment.py from the repository file tree.
• Delete tests/test_leakage_proxy.py from the repository file tree.
• Delete reports/v2/leakage_proxy_experiment.json from the repository file tree.

README cleanup:

• In README.md, remove only the content from <!-- LEVEL-4-FEATURES:START --> through <!-- LEVEL-4-FEATURES:END -->.
• Save README.md.
• Confirm the nine deleted files no longer appear in the repository file tree.
• Confirm README.md contains no Level 4 marker block.

The Level 4 additions are gone. Your immutable v1 evidence remains in place.

Nice Work!

Nice Work!

You did it. You built a reproducible Level 4 feature-engineering experiment that preserves the consumed-holdout boundary.

You've learned how to:

• Protect the immutable evidence boundary with a fail-closed SHA-256 audit. Preserve every consumed-holdout restriction before experimentation.
• Build a prediction-time schema gate that derives only supported synthetic predictors. Record prior_week_load_ratio as skipped because prior_week_km is absent. Compare raw features with engineered features through group-safe cross-validation on identical runner-safe folds.
• Keep imputation inside each training fold. Keep feature selection inside each training fold. Publish observed metrics. Measure generalization gaps. Report feature stability. Back the result with deterministic tests. Generate reviewer-ready documentation. Create a fingerprinted Level 5 handoff.
• Secret Mission: Demonstrated how development_only_leakage_proxy produces a perfect but invalid development score. Proved the repaired schema gate rejects the proxy before fitting. Preserved the consumed v1 holdout throughout the attack.

Ready to quiz yourself?