wehub-resource-sync
1160 lines
39 KiB
Python
1160 lines
39 KiB
Python
# ---
|
||
# jupyter:
|
||
# jupytext:
|
||
# cell_metadata_filter: tags,-all
|
||
# text_representation:
|
||
# extension: .py
|
||
# format_name: percent
|
||
# format_version: '1.3'
|
||
# jupytext_version: 1.19.3
|
||
# kernelspec:
|
||
# display_name: Python 3 (ipykernel)
|
||
# language: python
|
||
# name: python3
|
||
# ---
|
||
|
||
# %% [markdown]
|
||
# # Preprocessing Pipeline
|
||
#
|
||
# **Docker image**: `ml4t`
|
||
#
|
||
# **Chapter 7: Defining the Learning Task**
|
||
# **Section Reference**: 7.1 - Data Preprocessing and Encodings
|
||
#
|
||
# ## Purpose
|
||
#
|
||
# This notebook implements **hands-on cleaning** for datasets that need it, plus
|
||
# demonstrates **split-aware preprocessing** mechanics. The key teaching artifact
|
||
# is the `SplitAwarePreprocessor` class that prevents lookahead bias.
|
||
#
|
||
# ## Learning Objectives
|
||
#
|
||
# 1. Apply domain filters, spike detection, and winsorization
|
||
# 2. Understand why preprocessing must be fit on training data only
|
||
# 3. Implement a reusable `SplitAwarePreprocessor` class
|
||
# 4. Clean US Equities and ETF datasets with audit trails
|
||
# 5. Perform cross-dataset alignment (different frequencies)
|
||
#
|
||
# ## Book Reference
|
||
#
|
||
# Section 7.1 emphasizes that **preprocessing choices affect model validity**.
|
||
# Fitting scalers or encoders on full data introduces lookahead bias.
|
||
#
|
||
# ## Prerequisites
|
||
#
|
||
# - `01_data_quality_diagnostics` — establishes the baseline coverage and
|
||
# outlier counts that motivate each cleaning step here.
|
||
# - Familiarity with leakage-aware splitting (Chapter 6 §6.3).
|
||
# - Polars expressions, Jupytext percent-format notebooks.
|
||
#
|
||
# ## Data Contract
|
||
#
|
||
# - **Input**: Raw datasets from data loaders
|
||
# - **Output**: In-memory cleaned DataFrames (teaching demonstration, not persisted)
|
||
|
||
# %%
|
||
"""Preprocessing Pipeline - Clean datasets with split-aware processing."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pickle
|
||
import tempfile
|
||
import warnings
|
||
from collections.abc import Callable
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Literal
|
||
|
||
import matplotlib.pyplot as plt
|
||
import numpy as np
|
||
import polars as pl
|
||
from IPython.display import display
|
||
from ml4t.engineer.preprocessing import StandardScaler
|
||
from sklearn.preprocessing import OneHotEncoder
|
||
|
||
from data import (
|
||
load_crypto_perps,
|
||
load_crypto_premium,
|
||
load_etfs,
|
||
load_firm_characteristics,
|
||
load_us_equities,
|
||
)
|
||
from utils.reproducibility import set_global_seeds
|
||
|
||
warnings.filterwarnings("ignore")
|
||
|
||
# %% tags=["parameters"]
|
||
# Production defaults
|
||
SEED = 42
|
||
US_EQUITIES_START_DATE = "1970-01-01"
|
||
ETF_START_DATE = "2015-01-01"
|
||
CRYPTO_START_DATE = "2021-01-01T00:00:00+00:00"
|
||
FIRM_CHARACTERISTICS_START_DATE = "1990-01-01"
|
||
|
||
# %%
|
||
set_global_seeds(SEED)
|
||
|
||
|
||
# %%
|
||
def ensure_symbol_alias(df: pl.DataFrame) -> pl.DataFrame:
|
||
"""Expose canonical asset identifiers under the symbol name when needed."""
|
||
if "asset" in df.columns and "symbol" not in df.columns:
|
||
return df.with_columns(pl.col("asset").alias("symbol"))
|
||
return df
|
||
|
||
|
||
# %%
|
||
def filter_from_start(df: pl.DataFrame, time_col: str, start_value: str) -> pl.DataFrame:
|
||
"""Apply a start-date filter without timezone/unit mismatches."""
|
||
start_date = datetime.fromisoformat(start_value).date()
|
||
dtype = df.schema[time_col]
|
||
if dtype == pl.Date:
|
||
return df.filter(pl.col(time_col) >= pl.lit(start_date).cast(pl.Date))
|
||
return df.filter(pl.col(time_col).dt.date() >= pl.lit(start_date).cast(pl.Date))
|
||
|
||
|
||
# %% [markdown]
|
||
# ## 1. Preprocessing Utilities
|
||
#
|
||
# Reusable functions for common data cleaning operations.
|
||
|
||
|
||
# %% [markdown]
|
||
# ### 1.1 Remove Duplicates
|
||
#
|
||
# Handles exact duplicates and near-duplicates with configurable strategy.
|
||
|
||
|
||
# %%
|
||
def remove_duplicates(
|
||
df: pl.DataFrame,
|
||
key_cols: list[str],
|
||
strategy: Literal["keep_first", "keep_last", "drop_all"] = "keep_last",
|
||
) -> tuple[pl.DataFrame, int]:
|
||
"""Remove duplicate rows based on key columns.
|
||
|
||
Args:
|
||
df: Input DataFrame
|
||
key_cols: Columns that define uniqueness
|
||
strategy: How to handle duplicates
|
||
- keep_first: Keep first occurrence
|
||
- keep_last: Keep last occurrence
|
||
- drop_all: Remove all duplicates
|
||
|
||
Returns:
|
||
Tuple of (cleaned DataFrame, number of rows removed)
|
||
"""
|
||
original_len = len(df)
|
||
|
||
if strategy == "keep_first":
|
||
result = df.unique(subset=key_cols, keep="first")
|
||
elif strategy == "keep_last":
|
||
result = df.unique(subset=key_cols, keep="last")
|
||
else: # drop_all
|
||
# Count occurrences and keep only unique rows
|
||
counts = df.group_by(key_cols).len()
|
||
unique_keys = counts.filter(pl.col("len") == 1).drop("len")
|
||
result = df.join(unique_keys, on=key_cols, how="inner")
|
||
|
||
n_removed = original_len - len(result)
|
||
return result, n_removed
|
||
|
||
|
||
# %% [markdown]
|
||
# ### 1.2 Fill Expected Gaps
|
||
#
|
||
# Distinguishes between expected gaps (weekends, holidays) and unexpected gaps.
|
||
|
||
|
||
# %%
|
||
def fill_expected_gaps(
|
||
df: pl.DataFrame,
|
||
time_col: str,
|
||
symbol_col: str | None,
|
||
method: Literal["ffill", "interpolate", "flag_only"] = "flag_only",
|
||
max_gap_days: int = 5,
|
||
) -> pl.DataFrame:
|
||
"""Handle gaps in time series data.
|
||
|
||
Args:
|
||
df: Input DataFrame
|
||
time_col: Time column name
|
||
symbol_col: Symbol column (None for single asset)
|
||
method: How to handle gaps
|
||
- ffill: Forward fill
|
||
- interpolate: Linear interpolation
|
||
- flag_only: Add flag column without filling
|
||
max_gap_days: Maximum gap to fill (longer gaps are flagged)
|
||
|
||
Returns:
|
||
DataFrame with gaps handled
|
||
"""
|
||
# Add gap flag column
|
||
if symbol_col and symbol_col in df.columns:
|
||
result = df.sort([symbol_col, time_col]).with_columns(
|
||
prev_date=pl.col(time_col).shift(1).over(symbol_col),
|
||
)
|
||
else:
|
||
result = df.sort(time_col).with_columns(
|
||
prev_date=pl.col(time_col).shift(1),
|
||
)
|
||
|
||
# Calculate gap in days
|
||
result = result.with_columns(gap_days=(pl.col(time_col) - pl.col("prev_date")).dt.total_days())
|
||
|
||
# Flag unexpected gaps (>1 day for daily, accounting for weekends)
|
||
result = result.with_columns(
|
||
is_gap=(pl.col("gap_days") > max_gap_days) & pl.col("prev_date").is_not_null()
|
||
)
|
||
|
||
# Clean up temporary columns
|
||
result = result.drop(["prev_date", "gap_days"])
|
||
|
||
return result
|
||
|
||
|
||
# %% [markdown]
|
||
# ### 1.3 Apply Domain Filters
|
||
#
|
||
# Remove rows with impossible values (negative prices, zero volume, etc.)
|
||
|
||
|
||
# %%
|
||
def apply_domain_filters(
|
||
df: pl.DataFrame,
|
||
rules: dict[str, Callable[[pl.Expr], pl.Expr]],
|
||
) -> tuple[pl.DataFrame, dict[str, int]]:
|
||
"""Apply domain validation rules and filter invalid rows.
|
||
|
||
Args:
|
||
df: Input DataFrame
|
||
rules: Dictionary of {rule_name: filter_expression}
|
||
Expression should return True for rows to KEEP
|
||
|
||
Returns:
|
||
Tuple of (filtered DataFrame, counts of removed rows per rule)
|
||
|
||
Example:
|
||
>>> rules = {
|
||
... "positive_close": lambda c: pl.col("close") > 0,
|
||
... "non_negative_volume": lambda c: pl.col("volume") >= 0,
|
||
... }
|
||
>>> clean_df, removed = apply_domain_filters(df, rules)
|
||
"""
|
||
removed_counts = {}
|
||
result = df
|
||
|
||
for rule_name, filter_fn in rules.items():
|
||
pre_len = len(result)
|
||
result = result.filter(filter_fn(None))
|
||
removed_counts[rule_name] = pre_len - len(result)
|
||
|
||
return result, removed_counts
|
||
|
||
|
||
# %% [markdown]
|
||
# ### 1.4 Spike Filter
|
||
#
|
||
# Detect and flag single-bar price reversals (potential data errors).
|
||
|
||
|
||
# %%
|
||
def spike_filter(
|
||
df: pl.DataFrame,
|
||
price_col: str = "close",
|
||
threshold: float = 0.5,
|
||
symbol_col: str | None = "symbol",
|
||
time_col: str = "timestamp",
|
||
action: Literal["flag", "remove", "replace"] = "flag",
|
||
) -> pl.DataFrame:
|
||
"""Detect and handle price spikes (single-bar reversals).
|
||
|
||
A spike is defined as:
|
||
- Return > threshold (e.g., 50% up)
|
||
- Followed by return < -threshold/(1+threshold) (reverting back)
|
||
|
||
Args:
|
||
df: Input DataFrame
|
||
price_col: Price column to check
|
||
threshold: Minimum return to be considered a spike
|
||
symbol_col: Symbol column for panel data
|
||
time_col: Time column for sorting
|
||
action: How to handle spikes
|
||
- flag: Add is_spike column
|
||
- remove: Remove spike rows
|
||
- replace: Replace spike with interpolated value
|
||
|
||
Returns:
|
||
DataFrame with spikes handled
|
||
"""
|
||
# Calculate returns
|
||
if symbol_col and symbol_col in df.columns:
|
||
result = df.sort([symbol_col, time_col]).with_columns(
|
||
ret=pl.col(price_col).pct_change().over(symbol_col),
|
||
ret_next=pl.col(price_col).pct_change().shift(-1).over(symbol_col),
|
||
)
|
||
else:
|
||
result = df.sort(time_col).with_columns(
|
||
ret=pl.col(price_col).pct_change(),
|
||
ret_next=pl.col(price_col).pct_change().shift(-1),
|
||
)
|
||
|
||
# Detect spikes (sharp move followed by reversion)
|
||
reversion_threshold = -threshold / (1 + threshold)
|
||
result = result.with_columns(
|
||
is_spike=((pl.col("ret") > threshold) & (pl.col("ret_next") < reversion_threshold))
|
||
| ((pl.col("ret") < -threshold / (1 + threshold)) & (pl.col("ret_next") > threshold))
|
||
)
|
||
|
||
if action == "remove":
|
||
result = result.filter(~pl.col("is_spike"))
|
||
elif action == "replace":
|
||
# Replace with geometric mean of before and after
|
||
if symbol_col and symbol_col in df.columns:
|
||
result = result.with_columns(
|
||
pl.when(pl.col("is_spike"))
|
||
.then(
|
||
(
|
||
pl.col(price_col).shift(1).over(symbol_col)
|
||
* pl.col(price_col).shift(-1).over(symbol_col)
|
||
).sqrt()
|
||
)
|
||
.otherwise(pl.col(price_col))
|
||
.alias(price_col)
|
||
)
|
||
else:
|
||
result = result.with_columns(
|
||
pl.when(pl.col("is_spike"))
|
||
.then((pl.col(price_col).shift(1) * pl.col(price_col).shift(-1)).sqrt())
|
||
.otherwise(pl.col(price_col))
|
||
.alias(price_col)
|
||
)
|
||
|
||
# Clean up temporary columns
|
||
result = result.drop(["ret", "ret_next"])
|
||
|
||
return result
|
||
|
||
|
||
# %% [markdown]
|
||
# ### 1.5 Winsorize Panel
|
||
#
|
||
# Clip extreme values at percentile thresholds, respecting panel structure.
|
||
|
||
|
||
# %%
|
||
def winsorize_panel(
|
||
df: pl.DataFrame,
|
||
fields: list[str],
|
||
limits: tuple[float, float] = (0.01, 0.99),
|
||
by_period: bool = True,
|
||
period_col: str = "timestamp",
|
||
) -> pl.DataFrame:
|
||
"""Winsorize (clip) extreme values at percentile bounds.
|
||
|
||
Args:
|
||
df: Input DataFrame
|
||
fields: Columns to winsorize
|
||
limits: (lower_percentile, upper_percentile) bounds
|
||
by_period: If True, compute bounds per time period (cross-sectional)
|
||
period_col: Column to group by for cross-sectional winsorization
|
||
|
||
Returns:
|
||
DataFrame with winsorized values
|
||
"""
|
||
result = df
|
||
|
||
for col_name in fields:
|
||
if col_name not in df.columns:
|
||
continue
|
||
|
||
if by_period:
|
||
# Cross-sectional winsorization (per date)
|
||
bounds = df.group_by(period_col).agg(
|
||
[
|
||
pl.col(col_name).quantile(limits[0]).alias("lower"),
|
||
pl.col(col_name).quantile(limits[1]).alias("upper"),
|
||
]
|
||
)
|
||
result = result.join(bounds, on=period_col, how="left")
|
||
result = result.with_columns(
|
||
pl.when(pl.col(col_name) < pl.col("lower"))
|
||
.then(pl.col("lower"))
|
||
.when(pl.col(col_name) > pl.col("upper"))
|
||
.then(pl.col("upper"))
|
||
.otherwise(pl.col(col_name))
|
||
.alias(col_name)
|
||
).drop(["lower", "upper"])
|
||
else:
|
||
# Global winsorization
|
||
lower = df[col_name].quantile(limits[0])
|
||
upper = df[col_name].quantile(limits[1])
|
||
result = result.with_columns(pl.col(col_name).clip(lower, upper))
|
||
|
||
return result
|
||
|
||
|
||
# %% [markdown]
|
||
# ## 2. SplitAwarePreprocessor
|
||
#
|
||
# The key pedagogical artifact: a preprocessing class that **learns parameters
|
||
# on training data only** and applies them to validation/test data.
|
||
#
|
||
# This prevents lookahead bias in:
|
||
# - Scaling (mean, std computed on train only)
|
||
# - Imputation (median/mode computed on train only)
|
||
# - Encoding (vocabulary built on train only)
|
||
|
||
|
||
# %%
|
||
@dataclass
|
||
class SplitAwarePreprocessor:
|
||
"""Preprocessor that fits on training data only to prevent lookahead bias.
|
||
|
||
Key principle: Any statistic used for preprocessing (mean, std, median,
|
||
encoder vocabulary) must be computed on training data only.
|
||
|
||
Example:
|
||
>>> preprocessor = SplitAwarePreprocessor(
|
||
... scale_cols=["returns", "volume"],
|
||
... winsorize_cols=["returns"],
|
||
... )
|
||
>>> preprocessor.fit(train_df)
|
||
>>> train_processed = preprocessor.transform(train_df)
|
||
>>> test_processed = preprocessor.transform(test_df)
|
||
"""
|
||
|
||
# Configuration
|
||
scale_cols: list[str] = field(default_factory=list)
|
||
winsorize_cols: list[str] = field(default_factory=list)
|
||
winsorize_limits: tuple[float, float] = (0.01, 0.99)
|
||
impute_cols: list[str] = field(default_factory=list)
|
||
impute_strategy: Literal["median", "mean", "zero"] = "median"
|
||
rank_cols: list[str] = field(default_factory=list)
|
||
|
||
# Learned parameters (set by fit)
|
||
_fitted: bool = field(default=False, init=False)
|
||
_scale_params: dict[str, dict[str, float]] = field(default_factory=dict, init=False)
|
||
_winsorize_params: dict[str, dict[str, float]] = field(default_factory=dict, init=False)
|
||
_impute_params: dict[str, float] = field(default_factory=dict, init=False)
|
||
|
||
def fit(self, train_df: pl.DataFrame) -> SplitAwarePreprocessor:
|
||
"""Learn preprocessing parameters from training data.
|
||
|
||
Args:
|
||
train_df: Training DataFrame
|
||
|
||
Returns:
|
||
Self (for method chaining)
|
||
"""
|
||
# Learn scaling parameters
|
||
for col in self.scale_cols:
|
||
if col in train_df.columns:
|
||
col_data = train_df[col].drop_nulls()
|
||
self._scale_params[col] = {
|
||
"mean": float(col_data.mean()),
|
||
"std": float(col_data.std()),
|
||
}
|
||
|
||
# Learn winsorization bounds
|
||
for col in self.winsorize_cols:
|
||
if col in train_df.columns:
|
||
col_data = train_df[col].drop_nulls()
|
||
self._winsorize_params[col] = {
|
||
"lower": float(col_data.quantile(self.winsorize_limits[0])),
|
||
"upper": float(col_data.quantile(self.winsorize_limits[1])),
|
||
}
|
||
|
||
# Learn imputation values
|
||
for col in self.impute_cols:
|
||
if col in train_df.columns:
|
||
col_data = train_df[col].drop_nulls()
|
||
if self.impute_strategy == "median":
|
||
self._impute_params[col] = float(col_data.median())
|
||
elif self.impute_strategy == "mean":
|
||
self._impute_params[col] = float(col_data.mean())
|
||
else:
|
||
self._impute_params[col] = 0.0
|
||
|
||
self._fitted = True
|
||
return self
|
||
|
||
def transform(self, df: pl.DataFrame) -> pl.DataFrame:
|
||
"""Apply learned preprocessing to any DataFrame.
|
||
|
||
Args:
|
||
df: DataFrame to transform
|
||
|
||
Returns:
|
||
Transformed DataFrame
|
||
"""
|
||
if not self._fitted:
|
||
raise ValueError("Preprocessor not fitted. Call fit() first.")
|
||
|
||
result = df
|
||
|
||
# Apply imputation
|
||
for col, fill_value in self._impute_params.items():
|
||
if col in result.columns:
|
||
result = result.with_columns(pl.col(col).fill_null(fill_value))
|
||
|
||
# Apply winsorization
|
||
for col, bounds in self._winsorize_params.items():
|
||
if col in result.columns:
|
||
result = result.with_columns(pl.col(col).clip(bounds["lower"], bounds["upper"]))
|
||
|
||
# Apply scaling
|
||
for col, params in self._scale_params.items():
|
||
if col in result.columns and params["std"] > 0:
|
||
result = result.with_columns(
|
||
((pl.col(col) - params["mean"]) / params["std"]).alias(col)
|
||
)
|
||
|
||
# Apply cross-sectional ranking (doesn't require fitting)
|
||
for col in self.rank_cols:
|
||
if col in result.columns and "timestamp" in result.columns:
|
||
result = result.with_columns(
|
||
pl.col(col).rank().over("timestamp").alias(f"{col}_rank")
|
||
)
|
||
|
||
return result
|
||
|
||
def fit_transform(self, train_df: pl.DataFrame) -> pl.DataFrame:
|
||
"""Fit and transform in one step."""
|
||
return self.fit(train_df).transform(train_df)
|
||
|
||
def save(self, path: Path) -> None:
|
||
"""Save fitted preprocessor to disk."""
|
||
if not self._fitted:
|
||
raise ValueError("Cannot save unfitted preprocessor.")
|
||
with open(path, "wb") as f:
|
||
pickle.dump(self, f)
|
||
|
||
@classmethod
|
||
def load(cls, path: Path) -> SplitAwarePreprocessor:
|
||
"""Load fitted preprocessor from disk."""
|
||
with open(path, "rb") as f:
|
||
return pickle.load(f)
|
||
|
||
|
||
# %% [markdown]
|
||
# ## 3. US Equities Deep Clean
|
||
#
|
||
# The most complex dataset and the best teaching vehicle for preprocessing.
|
||
# We apply four sequential cleaning steps: penny stock filter, domain
|
||
# validation, extreme return removal, and spike detection.
|
||
|
||
# %%
|
||
us_equities = None
|
||
try:
|
||
us_equities = ensure_symbol_alias(load_us_equities())
|
||
us_equities = filter_from_start(us_equities, "timestamp", US_EQUITIES_START_DATE)
|
||
print(f"Loaded {len(us_equities):,} rows, {us_equities['symbol'].n_unique()} symbols")
|
||
except Exception as e:
|
||
print(f"Could not load US Equities: {type(e).__name__}")
|
||
|
||
# %% [markdown]
|
||
# ### Step 1: Remove penny stocks
|
||
#
|
||
# Stocks trading below $1 introduce microstructure noise that dominates
|
||
# cross-sectional models. Filter them early.
|
||
|
||
# %%
|
||
if us_equities is not None:
|
||
original_rows = len(us_equities)
|
||
cleaned = us_equities.filter(pl.col("close") >= 1.0)
|
||
penny_removed = original_rows - len(cleaned)
|
||
print(
|
||
f"Penny stocks removed: {penny_removed:,} rows ({100 * penny_removed / original_rows:.1f}%)"
|
||
)
|
||
|
||
# %% [markdown]
|
||
# ### Step 2: Domain filters
|
||
#
|
||
# Remove rows violating physical constraints: negative prices, negative volume,
|
||
# and OHLC inconsistency (high < low, etc.).
|
||
|
||
# %%
|
||
if us_equities is not None:
|
||
domain_rules = {
|
||
"positive_prices": lambda _: (
|
||
(pl.col("open") > 0)
|
||
& (pl.col("high") > 0)
|
||
& (pl.col("low") > 0)
|
||
& (pl.col("close") > 0)
|
||
),
|
||
"non_negative_volume": lambda _: pl.col("volume") >= 0,
|
||
"ohlc_consistency": lambda _: (
|
||
(pl.col("low") <= pl.col("open"))
|
||
& (pl.col("low") <= pl.col("close"))
|
||
& (pl.col("high") >= pl.col("open"))
|
||
& (pl.col("high") >= pl.col("close"))
|
||
),
|
||
}
|
||
cleaned, removed_counts = apply_domain_filters(cleaned, domain_rules)
|
||
for rule, count in removed_counts.items():
|
||
if count > 0:
|
||
print(f"{rule}: removed {count:,} rows")
|
||
|
||
# %% [markdown]
|
||
# ### Step 3: Extreme returns
|
||
#
|
||
# Daily returns exceeding 200% typically indicate stock splits or data errors.
|
||
|
||
# %%
|
||
if us_equities is not None:
|
||
cleaned = cleaned.sort(["symbol", "timestamp"]).with_columns(
|
||
returns=pl.col("close").pct_change().over("symbol")
|
||
)
|
||
extreme_returns = cleaned.filter(pl.col("returns").abs() > 2.0)
|
||
print(f"Extreme returns (>200%): {len(extreme_returns):,} rows")
|
||
cleaned = cleaned.filter(pl.col("returns").is_null() | (pl.col("returns").abs() <= 2.0))
|
||
|
||
# %% [markdown]
|
||
# ### Step 4: Spike detection
|
||
#
|
||
# Single-bar reversals (sharp move followed by immediate reversion) are
|
||
# likely data errors rather than genuine price moves.
|
||
|
||
# %%
|
||
if us_equities is not None:
|
||
cleaned = spike_filter(cleaned, "close", threshold=0.5, action="flag")
|
||
n_spikes = cleaned["is_spike"].sum()
|
||
print(f"Flagged spikes: {n_spikes}")
|
||
print(f"\nFinal: {len(cleaned):,} rows, {cleaned['symbol'].n_unique()} symbols")
|
||
print(f"Retention: {100 * len(cleaned) / original_rows:.1f}%")
|
||
us_equities_cleaned = cleaned
|
||
|
||
# %% [markdown]
|
||
# ### Winsorization effect
|
||
#
|
||
# Before saving, visualize how winsorization compresses the return distribution
|
||
# tails. This motivates the robust scaling discussion in Section 7.1.
|
||
|
||
# %%
|
||
if us_equities is not None:
|
||
raw_returns = us_equities_cleaned["returns"].drop_nulls().to_numpy()
|
||
winsorized = winsorize_panel(
|
||
us_equities_cleaned.filter(pl.col("returns").is_not_null()),
|
||
fields=["returns"],
|
||
limits=(0.01, 0.99),
|
||
by_period=False,
|
||
)["returns"].to_numpy()
|
||
|
||
fig, axes = plt.subplots(1, 2, figsize=(12, 4), sharey=True)
|
||
|
||
bin_edges = np.linspace(-0.15, 0.15, 201)
|
||
axes[0].hist(raw_returns, bins=bin_edges, alpha=0.8)
|
||
axes[0].set_title("Raw Returns")
|
||
axes[0].set_xlabel("Daily Return")
|
||
axes[0].set_ylabel("Count")
|
||
|
||
axes[1].hist(winsorized, bins=bin_edges, alpha=0.8)
|
||
axes[1].set_title("Winsorized Returns (1st/99th)")
|
||
axes[1].set_xlabel("Daily Return")
|
||
|
||
fig.suptitle("Effect of Winsorization on US Equities Return Distribution")
|
||
fig.tight_layout()
|
||
fig.show()
|
||
|
||
# %% [markdown]
|
||
# Winsorization clips the extreme tails without distorting the bulk of the
|
||
# distribution. The 1st/99th percentile bounds remove genuine outliers while
|
||
# preserving the fat-tailed shape that characterizes equity returns.
|
||
|
||
# %% [markdown]
|
||
# ### Cleaned US Equities Summary
|
||
#
|
||
# The cleaning pipeline is a teaching demonstration — downstream case study
|
||
# notebooks apply their own cleaning via loaders and feature engineering.
|
||
|
||
# %%
|
||
if us_equities is not None:
|
||
save_cols = [
|
||
c for c in us_equities_cleaned.columns if c not in ["returns", "is_spike", "is_gap"]
|
||
]
|
||
cleaned = us_equities_cleaned.select(save_cols)
|
||
print(f"Cleaned US Equities: {len(cleaned):,} rows, {cleaned['symbol'].n_unique()} symbols")
|
||
print(f"Date range: {cleaned['timestamp'].min()} to {cleaned['timestamp'].max()}")
|
||
|
||
|
||
# %% [markdown]
|
||
# ## 4. ETF Universe Cleanup
|
||
#
|
||
# Yahoo Finance data has specific issues: adjustment artifacts from splits
|
||
# and distributions, ticker changes, and occasional data gaps.
|
||
|
||
# %%
|
||
etfs = None
|
||
try:
|
||
etfs = ensure_symbol_alias(load_etfs())
|
||
if "timestamp" in etfs.columns:
|
||
etfs = etfs.with_columns(pl.col("timestamp").dt.date().alias("timestamp"))
|
||
etfs = filter_from_start(etfs, "timestamp", ETF_START_DATE)
|
||
print(f"Loaded {len(etfs):,} rows, {etfs['symbol'].n_unique()} symbols")
|
||
except Exception as e:
|
||
print(f"Could not load ETFs: {type(e).__name__}")
|
||
|
||
# %%
|
||
if etfs is not None:
|
||
original_rows = len(etfs)
|
||
|
||
# Detect overnight jumps matching common split ratios
|
||
etfs_sorted = etfs.sort(["symbol", "timestamp"]).with_columns(
|
||
overnight_return=(pl.col("open") / pl.col("close").shift(1).over("symbol") - 1)
|
||
)
|
||
split_ratios = [0.5, 0.333, 0.25, 2.0, 3.0, 4.0]
|
||
tolerance = 0.01
|
||
potential_issues = etfs_sorted.filter(
|
||
pl.any_horizontal(
|
||
[(pl.col("overnight_return") - (ratio - 1)).abs() < tolerance for ratio in split_ratios]
|
||
)
|
||
)
|
||
print(f"Adjustment artifacts: {len(potential_issues)} potential split-like jumps")
|
||
|
||
# Domain filters
|
||
domain_rules = {
|
||
"positive_prices": lambda _: pl.col("close") > 0,
|
||
"non_negative_volume": lambda _: pl.col("volume") >= 0,
|
||
}
|
||
etfs_cleaned, removed_counts = apply_domain_filters(etfs_sorted, domain_rules)
|
||
for rule, count in removed_counts.items():
|
||
if count > 0:
|
||
print(f"{rule}: removed {count:,} rows")
|
||
|
||
# Gap check
|
||
etfs_cleaned = fill_expected_gaps(
|
||
etfs_cleaned, "timestamp", "symbol", method="flag_only", max_gap_days=5
|
||
)
|
||
n_gaps = etfs_cleaned["is_gap"].sum()
|
||
print(f"Unexpected gaps (>5 days): {n_gaps}")
|
||
print(
|
||
f"Final: {len(etfs_cleaned):,} rows ({100 * len(etfs_cleaned) / original_rows:.1f}% retained)"
|
||
)
|
||
|
||
etfs_cleaned = etfs_cleaned.drop(["overnight_return"])
|
||
|
||
# %%
|
||
if etfs is not None:
|
||
save_cols = [c for c in etfs_cleaned.columns if c not in ["is_gap"]]
|
||
cleaned = etfs_cleaned.select(save_cols)
|
||
print(f"Cleaned ETFs: {len(cleaned):,} rows, {cleaned['symbol'].n_unique()} symbols")
|
||
print(f"Date range: {cleaned['timestamp'].min()} to {cleaned['timestamp'].max()}")
|
||
|
||
# %% [markdown]
|
||
# ETF cleaning is lighter than US Equities because the universe is curated
|
||
# (100 liquid ETFs). The main concerns are adjustment artifacts from Yahoo
|
||
# Finance and occasional data gaps around holidays or ticker changes.
|
||
|
||
|
||
# %% [markdown]
|
||
# ## 5. Cross-Dataset Alignment Demo
|
||
#
|
||
# When combining datasets with different frequencies (daily equities with
|
||
# monthly characteristics, or 8-hour crypto bars), alignment must preserve
|
||
# point-in-time correctness.
|
||
|
||
|
||
# %% [markdown]
|
||
# ### 5.1 Crypto Spot + Perps Alignment
|
||
#
|
||
# Aligning 8-hour bars for basis computation: the premium index
|
||
# captures the funding rate differential between spot and perpetual futures.
|
||
|
||
# %%
|
||
crypto_perps = None
|
||
crypto_premium = None
|
||
|
||
try:
|
||
crypto_perps = ensure_symbol_alias(load_crypto_perps(frequency="8h"))
|
||
crypto_premium = ensure_symbol_alias(load_crypto_premium(frequency="8h"))
|
||
crypto_perps = filter_from_start(crypto_perps, "timestamp", CRYPTO_START_DATE)
|
||
crypto_premium = filter_from_start(crypto_premium, "timestamp", CRYPTO_START_DATE)
|
||
print(f"Perps: {len(crypto_perps):,} rows | Premium: {len(crypto_premium):,} rows")
|
||
except Exception as e:
|
||
print(f"Could not load crypto data: {type(e).__name__}")
|
||
|
||
# %%
|
||
if crypto_perps is not None and crypto_premium is not None:
|
||
aligned = crypto_perps.join(
|
||
crypto_premium.select(
|
||
["timestamp", "symbol", pl.col("premium_index_close").alias("premium")]
|
||
),
|
||
on=["timestamp", "symbol"],
|
||
how="inner",
|
||
)
|
||
aligned = aligned.with_columns(basis=pl.col("premium") * 100)
|
||
|
||
print(f"Aligned: {len(aligned):,} rows, {aligned['symbol'].n_unique()} symbols")
|
||
display(aligned.select("basis").describe())
|
||
|
||
# %% [markdown]
|
||
# The inner join ensures we only keep timestamps where both datasets have
|
||
# observations. The basis (premium × 100) converts the raw premium index
|
||
# to percentage points for readability.
|
||
|
||
# %% [markdown]
|
||
# ### 5.2 Equity + Firm Characteristics As-Of Join
|
||
#
|
||
# Monthly characteristics must be joined to daily prices using point-in-time
|
||
# logic: each daily observation gets the most recent monthly snapshot.
|
||
|
||
# %%
|
||
firm_char = None
|
||
try:
|
||
firm_char = load_firm_characteristics()
|
||
firm_char = firm_char.filter(
|
||
pl.col("timestamp") >= datetime.fromisoformat(FIRM_CHARACTERISTICS_START_DATE)
|
||
)
|
||
print(f"Firm chars: {len(firm_char):,} rows, {len(firm_char.columns)} columns")
|
||
except Exception as e:
|
||
print(f"Could not load firm characteristics: {type(e).__name__}")
|
||
|
||
# %%
|
||
if firm_char is not None:
|
||
char_cols = sorted(c for c in firm_char.columns if c not in ["timestamp", "split", "ret"])[:5]
|
||
split_counts = firm_char.group_by("split").agg(pl.len().alias("n_obs")).sort("split")
|
||
display(split_counts)
|
||
print(f"Sample characteristics: {char_cols}")
|
||
print(f"Date range: {firm_char['timestamp'].min()} to {firm_char['timestamp'].max()}")
|
||
|
||
# %% [markdown]
|
||
# The as-of join pattern for monthly characteristics:
|
||
#
|
||
# ```python
|
||
# daily_df.with_columns(month=pl.col('timestamp').dt.truncate('1mo'))
|
||
# daily_df.join(monthly_chars, left_on=['permno', 'month'], right_on=['permno', 'timestamp'])
|
||
# ```
|
||
#
|
||
# This ensures each daily observation sees only information available at
|
||
# that point in time, preventing lookahead from future characteristic updates.
|
||
|
||
|
||
# %% [markdown]
|
||
# ## 6. Categorical Encoding Demo
|
||
#
|
||
# Section 7.1 discusses categorical encodings (one-hot, ordinal, hashing).
|
||
# The key constraint: **fit the encoder on training data only** so that
|
||
# unseen categories in the test set are handled gracefully.
|
||
|
||
# %%
|
||
# Create sample data with a categorical field (NumPy RNG is seeded globally in the preamble)
|
||
n_samples = 500
|
||
sectors = ["Technology", "Finance", "Healthcare", "Energy", "Consumer"]
|
||
train_sectors = np.random.choice(sectors, size=n_samples)
|
||
test_sectors = np.random.choice(sectors + ["Utilities"], size=100) # Unseen category
|
||
|
||
cat_train = pl.DataFrame({"sector": train_sectors, "returns": np.random.randn(n_samples) * 0.02})
|
||
cat_test = pl.DataFrame({"sector": test_sectors, "returns": np.random.randn(100) * 0.02})
|
||
|
||
print(f"Train sectors: {sorted(str(s) for s in set(train_sectors))}")
|
||
print(f"Test sectors: {sorted(str(s) for s in set(test_sectors))}")
|
||
|
||
# %%
|
||
# Fit encoder on training data only
|
||
encoder = OneHotEncoder(sparse_output=False, handle_unknown="ignore")
|
||
encoder.fit(cat_train.select("sector").to_pandas())
|
||
|
||
# Transform both splits
|
||
train_encoded = encoder.transform(cat_train.select("sector").to_pandas())
|
||
test_encoded = encoder.transform(cat_test.select("sector").to_pandas())
|
||
|
||
print(f"Encoded columns: {encoder.get_feature_names_out().tolist()}")
|
||
print(f"Train shape: {train_encoded.shape}")
|
||
print(f"Test shape: {test_encoded.shape}")
|
||
|
||
# %%
|
||
# Show how unseen "Utilities" category is handled
|
||
utilities_mask = cat_test["sector"] == "Utilities"
|
||
n_utilities = utilities_mask.sum()
|
||
utilities_encoded = test_encoded[utilities_mask.to_numpy()]
|
||
print(f"\nUnseen 'Utilities' rows: {n_utilities}")
|
||
print(f"Encoding (all zeros): {utilities_encoded[0] if n_utilities > 0 else 'N/A'}")
|
||
|
||
# %% [markdown]
|
||
# The `handle_unknown="ignore"` setting maps unseen categories to all-zero
|
||
# vectors. This is the correct behavior for split-aware encoding: the model
|
||
# sees a neutral representation for categories it wasn't trained on, rather
|
||
# than crashing or silently misencoding.
|
||
|
||
|
||
# %% [markdown]
|
||
# ## 7. Split-Aware Preprocessing Demo
|
||
#
|
||
# The critical lesson: **preprocessing parameters must be learned on training
|
||
# data only**. Fitting on the full dataset leaks future information into the
|
||
# training representation.
|
||
|
||
|
||
# %% [markdown]
|
||
# ### 7.1 Correct Approach: Fit on Train Only
|
||
|
||
# %%
|
||
n_train, n_test = 1000, 200
|
||
|
||
sample_data = pl.DataFrame(
|
||
{
|
||
"timestamp": pl.date_range(datetime(2020, 1, 1), datetime(2023, 12, 31), eager=True)[
|
||
: n_train + n_test
|
||
],
|
||
"returns": np.random.randn(n_train + n_test) * 0.02,
|
||
"volume": np.random.exponential(1e6, n_train + n_test),
|
||
}
|
||
)
|
||
|
||
# Add regime-dependent extreme values (March = high volatility)
|
||
sample_data = sample_data.with_columns(
|
||
returns=pl.when(pl.col("timestamp").dt.month() == 3)
|
||
.then(pl.col("returns") * 3)
|
||
.otherwise(pl.col("returns"))
|
||
)
|
||
|
||
train_df = sample_data.head(n_train)
|
||
test_df = sample_data.tail(n_test)
|
||
print(f"Train: {len(train_df)} rows | Test: {len(test_df)} rows")
|
||
|
||
# %%
|
||
preprocessor = SplitAwarePreprocessor(
|
||
scale_cols=["returns"],
|
||
winsorize_cols=["returns"],
|
||
winsorize_limits=(0.01, 0.99),
|
||
)
|
||
preprocessor.fit(train_df)
|
||
|
||
train_processed = preprocessor.transform(train_df)
|
||
test_processed = preprocessor.transform(test_df)
|
||
|
||
print("Learned from training data:")
|
||
print(f" Mean: {preprocessor._scale_params['returns']['mean']:.6f}")
|
||
print(f" Std: {preprocessor._scale_params['returns']['std']:.6f}")
|
||
print(
|
||
f" Winsorize: [{preprocessor._winsorize_params['returns']['lower']:.4f}, "
|
||
f"{preprocessor._winsorize_params['returns']['upper']:.4f}]"
|
||
)
|
||
print(
|
||
f"\nTrain processed: mean={train_processed['returns'].mean():.4f}, "
|
||
f"std={train_processed['returns'].std():.4f}"
|
||
)
|
||
print(
|
||
f"Test processed: mean={test_processed['returns'].mean():.4f}, "
|
||
f"std={test_processed['returns'].std():.4f}"
|
||
)
|
||
|
||
# %% [markdown]
|
||
# The test set mean and standard deviation differ from 0 and 1 because the
|
||
# scaler uses training-period parameters. This is correct behavior: the
|
||
# test set represents unseen future data with potentially different statistics.
|
||
|
||
|
||
# %% [markdown]
|
||
# ### 7.2 WRONG Approach: Fit on Full Data (Leakage Demo)
|
||
#
|
||
# What happens when we cheat and fit on all data including the test set?
|
||
|
||
# %%
|
||
full_data = pl.concat([train_df, test_df])
|
||
leaky_preprocessor = SplitAwarePreprocessor(scale_cols=["returns"])
|
||
leaky_preprocessor.fit(full_data) # BUG: includes future data!
|
||
|
||
correct_mean = preprocessor._scale_params["returns"]["mean"]
|
||
leaky_mean = leaky_preprocessor._scale_params["returns"]["mean"]
|
||
correct_std = preprocessor._scale_params["returns"]["std"]
|
||
leaky_std = leaky_preprocessor._scale_params["returns"]["std"]
|
||
|
||
print(
|
||
f"Correct mean: {correct_mean:.6f} | Leaky mean: {leaky_mean:.6f} | Diff: {abs(correct_mean - leaky_mean):.6f}"
|
||
)
|
||
print(
|
||
f"Correct std: {correct_std:.6f} | Leaky std: {leaky_std:.6f} | Diff: {abs(correct_std - leaky_std):.6f}"
|
||
)
|
||
|
||
# Transform test set with leaky parameters for comparison
|
||
leaky_test_processed = leaky_preprocessor.transform(test_df)
|
||
|
||
# %% [markdown]
|
||
# The parameter differences appear small in this synthetic example. But across
|
||
# dozens of features and thousands of test observations, the cumulative effect
|
||
# of leaking future information can meaningfully inflate apparent performance.
|
||
# More fundamentally, it violates the train/test boundary that gives evaluation
|
||
# its meaning.
|
||
|
||
# %% [markdown]
|
||
# ### Leakage comparison figure
|
||
#
|
||
# Visualize how the two scaling approaches produce different test-set distributions.
|
||
|
||
# %%
|
||
fig, axes = plt.subplots(1, 2, figsize=(12, 4), sharey=True)
|
||
|
||
correct_vals = test_processed["returns"].to_numpy()
|
||
leaky_vals = leaky_test_processed["returns"].to_numpy()
|
||
|
||
axes[0].hist(correct_vals, bins=np.linspace(correct_vals.min(), correct_vals.max(), 51), alpha=0.8)
|
||
axes[0].axvline(
|
||
correct_vals.mean(),
|
||
color="C1",
|
||
linestyle="--",
|
||
linewidth=2,
|
||
label=f"Mean={correct_vals.mean():.3f}",
|
||
)
|
||
axes[0].set_title("Correct: Train-Only Fit")
|
||
axes[0].set_xlabel("Scaled Return")
|
||
axes[0].set_ylabel("Count")
|
||
axes[0].legend()
|
||
|
||
axes[1].hist(leaky_vals, bins=np.linspace(leaky_vals.min(), leaky_vals.max(), 51), alpha=0.8)
|
||
axes[1].axvline(
|
||
leaky_vals.mean(),
|
||
color="C1",
|
||
linestyle="--",
|
||
linewidth=2,
|
||
label=f"Mean={leaky_vals.mean():.3f}",
|
||
)
|
||
axes[1].set_title("Leaky: Full-Data Fit")
|
||
axes[1].set_xlabel("Scaled Return")
|
||
axes[1].legend()
|
||
|
||
fig.suptitle("Distribution of Scaled Test Returns: Correct vs Leaky Preprocessing")
|
||
fig.tight_layout()
|
||
fig.show()
|
||
|
||
# %% [markdown]
|
||
# The location and scale shifts between the two panels demonstrate information
|
||
# leakage. In production with many features, this systematic bias accumulates
|
||
# and can meaningfully inflate Sharpe ratios and IC estimates.
|
||
|
||
|
||
# %% [markdown]
|
||
# ### 7.3 Walk-Forward Refit Demo
|
||
#
|
||
# In walk-forward evaluation (Chapter 6), the preprocessor must be **refit
|
||
# at each fold boundary** using only data available up to that point.
|
||
# Parameters drift over time as the data distribution evolves.
|
||
|
||
# %%
|
||
# Simple 3-fold walk-forward loop
|
||
fold_size = len(sample_data) // 4
|
||
folds = [
|
||
(sample_data[:fold_size], sample_data[fold_size : 2 * fold_size]),
|
||
(sample_data[: 2 * fold_size], sample_data[2 * fold_size : 3 * fold_size]),
|
||
(sample_data[: 3 * fold_size], sample_data[3 * fold_size :]),
|
||
]
|
||
|
||
fold_rows = []
|
||
for i, (fold_train, fold_test) in enumerate(folds, 1):
|
||
fold_pp = SplitAwarePreprocessor(
|
||
scale_cols=["returns"],
|
||
winsorize_cols=["returns"],
|
||
winsorize_limits=(0.01, 0.99),
|
||
)
|
||
fold_pp.fit(fold_train)
|
||
params = fold_pp._scale_params["returns"]
|
||
wparams = fold_pp._winsorize_params["returns"]
|
||
fold_rows.append(
|
||
{
|
||
"fold": i,
|
||
"n_train": len(fold_train),
|
||
"n_test": len(fold_test),
|
||
"mean": round(params["mean"], 6),
|
||
"std": round(params["std"], 6),
|
||
"winsor_lo": round(wparams["lower"], 4),
|
||
"winsor_hi": round(wparams["upper"], 4),
|
||
}
|
||
)
|
||
|
||
fold_summary = pl.DataFrame(fold_rows)
|
||
fold_summary
|
||
|
||
# %% [markdown]
|
||
# Parameters vary across folds as the expanding training window incorporates
|
||
# new periods. The variation here is small because the synthetic data is
|
||
# stationary by construction; in real markets, regime shifts produce larger
|
||
# swings in mean, variance, and tail bounds. The walk-forward protocol ensures
|
||
# each test fold sees parameters estimated only from its past, so any drift
|
||
# (small or large) is handled correctly.
|
||
|
||
|
||
# %% [markdown]
|
||
# ### 7.4 Preprocessor Serialization Demo
|
||
#
|
||
# Demonstrate save/load round-trip. Note: `pickle` is used here for
|
||
# simplicity. For production systems, prefer `ml4t-engineer` serialization.
|
||
|
||
# %%
|
||
with tempfile.NamedTemporaryFile(suffix=".pkl", delete=True) as tmp:
|
||
preprocessor.save(tmp.name)
|
||
loaded_preprocessor = SplitAwarePreprocessor.load(tmp.name)
|
||
print(f"Round-trip OK: fitted={loaded_preprocessor._fitted}")
|
||
print(f" Scale params: {loaded_preprocessor._scale_params}")
|
||
|
||
|
||
# %% [markdown]
|
||
# ### 7.5 Production Alternative: ml4t-engineer
|
||
#
|
||
# The manual `SplitAwarePreprocessor` above teaches the principle. In
|
||
# practice, use the tested library version which provides the same
|
||
# split-aware semantics with better performance and serialization.
|
||
|
||
# %%
|
||
# Compare manual scale-only (no winsorization) against library
|
||
manual_scale_only = SplitAwarePreprocessor(scale_cols=["returns"])
|
||
manual_scale_only.fit(train_df)
|
||
manual_test_scaled = manual_scale_only.transform(test_df)
|
||
|
||
lib_scaler = StandardScaler(columns=["returns"])
|
||
train_lib = lib_scaler.fit_transform(train_df)
|
||
test_lib = lib_scaler.transform(test_df)
|
||
|
||
print(
|
||
f"Library scaler (train): mean={train_lib['returns'].mean():.4f}, std={train_lib['returns'].std():.4f}"
|
||
)
|
||
print(
|
||
f"Library scaler (test): mean={test_lib['returns'].mean():.4f}, std={test_lib['returns'].std():.4f}"
|
||
)
|
||
|
||
manual_test_mean = manual_test_scaled["returns"].mean()
|
||
lib_test_mean = test_lib["returns"].mean()
|
||
print(f"\nManual vs library difference: {abs(manual_test_mean - lib_test_mean):.6f}")
|
||
|
||
# %% [markdown]
|
||
# Comparing the manual scaler (without winsorization) against the library
|
||
# implementation confirms they produce consistent results. Any small
|
||
# differences reflect implementation details in standard deviation computation
|
||
# (sample vs population).
|
||
|
||
# %% [markdown]
|
||
# ## 8. Final Verification
|
||
#
|
||
# Quick quality check on the cleaned data (in-memory, not persisted).
|
||
|
||
# %%
|
||
for name, df in [("US Equities", us_equities_cleaned), ("ETFs", etfs_cleaned)]:
|
||
if df is not None:
|
||
null_close = df["close"].is_null().sum()
|
||
neg_close = (df["close"] < 0).sum()
|
||
print(
|
||
f"{name}: {len(df):,} rows, {df['symbol'].n_unique()} symbols, "
|
||
f"null_close={null_close}, neg_close={neg_close}"
|
||
)
|
||
|
||
# %% [markdown]
|
||
# Both cleaned datasets pass basic quality checks: no null or negative
|
||
# prices remain. This notebook demonstrates the cleaning techniques —
|
||
# downstream case study notebooks apply their own cleaning via loaders.
|
||
|
||
|
||
# %% [markdown]
|
||
# ## Key Takeaways
|
||
#
|
||
# 1. **Preprocessing must be split-aware** — fit parameters on training data only
|
||
# 2. **Domain filters catch obvious errors** — negative prices, impossible OHLC relations
|
||
# 3. **Spike detection identifies data artifacts** — single-bar reversals often indicate errors
|
||
# 4. **Winsorization handles outliers** — but bounds must come from training data
|
||
# 5. **Categorical encoding** must handle unseen categories at test time
|
||
# 6. **Walk-forward refit** captures parameter drift across market regimes
|
||
# 7. **Cross-dataset alignment requires care** — use as-of joins for different frequencies
|
||
#
|
||
# **Next**: See `05_signal_evaluation` for IC-based signal quality assessment.
|
||
# **Book**: Section 7.1 discusses why preprocessing choices affect model validity.
|