Build a Group-Safe Data Handoff
Create a leakage-safe, versioned running data handoff for model training.
Introduction
30 Second Summary
A runner's daily record grows throughout the day while tomorrow's recovery need remains unknown. Mixing later information into an earlier decision makes the result look more trustworthy than it is.
In this project, you will build the v1 data handoff for a future next-day running-readiness classifier. The handoff keeps future information and held-out runners out of learned preprocessing state.
What You'll Build
You will run one local pipeline that turns deliberately dirty runner-day records into a tested v1 handoff that Level 2 can load directly.
By the end of this project, you'll have:
- A prediction-time contract that identifies which runner-day measurements are available at 20:00 on day D.
- A data leakage experiment that shows why global imputation changes when held-out runner values change.
- A versioned Level 2 handoff whose seven automated tests prove that its outputs remain deterministic and group-safe.
- Secret Mission: A schema drift gate that accepts the normal v1 test batch but rejects a malformed batch with one missing feature plus one unexpected future column.
Are there any prerequisites?
You need Python 3.12 or newer plus VS Code. Step 1 installs the pinned data libraries inside a local virtual environment at no cost.
The synthetic target and ranges are educational examples with no medical or coaching purpose.
Before We Start
Before any code, this checkpoint locks in the prediction question. It also fixes the time boundary and runner-level holdout rule that prevent data leakage.
Set Up the Reproducible Data Lab
Your prediction contract now defines the boundary between valid features and future information. The next risk is a data experiment that works once but cannot be recreated.
A reproducible lab locks the Python runtime to exact dependency versions. Deterministic generation controls the dataset itself.
Stable versioned paths give every later run the same destinations. Together, these controls make the complete environment repeatable on another machine.
In this step, get ready to:
- Create the versioned level-1-data workspace.
- Isolate Python inside an activated .venv environment.
- Install the exact data-library versions required by the pipeline.
Create the versioned workspace
A structured workspace gives source code plus every versioned output a fixed home. VS Code keeps the complete layout visible in its Explorer view.
- Press Cmd+Space on macOS or the Windows key on Windows to open system search.
- Type VS Code into the search bar.
- Press Enter to open VS Code.
- Select File from the top menu.
- Select Open Folder....
- Choose your Desktop as the location.
- Select New Folder.
- Enter level-1-data as the folder name.
- Select Open to load the new folder as your workspace.
- Select Yes, I trust the authors if the Workspace Trust dialog appears.
You now have a dedicated level-1-data workspace on your Desktop. The Explorer sidebar should show that folder name.
- Select View from the top menu.
- Select Terminal to open the integrated terminal inside level-1-data.
- Create the reserved directories and starter files by running these commands:
mkdir -p src tests data/v1 artifacts/v1 reports/v1 contracts/v1 evidence/v1
touch requirements.txt .gitignore README.md src/__init__.py tests/__init__.py
ls -a src tests data/v1 artifacts/v1 reports/v1 contracts/v1 evidence/v1What do these commands create?
- The first command reserves separate paths for source code and tests.
- It also reserves versioned paths for data artifacts and reports.
- The second command creates the three setup documents.
- It also creates the empty Python package markers in src and tests.
- The final command lists each path so you can confirm that it exists.
The terminal should print a heading for every reserved directory. Empty directories show only their current-directory entries.
- Expand src in the Explorer sidebar to confirm that src/__init__.py exists.
- Expand tests to confirm that tests/__init__.py exists.
- Expand each output directory to confirm that it contains a v1 folder.
Missing a directory or file?
Confirm that the terminal prompt ends in level-1-data. The VS Code terminal starts inside the folder shown in Explorer.
Compare any missing name with the creation command above. Help me diagnose my project structure.
Check Python and install the pinned dependencies
Python 3.12 is the compatibility floor for this lab. The pinned NumPy release requires that version or newer.
A virtual environment gives this project its own interpreter state. Packages installed there stay separate from your other Python projects.
- Check the Python version available to this workspace by running this command:
python --versionWhat does this check show?
The version option prints the interpreter version used by the python command. That interpreter creates the project environment in the next action.
✔️ I see version 3.12 or higher
Your interpreter meets the dependency requirement. Continue with the environment creation below.
ⓧ I see an older version
The current interpreter cannot support every pinned dependency in this project. Install Python 3.12 or newer before creating the environment.
- Visit the official Python downloads page.
- Download a macOS release at 3.12 or newer.
- Complete the guided installer.
- Return to VS Code.
- Close the existing integrated terminal with its trash icon.
- Select Terminal from the top menu.
- Select New Terminal.
- Repeat the version check shown above.
ⓧ Command not found
Your terminal cannot locate Python through the expected command. Install Python 3.12 or newer before continuing.
- Visit the official Python downloads page.
- Download a macOS release at 3.12 or newer.
- Complete the guided installer.
- Return to VS Code.
- Close the existing integrated terminal with its trash icon.
- Select Terminal from the top menu.
- Select New Terminal.
- Repeat the version check shown above.
Your compatible interpreter can now create the isolated runtime. Activation makes every later Python command use that runtime.
- Create the virtual environment and activate it by running these commands:
python -m venv .venv
source .venv/bin/activateHow does the environment work?
The first command creates the environment inside .venv. Its interpreter state belongs to this project.
The second command activates that environment in the current terminal. Later installations now target level-1-data/.venv.
- Confirm that the terminal prompt begins with the active environment name.
- Confirm that .venv appears in the Explorer sidebar.
The environment marker in the prompt confirms that isolation is active. Your pinned packages now have a dedicated installation location.
Environment name missing?
Confirm that the .venv directory exists in the Explorer sidebar. A missing directory means the creation command did not complete.
Repeat the activation command if the directory exists. Help me activate my virtual environment.
Exact dependency pins turn the runtime into a reproducible project input. A future installation receives the same numerical libraries used by your pipeline.
- Select requirements.txt in the Explorer sidebar.
- Paste these dependency pins into requirements.txt:
numpy==2.5.4
pandas==3.1.0
scikit-learn==1.9.1
Why pin these versions?
NumPy 2.5.4 supplies numerical operations for the future pipeline. pandas 3.1.0 supplies tabular data structures.
scikit-learn 1.9.1 supplies the group-aware split used later. Exact pins prevent those runtime inputs from changing silently.
- Save requirements.txt.
- Confirm that the editor shows exactly three dependency lines.
The first installation can take a few minutes while the packages download. Quiet periods during that download are normal.
- Install the pinned dependencies into the active environment by running this command:
python -m pip install -r requirements.txtWhat does this command install?
The pip module reads each exact pin from requirements.txt. It installs the matching packages into the active virtual environment.
The terminal should finish with a successful installation summary. Your isolated environment now contains the three libraries required by the data pipeline.
Dependency installation failed?
Confirm that the terminal prompt still shows the active environment name. A missing marker means the installation may be targeting a different interpreter.
A compatibility message can indicate an older Python version. Help me debug the dependency installation.
Document and verify the lab
The project still needs rules for generated Python files plus instructions for recreating the environment. These files preserve the boundary between source-controlled inputs and local runtime state.
Git ignore rules keep the environment plus compiled caches out of version control. The README records the Level 1 scope and local commands.
- Select .gitignore in the Explorer sidebar.
- Paste these exclusion patterns into .gitignore:
.venv/
__pycache__/
*.pyc
What do these rules exclude?
- The first rule excludes the local virtual environment.
- The second rule excludes Python bytecode cache directories.
- The final rule excludes compiled Python files.
- Save .gitignore.
- Confirm that the editor shows exactly three exclusion patterns.
Can't find the ignore file?
Confirm that the filename begins with one period. The VS Code Explorer displays dotfiles inside the open workspace.
Check that the file sits directly inside level-1-data. Help me locate my ignore file.
- Select README.md in the Explorer sidebar.
- Paste this project scope and local workflow into README.md:
# Level 1 Data: Time-Aware, Group-Safe Run Readiness Handoff
This project creates the `v1` data contract for a future next-day running-readiness classifier. It prepares and validates data only. Model fitting and baseline evaluation begin in Level 2.
The synthetic target and ranges are educational and are not medical or coaching advice.
## Run locally
```bash
python -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
```
What does the README preserve?
The opening defines Level 1 as a data-contract project. It reserves model fitting for Level 2.
The Markdown command block records the environment workflow you just completed. A future contributor can recreate the lab from the repository.
- Save README.md.
- Confirm that the editor shows the project title.
- Confirm that the editor shows the Run locally section.
README formatting looks broken?
Confirm that the command section begins with three backticks followed by bash. Confirm that another three backticks close the section.
Preserve the blank lines around each heading. Help me fix the README formatting.
That completes the reproducibility files. Your environment paths and local workflow are now explicit project artifacts.
✔️ Awesome, I've got everything!
Great. Confirm that every setup file is saved and the virtual environment remains active in the terminal.
ⓧ I'd like to double check the full code
Compare each populated setup file with the complete versions below. The src/__init__.py and tests/__init__.py files should both remain empty.
numpy==2.5.4
pandas==3.1.0
scikit-learn==1.9.1
Requirements file check
This file contains the three exact dependency pins used by the project environment.
.venv/
__pycache__/
*.pyc
Ignore file check
This file excludes the virtual environment plus generated Python cache files.
# Level 1 Data: Time-Aware, Group-Safe Run Readiness Handoff
This project creates the `v1` data contract for a future next-day running-readiness classifier. It prepares and validates data only. Model fitting and baseline evaluation begin in Level 2.
The synthetic target and ranges are educational and are not medical or coaching advice.
## Run locally
```bash
python -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
```
README file check
This file records the Level 1 scope plus the three local environment commands.
- Leave src/__init__.py empty.
- Leave tests/__init__.py empty.
Before you run the final check, do you expect the active environment to show flexible package versions or the three exact pins?
- Verify the installed packages from the active environment by running this command:
python -m pip listWhat does this final check prove?
The command lists packages installed for the active Python interpreter. Matching versions prove that the dependencies were installed inside the environment you activated.
You should see numpy 2.5.4, pandas 3.1.0, plus scikit-learn 1.9.1 in the package list.
Pinned package missing?
Confirm that the terminal prompt still shows the active environment name. Repeat the installation command if one of the three packages is absent.
A different version can indicate that the terminal is using another Python installation. Help me reconcile my package list.
Your reproducible data lab is ready. Next, you will define the runner-day rows and expose the dataset's intentional quality failures.
Define the Prediction Point and Expose Dirty Data
Your reproducible Python lab is ready. The pinned environment gives every learner the same foundation for generating the running dataset.
A runner-day row only makes sense when its decision time is explicit. Without a prediction timestamp, future information can slip into features and create data leakage.
In this step, get ready to:
- Define the runner-day prediction contract and its five allowed features.
- Generate repeated observations with deliberate data-quality defects.
- Write the raw dataset and reveal its failing quality audit.
Define the runner-day prediction contract
Each row represents one runner on one observation day. The application predicts at 20:00 on day D before the target becomes observable at 08:00 on D+1.
- Select the existing src directory in the VS Code Explorer.
- Click the New File button in the Explorer toolbar.
- Enter data_pipeline.py as the file name.
- Add the pipeline imports by pasting this code into src/data_pipeline.py:
from __future__ import annotations
import json
from pathlib import Path
import numpy as np
import pandas as pd
from sklearn.model_selection import GroupShuffleSplitWhat do these imports provide?
- The standard-library imports support JSON output and stable project paths.
- NumPy and pandas provide missing-value handling and tabular data operations.
- The GroupShuffleSplit import comes from scikit-learn. It supports the group-aware holdout added later in the pipeline.
- Save src/data_pipeline.py.
- Confirm data_pipeline.py appears beneath src in the Explorer.
Can't find the new file?
Check that data_pipeline.py is nested inside src. A file beside the src directory has the wrong path.
Use Help me place data_pipeline.py inside src. if your Explorer layout looks different.
The contract constants separate model features from identifiers and supervision metadata. This boundary keeps the target out of FEATURE_COLUMNS.
- Add the dataset identity and feature constants beneath the imports by pasting this code:
DATASET_VERSION = "v1"
SCHEMA_VERSION = "1.0.0"
RANDOM_STATE = 42
GROUP_COLUMN = "runner_id"
TARGET_COLUMN = "needs_recovery_next_day"
KEY_COLUMNS = ["runner_id", "observation_date"]
FEATURE_COLUMNS = [
"weekly_km_to_date",
"sleep_hours_last_night",
"resting_hr_morning",
"soreness_score_evening",
"prior_injury",
]What does this contract establish?
- The version constants identify the dataset and schema independently.
- The runner_id value identifies the runner group.
- The two key columns identify one runner-day observation.
- The five feature columns contain information available by the 20:00 prediction point.
- Save src/data_pipeline.py.
- Confirm TARGET_COLUMN does not appear inside FEATURE_COLUMNS.
- Add the complete column order and synthetic validity ranges beneath FEATURE_COLUMNS by pasting this code:
ALL_COLUMNS = [
"runner_id",
"observation_date",
"prediction_timestamp",
*FEATURE_COLUMNS,
TARGET_COLUMN,
"target_available_timestamp",
]
VALID_RANGES = {
"weekly_km_to_date": (0.0, 200.0),
"sleep_hours_last_night": (0.0, 14.0),
"resting_hr_morning": (30.0, 220.0),
"soreness_score_evening": (0.0, 10.0),
"prior_injury": (0.0, 1.0),
}Why keep columns and ranges together?
The ordered column list produces a stable CSV layout. The range dictionary gives the audit one source for detecting impossible synthetic values.
These ranges support an educational data-quality exercise. They do not provide medical or coaching guidance.
- Save src/data_pipeline.py.
- Confirm ALL_COLUMNS places the target after all five features.
The prediction contract also belongs in the project documentation. A reader should understand the time boundary before running the pipeline.
- Replace the contents of README.md with this cumulative version:
# Level 1 Data: Time-Aware, Group-Safe Run Readiness Handoff
This project creates the `v1` data contract for a future next-day running-readiness classifier. It prepares and validates data only. Model fitting and baseline evaluation begin in Level 2.
The synthetic target and ranges are educational and are not medical or coaching advice.
## Prediction contract
At 20:00 local time on observation day D, predict `needs_recovery_next_day` for D+1. The label becomes observable at 08:00 on D+1.
## Run locally
```bash
python -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python src/data_pipeline.py
```Why document the timestamps?
The prediction timestamp defines when feature information stops. The target-availability timestamp shows why the outcome belongs to the future.
This written boundary lets later contributors review features against the same decision point.
- Save README.md.
- Confirm the prediction contract names both 20:00 on day D and 08:00 on D+1.
Generate repeated runner-day observations
Repeated observations let you test whether a runner can leak across dataset partitions. The deterministic formulas produce the same 24 runners and six dates on every run.
- Add the first part of build_raw_dataset() beneath VALID_RANGES by pasting this code:
def build_raw_dataset() -> pd.DataFrame:
records: list[dict[str, object]] = []
for runner_number in range(1, 25):
baseline_hr = 45 + (runner_number % 10)
prior_injury = int(runner_number % 7 == 0)
for day in range(1, 7):
weekly_km = 18.0 + (runner_number % 6) * 3.0 + day * 2.0
sleep_hours = 5.5 + ((runner_number + day) % 6) * 0.5
resting_hr = baseline_hr + (day % 3) * 2
soreness = (runner_number * 2 + day) % 11
needs_recovery = int(
sleep_hours < 6.5
or soreness >= 8
or resting_hr >= baseline_hr + 4
or weekly_km >= 45.0
or (prior_injury == 1 and soreness >= 6)
)
return pd.DataFrame(records, columns=ALL_COLUMNS)How is each observation determined?
- The outer loop creates runner numbers from 1 through 24.
- The inner loop creates six observation days for each runner.
- The formulas derive stable feature values from the runner number and day.
- The synthetic target becomes positive when at least one recovery condition is true.
- Save src/data_pipeline.py.
- Confirm the editor shows one outer runner loop and one nested day loop.
The loops now calculate each runner-day. Each calculated observation needs a dictionary containing its identifiers, timestamps, features, and target.
- Insert this record-building statement after the needs_recovery calculation and before the final return statement:
records.append(
{
"runner_id": f"runner-{runner_number:02d}",
"observation_date": f"2026-01-{day:02d}",
"prediction_timestamp": (
f"2026-01-{day:02d}T20:00:00"
),
"weekly_km_to_date": weekly_km,
"sleep_hours_last_night": sleep_hours,
"resting_hr_morning": resting_hr,
"soreness_score_evening": soreness,
"prior_injury": prior_injury,
TARGET_COLUMN: needs_recovery,
"target_available_timestamp": (
f"2026-01-{day + 1:02d}T08:00:00"
),
}
)What does each record capture?
Every dictionary represents one runner-day. The prediction timestamp stays on day D at 20:00.
The target timestamp moves to the next calendar day at 08:00. That ordering makes the feature boundary testable.
- Save src/data_pipeline.py.
- Confirm the record dictionary contains every name from ALL_COLUMNS.
Controlled defects make the audit useful on its first run. The same rows receive the same defects because their positions are fixed.
- Insert the deliberate defect assignments immediately before the return pd.DataFrame(records, columns=ALL_COLUMNS) line:
records[7]["sleep_hours_last_night"] = None
records[61]["resting_hr_morning"] = None
records[103]["soreness_score_evening"] = None
records[15]["weekly_km_to_date"] = -3.0
records[95]["sleep_hours_last_night"] = 16.0
records.append(records[20].copy())What defects enter the raw data?
- Three feature values become missing.
- Two numeric feature values move outside their allowed ranges.
- One existing runner-day is copied to create a duplicate.
- The timestamp fields remain unchanged.
- Save src/data_pipeline.py.
- Confirm the six defect statements sit inside build_raw_dataset() immediately before its return statement.
Write and run the failing raw audit
A quality report turns vague dirtiness into measurable evidence. It checks completeness, key uniqueness, valid ranges, and temporal ordering separately.
- Add the audit calculations beneath build_raw_dataset() by pasting this code:
def quality_report(frame: pd.DataFrame) -> dict[str, int | float]:
missing_features = int(frame[FEATURE_COLUMNS].isna().sum().sum())
duplicate_records = int(frame.duplicated(subset=KEY_COLUMNS).sum())
invalid_values = 0
for column, (minimum, maximum) in VALID_RANGES.items():
invalid_values += int(
((frame[column] < minimum) | (frame[column] > maximum)).sum()
)
checked_values = frame.shape[0] * len(FEATURE_COLUMNS)
temporal_violations = int(
(
frame["prediction_timestamp"]
>= frame["target_available_timestamp"]
).sum()
)
return {
"row_count": int(frame.shape[0]),
}How does the audit separate failures?
- Missing-value detection counts empty cells across feature columns only.
- Duplicate detection uses the runner-day key.
- Range checks compare each feature with its declared minimum and maximum.
- Temporal validation counts rows whose target is available too early.
- Save src/data_pipeline.py.
- Confirm quality_report() appears beneath build_raw_dataset().
The report already calculates every audit value. Its return dictionary now needs to expose those values as JSON-friendly metrics.
- Replace the current one-metric return dictionary in quality_report() with this complete dictionary:
return {
"row_count": int(frame.shape[0]),
"runner_count": len(set(frame[GROUP_COLUMN])),
"missing_feature_values": missing_features,
"feature_completeness_ratio": round(
1 - missing_features / checked_values, 4
),
"duplicate_runner_days": duplicate_records,
"invalid_values": invalid_values,
"temporal_order_violations": temporal_violations,
"target_positive_rate": round(float(frame[TARGET_COLUMN].mean()), 4),
}What does the completed report show?
The report combines dataset size with separate quality dimensions. A passing temporal check cannot hide missing values or duplicate keys.
The target-positive rate describes the synthetic label balance. It does not evaluate a model.
- Save src/data_pipeline.py.
- Confirm the return dictionary exposes eight named metrics.
The final entry point creates the raw output directory when needed. It writes the generated table before printing its audit as formatted JSON.
- Add the runnable entry point at the bottom of src/data_pipeline.py by pasting this code:
def main() -> None:
project_root = Path(__file__).resolve().parents[1]
raw = build_raw_dataset()
raw_path = (
project_root
/ "data"
/ DATASET_VERSION
/ "raw"
/ "run_readiness_observations.csv"
)
raw_path.parent.mkdir(parents=True, exist_ok=True)
raw.to_csv(raw_path, index=False)
print(json.dumps(quality_report(raw), indent=2))
if __name__ == "__main__":
main()What happens when the script runs?
- The project path is resolved from the location of data_pipeline.py.
- The raw directory is created if it does not exist.
- The DataFrame is written without a pandas row index.
- The terminal receives the complete raw quality report.
- Save src/data_pipeline.py.
Before you run the audit, which checks do you expect to fail? Do you think timestamp ordering fails too?
- Generate the raw CSV and reveal the audit by running this command in the activated terminal:
python src/data_pipeline.pyWhat should you see?
The report shows 145 rows across 24 runners. It reports three missing feature values, one duplicate runner-day, and two invalid values.
The temporal-order count remains zero. This proves the prediction and target timestamps preserve the contract even while other quality checks fail.
You now have the intended failure evidence. The raw dataset exposes structural defects without violating the prediction-time boundary.
- Expand data/v1/raw in the VS Code Explorer.
- Confirm run_readiness_observations.csv appears inside the raw directory.
Audit command not working?
- Confirm the terminal prompt shows the activated virtual environment before rerunning the command.
- Check that main() sits below both function definitions.
- Compare the indentation around records.append() with the nested day loop.
- Use Help me debug the raw data pipeline. for targeted help.
✔️ Awesome, I've got everything!
Your raw CSV now contains 145 rows. Your terminal also shows the deliberate quality failures and zero temporal-order violations.
- Keep src/data_pipeline.py and README.md saved.
ⓧ I'd like to double check the full code
Compare your cumulative src/data_pipeline.py with this reference:
from __future__ import annotations
import json
from pathlib import Path
import numpy as np
import pandas as pd
from sklearn.model_selection import GroupShuffleSplit
DATASET_VERSION = "v1"
SCHEMA_VERSION = "1.0.0"
RANDOM_STATE = 42
GROUP_COLUMN = "runner_id"
TARGET_COLUMN = "needs_recovery_next_day"
KEY_COLUMNS = ["runner_id", "observation_date"]
FEATURE_COLUMNS = [
"weekly_km_to_date",
"sleep_hours_last_night",
"resting_hr_morning",
"soreness_score_evening",
"prior_injury",
]
ALL_COLUMNS = [
"runner_id",
"observation_date",
"prediction_timestamp",
*FEATURE_COLUMNS,
TARGET_COLUMN,
"target_available_timestamp",
]
VALID_RANGES = {
"weekly_km_to_date": (0.0, 200.0),
"sleep_hours_last_night": (0.0, 14.0),
"resting_hr_morning": (30.0, 220.0),
"soreness_score_evening": (0.0, 10.0),
"prior_injury": (0.0, 1.0),
}
def build_raw_dataset() -> pd.DataFrame:
records: list[dict[str, object]] = []
for runner_number in range(1, 25):
baseline_hr = 45 + (runner_number % 10)
prior_injury = int(runner_number % 7 == 0)
for day in range(1, 7):
weekly_km = 18.0 + (runner_number % 6) * 3.0 + day * 2.0
sleep_hours = 5.5 + ((runner_number + day) % 6) * 0.5
resting_hr = baseline_hr + (day % 3) * 2
soreness = (runner_number * 2 + day) % 11
needs_recovery = int(
sleep_hours < 6.5
or soreness >= 8
or resting_hr >= baseline_hr + 4
or weekly_km >= 45.0
or (prior_injury == 1 and soreness >= 6)
)
records.append(
{
"runner_id": f"runner-{runner_number:02d}",
"observation_date": f"2026-01-{day:02d}",
"prediction_timestamp": (
f"2026-01-{day:02d}T20:00:00"
),
"weekly_km_to_date": weekly_km,
"sleep_hours_last_night": sleep_hours,
"resting_hr_morning": resting_hr,
"soreness_score_evening": soreness,
"prior_injury": prior_injury,
TARGET_COLUMN: needs_recovery,
"target_available_timestamp": (
f"2026-01-{day + 1:02d}T08:00:00"
),
}
)
records[7]["sleep_hours_last_night"] = None
records[61]["resting_hr_morning"] = None
records[103]["soreness_score_evening"] = None
records[15]["weekly_km_to_date"] = -3.0
records[95]["sleep_hours_last_night"] = 16.0
records.append(records[20].copy())
return pd.DataFrame(records, columns=ALL_COLUMNS)
def quality_report(frame: pd.DataFrame) -> dict[str, int | float]:
missing_features = int(frame[FEATURE_COLUMNS].isna().sum().sum())
duplicate_records = int(frame.duplicated(subset=KEY_COLUMNS).sum())
invalid_values = 0
for column, (minimum, maximum) in VALID_RANGES.items():
invalid_values += int(
((frame[column] < minimum) | (frame[column] > maximum)).sum()
)
checked_values = frame.shape[0] * len(FEATURE_COLUMNS)
temporal_violations = int(
(
frame["prediction_timestamp"]
>= frame["target_available_timestamp"]
).sum()
)
return {
"row_count": int(frame.shape[0]),
"runner_count": len(set(frame[GROUP_COLUMN])),
"missing_feature_values": missing_features,
"feature_completeness_ratio": round(
1 - missing_features / checked_values, 4
),
"duplicate_runner_days": duplicate_records,
"invalid_values": invalid_values,
"temporal_order_violations": temporal_violations,
"target_positive_rate": round(float(frame[TARGET_COLUMN].mean()), 4),
}
def main() -> None:
project_root = Path(__file__).resolve().parents[1]
raw = build_raw_dataset()
raw_path = (
project_root
/ "data"
/ DATASET_VERSION
/ "raw"
/ "run_readiness_observations.csv"
)
raw_path.parent.mkdir(parents=True, exist_ok=True)
raw.to_csv(raw_path, index=False)
print(json.dumps(quality_report(raw), indent=2))
if __name__ == "__main__":
main()This reference matches the runnable pipeline state for this step.
The raw audit now separates data defects from timestamp validity. Next, you will encode that distinction in a schema contract and repair the structural failures.
Enforce the Data and Time Contract
Your raw runner-day dataset now exposes missing values, impossible ranges, and a duplicate record. The timestamp audit also confirms that every target becomes available after its prediction point.
Those checks cover observed values. A schema contract adds the structural rules that stop future information from entering silently. You will preserve unresolved missingness until the runner-level split can protect the held-out data.
In this step, get ready to:
- Define the column roles and prediction-time availability rules.
- Remove duplicate runner-days and replace impossible values with missing values.
- Export the cleaned dataset and its versioned schema contract.
Define the prediction schema
A schema records more than column names. It identifies each column's role. It also states whether that column is available when the prediction is made.
- In src/data_pipeline.py, find def build_raw_dataset().
- Insert the first part of build_schema_contract() directly above that function:
def build_schema_contract() -> dict[str, object]:
roles = {
"runner_id": "group_identifier",
"observation_date": "context",
"prediction_timestamp": "context",
TARGET_COLUMN: "target",
"target_available_timestamp": "target_metadata",
}
data_types = {
"runner_id": "string",
"observation_date": "string",
"prediction_timestamp": "string",
"weekly_km_to_date": "float",
"sleep_hours_last_night": "float",
"resting_hr_morning": "float",
"soreness_score_evening": "float",
"prior_injury": "integer",
TARGET_COLUMN: "integer",
"target_available_timestamp": "string",
}What does this code define?
- The roles mapping separates features from identifiers and supervision metadata.
- The data_types mapping gives future consumers a stable expectation for each field.
- The target receives its role through TARGET_COLUMN. This keeps the contract aligned with the pipeline constant.
- Continue build_schema_contract() by adding this column-specification loop below the data_types mapping:
columns = []
for name in ALL_COLUMNS:
role = roles.get(name, "feature")
columns.append(
{
"name": name,
"role": role,
"data_type": data_types[name],
"available_at_prediction": role
in {"group_identifier", "context", "feature"},
}
)How are feature-time roles assigned?
Columns without a special role become features. Features and context fields receive available_at_prediction as true.
The target and its availability timestamp receive false. That distinction prevents outcome information from being treated as a valid input.
- Complete build_schema_contract() by adding its returned contract below the loop:
return {
"schema_version": SCHEMA_VERSION,
"dataset_version": DATASET_VERSION,
"prediction_definition": {
"entity": "runner-day",
"prediction_timestamp": "20:00 local time on observation day D",
"target": TARGET_COLUMN,
"prediction_horizon": "next calendar day D+1",
"target_available_timestamp": (
"08:00 local time on next calendar day D+1"
),
},
"primary_key": KEY_COLUMNS,
"group_column": GROUP_COLUMN,
"feature_columns": FEATURE_COLUMNS,
"target_column": TARGET_COLUMN,
"required_columns": ALL_COLUMNS,
"columns": columns,
}What does the contract preserve?
- The prediction definition fixes the decision point at 20:00 on day D.
- The target availability definition fixes the outcome point at 08:00 on day D+1.
- The primary key identifies one record with runner_id plus observation_date.
- The group field keeps repeated observations from the same runner connected.
- Save src/data_pipeline.py.
- Confirm the completed function has valid syntax by running:
python src/data_pipeline.pyYou should see the existing raw audit finish with 145 rows. This confirms that the new contract definition parses without changing the raw dataset.
Seeing a syntax error in the contract?
- Check that the three snippets remain inside one build_schema_contract() function.
- Confirm that the columns loop and returned dictionary use four spaces of indentation.
- Ask for help with the contract indentation.
The contract describes the expected structure. The next check compares an observed pandas DataFrame against those required columns.
- In src/data_pipeline.py, find the end of build_raw_dataset().
- Add validate_schema() directly below that function:
def validate_schema(
frame: pd.DataFrame,
expected_columns: list[str] | None = None,
) -> dict[str, object]:
required = expected_columns or ALL_COLUMNS
expected = set(required)
observed = set(frame.columns)
missing = sorted(expected - observed)
unexpected = sorted(observed - expected)
return {
"valid": not missing and not unexpected,
"missing": missing,
"unexpected": unexpected,
}What does schema validation check?
The function compares expected column names with observed column names. It reports missing fields and unexpected fields separately.
The valid result becomes true only when both lists are empty. This gives downstream code a machine-readable gate.
- Save src/data_pipeline.py.
- Check the expanded file for syntax errors by running:
python src/data_pipeline.pyYou should see the raw audit complete again. The validator is now ready for the export path that you will wire in shortly.
Does the validator stop the script?
- Confirm that validate_schema() sits outside build_raw_dataset().
- Check that missing and unexpected are defined before the returned dictionary.
- Ask for help with the schema validator.
Clean structural defects without imputation
Duplicate runner-days and impossible numeric values violate the data contract. Unknown values have a different meaning. They must stay missing until preprocessing can be fitted on training runners only.
- In src/data_pipeline.py, find the end of quality_report().
- Add clean_structural_issues() directly below it:
def clean_structural_issues(frame: pd.DataFrame) -> pd.DataFrame:
cleaned = frame.drop_duplicates(subset=KEY_COLUMNS, keep="first").copy()
for column, (minimum, maximum) in VALID_RANGES.items():
out_of_range = (cleaned[column] < minimum) | (
cleaned[column] > maximum
)
cleaned.loc[out_of_range, column] = np.nan
return cleanedWhat does structural cleaning change?
- The duplicate check uses KEY_COLUMNS to retain one record per runner-day.
- The range checks inspect every field listed in VALID_RANGES.
- Values outside a valid range become np.nan. This preserves missingness for train-only preprocessing.
- The copied DataFrame keeps the original raw dataset unchanged.
The prediction contract also belongs in the project documentation. A reader should be able to identify every allowed input without opening the pipeline.
- In README.md, find the paragraph ending with 08:00 on D+1..
- Add this allowed-feature statement directly below that paragraph:
Allowed features are weekly distance through D, previous-night sleep, morning resting heart rate, evening soreness recorded by 20:00, and prior injury history.Why document the feature list?
The sentence gives future model developers the same feature boundary encoded in FEATURE_COLUMNS. Documentation and executable metadata now describe one prediction contract.
- Save README.md.
- Save src/data_pipeline.py.
- Confirm that the raw export still remains reproducible by running:
python src/data_pipeline.pyYou should still see 145 raw rows with one duplicate and two invalid values. That unchanged result proves the cleaner does not overwrite the raw evidence.
Did the raw audit change?
- Confirm that clean_structural_issues() starts by creating cleaned from a copy.
- Check that the function returns cleaned instead of modifying frame directly.
- Ask for help preserving the raw dataset.
Export and compare the versioned artifacts
The new functions become useful when one pipeline run applies them and records their evidence. The exported contract and cleaned CSV give the next level stable inputs.
- In src/data_pipeline.py, add build_outputs() below clean_structural_issues().
def build_outputs() -> dict[str, object]:
contract = build_schema_contract()
raw = build_raw_dataset()
cleaned = clean_structural_issues(raw)
report = {
"dataset_version": DATASET_VERSION,
"schema_validation": validate_schema(
raw, contract["required_columns"]
),
"prediction_definition": contract["prediction_definition"],
"raw_quality": quality_report(raw),
"cleaned_quality": quality_report(cleaned),
}
return {
"raw": raw,
"cleaned": cleaned,
"report": report,
"contract": contract,
}What does the output builder assemble?
The function creates the raw and cleaned datasets from the same deterministic source. It also validates the raw columns against the contract.
Both quality reports sit in one dictionary. Their side-by-side metrics make every structural repair visible.
- Save src/data_pipeline.py.
- Confirm that the file still parses before replacing its entry point by running:
python src/data_pipeline.pyYou should see the raw audit complete with 145 rows. The output builder is now ready to replace the earlier raw-only export path.
- At the bottom of src/data_pipeline.py, replace the current raw-only export section with the first part of run_pipeline() below:
def run_pipeline(project_root: Path) -> dict[str, object]:
outputs = build_outputs()
paths = {
"raw": project_root
/ "data"
/ DATASET_VERSION
/ "raw"
/ "run_readiness_observations.csv",
"cleaned": project_root
/ "data"
/ DATASET_VERSION
/ "cleaned"
/ "run_readiness_observations.csv",
"contract": project_root
/ "contracts"
/ DATASET_VERSION
/ "schema_contract.json",
}Why centralize the artifact paths?
The paths dictionary places every reusable output under v1. A later dataset version can use a new directory without overwriting this handoff.
- Complete the replacement by adding the export logic and entry point directly below the paths dictionary:
for path in paths.values():
path.parent.mkdir(parents=True, exist_ok=True)
for name in ("raw", "cleaned"):
outputs[name].to_csv(paths[name], index=False)
paths["contract"].write_text(
json.dumps(outputs["contract"], indent=2), encoding="utf-8"
)
return outputs
def main() -> None:
project_root = Path(__file__).resolve().parents[1]
outputs = run_pipeline(project_root)
print(json.dumps(outputs["report"], indent=2))
if __name__ == "__main__":
main()What does the runnable pipeline export?
- Each parent directory is created before its artifact is written.
- The raw and cleaned DataFrames are written as separate CSV files.
- The contract is serialized as formatted JSON.
- The terminal receives the schema result plus both quality reports.
Before you run the completed pipeline, which raw defects do you expect to disappear from the cleaned report?
- Save src/data_pipeline.py.
- Generate the versioned artifacts and reveal the cleaned audit by running:
python src/data_pipeline.pyYou should see schema_validation report valid as true. The cleaned report should show 144 rows with zero duplicate runner-days.
You should also see zero invalid values and zero temporal-order violations. Five missing feature values remain because the three original gaps now include the two impossible values converted to np.nan.
- In the VS Code Explorer sidebar, expand data/v1/cleaned to confirm run_readiness_observations.csv exists.
- In the same sidebar, expand contracts/v1 to confirm schema_contract.json exists.
Strong work. Your pipeline now repairs structural defects while preserving the missingness boundary needed for leakage-safe preprocessing.
Missing an export or seeing the wrong metrics?
- Confirm that run_pipeline() passes project_root into every path.
- Check that the cleaned CSV uses outputs["cleaned"] instead of outputs["raw"].
- Ask for help with the artifact export.
✔️ Awesome, I've got everything!
Your cleaned audit passes its structural and temporal checks. Double check that src/data_pipeline.py and README.md are saved.
ⓧ I'd like to double check the full code
Compare your cumulative src/data_pipeline.py with this Step 3 version:
from __future__ import annotations
import json
from pathlib import Path
import numpy as np
import pandas as pd
from sklearn.model_selection import GroupShuffleSplit
DATASET_VERSION = "v1"
SCHEMA_VERSION = "1.0.0"
RANDOM_STATE = 42
GROUP_COLUMN = "runner_id"
TARGET_COLUMN = "needs_recovery_next_day"
KEY_COLUMNS = ["runner_id", "observation_date"]
FEATURE_COLUMNS = [
"weekly_km_to_date",
"sleep_hours_last_night",
"resting_hr_morning",
"soreness_score_evening",
"prior_injury",
]
ALL_COLUMNS = [
"runner_id",
"observation_date",
"prediction_timestamp",
*FEATURE_COLUMNS,
TARGET_COLUMN,
"target_available_timestamp",
]
VALID_RANGES = {
"weekly_km_to_date": (0.0, 200.0),
"sleep_hours_last_night": (0.0, 14.0),
"resting_hr_morning": (30.0, 220.0),
"soreness_score_evening": (0.0, 10.0),
"prior_injury": (0.0, 1.0),
}
def build_schema_contract() -> dict[str, object]:
roles = {
"runner_id": "group_identifier",
"observation_date": "context",
"prediction_timestamp": "context",
TARGET_COLUMN: "target",
"target_available_timestamp": "target_metadata",
}
data_types = {
"runner_id": "string",
"observation_date": "string",
"prediction_timestamp": "string",
"weekly_km_to_date": "float",
"sleep_hours_last_night": "float",
"resting_hr_morning": "float",
"soreness_score_evening": "float",
"prior_injury": "integer",
TARGET_COLUMN: "integer",
"target_available_timestamp": "string",
}
columns = []
for name in ALL_COLUMNS:
role = roles.get(name, "feature")
columns.append(
{
"name": name,
"role": role,
"data_type": data_types[name],
"available_at_prediction": role
in {"group_identifier", "context", "feature"},
}
)
return {
"schema_version": SCHEMA_VERSION,
"dataset_version": DATASET_VERSION,
"prediction_definition": {
"entity": "runner-day",
"prediction_timestamp": "20:00 local time on observation day D",
"target": TARGET_COLUMN,
"prediction_horizon": "next calendar day D+1",
"target_available_timestamp": (
"08:00 local time on next calendar day D+1"
),
},
"primary_key": KEY_COLUMNS,
"group_column": GROUP_COLUMN,
"feature_columns": FEATURE_COLUMNS,
"target_column": TARGET_COLUMN,
"required_columns": ALL_COLUMNS,
"columns": columns,
}
def build_raw_dataset() -> pd.DataFrame:
records: list[dict[str, object]] = []
for runner_number in range(1, 25):
baseline_hr = 45 + (runner_number % 10)
prior_injury = int(runner_number % 7 == 0)
for day in range(1, 7):
weekly_km = 18.0 + (runner_number % 6) * 3.0 + day * 2.0
sleep_hours = 5.5 + ((runner_number + day) % 6) * 0.5
resting_hr = baseline_hr + (day % 3) * 2
soreness = (runner_number * 2 + day) % 11
needs_recovery = int(
sleep_hours < 6.5
or soreness >= 8
or resting_hr >= baseline_hr + 4
or weekly_km >= 45.0
or (prior_injury == 1 and soreness >= 6)
)
records.append(
{
"runner_id": f"runner-{runner_number:02d}",
"observation_date": f"2026-01-{day:02d}",
"prediction_timestamp": (
f"2026-01-{day:02d}T20:00:00"
),
"weekly_km_to_date": weekly_km,
"sleep_hours_last_night": sleep_hours,
"resting_hr_morning": resting_hr,
"soreness_score_evening": soreness,
"prior_injury": prior_injury,
TARGET_COLUMN: needs_recovery,
"target_available_timestamp": (
f"2026-01-{day + 1:02d}T08:00:00"
),
}
)
records[7]["sleep_hours_last_night"] = None
records[61]["resting_hr_morning"] = None
records[103]["soreness_score_evening"] = None
records[15]["weekly_km_to_date"] = -3.0
records[95]["sleep_hours_last_night"] = 16.0
records.append(records[20].copy())
return pd.DataFrame(records, columns=ALL_COLUMNS)
def validate_schema(
frame: pd.DataFrame,
expected_columns: list[str] | None = None,
) -> dict[str, object]:
required = expected_columns or ALL_COLUMNS
expected = set(required)
observed = set(frame.columns)
missing = sorted(expected - observed)
unexpected = sorted(observed - expected)
return {
"valid": not missing and not unexpected,
"missing": missing,
"unexpected": unexpected,
}
def quality_report(frame: pd.DataFrame) -> dict[str, int | float]:
missing_features = int(frame[FEATURE_COLUMNS].isna().sum().sum())
duplicate_records = int(frame.duplicated(subset=KEY_COLUMNS).sum())
invalid_values = 0
for column, (minimum, maximum) in VALID_RANGES.items():
invalid_values += int(
((frame[column] < minimum) | (frame[column] > maximum)).sum()
)
checked_values = frame.shape[0] * len(FEATURE_COLUMNS)
temporal_violations = int(
(
frame["prediction_timestamp"]
>= frame["target_available_timestamp"]
).sum()
)
return {
"row_count": int(frame.shape[0]),
"runner_count": len(set(frame[GROUP_COLUMN])),
"missing_feature_values": missing_features,
"feature_completeness_ratio": round(
1 - missing_features / checked_values, 4
),
"duplicate_runner_days": duplicate_records,
"invalid_values": invalid_values,
"temporal_order_violations": temporal_violations,
"target_positive_rate": round(float(frame[TARGET_COLUMN].mean()), 4),
}
def clean_structural_issues(frame: pd.DataFrame) -> pd.DataFrame:
cleaned = frame.drop_duplicates(subset=KEY_COLUMNS, keep="first").copy()
for column, (minimum, maximum) in VALID_RANGES.items():
out_of_range = (cleaned[column] < minimum) | (
cleaned[column] > maximum
)
cleaned.loc[out_of_range, column] = np.nan
return cleaned
def build_outputs() -> dict[str, object]:
contract = build_schema_contract()
raw = build_raw_dataset()
cleaned = clean_structural_issues(raw)
report = {
"dataset_version": DATASET_VERSION,
"schema_validation": validate_schema(
raw, contract["required_columns"]
),
"prediction_definition": contract["prediction_definition"],
"raw_quality": quality_report(raw),
"cleaned_quality": quality_report(cleaned),
}
return {
"raw": raw,
"cleaned": cleaned,
"report": report,
"contract": contract,
}
def run_pipeline(project_root: Path) -> dict[str, object]:
outputs = build_outputs()
paths = {
"raw": project_root
/ "data"
/ DATASET_VERSION
/ "raw"
/ "run_readiness_observations.csv",
"cleaned": project_root
/ "data"
/ DATASET_VERSION
/ "cleaned"
/ "run_readiness_observations.csv",
"contract": project_root
/ "contracts"
/ DATASET_VERSION
/ "schema_contract.json",
}
for path in paths.values():
path.parent.mkdir(parents=True, exist_ok=True)
for name in ("raw", "cleaned"):
outputs[name].to_csv(paths[name], index=False)
paths["contract"].write_text(
json.dumps(outputs["contract"], indent=2), encoding="utf-8"
)
return outputs
def main() -> None:
project_root = Path(__file__).resolve().parents[1]
outputs = run_pipeline(project_root)
print(json.dumps(outputs["report"], indent=2))
if __name__ == "__main__":
main()What should match?
Confirm that the contract functions appear before the reporting and export functions. Also confirm that the runnable entry point prints the report returned by run_pipeline().
Your cumulative README.md should now contain this content:
# Level 1 Data: Time-Aware, Group-Safe Run Readiness Handoff
This project creates the `v1` data contract for a future next-day running-readiness classifier. It prepares and validates data only. Model fitting and baseline evaluation begin in Level 2.
The synthetic target and ranges are educational and are not medical or coaching advice.
## Prediction contract
At 20:00 local time on observation day D, predict `needs_recovery_next_day` for D+1. The label becomes observable at 08:00 on D+1.
Allowed features are weekly distance through D, previous-night sleep, morning resting heart rate, evening soreness recorded by 20:00, and prior injury history.
## Run locally
```bash
python -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python src/data_pipeline.py
```What changed in the documentation?
The prediction section now lists the five permitted feature categories. The setup commands remain unchanged.
Your versioned data contract now separates structural repair from learned preprocessing. Next, you will deliberately contaminate a global statistic before building the runner-safe split that fixes it.
Split by Runner and Fit Preprocessing Safely
Your cleaned dataset now respects the structural contract. Its unresolved missing values create the next leakage risk because any statistics learned from every runner would include information from the holdout set.
In this step, you will expose that data leakage path. You will use scikit-learn to keep complete runner groups on one side of the split. You will then fit imputation values from training runners only.
In this step, get ready to:
- Reproduce a global preprocessing fit that changes when held-out runner data changes.
- Separate complete runners with a deterministic group-aware split.
- Export processed datasets with train-fitted imputation state and zero runner overlap.
Split complete runner groups
Each runner contributes six related observations. A row-level split could place one runner's history in both partitions. GroupShuffleSplit assigns complete runner groups to one side of the holdout boundary.
- In src/data_pipeline.py, add the group-aware splitter import below the existing pandas import:
from sklearn.model_selection import GroupShuffleSplitWhy use a group-aware split?
The splitter receives one group label for every row. Its configured holdout size applies to runner groups. This keeps a runner's repeated observations together.
- In src/data_pipeline.py, find the closing line of clean_structural_issues().
- Add the group-splitting function directly below it by pasting this code:
def split_by_runner(
cleaned: pd.DataFrame,
) -> tuple[pd.DataFrame, pd.DataFrame]:
splitter = GroupShuffleSplit(
n_splits=1,
test_size=0.25,
random_state=RANDOM_STATE,
)
train_indices, test_indices = next(
splitter.split(
cleaned,
cleaned[TARGET_COLUMN],
groups=cleaned[GROUP_COLUMN],
)
)
return (
cleaned.iloc[train_indices].copy(),
cleaned.iloc[test_indices].copy(),
)What does this split do?
- The groups argument identifies every row belonging to the same runner.
- The test_size value reserves one quarter of the runner groups for testing.
- The integer random_state makes the same runner assignment reproducible.
- The two iloc selections turn the generated row positions into independent DataFrames.
- Add the preprocessing helpers directly below split_by_runner() by pasting this code:
def fit_preprocessing(frame: pd.DataFrame) -> pd.Series:
return frame[FEATURE_COLUMNS].mean()
def apply_preprocessing(
frame: pd.DataFrame, fill_values: pd.Series
) -> pd.DataFrame:
ready = frame.copy()
ready[FEATURE_COLUMNS] = ready[FEATURE_COLUMNS].fillna(fill_values)
return ready
def as_float_dict(values: pd.Series) -> dict[str, float]:
return {name: float(value) for name, value in values.items()}How does the preprocessing state flow?
- The fit_preprocessing() function learns one mean for each allowed feature.
- The apply_preprocessing() function copies a partition before filling its missing feature values.
- The same training-fitted Series is applied to both partitions.
- The as_float_dict() helper converts the learned values into JSON-compatible numbers.
Seeing a splitter or import error?
- Confirm the virtual environment from the earlier setup step remains activated.
- Check that the import uses sklearn.model_selection exactly.
- Compare the group argument with groups=cleaned[GROUP_COLUMN].
- Ask for help with the exact traceback: Help me debug my GroupShuffleSplit setup.
Expose the global preprocessing leak
An imputation mean is learned preprocessing state. If you calculate it from all cleaned rows, held-out runners influence a value used to prepare the training data.
The leakage experiment changes only held-out sleep_hours_last_night values. A global mean that changes in response provides direct evidence of contamination.
- Add the leakage experiment below as_float_dict() by pasting this initial global-fit version:
def demonstrate_preprocessing_leakage(
cleaned: pd.DataFrame,
train_raw: pd.DataFrame,
test_raw: pd.DataFrame,
) -> dict[str, object]:
leaky_global_values = fit_preprocessing(cleaned)
perturbed_test = test_raw.copy()
perturbed_test.loc[:, "sleep_hours_last_night"] = 14.0
perturbed_all = pd.concat(
[train_raw, perturbed_test], ignore_index=True
)
perturbed_global_values = fit_preprocessing(perturbed_all)
global_before = float(
leaky_global_values["sleep_hours_last_night"]
)
global_after = float(
perturbed_global_values["sleep_hours_last_night"]
)
return {
"global_sleep_mean_before": round(global_before, 4),
"global_sleep_mean_after_test_perturbation": round(
global_after, 4
),
"global_fit_changed_after_test_perturbation": bool(
abs(global_before - global_after) > 1e-9
),
}What does the experiment measure?
The first mean uses every cleaned runner-day. The experiment then replaces only held-out sleep values with 14.0 before recalculating that global mean.
A changed global statistic proves that held-out runner data can alter preprocessing state learned before the split.
- In demonstrate_preprocessing_leakage(), add the train-only assignments beside their matching global assignments using these reference lines:
leaky_global_values = fit_preprocessing(cleaned)
train_only_values = fit_preprocessing(train_raw)
perturbed_global_values = fit_preprocessing(perturbed_all)
train_values_after_test_change = fit_preprocessing(train_raw)
global_after = float(
perturbed_global_values["sleep_hours_last_night"]
)
train_before = float(train_only_values["sleep_hours_last_night"])
train_after = float(
train_values_after_test_change["sleep_hours_last_night"]
)Why calculate the training mean twice?
Both training calculations read the unchanged training partition. The held-out perturbation cannot affect either value because the test rows never enter fit_preprocessing(train_raw).
- Find the current return dictionary at the end of demonstrate_preprocessing_leakage().
- Replace that dictionary with the complete comparison below:
return {
"wrong_order": "fit preprocessing on all rows, then split",
"correct_order": "split runners, fit on train, apply to both",
"leaky_global_fill_values": as_float_dict(leaky_global_values),
"train_only_fill_values": as_float_dict(train_only_values),
"global_sleep_mean_before": round(global_before, 4),
"global_sleep_mean_after_test_perturbation": round(
global_after, 4
),
"train_sleep_mean_before": round(train_before, 4),
"train_sleep_mean_after_test_perturbation": round(
train_after, 4
),
"global_fit_changed_after_test_perturbation": bool(
abs(global_before - global_after) > 1e-9
),
"train_fit_changed_after_test_perturbation": bool(
abs(train_before - train_after) > 1e-9
),
}How does this prove the safe order?
The two boolean fields compare learned state before and after a test-only change. The global fit should change. The training-only fit should remain fixed.
That contrast turns an abstract leakage warning into reproducible evidence.
Build and verify the group-safe handoff
The pipeline now needs to use the group split as its only partitioning path. It also needs to export the learned training statistics so Level 2 can reuse them without refitting on held-out data.
- In build_outputs(), find the line cleaned = clean_structural_issues(raw).
- Add the split, leakage experiment, imputation flow, runner sets, and preprocessing artifact directly below that line:
train_raw, test_raw = split_by_runner(cleaned)
leakage_demo = demonstrate_preprocessing_leakage(
cleaned, train_raw, test_raw
)
train_fill_values = fit_preprocessing(train_raw)
train_ready = apply_preprocessing(train_raw, train_fill_values)
test_ready = apply_preprocessing(test_raw, train_fill_values)
train_runners = set(train_ready[GROUP_COLUMN])
test_runners = set(test_ready[GROUP_COLUMN])
runner_overlap = sorted(train_runners.intersection(test_runners))
preprocessing_artifact = {
"artifact_version": "1.0.0",
"dataset_version": DATASET_VERSION,
"strategy": "mean imputation",
"fit_scope": "training runners only",
"feature_columns": FEATURE_COLUMNS,
"fill_values": as_float_dict(train_fill_values),
}What becomes reusable state?
- The raw train and test partitions retain missing values until after the split.
- The training partition supplies the only learned fill values.
- The processed partitions receive the same fixed values.
- The preprocessing artifact records its version, strategy, scope, feature order, and numeric means.
- Replace the existing report dictionary inside build_outputs() with this expanded report:
report = {
"dataset_version": DATASET_VERSION,
"schema_validation": validate_schema(
raw, contract["required_columns"]
),
"prediction_definition": contract["prediction_definition"],
"raw_quality": quality_report(raw),
"cleaned_quality": quality_report(cleaned),
"train_quality": quality_report(train_ready),
"test_quality": quality_report(test_ready),
"split": {
"method": "GroupShuffleSplit",
"group_column": GROUP_COLUMN,
"random_state": RANDOM_STATE,
"test_group_fraction": 0.25,
"train_runner_count": len(train_runners),
"test_runner_count": len(test_runners),
"runner_overlap": runner_overlap,
},
"leakage_demo": leakage_demo,
}What evidence does the report preserve?
The report keeps quality metrics for every dataset stage. Its split section records the reproducibility settings plus the observed runner overlap.
The leakage section preserves both experiments so the preprocessing decision remains auditable.
- Replace the current return dictionary at the end of build_outputs() with the complete output map:
return {
"raw": raw,
"cleaned": cleaned,
"train": train_ready,
"test": test_ready,
"preprocessing": preprocessing_artifact,
"report": report,
"contract": contract,
}Why return every output?
The output map gives the exporter one stable interface for datasets, learned preprocessing state, validation evidence, and the schema contract.
- In the paths dictionary inside run_pipeline(), add the processed training and test paths below the cleaned path:
"train": project_root
/ "data"
/ DATASET_VERSION
/ "processed"
/ "train.csv",
"test": project_root
/ "data"
/ DATASET_VERSION
/ "processed"
/ "test.csv",Why version the partitions?
The v1 path ties both partitions to the same dataset contract. Level 2 can consume those files without silently generating a different holdout.
- In the same paths dictionary, add the preprocessing and report paths before the contract path:
"preprocessing": project_root
/ "artifacts"
/ DATASET_VERSION
/ "preprocessing.json",
"report": project_root
/ "reports"
/ DATASET_VERSION
/ "data_quality_report.json",Why export state separately?
The preprocessing artifact stores values that transform future data. The quality report stores evidence about how those values were produced.
- Replace the current CSV and JSON write statements in run_pipeline() with these export loops:
for name in ("raw", "cleaned", "train", "test"):
outputs[name].to_csv(paths[name], index=False)
for name in ("preprocessing", "report", "contract"):
paths[name].write_text(
json.dumps(outputs[name], indent=2), encoding="utf-8"
)What do the export loops guarantee?
Every dataset uses the same CSV export path. Every reusable metadata object uses indented JSON with UTF-8 encoding.
- In README.md, add the leakage experiment section between the prediction contract and the local run instructions:
## Leakage experiment
The wrong path fits feature means on all cleaned rows before splitting. Changing held-out runner values changes that global learned state. The correct path splits runner groups first, fits means only on training runners, and applies those fixed values to both partitions.Why document the experiment?
The report supplies machine-readable evidence. The README gives the next engineer the reasoning behind the exported preprocessing state.
Before you run the pipeline, do you expect a test-only sleep change to affect the global mean, the train-only mean, or both?
- Save src/data_pipeline.py.
- Save README.md.
- Run the completed pipeline from the level-1-data folder using this command:
python src/data_pipeline.pyWhat should the report prove?
- The runner_overlap list is empty.
- The processed train and test quality sections each report zero missing feature values.
- The global_fit_changed_after_test_perturbation field is true.
- The train_fit_changed_after_test_perturbation field is false.
That result exposes the intended shortfall in global preprocessing. Held-out runners changed learned state that would have been used for training.
The training-only statistic stayed fixed because the test partition had no path back into the fit. You now have direct evidence for the split-first rule.
- Confirm data/v1/processed/train.csv appears in the VS Code Explorer.
- Confirm data/v1/processed/test.csv appears in the VS Code Explorer.
- Confirm artifacts/v1/preprocessing.json records training runners only as its fit scope.
- Confirm reports/v1/data_quality_report.json records an empty runner-overlap list.
Do the verification fields look wrong?
- Check that fit_preprocessing() receives train_raw when creating the exported fill values.
- Check that the perturbation changes only perturbed_test.
- Check that both processed partitions receive train_fill_values.
- Ask for a focused comparison: Help me trace why my leakage flags or runner overlap are incorrect.
✔️ My split and leakage evidence are correct
Strong work. Your processed datasets now isolate complete runners while sharing one immutable train-fitted preprocessing artifact.
ⓧ I'd like to double check the full code
Compare your completed files with these cumulative versions. Save any corrections before running the pipeline again.
from __future__ import annotations
import json
from pathlib import Path
import numpy as np
import pandas as pd
from sklearn.model_selection import GroupShuffleSplit
DATASET_VERSION = "v1"
SCHEMA_VERSION = "1.0.0"
RANDOM_STATE = 42
GROUP_COLUMN = "runner_id"
TARGET_COLUMN = "needs_recovery_next_day"
KEY_COLUMNS = ["runner_id", "observation_date"]
FEATURE_COLUMNS = [
"weekly_km_to_date",
"sleep_hours_last_night",
"resting_hr_morning",
"soreness_score_evening",
"prior_injury",
]
ALL_COLUMNS = [
"runner_id",
"observation_date",
"prediction_timestamp",
*FEATURE_COLUMNS,
TARGET_COLUMN,
"target_available_timestamp",
]
VALID_RANGES = {
"weekly_km_to_date": (0.0, 200.0),
"sleep_hours_last_night": (0.0, 14.0),
"resting_hr_morning": (30.0, 220.0),
"soreness_score_evening": (0.0, 10.0),
"prior_injury": (0.0, 1.0),
}
def build_schema_contract() -> dict[str, object]:
roles = {
"runner_id": "group_identifier",
"observation_date": "context",
"prediction_timestamp": "context",
TARGET_COLUMN: "target",
"target_available_timestamp": "target_metadata",
}
data_types = {
"runner_id": "string",
"observation_date": "string",
"prediction_timestamp": "string",
"weekly_km_to_date": "float",
"sleep_hours_last_night": "float",
"resting_hr_morning": "float",
"soreness_score_evening": "float",
"prior_injury": "integer",
TARGET_COLUMN: "integer",
"target_available_timestamp": "string",
}
columns = []
for name in ALL_COLUMNS:
role = roles.get(name, "feature")
columns.append(
{
"name": name,
"role": role,
"data_type": data_types[name],
"available_at_prediction": role
in {"group_identifier", "context", "feature"},
}
)
return {
"schema_version": SCHEMA_VERSION,
"dataset_version": DATASET_VERSION,
"prediction_definition": {
"entity": "runner-day",
"prediction_timestamp": "20:00 local time on observation day D",
"target": TARGET_COLUMN,
"prediction_horizon": "next calendar day D+1",
"target_available_timestamp": (
"08:00 local time on next calendar day D+1"
),
},
"primary_key": KEY_COLUMNS,
"group_column": GROUP_COLUMN,
"feature_columns": FEATURE_COLUMNS,
"target_column": TARGET_COLUMN,
"required_columns": ALL_COLUMNS,
"columns": columns,
}
def build_raw_dataset() -> pd.DataFrame:
records: list[dict[str, object]] = []
for runner_number in range(1, 25):
baseline_hr = 45 + (runner_number % 10)
prior_injury = int(runner_number % 7 == 0)
for day in range(1, 7):
weekly_km = 18.0 + (runner_number % 6) * 3.0 + day * 2.0
sleep_hours = 5.5 + ((runner_number + day) % 6) * 0.5
resting_hr = baseline_hr + (day % 3) * 2
soreness = (runner_number * 2 + day) % 11
needs_recovery = int(
sleep_hours < 6.5
or soreness >= 8
or resting_hr >= baseline_hr + 4
or weekly_km >= 45.0
or (prior_injury == 1 and soreness >= 6)
)
records.append(
{
"runner_id": f"runner-{runner_number:02d}",
"observation_date": f"2026-01-{day:02d}",
"prediction_timestamp": (
f"2026-01-{day:02d}T20:00:00"
),
"weekly_km_to_date": weekly_km,
"sleep_hours_last_night": sleep_hours,
"resting_hr_morning": resting_hr,
"soreness_score_evening": soreness,
"prior_injury": prior_injury,
TARGET_COLUMN: needs_recovery,
"target_available_timestamp": (
f"2026-01-{day + 1:02d}T08:00:00"
),
}
)
records[7]["sleep_hours_last_night"] = None
records[61]["resting_hr_morning"] = None
records[103]["soreness_score_evening"] = None
records[15]["weekly_km_to_date"] = -3.0
records[95]["sleep_hours_last_night"] = 16.0
records.append(records[20].copy())
return pd.DataFrame(records, columns=ALL_COLUMNS)
def validate_schema(
frame: pd.DataFrame,
expected_columns: list[str] | None = None,
) -> dict[str, object]:
required = expected_columns or ALL_COLUMNS
expected = set(required)
observed = set(frame.columns)
missing = sorted(expected - observed)
unexpected = sorted(observed - expected)
return {
"valid": not missing and not unexpected,
"missing": missing,
"unexpected": unexpected,
}
def quality_report(frame: pd.DataFrame) -> dict[str, int | float]:
missing_features = int(frame[FEATURE_COLUMNS].isna().sum().sum())
duplicate_records = int(frame.duplicated(subset=KEY_COLUMNS).sum())
invalid_values = 0
for column, (minimum, maximum) in VALID_RANGES.items():
invalid_values += int(
((frame[column] < minimum) | (frame[column] > maximum)).sum()
)
checked_values = frame.shape[0] * len(FEATURE_COLUMNS)
temporal_violations = int(
(
frame["prediction_timestamp"]
>= frame["target_available_timestamp"]
).sum()
)
return {
"row_count": int(frame.shape[0]),
"runner_count": len(set(frame[GROUP_COLUMN])),
"missing_feature_values": missing_features,
"feature_completeness_ratio": round(
1 - missing_features / checked_values, 4
),
"duplicate_runner_days": duplicate_records,
"invalid_values": invalid_values,
"temporal_order_violations": temporal_violations,
"target_positive_rate": round(float(frame[TARGET_COLUMN].mean()), 4),
}
def clean_structural_issues(frame: pd.DataFrame) -> pd.DataFrame:
cleaned = frame.drop_duplicates(subset=KEY_COLUMNS, keep="first").copy()
for column, (minimum, maximum) in VALID_RANGES.items():
out_of_range = (cleaned[column] < minimum) | (
cleaned[column] > maximum
)
cleaned.loc[out_of_range, column] = np.nan
return cleaned
def split_by_runner(
cleaned: pd.DataFrame,
) -> tuple[pd.DataFrame, pd.DataFrame]:
splitter = GroupShuffleSplit(
n_splits=1,
test_size=0.25,
random_state=RANDOM_STATE,
)
train_indices, test_indices = next(
splitter.split(
cleaned,
cleaned[TARGET_COLUMN],
groups=cleaned[GROUP_COLUMN],
)
)
return (
cleaned.iloc[train_indices].copy(),
cleaned.iloc[test_indices].copy(),
)
def fit_preprocessing(frame: pd.DataFrame) -> pd.Series:
return frame[FEATURE_COLUMNS].mean()
def apply_preprocessing(
frame: pd.DataFrame, fill_values: pd.Series
) -> pd.DataFrame:
ready = frame.copy()
ready[FEATURE_COLUMNS] = ready[FEATURE_COLUMNS].fillna(fill_values)
return ready
def as_float_dict(values: pd.Series) -> dict[str, float]:
return {name: float(value) for name, value in values.items()}
def demonstrate_preprocessing_leakage(
cleaned: pd.DataFrame,
train_raw: pd.DataFrame,
test_raw: pd.DataFrame,
) -> dict[str, object]:
leaky_global_values = fit_preprocessing(cleaned)
train_only_values = fit_preprocessing(train_raw)
perturbed_test = test_raw.copy()
perturbed_test.loc[:, "sleep_hours_last_night"] = 14.0
perturbed_all = pd.concat(
[train_raw, perturbed_test], ignore_index=True
)
perturbed_global_values = fit_preprocessing(perturbed_all)
train_values_after_test_change = fit_preprocessing(train_raw)
global_before = float(
leaky_global_values["sleep_hours_last_night"]
)
global_after = float(
perturbed_global_values["sleep_hours_last_night"]
)
train_before = float(train_only_values["sleep_hours_last_night"])
train_after = float(
train_values_after_test_change["sleep_hours_last_night"]
)
return {
"wrong_order": "fit preprocessing on all rows, then split",
"correct_order": "split runners, fit on train, apply to both",
"leaky_global_fill_values": as_float_dict(leaky_global_values),
"train_only_fill_values": as_float_dict(train_only_values),
"global_sleep_mean_before": round(global_before, 4),
"global_sleep_mean_after_test_perturbation": round(
global_after, 4
),
"train_sleep_mean_before": round(train_before, 4),
"train_sleep_mean_after_test_perturbation": round(
train_after, 4
),
"global_fit_changed_after_test_perturbation": bool(
abs(global_before - global_after) > 1e-9
),
"train_fit_changed_after_test_perturbation": bool(
abs(train_before - train_after) > 1e-9
),
}
def build_outputs() -> dict[str, object]:
contract = build_schema_contract()
raw = build_raw_dataset()
cleaned = clean_structural_issues(raw)
train_raw, test_raw = split_by_runner(cleaned)
leakage_demo = demonstrate_preprocessing_leakage(
cleaned, train_raw, test_raw
)
train_fill_values = fit_preprocessing(train_raw)
train_ready = apply_preprocessing(train_raw, train_fill_values)
test_ready = apply_preprocessing(test_raw, train_fill_values)
train_runners = set(train_ready[GROUP_COLUMN])
test_runners = set(test_ready[GROUP_COLUMN])
runner_overlap = sorted(train_runners.intersection(test_runners))
preprocessing_artifact = {
"artifact_version": "1.0.0",
"dataset_version": DATASET_VERSION,
"strategy": "mean imputation",
"fit_scope": "training runners only",
"feature_columns": FEATURE_COLUMNS,
"fill_values": as_float_dict(train_fill_values),
}
report = {
"dataset_version": DATASET_VERSION,
"schema_validation": validate_schema(
raw, contract["required_columns"]
),
"prediction_definition": contract["prediction_definition"],
"raw_quality": quality_report(raw),
"cleaned_quality": quality_report(cleaned),
"train_quality": quality_report(train_ready),
"test_quality": quality_report(test_ready),
"split": {
"method": "GroupShuffleSplit",
"group_column": GROUP_COLUMN,
"random_state": RANDOM_STATE,
"test_group_fraction": 0.25,
"train_runner_count": len(train_runners),
"test_runner_count": len(test_runners),
"runner_overlap": runner_overlap,
},
"leakage_demo": leakage_demo,
}
return {
"raw": raw,
"cleaned": cleaned,
"train": train_ready,
"test": test_ready,
"preprocessing": preprocessing_artifact,
"report": report,
"contract": contract,
}
def run_pipeline(project_root: Path) -> dict[str, object]:
outputs = build_outputs()
paths = {
"raw": project_root
/ "data"
/ DATASET_VERSION
/ "raw"
/ "run_readiness_observations.csv",
"cleaned": project_root
/ "data"
/ DATASET_VERSION
/ "cleaned"
/ "run_readiness_observations.csv",
"train": project_root
/ "data"
/ DATASET_VERSION
/ "processed"
/ "train.csv",
"test": project_root
/ "data"
/ DATASET_VERSION
/ "processed"
/ "test.csv",
"preprocessing": project_root
/ "artifacts"
/ DATASET_VERSION
/ "preprocessing.json",
"report": project_root
/ "reports"
/ DATASET_VERSION
/ "data_quality_report.json",
"contract": project_root
/ "contracts"
/ DATASET_VERSION
/ "schema_contract.json",
}
for path in paths.values():
path.parent.mkdir(parents=True, exist_ok=True)
for name in ("raw", "cleaned", "train", "test"):
outputs[name].to_csv(paths[name], index=False)
for name in ("preprocessing", "report", "contract"):
paths[name].write_text(
json.dumps(outputs[name], indent=2), encoding="utf-8"
)
return outputs
def main() -> None:
project_root = Path(__file__).resolve().parents[1]
outputs = run_pipeline(project_root)
print(json.dumps(outputs["report"], indent=2))
if __name__ == "__main__":
main()# Level 1 Data: Time-Aware, Group-Safe Run Readiness Handoff
This project creates the `v1` data contract for a future next-day running-readiness classifier. It prepares and validates data only. Model fitting and baseline evaluation begin in Level 2.
The synthetic target and ranges are educational and are not medical or coaching advice.
## Prediction contract
At 20:00 local time on observation day D, predict `needs_recovery_next_day` for D+1. The label becomes observable at 08:00 on D+1.
Allowed features are weekly distance through D, previous-night sleep, morning resting heart rate, evening soreness recorded by 20:00, and prior injury history.
## Leakage experiment
The wrong path fits feature means on all cleaned rows before splitting. Changing held-out runner values changes that global learned state. The correct path splits runner groups first, fits means only on training runners, and applies those fixed values to both partitions.
## Run locally
```bash
python -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python src/data_pipeline.py
```
You have a deterministic runner-level holdout with no shared runners. Next, you will turn this handoff into a tested data product that Level 2 can trust.
Export and Test the Level 2 Handoff
Your runner-level split now keeps every runner on one side of the holdout. Your train-fitted preprocessing state also stays unchanged when held-out data changes.
Level 2 needs an immutable schema contract with verified datasets. This step turns your pipeline decisions into a tested handoff that Level 2 can consume without repeating hidden preparation work.
In this step, get ready to:
- Export the complete versioned handoff.
- Add seven automated tests for the data contract.
- Document the artifact inventory plus the Level 2 boundary.
Export the complete v1 handoff
A reproducible data product includes its datasets plus every decision needed to interpret them. Running the complete pipeline writes the raw data through the same deterministic path as the processed outputs.
Before you run the pipeline, do you expect any artifact paths or split evidence to change?
- Regenerate every v1 artifact by running this command:
python src/data_pipeline.pyYou should see the complete report in your terminal. Its split section shows "runner_overlap": [].
The leakage section shows global_fit_changed_after_test_perturbation as true. It shows train_fit_changed_after_test_perturbation as false.
Pipeline command failing?
- Confirm your terminal prompt still shows the activated .venv environment.
- Confirm the terminal is inside the level-1-data folder that contains src.
- Ask for help with the failing pipeline command.
- Return to VS Code.
- Confirm data/v1/raw/run_readiness_observations.csv appears in the file tree.
- Confirm data/v1/cleaned/run_readiness_observations.csv appears in the file tree.
- Confirm data/v1/processed/train.csv appears in the file tree.
- Confirm data/v1/processed/test.csv appears in the file tree.
- Confirm artifacts/v1/preprocessing.json appears in the file tree.
- Confirm reports/v1/data_quality_report.json appears in the file tree.
- Confirm contracts/v1/schema_contract.json appears in the file tree.
Your complete v1 artifact set is now reproducible from one pipeline run.
Test the complete data contract
An automated test suite makes the handoff executable. Python's unittest framework checks the schema boundary plus the structural guarantees that Level 2 depends on.
- Select the tests folder in the Explorer view.
- Select New File.
- Enter test_data_pipeline.py.
- Add the test dependencies by pasting these imports:
import unittest
from src.data_pipeline import (
FEATURE_COLUMNS,
GROUP_COLUMN,
TARGET_COLUMN,
build_outputs,
build_raw_dataset,
clean_structural_issues,
fit_preprocessing,
split_by_runner,
)What do these imports provide?
- The unittest import supplies the test case class plus its assertion methods.
- The constants expose the feature boundary plus the runner group column.
- The pipeline functions let each test inspect deterministic in-memory outputs.
- Save tests/test_data_pipeline.py.
- Check that the import scaffold loads by running the test module:
python -m unittest -v tests/test_data_pipeline.pyThe command should complete with OK. The file does not contain test methods yet.
Imports not loading?
- Check that test_data_pipeline.py is inside the existing tests folder.
- Check that your terminal is inside level-1-data.
- Ask for help with the test imports.
- Add the test fixture plus the schema and feature-time test below the imports:
class DataPipelineTests(unittest.TestCase):
def setUp(self) -> None:
self.outputs = build_outputs()
def test_schema_and_feature_time_contract_are_valid(self) -> None:
contract = self.outputs["contract"]
raw = self.outputs["raw"]
self.assertTrue(self.outputs["report"]["schema_validation"]["valid"])
self.assertNotIn(TARGET_COLUMN, FEATURE_COLUMNS)
self.assertTrue(
all(
raw["prediction_timestamp"]
< raw["target_available_timestamp"]
)
)
feature_specs = [
column
for column in contract["columns"]
if column["role"] == "feature"
]
self.assertTrue(
all(column["available_at_prediction"] for column in feature_specs)
)What does this test prove?
- The setUp() fixture rebuilds the deterministic outputs before each test.
- The test confirms the exported schema is valid.
- The assertions keep the target out of FEATURE_COLUMNS.
- The timestamp check proves every target becomes available after its prediction point.
- Save tests/test_data_pipeline.py.
- Run the test module again to execute the first contract test:
python -m unittest -v tests/test_data_pipeline.pyYou should see the schema and feature-time test pass. The run finishes with OK.
Schema test failing?
- Compare the constant names in your test with FEATURE_COLUMNS plus TARGET_COLUMN in src/data_pipeline.py.
- Ask for help tracing the failed schema assertion.
- Add the structural-cleaning test below the first test method:
def test_cleaning_removes_duplicates_and_invalid_values(self) -> None:
cleaned_quality = self.outputs["report"]["cleaned_quality"]
self.assertEqual(cleaned_quality["duplicate_runner_days"], 0)
self.assertEqual(cleaned_quality["invalid_values"], 0)
self.assertGreater(cleaned_quality["missing_feature_values"], 0)What does the cleaning test protect?
This test separates repaired structural defects from unresolved missingness. Duplicate runner-days plus invalid ranges must reach zero before split-time imputation begins.
- Save tests/test_data_pipeline.py.
- Run the expanded test module:
python -m unittest -v tests/test_data_pipeline.pyYou should see both tests pass. The command finishes with OK.
Cleaning test failing?
- Confirm the test reads cleaned_quality from the report.
- Ask for help comparing the cleaned quality metrics.
- Add the processed-missingness test below the cleaning test:
def test_processed_splits_have_no_missing_features(self) -> None:
train = self.outputs["train"]
test = self.outputs["test"]
self.assertEqual(int(train[FEATURE_COLUMNS].isna().sum().sum()), 0)
self.assertEqual(int(test[FEATURE_COLUMNS].isna().sum().sum()), 0)What does the missingness test prove?
The test checks only the permitted feature columns in both processed partitions. It proves the train-fitted means were applied to each side of the split.
- Save tests/test_data_pipeline.py.
- Run the three-test suite:
python -m unittest -v tests/test_data_pipeline.pyYou should see the processed-missingness test join the two passing contract tests. The run ends with OK.
Processed values still missing?
- Confirm the assertions inspect self.outputs["train"] plus self.outputs["test"].
- Ask for help tracing missing processed features.
- Add the runner-separation test below the processed-missingness test:
def test_runners_are_separated_between_train_and_test(self) -> None:
train_runners = set(self.outputs["train"][GROUP_COLUMN])
test_runners = set(self.outputs["test"][GROUP_COLUMN])
self.assertEqual(train_runners.intersection(test_runners), set())
self.assertEqual(
self.outputs["report"]["split"]["runner_overlap"], []
)What does the group test protect?
This test checks the actual runner sets plus the exported report evidence. A runner appearing in both partitions would fail the handoff immediately.
- Save tests/test_data_pipeline.py.
- Run the four-test suite:
python -m unittest -v tests/test_data_pipeline.pyYou should see the runner-separation test pass. The command finishes with OK.
Runner overlap detected?
- Confirm the test builds sets from GROUP_COLUMN on both processed partitions.
- Ask for help inspecting the split groups.
- Add the train-only preprocessing test below the runner-separation test:
def test_preprocessing_is_fitted_on_training_runners_only(self) -> None:
cleaned = clean_structural_issues(build_raw_dataset())
train_raw, _ = split_by_runner(cleaned)
expected = fit_preprocessing(train_raw)
actual = self.outputs["preprocessing"]["fill_values"]
self.assertEqual(
self.outputs["preprocessing"]["fit_scope"],
"training runners only",
)
for feature in FEATURE_COLUMNS:
self.assertAlmostEqual(actual[feature], float(expected[feature]))What does the preprocessing test prove?
- The test reconstructs the deterministic raw training partition.
- It independently calculates the expected training means.
- Each exported fill value must match that train-only calculation.
- Save tests/test_data_pipeline.py.
- Run the five-test suite:
python -m unittest -v tests/test_data_pipeline.pyYou should see the train-only preprocessing test pass. The suite ends with OK.
Fill values do not match?
- Confirm expected comes from fit_preprocessing(train_raw).
- Ask for help comparing the preprocessing artifact with the training means.
- Add the leakage-evidence test below the preprocessing test:
def test_leakage_experiment_detects_test_contamination(self) -> None:
experiment = self.outputs["report"]["leakage_demo"]
self.assertTrue(
experiment["global_fit_changed_after_test_perturbation"]
)
self.assertFalse(
experiment["train_fit_changed_after_test_perturbation"]
)What does the leakage test preserve?
The wrong global statistic must react to a held-out change. The correct training statistic must remain fixed.
- Save tests/test_data_pipeline.py.
- Run the six-test suite:
python -m unittest -v tests/test_data_pipeline.pyYou should see the leakage experiment test pass. The suite finishes with OK.
Leakage evidence test failing?
- Confirm the assertions use the two exact evidence keys from leakage_demo.
- Ask for help comparing the before-and-after statistics.
- Add the deterministic-output test plus the module runner below the leakage test:
def test_outputs_are_deterministic(self) -> None:
second = build_outputs()
for name in ("raw", "cleaned", "train", "test"):
self.assertTrue(self.outputs[name].equals(second[name]))
for name in ("preprocessing", "report", "contract"):
self.assertEqual(self.outputs[name], second[name])
if __name__ == "__main__":
unittest.main()What does the final test prove?
- The test builds the complete output collection a second time.
- Every dataset must contain the same values in the same shape.
- Every reusable artifact must contain identical structured data.
- Save tests/test_data_pipeline.py.
- Run the complete seven-test suite:
python -m unittest -v tests/test_data_pipeline.pyAll seven tests should pass. The final line is OK.
Deterministic test failing?
- Confirm the test calls build_outputs() without changing the random state.
- Ask for help locating nondeterministic output.
✔️ Awesome, I've got everything!
Your tests/test_data_pipeline.py file now protects all seven handoff guarantees.
ⓧ I'd like to double check the full code
Compare your test module with this complete version.
import unittest
from src.data_pipeline import (
FEATURE_COLUMNS,
GROUP_COLUMN,
TARGET_COLUMN,
build_outputs,
build_raw_dataset,
clean_structural_issues,
fit_preprocessing,
split_by_runner,
)
class DataPipelineTests(unittest.TestCase):
def setUp(self) -> None:
self.outputs = build_outputs()
def test_schema_and_feature_time_contract_are_valid(self) -> None:
contract = self.outputs["contract"]
raw = self.outputs["raw"]
self.assertTrue(self.outputs["report"]["schema_validation"]["valid"])
self.assertNotIn(TARGET_COLUMN, FEATURE_COLUMNS)
self.assertTrue(
all(
raw["prediction_timestamp"]
< raw["target_available_timestamp"]
)
)
feature_specs = [
column
for column in contract["columns"]
if column["role"] == "feature"
]
self.assertTrue(
all(column["available_at_prediction"] for column in feature_specs)
)
def test_cleaning_removes_duplicates_and_invalid_values(self) -> None:
cleaned_quality = self.outputs["report"]["cleaned_quality"]
self.assertEqual(cleaned_quality["duplicate_runner_days"], 0)
self.assertEqual(cleaned_quality["invalid_values"], 0)
self.assertGreater(cleaned_quality["missing_feature_values"], 0)
def test_processed_splits_have_no_missing_features(self) -> None:
train = self.outputs["train"]
test = self.outputs["test"]
self.assertEqual(int(train[FEATURE_COLUMNS].isna().sum().sum()), 0)
self.assertEqual(int(test[FEATURE_COLUMNS].isna().sum().sum()), 0)
def test_runners_are_separated_between_train_and_test(self) -> None:
train_runners = set(self.outputs["train"][GROUP_COLUMN])
test_runners = set(self.outputs["test"][GROUP_COLUMN])
self.assertEqual(train_runners.intersection(test_runners), set())
self.assertEqual(
self.outputs["report"]["split"]["runner_overlap"], []
)
def test_preprocessing_is_fitted_on_training_runners_only(self) -> None:
cleaned = clean_structural_issues(build_raw_dataset())
train_raw, _ = split_by_runner(cleaned)
expected = fit_preprocessing(train_raw)
actual = self.outputs["preprocessing"]["fill_values"]
self.assertEqual(
self.outputs["preprocessing"]["fit_scope"],
"training runners only",
)
for feature in FEATURE_COLUMNS:
self.assertAlmostEqual(actual[feature], float(expected[feature]))
def test_leakage_experiment_detects_test_contamination(self) -> None:
experiment = self.outputs["report"]["leakage_demo"]
self.assertTrue(
experiment["global_fit_changed_after_test_perturbation"]
)
self.assertFalse(
experiment["train_fit_changed_after_test_perturbation"]
)
def test_outputs_are_deterministic(self) -> None:
second = build_outputs()
for name in ("raw", "cleaned", "train", "test"):
self.assertTrue(self.outputs[name].equals(second[name]))
for name in ("preprocessing", "report", "contract"):
self.assertEqual(self.outputs[name], second[name])
if __name__ == "__main__":
unittest.main()Document and verify the Level 2 boundary
Tests prove the handoff mechanically. The README explains the same boundary to the person or system that consumes it next.
- Select the existing README.md file in the Explorer view.
- Replace its current contents with this opening documentation:
# Level 1 Data: Time-Aware, Group-Safe Run Readiness Handoff
This project creates the `v1` data contract for a future next-day running-readiness classifier. It prepares and validates data only. Model fitting and baseline evaluation begin in Level 2.
The synthetic target and ranges are educational and are not medical or coaching advice.
## Prediction contract
At 20:00 local time on observation day D, predict `needs_recovery_next_day` for D+1. The label becomes observable at 08:00 on D+1.
Allowed features contain information available by 20:00 on D:
- distance accumulated through day D
- sleep from the previous night
- morning resting heart rate from day D
- evening soreness recorded by 20:00 on day D
- prior injury history
Runner ID, dates, timestamps, and the target are context or supervision metadata, not model features.
## Core concepts
- Prediction time defines the boundary between valid features and future leakage.
- Preprocessing statistics are learned state and must be fitted after splitting.
- Repeated observations from one runner are dependent, so runner groups must not cross the holdout boundary.
- A reusable ML handoff includes datasets, preprocessing state, a schema contract, quality evidence, tests, and documentation.What does this section establish?
- The prediction contract fixes the decision point at 20:00 on day D.
- The feature list defines what is available before that decision.
- The core concepts state why time plus runner groups shape the split.
- Save README.md.
- Check the saved file for the Prediction contract plus Core concepts headings.
README opening incomplete?
- Confirm the five feature bullets appear below the prediction contract.
- Ask for help comparing the documentation opening.
- Append the leakage experiment plus the versioned handoff inventory below the core concepts:
## Leakage experiment
The intentionally wrong path computes feature means from all cleaned rows before splitting. The experiment then changes only held-out runner values and shows that the global mean changes.
The correct path first splits runners, computes means from training runners only, and applies those fixed values to both partitions. Changing held-out data cannot change the training artifact.
## Run locally
```bash
python -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python src/data_pipeline.py
python -m unittest -v tests/test_data_pipeline.py
```
## Versioned Level 2 handoff
- `data/v1/raw/run_readiness_observations.csv`
- `data/v1/cleaned/run_readiness_observations.csv`
- `data/v1/processed/train.csv`
- `data/v1/processed/test.csv`
- `artifacts/v1/preprocessing.json`
- `reports/v1/data_quality_report.json`
- `contracts/v1/schema_contract.json`What does the handoff inventory provide?
The leakage section records the failed global-fit approach plus the correct train-only sequence. The inventory gives Level 2 stable paths for every dataset plus reusable artifact.
- Save README.md.
- Check the saved file for all seven versioned output paths.
Handoff path missing?
- Compare each README path with the corresponding file in your project tree.
- Ask for help reconciling the artifact inventory.
- Append the project limitations plus the Level 2 boundary below the artifact inventory:
## Limitations
- Data, target logic, and validity ranges are synthetic.
- Mean imputation is chosen for a transparent leakage experiment, not because it is universally optimal.
- The group holdout evaluates unseen runners, but model performance is intentionally not measured in Level 1.
- Level 2 should treat the `v1` contract and split as inputs rather than silently regenerating them.
## Next level
Level 2 Train will load the processed data and contract, build a naive reference, fit a simple scikit-learn baseline, and evaluate whether learning adds measurable value.Why state these limitations?
The limitations prevent synthetic target rules plus mean imputation from being mistaken for universal choices. The final boundary keeps model fitting out of Level 1.
- Save README.md.
- Check the final section for a clear statement that baseline modeling begins in Level 2.
Level boundary unclear?
- Confirm the limitations say model performance is intentionally unmeasured in Level 1.
- Ask for help checking the Level 2 boundary.
✔️ Awesome, I've got everything!
Your README.md now documents the contract plus every artifact that Level 2 receives.
ⓧ I'd like to double check the full code
Compare your README with this complete version.
# Level 1 Data: Time-Aware, Group-Safe Run Readiness Handoff
This project creates the `v1` data contract for a future next-day running-readiness classifier. It prepares and validates data only. Model fitting and baseline evaluation begin in Level 2.
The synthetic target and ranges are educational and are not medical or coaching advice.
## Prediction contract
At 20:00 local time on observation day D, predict `needs_recovery_next_day` for D+1. The label becomes observable at 08:00 on D+1.
Allowed features contain information available by 20:00 on D:
- distance accumulated through day D
- sleep from the previous night
- morning resting heart rate from day D
- evening soreness recorded by 20:00 on day D
- prior injury history
Runner ID, dates, timestamps, and the target are context or supervision metadata, not model features.
## Core concepts
- Prediction time defines the boundary between valid features and future leakage.
- Preprocessing statistics are learned state and must be fitted after splitting.
- Repeated observations from one runner are dependent, so runner groups must not cross the holdout boundary.
- A reusable ML handoff includes datasets, preprocessing state, a schema contract, quality evidence, tests, and documentation.
## Leakage experiment
The intentionally wrong path computes feature means from all cleaned rows before splitting. The experiment then changes only held-out runner values and shows that the global mean changes.
The correct path first splits runners, computes means from training runners only, and applies those fixed values to both partitions. Changing held-out data cannot change the training artifact.
## Run locally
```bash
python -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python src/data_pipeline.py
python -m unittest -v tests/test_data_pipeline.py
```
## Versioned Level 2 handoff
- `data/v1/raw/run_readiness_observations.csv`
- `data/v1/cleaned/run_readiness_observations.csv`
- `data/v1/processed/train.csv`
- `data/v1/processed/test.csv`
- `artifacts/v1/preprocessing.json`
- `reports/v1/data_quality_report.json`
- `contracts/v1/schema_contract.json`
## Limitations
- Data, target logic, and validity ranges are synthetic.
- Mean imputation is chosen for a transparent leakage experiment, not because it is universally optimal.
- The group holdout evaluates unseen runners, but model performance is intentionally not measured in Level 1.
- Level 2 should treat the `v1` contract and split as inputs rather than silently regenerating them.
## Next level
Level 2 Train will load the processed data and contract, build a naive reference, fit a simple scikit-learn baseline, and evaluate whether learning adds measurable value.Before the final run, do you expect any contract test to depend on a fitted model?
- Run the complete handoff test suite one final time:
python -m unittest -v tests/test_data_pipeline.pyYou should see all seven tests pass. The run finishes with OK.
One of the seven tests failing?
- Read the failed test name to identify the broken handoff guarantee.
- Compare that method with the complete test module above.
- Ask for help interpreting the failed assertion.
That is the full Level 1 handoff complete. Level 2 can now consume versioned datasets plus explicit preprocessing state without rebuilding your data preparation.
Secret mission
Prove the Schema Drift Gate
Your versioned contract defines the columns that Level 2 can trust. In this mission, you will build a gate that accepts a compatible batch. The same gate will reject schema drift before preprocessing begins.
Clean Up Your Resources
Clean Up Your Resources
Every resource from this project lives on your machine. No ongoing costs remain.
Choose the cleanup option that matches your plans for the v1 handoff.
Resources you used:
- Local Python virtual environment in .venv/.
- Four versioned CSV datasets at data/v1/raw/run_readiness_observations.csv, data/v1/cleaned/run_readiness_observations.csv, data/v1/processed/train.csv, and data/v1/processed/test.csv.
- Reusable JSON handoff artifacts at artifacts/v1/preprocessing.json and contracts/v1/schema_contract.json.
- Quality evidence at reports/v1/data_quality_report.json and evidence/v1/schema_drift_evidence.json.
- Project source code in src/.
- Automated tests in tests/.
- Project documentation in README.md, requirements.txt, and .gitignore.
Keep everything running
No action needed. Choose this if Level 2 will consume the unchanged handoff.
- Keep every v1 output as the immutable input contract for Level 2.
- Leave .venv/ in place so the pinned dependencies remain reproducible.
- Retain evidence/v1/schema_drift_evidence.json as proof that the schema drift gate accepts valid input.
- Keep the local files without worrying about ongoing charges.
Pause - I'll come back to this later
Shut down the activated virtual environment while keeping every project file available.
- Deactivate the environment by running this command:
deactivateWhat does this command do?
The deactivate command leaves the virtual environment for the current terminal session. The .venv/ folder stays on disk. Every v1 artifact remains unchanged.
You're safely paused. The complete handoff remains ready for your next session.
Delete - I don't want to use this again
Deleting the folder is permanent. This affects only the local level-1-data project.
- Return to the terminal panel from earlier.
- Remove the complete local project by running this cleanup sequence:
deactivate
cd ..
rm -rf level-1-dataWhat does this cleanup do?
- The first line exits the active Python environment.
- The second line moves the terminal to the folder that contains level-1-data.
- The final line removes level-1-data with every nested artifact.
- Confirm the project folder is gone by running this command:
lsWhat should you see?
The output should no longer list level-1-data. This confirms that the environment and project artifacts are removed.
Still see the project folder?
The deletion only works from the directory that contains level-1-data. If the folder remains, get help checking your terminal location.
That cleanup is complete. The local project environment and handoff are now removed.
Nice Work!
Nice Work!
You did it! Your tested Level 1 handoff gives Level 2 an explicit prediction-time boundary with fully separated runners.
You've learned how to:
- Define a precise next-day target at 20:00 on day D. Restrict every feature to information available at that prediction point.
- Expose data leakage by perturbing held-out runner values. Prove that training-only imputation keeps learned state unchanged. Create a deterministic runner-level holdout with zero runner overlap.
- Package every dataset in a versioned Level 2 handoff. Include reusable preprocessing state. Include quality evidence. Include a schema contract. Verify deterministic behavior with seven automated pipeline tests.
- Secret Mission: Prove the schema drift gate accepts valid input. Show that it rejects a malformed batch with one missing feature plus one unexpected future column.
Ready to quiz yourself?