金融时间序列的防泄漏预处理

# 02_preprocessing_pipeline.py


```py
# ---
# 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
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
from utils.style import show_with_alt

# %% 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]
# ## Preprocessing Utilities
#
# Reusable functions for common data cleaning operations.


# %% [markdown]
# ### 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]
# ### 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]
# ### 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]
# ### 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]
# ### 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]
# ## 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]
# ## US Equities Deep Clean
#
# The most complex dataset here, and the one where every cleaning step has something to
# remove. We apply four sequential 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: Implausibly large one-day moves
#
# A raw return is bounded below by $-1$, so a threshold on $|r|$ set above 1 can only ever
# fire on the upside. `01_data_quality_diagnostics` works that through on this same panel:
# the rule catches most reverse splits and essentially no forward ones, and forward splits
# are the large majority.
#
# Read the two directions separately, because they do not agree here. Most of the rows this
# step removes *are* split days - the cell prints how many, from `split_ratio` itself - so
# the rule is reasonably precise about what it takes out. What it cannot do is find splits,
# because the kind that divides the price can never clear the threshold. A filter you
# cannot use as a detector is still a usable filter, and that is all this step is: it drops
# implausibly large upward jumps before anything fits a scaler. The smallest return it
# removes cannot be negative for any threshold at or above 1, and this one is well above.

# %%
MAX_ABS_DAILY_RETURN = 2.0

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() > MAX_ABS_DAILY_RETURN)
    on_a_split_day = extreme_returns.filter(
        (pl.col("split_ratio") != 1) & pl.col("split_ratio").is_not_null()
    ).height
    smallest_removed = extreme_returns["returns"].min()
    print(f"Rows with |return| > {MAX_ABS_DAILY_RETURN}: {len(extreme_returns):,}")
    print(f"  that split_ratio records as a split: {on_a_split_day:,}")
    if smallest_removed is None:
        print("  smallest return removed:             none, the filter removed nothing")
    else:
        print(f"  smallest return removed:             {smallest_removed:+.4f}")
    cleaned = cleaned.filter(
        pl.col("returns").is_null() | (pl.col("returns").abs() <= MAX_ABS_DAILY_RETURN)
    )

# %% [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("US equities daily returns, raw and winsorized at the 1st/99th")
    show_with_alt(
        fig,
        alt=(
            "Two histograms of daily returns side by side on identical bins and a shared "
            "count axis, raw on the left and winsorized at the 1st and 99th percentiles "
            "on the right. Both are sharply peaked at zero with a very tall narrow spike "
            "exactly at zero. The left panel's thin tails run out to the edges of the "
            "range; in the right panel they stop, and a narrow spike stands at each "
            "clipping bound where that tail mass has been piled up. The central bulk is "
            "indistinguishable between the two."
        ),
    )

# %% [markdown]
# The bulk of the distribution is untouched. What moves is the far tail, and it does not
# move by being discarded: winsorizing reassigns every row past the 1st or 99th percentile
# onto that percentile, which is why the right panel grows a narrow spike at each bound
# where the left panel has a thin tail. The fat-tailed shape is intact *inside* the bounds
# and gone outside them, replaced by two point masses. That is the trade the method
# makes - a scaler fitted on the clipped series is no longer dragged by a handful of
# extreme days, and in exchange the series can no longer say how extreme those days were.
#
# Both panels also carry a tall spike at exactly zero. Those are days on which the close
# did not move at all. They are a property of the raw panel, not of anything winsorizing
# did to it.

# %% [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]
# ## 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]
# ## 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]
# ### Crypto Spot and 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]
# ### Equity and 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]
# ## 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]
# ## 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]
# ### 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]
# ### Wrong Approach: Fit on Full Data (Leakage Demo)
#
# What happens when we cheat and fit on all data including the test set? The preprocessor
# below is configured exactly like the correct one - same winsorize limits, same scaled
# column - so the fit data is the only thing that differs, and the leakage effect is the
# only thing the comparison can be measuring.

# %%
full_data = pl.concat([train_df, test_df])
leaky_preprocessor = SplitAwarePreprocessor(
    scale_cols=["returns"],
    winsorize_cols=["returns"],
    winsorize_limits=(0.01, 0.99),
)
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. The
# two panels share their bins and both axes, so the only thing that can differ between
# them is the fit data - not the binning.

# %%
fig, axes = plt.subplots(1, 2, figsize=(12, 4), sharex=True, sharey=True)

correct_vals = test_processed["returns"].to_numpy()
leaky_vals = leaky_test_processed["returns"].to_numpy()

bin_edges = np.linspace(
    min(correct_vals.min(), leaky_vals.min()),
    max(correct_vals.max(), leaky_vals.max()),
    51,
)

axes[0].hist(correct_vals, bins=bin_edges, 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=bin_edges, 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("Scaled test returns under a train-only and a full-data scaler fit")
show_with_alt(
    fig,
    alt=(
        "Two histograms of the scaled test returns on identical bins and shared axes, the "
        "train-only fit on the left and the full-data fit on the right, each with a dashed "
        "vertical line at its own mean and that mean in the legend. The two distributions "
        "have the same overall spread and the two mean lines both sit just above zero, "
        "close enough that the shift is visible only by reading the legend."
    ),
)

# %% [markdown]
# With everything else held equal (both panels winsorized, identical bins and
# axes), the small shift in the mean line is the pure signature of leakage:
# fitting scaling parameters on future data nudges the test distribution. The
# effect is modest here, but in production with many features this systematic
# bias accumulates and can meaningfully inflate Sharpe ratios and IC estimates.


# %% [markdown]
# ### 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]
# ### 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]
# ### 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]
# ## 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.

```