Kiểm tra chất lượng dữ liệu bảng tài chính: giá trị thiếu, trùng lặp và thứ tự thời gian

# Data Quality Diagnostics


# Data Quality Diagnostics

**Docker image**: `ml4t`

**Chapter 7: Defining the Learning Task**
**Section Reference**: 7.1 - Data Preprocessing and Encodings

## Purpose

This notebook provides a **standardized diagnostic survey** across all ML4T datasets.
No cleaning is performed - just assessment. The output is a "health report" identifying
which datasets need active preprocessing versus which are already in good shape.

## Learning Objectives

1. Build reusable diagnostic functions for financial panel data
2. Understand common data quality issues: missingness, duplicates, outliers, gaps
3. Identify dataset-specific quirks that affect ML pipeline design
4. Create a systematic framework for data validation

## Book Reference

Section 7.1 emphasizes that **data quality determines model quality**. This notebook
operationalizes that principle with concrete diagnostics before any feature engineering.

## Prerequisites

- Familiarity with the seven ML4T datasets and their loaders (`data` module).
- Polars basics (`pl.DataFrame`, `group_by`, `with_columns`).
- No prior notebook output is required; this is the entry point of the pipeline.

## Data Contract

- **Input**: Raw datasets from data loaders (ETFs, US Equities, Crypto, etc.)
- **Output**: Diagnostic reports and quality assessments (no mutations)

```python
"""Data Quality Diagnostics - Survey all ML4T datasets."""

from __future__ import annotations

import logging
import warnings
from datetime import datetime
from typing import Any, Literal

import numpy as np
import plotly.graph_objects as go
import polars as pl
from IPython.display import display

from data import (
    load_cme_futures,
    load_crypto_perps,
    load_crypto_premium,
    load_etfs,
    load_firm_characteristics,
    load_fx_pairs,
    load_us_equities,
)
from utils.style import COLORS, show_plotly_with_alt  # activates the ml4t Plotly template


def quiet_ml4t_logging() -> None:
    """Keep ml4t.diagnostic's INFO narration out of the output cells.

    Each of its modules builds a `DiagnosticLogger`, which sets INFO on its own leaf
    logger and attaches its own stderr handler, so a level set on the `ml4t` parent
    never reaches them. The leaves come into existence when their module is imported,
    which is why this is called again after the two ml4t imports further down rather
    than once here.
    """
    for name, logger in logging.root.manager.loggerDict.items():
        if name.startswith("ml4t") and isinstance(logger, logging.Logger):
            logger.setLevel(logging.WARNING)


quiet_ml4t_logging()
```

```python
# Production defaults
ETF_START_DATE = "2015-01-01"
US_EQUITIES_START_DATE = "1970-01-01"
CRYPTO_START_DATE = "2021-01-01T00:00:00+00:00"
CME_FUTURES_START_DATE = "2015-01-01"
FX_START_DATE = "2015-01-01T00:00:00+00:00"
FIRM_CHARACTERISTICS_START_DATE = "1990-01-01"
```

## Diagnostic Function Library

These functions are designed to be **reusable** across datasets and chapters.
They follow a consistent interface: input DataFrame, return diagnostic results.

### Index Integrity Check

Validates that the time index has correct dtype, monotonicity, and uniqueness.
For panel data, also checks that (date, symbol) pairs are unique.

The monotonicity check reads each symbol's rows in the order they arrive. Sorting the
frame by symbol and time before asking whether time is sorted would return True for
every input, including a frame whose rows are in the wrong order - a check that cannot
fail is the defect this notebook exists to find, and it is an easy one to commit here.

```python
def check_index_integrity(
    df: pl.DataFrame,
    time_col: str = "timestamp",
    symbol_col: str | None = "symbol",
) -> dict[str, Any]:
    """Check index integrity: dtype, monotonicity, uniqueness.

    Args:
        df: Input DataFrame
        time_col: Name of the time column
        symbol_col: Name of the symbol column (None for single-asset data)

    Returns:
        Dictionary with diagnostic results
    """
    results = {
        "time_col": time_col,
        "symbol_col": symbol_col,
        "n_rows": len(df),
        "checks": {},
    }

    # Check time column dtype
    time_dtype = df[time_col].dtype
    results["time_dtype"] = str(time_dtype)
    results["checks"]["valid_time_dtype"] = time_dtype in [pl.Date, pl.Datetime]

    # Check for null time values
    null_times = df[time_col].is_null().sum()
    results["null_times"] = null_times
    results["checks"]["no_null_times"] = null_times == 0

    if symbol_col and symbol_col in df.columns:
        # Panel data: check uniqueness of (date, symbol)
        n_unique = df.select([time_col, symbol_col]).unique().height
        results["unique_pairs"] = n_unique
        results["checks"]["unique_date_symbol"] = n_unique == len(df)

        # `maintain_order=True` keeps each group's rows in their arrival sequence,
        # which is what makes the comparison capable of failing.
        mono_check = df.group_by(symbol_col, maintain_order=True).agg(
            is_mono=(pl.col(time_col) == pl.col(time_col).sort()).all()
        )
        all_mono = mono_check["is_mono"].all()
        results["checks"]["monotonic_per_symbol"] = all_mono

        # Symbol statistics
        n_symbols = df[symbol_col].n_unique()
        results["n_symbols"] = n_symbols
    else:
        # Single asset: check global monotonicity
        is_sorted = (df[time_col] == df[time_col].sort()).all()
        results["checks"]["monotonic"] = is_sorted

        # Check uniqueness
        n_unique = df[time_col].n_unique()
        results["unique_times"] = n_unique
        results["checks"]["unique_times"] = n_unique == len(df)

    # Date range
    results["date_min"] = str(df[time_col].min())
    results["date_max"] = str(df[time_col].max())

    # Overall pass/fail
    results["passed"] = all(results["checks"].values())

    return results
```

### Duplicate Detection

Finds exact duplicates (identical rows) and near-duplicates (same index, different values).

```python
def check_duplicates(
    df: pl.DataFrame,
    key_cols: list[str],
    value_cols: list[str] | None = None,
) -> dict[str, Any]:
    """Detect exact and near-duplicates.

    Args:
        df: Input DataFrame
        key_cols: Columns that define uniqueness (e.g., ['timestamp', 'symbol'])
        value_cols: Columns to check for value differences in near-duplicates

    Returns:
        Dictionary with duplicate counts and examples
    """
    results = {"key_cols": key_cols, "n_rows": len(df)}

    # Exact duplicates (all columns identical)
    n_exact = len(df) - df.unique().height
    results["exact_duplicates"] = n_exact

    # Key duplicates (same key, potentially different values)
    key_counts = df.group_by(key_cols).len()
    dupes = key_counts.filter(pl.col("len") > 1)
    n_key_dupes = dupes["len"].sum() - len(dupes) if len(dupes) > 0 else 0
    results["key_duplicates"] = n_key_dupes

    # If there are key duplicates and value columns specified, check for near-dupes
    if n_key_dupes > 0 and value_cols:
        # Near-duplicates: same key, different values
        dupe_keys = dupes.select(key_cols)
        dupe_rows = df.join(dupe_keys, on=key_cols, how="inner")

        # Group by key and check if values vary
        near_dupes = dupe_rows.group_by(key_cols).agg(
            [pl.col(c).n_unique().alias(f"{c}_nunique") for c in value_cols]
        )
        # Count rows where any value column has more than 1 unique value
        vary_cols = [f"{c}_nunique" for c in value_cols]
        has_variation = near_dupes.select([(pl.col(c) > 1) for c in vary_cols]).select(
            pl.any_horizontal(pl.all())
        )

        n_near = has_variation.sum().item() if len(has_variation) > 0 else 0
        results["near_duplicates"] = n_near

        # Sample of duplicates
        if n_key_dupes > 0 and len(dupe_rows) > 0:
            results["sample_duplicates"] = dupe_rows.head(5).to_dicts()
    else:
        results["near_duplicates"] = 0

    results["passed"] = n_exact == 0 and n_key_dupes == 0

    return results
```

### Coverage Report

Analyzes missingness patterns by field, by asset, and by time period.

```python
def coverage_report(
    df: pl.DataFrame,
    time_col: str = "timestamp",
    symbol_col: str | None = "symbol",
    value_cols: list[str] | None = None,
) -> dict[str, Any]:
    """Generate coverage report: missingness by field, asset, time.

    Args:
        df: Input DataFrame
        time_col: Time column name
        symbol_col: Symbol column name (None for single-asset)
        value_cols: Columns to check for missingness (default: numeric columns)

    Returns:
        Dictionary with coverage statistics
    """
    results = {"n_rows": len(df), "n_cols": len(df.columns)}

    # Default to numeric columns if not specified
    if value_cols is None:
        value_cols = [
            c
            for c in df.columns
            if df[c].dtype in [pl.Float32, pl.Float64, pl.Int32, pl.Int64]
            and c not in [time_col, symbol_col]
        ]

    results["checked_cols"] = value_cols

    # Missingness by column
    col_missing = {}
    for col in value_cols:
        n_null = df[col].is_null().sum()
        pct_null = 100.0 * n_null / len(df) if len(df) > 0 else 0
        col_missing[col] = {"n_null": n_null, "pct_null": round(pct_null, 2)}

    results["by_column"] = col_missing

    # Worst columns
    sorted_cols = sorted(col_missing.items(), key=lambda x: x[1]["pct_null"], reverse=True)
    results["worst_columns"] = sorted_cols[:5]

    # Missingness by asset (if panel)
    if symbol_col and symbol_col in df.columns:
        # For each symbol, count rows with any missing value
        symbol_missing = df.group_by(symbol_col).agg(
            [
                pl.len().alias("n_rows"),
                *[pl.col(c).is_null().sum().alias(f"{c}_null") for c in value_cols[:5]],
            ]
        )
        results["n_symbols"] = df[symbol_col].n_unique()

        # Find symbols with most missing data
        null_cols = [f"{c}_null" for c in value_cols[:5]]
        if null_cols:
            symbol_missing = symbol_missing.with_columns(
                total_null=pl.sum_horizontal(null_cols)
            ).sort("total_null", descending=True)
            results["worst_symbols"] = symbol_missing.head(5).to_dicts()

    # Missingness by time period
    time_missing = df.group_by(time_col).agg(
        [pl.col(c).is_null().sum().alias(f"{c}_null") for c in value_cols[:3]]
    )
    if value_cols:
        null_col = f"{value_cols[0]}_null"
        worst_dates = time_missing.sort(null_col, descending=True).head(5)
        results["worst_dates"] = worst_dates.to_dicts()

    # Overall coverage
    total_cells = len(df) * len(value_cols)
    total_missing = sum(v["n_null"] for v in col_missing.values())
    results["overall_coverage_pct"] = (
        round(100 * (1 - total_missing / total_cells), 2) if total_cells > 0 else 100
    )

    return results
```

### Coverage Heatmap

Visualizes data availability across time × asset for panel data.

The pivot underneath emits one row per month that *has* data, so a month in which the
whole panel is missing - a vendor outage, an ingestion gap - would be absent from the
matrix rather than present as an all-zero row. Any count of "months missing for every
asset" taken off that matrix is therefore zero by construction. Reindexing onto a
continuous month range first is what makes the count capable of firing.

```python
def coverage_heatmap(
    df: pl.DataFrame,
    time_col: str = "timestamp",
    symbol_col: str = "symbol",
    value_col: str = "close",
    title: str = "Data Coverage Heatmap",
    max_symbols: int = 50,
    order_by: Literal["symbol", "first_observation"] = "first_observation",
) -> tuple[go.Figure, np.ndarray]:
    """Create time × asset coverage heatmap.

    Args:
        df: Input DataFrame
        time_col: Time column name
        symbol_col: Symbol column name
        value_col: Column to check for presence
        title: Plot title
        max_symbols: Maximum symbols to show (for readability)
        order_by: Column order along the x axis. Alphabetical order scatters entry
            dates at random, so a survivorship-free panel reads as a ragged skyline
            with no visible structure. Ordering by first observation is what turns
            the same data into the listing staircase the pattern actually is.

    Returns:
        The figure, and the presence matrix it draws (months x assets), so the
        notebook can quantify the pattern rather than assert it.
    """
    # Aggregate to reduce data size - use monthly periods
    coverage_df = (
        df.with_columns(period=pl.col(time_col).dt.truncate("1mo"))
        .group_by(["period", symbol_col])
        .agg(has_data=(pl.col(value_col).is_not_null().sum() > 0).cast(pl.Int8))
    )

    # Column order
    if order_by == "first_observation":
        symbols = (
            coverage_df.filter(pl.col("has_data") == 1)
            .group_by(symbol_col)
            .agg(first_seen=pl.col("period").min())
            .sort(["first_seen", symbol_col])[symbol_col]
            .to_list()
        )
    else:
        symbols = coverage_df[symbol_col].unique().sort().to_list()

    if len(symbols) > max_symbols:
        # Even sample across the ordering, so the thinned axis keeps its shape
        symbols = symbols[:: len(symbols) // max_symbols][:max_symbols]
        coverage_df = coverage_df.filter(pl.col(symbol_col).is_in(symbols))

    pivot_df = coverage_df.pivot(
        values="has_data",
        index="period",
        on=symbol_col,
    ).sort("period")

    # Reindex onto a continuous month range before anything reads the matrix.
    if len(pivot_df):
        full_months = pl.DataFrame(
            {
                "period": pl.datetime_range(
                    pivot_df["period"].min(),
                    pivot_df["period"].max(),
                    interval="1mo",
                    eager=True,
                ).cast(pivot_df["period"].dtype)
            }
        )
        pivot_df = full_months.join(pivot_df, on="period", how="left").fill_null(0).sort("period")

    # Pivot emits columns in order of first appearance, which is not `symbols`.
    # Select explicitly, or the x labels would name different assets than the
    # columns they sit under.
    symbols = [s for s in symbols if s in pivot_df.columns]
    periods = pivot_df["period"].to_list()
    period_labels = [str(p)[:7] for p in periods]  # YYYY-MM format
    matrix = pivot_df.select(symbols).to_numpy()

    # Create heatmap - present bars are filled (blue), missing periods stay white, so the
    # listing/delisting staircase reads as "where do we have data" rather than the reverse.
    fig = go.Figure(
        data=go.Heatmap(
            z=matrix,
            x=symbols,
            y=period_labels,
            colorscale=[[0, "white"], [1, COLORS["blue"]]],
            showscale=False,
            hovertemplate="Symbol: %{x}<br>Period: %{y}<br>Has data: %{z}<extra></extra>",
        )
    )

    # Cap the pixel height so a multi-decade monthly panel renders as a readable block
    # instead of an unusable vertical strip (588 months × 15px would be > 8000px tall).
    fig.update_layout(
        title=title,
        xaxis_title="Symbol",
        yaxis_title="Period",
        height=min(800, max(400, len(periods) * 15)),
        width=min(900, max(600, len(symbols) * 15)),
    )

    return fig, matrix
```

### Distribution Summary

Computes summary statistics and flags extreme values.

```python
def distribution_summary(
    df: pl.DataFrame,
    fields: list[str],
    extreme_threshold: float = 5.0,
) -> pl.DataFrame:
    """Compute distribution statistics with extreme value flags.

    Args:
        df: Input DataFrame
        fields: Columns to summarize
        extreme_threshold: Z-score threshold for flagging extremes

    Returns:
        DataFrame with summary statistics
    """
    stats_list = []

    for field in fields:
        if field not in df.columns:
            continue

        col = df[field].drop_nulls()
        if len(col) == 0:
            continue

        # Basic stats
        stats = {
            "field": field,
            "count": len(col),
            "null_pct": round(100 * df[field].is_null().sum() / len(df), 2),
            "mean": round(col.mean(), 6),
            "std": round(col.std(), 6),
            "min": round(col.min(), 6),
            "q25": round(col.quantile(0.25), 6),
            "median": round(col.median(), 6),
            "q75": round(col.quantile(0.75), 6),
            "max": round(col.max(), 6),
        }

        # Skewness and kurtosis (using numpy for robustness)
        arr = col.to_numpy()
        if len(arr) > 2 and np.std(arr) > 0:
            stats["skew"] = round(float(np.mean(((arr - np.mean(arr)) / np.std(arr)) ** 3)), 4)
            stats["kurtosis"] = round(
                float(np.mean(((arr - np.mean(arr)) / np.std(arr)) ** 4) - 3), 4
            )
        else:
            stats["skew"] = 0.0
            stats["kurtosis"] = 0.0

        # Extreme value count
        if stats["std"] > 0:
            z_scores = np.abs((arr - np.mean(arr)) / np.std(arr))
            n_extreme = int(np.sum(z_scores > extreme_threshold))
            stats["n_extreme"] = n_extreme
            stats["pct_extreme"] = round(100 * n_extreme / len(arr), 4)
        else:
            stats["n_extreme"] = 0
            stats["pct_extreme"] = 0.0

        stats_list.append(stats)

    return pl.DataFrame(stats_list)
```

### Outlier Flags

Detects domain violations and spike anomalies (single-bar reversals).

```python
def outlier_flags(
    df: pl.DataFrame,
    price_cols: list[str] | None = None,
    volume_col: str | None = "volume",
    return_col: str | None = None,
    max_return_threshold: float = 2.0,
    high_spike_ratio: float = 1.5,
    low_spike_ratio: float = 0.5,
) -> dict[str, Any]:
    """Flag domain violations and spike anomalies.

    Args:
        df: Input DataFrame
        price_cols: Price columns to check for domain violations
        volume_col: Volume column to check for negative values
        return_col: Return column to check for extreme spikes
        max_return_threshold: Maximum plausible return, as a fraction of the prior price
        high_spike_ratio: Flag a bar whose high exceeds this multiple of its close
        low_spike_ratio: Flag a bar whose low falls below this multiple of its close

    Returns:
        Dictionary with outlier counts and examples
    """
    results = {"n_rows": len(df), "flags": {}}

    # Check for negative prices
    if price_cols:
        for col in price_cols:
            if col in df.columns:
                n_negative = (df[col] < 0).sum()
                n_zero = (df[col] == 0).sum()
                results["flags"][f"{col}_negative"] = n_negative
                results["flags"][f"{col}_zero"] = n_zero

    # Check for negative volume
    if volume_col and volume_col in df.columns:
        n_neg_vol = (df[volume_col] < 0).sum()
        results["flags"]["negative_volume"] = n_neg_vol

    # Check for impossible returns
    if return_col and return_col in df.columns:
        impossible = df[return_col].abs() > max_return_threshold
        n_impossible = impossible.sum()
        results["flags"]["impossible_returns"] = n_impossible

        if n_impossible > 0:
            examples = df.filter(impossible).head(5)
            results["impossible_return_examples"] = examples.to_dicts()

    # Resolve the high/low/close names so roll-adjusted panels (adj_*) are covered too:
    # every OHLC field in a bar shares one adjustment factor, so the ratios below are
    # adjust-invariant and the same thresholds apply to raw and adjusted panels alike.
    def _resolve(*names: str) -> str | None:
        return next((n for n in names if n in df.columns), None)

    hi_col = _resolve("high", "adj_high")
    lo_col = _resolve("low", "adj_low")
    cl_col = _resolve("close", "adj_close")
    if hi_col and lo_col and cl_col:
        high_spike = df[hi_col] > df[cl_col] * high_spike_ratio
        low_spike = df[lo_col] < df[cl_col] * low_spike_ratio
        n_spikes = (high_spike | low_spike).sum()
        results["flags"]["price_spikes"] = n_spikes

    # Count total flags
    results["total_flags"] = sum(results["flags"].values())
    results["passed"] = results["total_flags"] == 0

    return results
```

## Dataset Registry

Central registry mapping dataset names to loader functions and metadata.

```python
def load_dataset_safely(loader_func, *args, **kwargs):
    """Attempt to load a dataset, returning None with message if unavailable."""
    try:
        return loader_func(*args, **kwargs)
    except Exception as e:
        error_type = type(e).__name__
        print(f"  WARNING: Could not load: {error_type}")
        return None


# Define dataset registry
DATASET_REGISTRY = {
    "etfs": {
        "loader": load_etfs,
        "time_col": "timestamp",
        "symbol_col": "symbol",
        "freq": "daily",
        "price_cols": ["open", "high", "low", "close"],
        "volume_col": "volume",
        "description": "100 ETFs from Yahoo Finance",
    },
    "us_equities": {
        "loader": load_us_equities,
        "time_col": "timestamp",
        "symbol_col": "symbol",
        "freq": "daily",
        "price_cols": ["open", "high", "low", "close", "adj_close"],
        "volume_col": "volume",
        "description": "3,199 US equities (1962-2018)",
    },
    "crypto_perps": {
        "loader": lambda: load_crypto_perps(frequency="8h"),
        "time_col": "timestamp",
        "symbol_col": "symbol",
        "freq": "8h",
        "price_cols": ["open", "high", "low", "close"],
        "volume_col": "volume",
        "description": "Crypto perpetuals from Binance",
    },
    "crypto_premium": {
        "loader": lambda: load_crypto_premium(frequency="8h"),
        "time_col": "timestamp",
        "symbol_col": "symbol",
        "freq": "8h",
        "price_cols": [
            "premium_index_open",
            "premium_index_high",
            "premium_index_low",
            "premium_index_close",
        ],
        "volume_col": None,
        "description": "Crypto premium index from Binance",
    },
    "cme_futures": {
        "loader": lambda: load_cme_futures(tenors=[0]),
        "time_col": "session_date",
        "symbol_col": "product",
        "freq": "daily",
        # Diagnostics run on the adjusted series (the canonical downstream series);
        # per-row OHLC share one cum_ratio, so integrity checks are adjust-invariant.
        "price_cols": ["adj_open", "adj_high", "adj_low", "adj_close"],
        "volume_col": "volume",
        "description": "CME futures (front month)",
    },
    "fx_pairs": {
        "loader": lambda: load_fx_pairs(frequency="4h"),
        "time_col": "timestamp",
        "symbol_col": "symbol",
        "freq": "4h",
        "price_cols": ["open", "high", "low", "close"],
        "volume_col": "volume",
        "description": "FX pairs from OANDA",
    },
    "firm_characteristics": {
        "loader": load_firm_characteristics,
        "time_col": "timestamp",
        "symbol_col": "symbol",
        "freq": "monthly",
        "price_cols": [],
        "volume_col": None,
        "description": "Chen-Pelger-Zhu firm characteristics",
    },
}
```

## Per-Dataset Diagnostics

Run all diagnostics for each available dataset. Results are stored for the
summary dashboard.

```python
diagnostic_results = {}
```

```python
def ensure_symbol_alias(df: pl.DataFrame) -> pl.DataFrame:
    """Expose the canonical asset identifier 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
```

```python
def filter_from_start(df: pl.DataFrame, time_col: str, start_value: str) -> pl.DataFrame:
    """Apply a start-date filter without tripping over timezone/unit differences."""
    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))
```

### ETF Universe

Yahoo Finance data covering 100 ETFs across 9 asset classes. Key concerns:
adjustment artifacts from corporate actions and ticker changes.

```python
etfs = load_dataset_safely(DATASET_REGISTRY["etfs"]["loader"])

if etfs is not None:
    config = DATASET_REGISTRY["etfs"]
    tcol = config["time_col"]
    etfs = ensure_symbol_alias(etfs)
    etfs = filter_from_start(etfs, tcol, ETF_START_DATE)

    print(f"Loaded {len(etfs):,} rows, {etfs['symbol'].n_unique()} symbols")
    print(f"Date range: {etfs[tcol].min()} to {etfs[tcol].max()}")
```

```python
if etfs is not None:
    # Index integrity and duplicates
    idx_check = check_index_integrity(etfs, config["time_col"], config["symbol_col"])
    dupe_check = check_duplicates(
        etfs,
        key_cols=[config["time_col"], config["symbol_col"]],
        value_cols=config["price_cols"],
    )
    print(f"Index Integrity: {'PASSED' if idx_check['passed'] else 'FAILED'}")
    print(
        f"Duplicates:      {'PASSED' if dupe_check['passed'] else 'FAILED'} "
        f"(exact: {dupe_check['exact_duplicates']}, key: {dupe_check['key_duplicates']})"
    )
```

```python
if etfs is not None:
    # Coverage and outlier flags
    cov_report = coverage_report(
        etfs,
        config["time_col"],
        config["symbol_col"],
        config["price_cols"] + [config["volume_col"]],
    )
    outliers = outlier_flags(
        etfs,
        price_cols=config["price_cols"],
        volume_col=config["volume_col"],
    )
    print(f"Coverage: {cov_report['overall_coverage_pct']}%")
    print(f"Outliers: {'PASSED' if outliers['passed'] else 'FLAGGED'}")
    for flag, count in outliers["flags"].items():
        if count > 0:
            print(f"  {flag}: {count}")
```

```python
if etfs is not None:
    # Distribution summary
    dist_summary = distribution_summary(etfs, config["price_cols"][:3])
    display(dist_summary)

    diagnostic_results["etfs"] = {
        "index": idx_check,
        "duplicates": dupe_check,
        "coverage": cov_report,
        "outliers": outliers,
    }
```

ETFs pass index integrity and duplicate checks. Coverage is near-complete for
OHLCV fields. Any outlier flags likely reflect Yahoo Finance adjustment
artifacts (split ratios applied to historical bars) rather than genuine
price anomalies.

### US Equities

The longest panel in the survey, and the only survivorship-free one. Key concerns:
penny stocks, stock splits, and delisting events. The loaded span starts at
`US_EQUITIES_START_DATE` rather than at the vendor's first year.

```python
us_equities = load_dataset_safely(DATASET_REGISTRY["us_equities"]["loader"])

if us_equities is not None:
    config = DATASET_REGISTRY["us_equities"]
    us_equities = ensure_symbol_alias(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")
    print(f"Date range: {us_equities['timestamp'].min()} to {us_equities['timestamp'].max()}")
```

```python
if us_equities is not None:
    # Index integrity
    idx_check = check_index_integrity(us_equities, config["time_col"], config["symbol_col"])
    print(f"Index Integrity: {'PASSED' if idx_check['passed'] else 'FAILED'}")

    # Duplicates and coverage (same battery as the other panels; this is the
    # largest dataset and the one that most needs the full check).
    dupe_check = check_duplicates(
        us_equities,
        key_cols=[config["time_col"], config["symbol_col"]],
        value_cols=config["price_cols"],
    )
    cov_report = coverage_report(
        us_equities,
        config["time_col"],
        config["symbol_col"],
        config["price_cols"] + [config["volume_col"]],
    )
    print(
        f"Duplicates:      {'PASSED' if dupe_check['passed'] else 'FAILED'} "
        f"(exact: {dupe_check['exact_duplicates']}, key: {dupe_check['key_duplicates']})"
    )
    print(f"Coverage: {cov_report['overall_coverage_pct']}%")

    # Penny stocks and extreme returns
    penny = us_equities.filter(pl.col("close") < 1)
    us_equities_ret = us_equities.sort(["symbol", "timestamp"]).with_columns(
        returns=pl.col("close").pct_change().over("symbol")
    )
    extreme_ret = us_equities_ret.filter(pl.col("returns").abs() > 1.0)
    print(
        f"Penny stocks (close < $1): {len(penny):,} rows ({100 * len(penny) / len(us_equities):.1f}%)"
    )
    print(f"Extreme returns (>100%):   {len(extreme_ret):,} rows")
```

### A one-sided filter, and the column that shows it

It is tempting to read "extreme returns" as "corporate actions" and move on. This
dataset ships a `split_ratio` column, so we do not have to guess - and checking
turns the reading around.

A raw return is bounded below by $-1$, because a price cannot fall below zero. So a
symmetric-looking filter on $|r| > 1$ **can only ever fire on the upside**. A reverse
split, which multiplies the price, clears the threshold easily. A forward split, which
divides it, cannot reach $-1$ however large the split, so it is invisible to the filter
by construction rather than by accident.

```python
if us_equities is not None:
    splits = us_equities_ret.filter(
        (pl.col("split_ratio") != 1) & pl.col("split_ratio").is_not_null()
    )
    forward = splits.filter(pl.col("split_ratio") > 1)
    reverse = splits.filter(pl.col("split_ratio") < 1)

    print(f"Rows flagged by |return| > 1.0:  {len(extreme_ret):,}")
    print(f"  smallest flagged return:      {extreme_ret['returns'].min():+.4f}")
    print(
        f"  that are a split day:         {splits.filter(pl.col('returns').abs() > 1.0).height:,}"
    )
    print(f"\nRows with split_ratio != 1:      {len(splits):,}")
    print(
        f"  forward splits (ratio > 1):   {len(forward):,}, "
        f"median same-day return {forward['returns'].median():+.4f}, "
        f"caught by the filter: {forward.filter(pl.col('returns').abs() > 1.0).height:,}"
    )
    print(
        f"  reverse splits (ratio < 1):   {len(reverse):,}, "
        f"median same-day return {reverse['returns'].median():+.4f}, "
        f"caught by the filter: {reverse.filter(pl.col('returns').abs() > 1.0).height:,}"
    )
```

The smallest flagged return is positive, which is the whole story in one number:
not a single row was flagged for falling. The filter catches most reverse splits
and essentially none of the forward splits, even though forward splits are the far
more common event here by roughly eight to one.

So "extreme returns signal corporate actions" is the wrong direction of inference.
Most flagged rows are *not* splits, and the great majority of splits are *not*
flagged. The lesson is the general one: **when a dataset records an event
directly, detect it from that column rather than inferring it from a threshold on
something else** - and check the bounds of the quantity you are thresholding before
assuming a two-sided rule is two-sided.

```python
if us_equities is not None:
    # Outlier flags
    outliers = outlier_flags(
        us_equities,
        price_cols=["open", "high", "low", "close"],
        volume_col="volume",
    )
    print(f"Outliers: {'PASSED' if outliers['passed'] else 'FLAGGED'}")
    for flag, count in outliers["flags"].items():
        if count > 0:
            print(f"  {flag}: {count}")

    diagnostic_results["us_equities"] = {
        "index": idx_check,
        "duplicates": dupe_check,
        "coverage": cov_report,
        "outliers": outliers,
        "penny_stocks": len(penny),
        "extreme_returns": len(extreme_ret),
        "extreme_return_min": extreme_ret["returns"].min(),
    }
```

US Equities is the dataset that most needs active cleaning. Penny stocks should be
filtered so microstructure noise does not dominate cross-sectional models, and the
rows the return threshold flags need investigating rather than deleting: the section
above shows they are not the corporate actions they look like. See
`02_preprocessing_pipeline` for the full cleaning pipeline.

### Coverage Heatmap: US Equities

Visualize data availability across time and assets. Shaded cells mark months with
data; white marks their absence. **Assets are ordered left to right by the month
they first appear**, which is what makes the pattern legible: in alphabetical order
the same panel is a ragged skyline with no visible structure, because a ticker's
spelling says nothing about when it listed.

```python
if us_equities is not None:
    fig, matrix_for_summary = coverage_heatmap(
        us_equities,
        time_col="timestamp",
        symbol_col="symbol",
        value_col="close",
        title="US equities monthly coverage by asset, ordered by first month",
        max_symbols=50,
    )
    fig.update_layout(
        xaxis_title="Asset (sampled)",
        yaxis_title="Month",
    )
    show_plotly_with_alt(
        fig,
        alt=(
            "Monthly coverage grid for a sample of US equities, months running up the "
            "vertical axis and assets ordered left to right by the month each first "
            "traded. Filled cells mark months in which the asset has data. The boundary "
            "between filled and empty is a staircase rising to the right, so each column "
            "begins later than the one to its left. Notches along the top edge mark "
            "assets whose data stops before the panel ends. No row is empty across its "
            "full width."
        ),
    )
```

Two things to read off it. The **staircase rising to the right** is the listing
history: each column begins at the month its asset first traded, and because the
columns are ordered by that month the start dates form a step rather than
scattering. The **broken top edge** is delisting - a column that stops short of the
present is an asset that left the panel, and it is still in the data. That is what
survivorship-free means, and it is the reason a universe must be rebuilt as of each
decision date rather than taken from today's index membership.

What is *not* here is equally informative, but only one of the two absences is
something this figure can report. There are **no full-width white rows**: the month
axis is a continuous range built from the panel's own bounds, so a month in which
nothing traded would be drawn as a blank row, and none is - no vendor outage or
exchange closure is written into this panel.

The columns cannot answer the matching question. The symbol axis is assembled from
the assets that *appear in the data*, so an asset missing throughout would not be a
white column - it would not be a column at all. "No asset is entirely absent" is
therefore not a finding this figure can produce, and checking it needs an
independently defined expected universe to compare against, which this notebook
does not have. The cell below counts only what is countable here.

Interior gaps, where an asset goes quiet and comes back, do exist but are rare.
Missingness is overwhelmingly a question of when each asset entered and left, which
makes the Section 7.1 strategy a universe-construction problem rather than an
imputation one.

```python
# Quantify what the heatmap shows, so the reading above is checkable
if us_equities is not None:
    _m = np.nan_to_num(matrix_for_summary.astype(float), nan=0.0)
    _interior = 0
    for _j in range(_m.shape[1]):
        _idx = np.flatnonzero(_m[:, _j] == 1)
        if len(_idx):
            _interior += int((_m[_idx[0] : _idx[-1] + 1, _j] == 0).sum())
    print(f"Sampled panel: {_m.shape[0]} months x {_m.shape[1]} assets")
    # The month axis is a continuous range, so this count can actually fire; the
    # symbol axis is the observed universe, so "an asset absent in every month" is
    # not expressible here and is deliberately not printed as a guaranteed zero.
    print(f"  Months missing for every asset:       {int((_m.sum(axis=1) == 0).sum())}")
    print(
        f"  Assets listing after the first month: {int(sum(1 for j in range(_m.shape[1]) if _m[0, j] == 0))}"
    )
    print(
        f"  Assets ending before the last month:  {int(sum(1 for j in range(_m.shape[1]) if _m[-1, j] == 0))}"
    )
    print(f"  Interior gaps (absent, then back):    {_interior} cells")
```

### Crypto Perpetuals

24/7 trading with 8-hour funding settlement cycles. Bars should
align exactly to 00:00, 08:00, 16:00 UTC.

```python
crypto_perps = load_dataset_safely(DATASET_REGISTRY["crypto_perps"]["loader"])

if crypto_perps is not None:
    config = DATASET_REGISTRY["crypto_perps"]
    crypto_perps = ensure_symbol_alias(crypto_perps)
    crypto_perps = filter_from_start(crypto_perps, "timestamp", CRYPTO_START_DATE)

    print(f"Loaded {len(crypto_perps):,} rows, {crypto_perps['symbol'].n_unique()} symbols")

    idx_check = check_index_integrity(crypto_perps, config["time_col"], config["symbol_col"])
    cov_report = coverage_report(
        crypto_perps,
        config["time_col"],
        config["symbol_col"],
        config["price_cols"],
    )
    hours = crypto_perps["timestamp"].dt.hour()
    aligned = hours.is_in([0, 8, 16]).all()

    print(f"Index Integrity: {'PASSED' if idx_check['passed'] else 'FAILED'}")
    print(f"Coverage: {cov_report['overall_coverage_pct']}%")
    print(f"8-hour alignment: {'PASSED' if aligned else 'FAILED'}")

    diagnostic_results["crypto_perps"] = {
        "index": idx_check,
        "coverage": cov_report,
        "aligned_8h": aligned,
    }
```

### Crypto Premium Index

Premium index for funding arbitrage strategies. Same 8-hour frequency
as perpetuals; values represent the basis between spot and futures.

```python
crypto_premium = load_dataset_safely(DATASET_REGISTRY["crypto_premium"]["loader"])

if crypto_premium is not None:
    config = DATASET_REGISTRY["crypto_premium"]
    crypto_premium = ensure_symbol_alias(crypto_premium)
    crypto_premium = filter_from_start(crypto_premium, "timestamp", CRYPTO_START_DATE)

    print(f"Loaded {len(crypto_premium):,} rows, {crypto_premium['symbol'].n_unique()} symbols")

    idx_check = check_index_integrity(crypto_premium, config["time_col"], config["symbol_col"])
    dist = distribution_summary(crypto_premium, config["price_cols"])
    print(f"Index Integrity: {'PASSED' if idx_check['passed'] else 'FAILED'}")
    display(dist)

    diagnostic_results["crypto_premium"] = {"index": idx_check}
```

Both crypto datasets pass integrity checks. The 24/7 schedule means no
weekend gaps to worry about, but delisting events mid-panel (coins removed
from the exchange) can create sudden coverage drops.

### CME Futures

Session-aligned daily bars with roll continuity across contract months.

```python
cme_futures = load_dataset_safely(DATASET_REGISTRY["cme_futures"]["loader"])

if cme_futures is not None:
    config = DATASET_REGISTRY["cme_futures"]
    cme_futures = filter_from_start(cme_futures, "session_date", CME_FUTURES_START_DATE)

    products = cme_futures["product"].unique().sort().to_list()
    print(f"Loaded {len(cme_futures):,} rows, {len(products)} products")
    print(f"Products: {', '.join(products)}")

    idx_check = check_index_integrity(cme_futures, config["time_col"], config["symbol_col"])
    cov_by_product = (
        cme_futures.group_by("product")
        .agg(
            pl.len().alias("n_rows"),
            pl.col("adj_close").is_null().sum().alias("n_null_close"),
        )
        .sort("product")
    )
    print(f"Index Integrity: {'PASSED' if idx_check['passed'] else 'FAILED'}")
    display(cov_by_product)

    diagnostic_results["cme_futures"] = {"index": idx_check, "n_products": len(products)}
```

### FX Pairs

4-hour bars with expected weekend gaps: the market closes Friday 17:00 EST and reopens
Sunday 17:00 EST. That reopen lands between 21:00 and 22:00 UTC depending on daylight
saving, so a Sunday bar at or after the reopen hour is ordinary market data rather than
an anomaly. Saturday bars, and Sunday bars earlier in the day, fall outside the trading
week and are the ones worth filtering.

```python
fx_pairs = load_dataset_safely(DATASET_REGISTRY["fx_pairs"]["loader"])

if fx_pairs is not None:
    config = DATASET_REGISTRY["fx_pairs"]
    fx_pairs = ensure_symbol_alias(fx_pairs)
    fx_pairs = filter_from_start(fx_pairs, "timestamp", FX_START_DATE)

    print(f"Loaded {len(fx_pairs):,} rows, {fx_pairs['symbol'].n_unique()} pairs")

    idx_check = check_index_integrity(fx_pairs, config["time_col"], config["symbol_col"])
    # Polars dt.weekday() is ISO: Monday=1 ... Saturday=6, Sunday=7.
    FX_REOPEN_HOUR_UTC = 21
    fx_pairs_with_weekday = fx_pairs.with_columns(
        weekday=pl.col("timestamp").dt.weekday(),
        hour=pl.col("timestamp").dt.hour(),
    )
    weekend_data = fx_pairs_with_weekday.filter(pl.col("weekday").is_in([6, 7]))
    reopen_bars = weekend_data.filter(
        (pl.col("weekday") == 7) & (pl.col("hour") >= FX_REOPEN_HOUR_UTC)
    )
    anomalous_weekend = weekend_data.filter(
        (pl.col("weekday") == 6) | (pl.col("hour") < FX_REOPEN_HOUR_UTC)
    )
    cov_report = coverage_report(
        fx_pairs,
        config["time_col"],
        config["symbol_col"],
        config["price_cols"],
    )

    print(f"Index Integrity: {'PASSED' if idx_check['passed'] else 'FAILED'}")
    print(
        f"Weekend bars: {len(weekend_data)} "
        f"({len(reopen_bars)} Sunday reopen, expected; "
        f"{len(anomalous_weekend)} anomalous)"
    )
    print(f"Coverage: {cov_report['overall_coverage_pct']}%")

    diagnostic_results["fx_pairs"] = {
        "index": idx_check,
        "coverage": cov_report,
        "weekend_data": len(weekend_data),
        "reopen_bars": len(reopen_bars),
        "anomalous_weekend": len(anomalous_weekend),
    }
```

FX trades continuously from the Sunday 17:00 EST reopen to the Friday close, so
Sunday bars at or after 21:00 UTC are expected market data. Saturday bars and
Sunday-daytime bars fall outside the trading week and are the genuine anomalies
to filter.

### Firm Characteristics

Chen-Pelger-Zhu academic monthly panel with 46 pre-computed characteristics
(plus `ret`, `timestamp`, and a `split` indicator). Coverage is engineered
upstream: rows that fail the source paper's data-availability rules are
dropped before the panel is published, so the columns themselves arrive
dense - but the *firm-month* footprint shrinks substantially relative to a
raw CRSP–Compustat join.

```python
firm_char = load_dataset_safely(DATASET_REGISTRY["firm_characteristics"]["loader"])

if firm_char is not None:
    config = DATASET_REGISTRY["firm_characteristics"]
    firm_char = filter_from_start(firm_char, "timestamp", FIRM_CHARACTERISTICS_START_DATE)

    n_dates = firm_char["timestamp"].n_unique()
    print(f"Loaded {len(firm_char):,} rows, {n_dates} months, {len(firm_char.columns)} columns")

    idx_check = check_index_integrity(firm_char, config["time_col"], config["symbol_col"])
    print(f"Index integrity: {'PASSED' if idx_check['passed'] else 'FAILED'}")
    print(f"  {idx_check['n_symbols']:,} firms, {idx_check['unique_pairs']:,} unique firm-months")

    # Sort columns alphabetically before slicing for reproducibility
    char_cols = sorted(c for c in firm_char.columns if c not in ["timestamp", "split", "ret"])[:10]
    cov_report = coverage_report(firm_char, config["time_col"], None, char_cols)
    print(f"Coverage (sample of 10 chars): {cov_report['overall_coverage_pct']}%")
    for col, stats in cov_report["worst_columns"][:5]:
        print(f"  {col}: {stats['pct_null']:.1f}% null")

    diagnostic_results["firm_characteristics"] = {
        "index": idx_check,
        "coverage": cov_report,
        "n_characteristics": len(char_cols),
    }
```

The sampled characteristics show complete column-level coverage, because the upstream
filter only retains firm-months with valid characteristic vectors. The cost of that
filter is in the row count: many small-cap firm-months are excluded entirely, which is
the right framing for Section 7.1's discussion of "missing due to observed coverage
rules". Downstream notebooks should not attempt to reconstruct dropped rows; treat the
panel as the authors' own coverage rule applied at source.

## Summary Dashboard

Comparative view across all datasets.

```python
def rag_status(
    passed: bool | None, warning_threshold: float | None = None, value: float | None = None
) -> str:
    """Convert pass/fail to RAG status."""
    if passed is None:
        return "N/A"
    if passed:
        if warning_threshold is not None and value is not None and value < warning_threshold:
            return "WARN"
        return "OK"
    return "FAIL"
```

```python
summary_rows = []

for name, results in diagnostic_results.items():
    row = {"Dataset": name}

    if "index" in results:
        row["Index"] = rag_status(results["index"]["passed"])
    else:
        row["Index"] = "N/A"

    if "duplicates" in results:
        row["Duplicates"] = rag_status(results["duplicates"]["passed"])
    else:
        row["Duplicates"] = "N/A"

    if "coverage" in results:
        cov_pct = results["coverage"]["overall_coverage_pct"]
        passed = cov_pct >= 95
        row["Coverage"] = f"{rag_status(passed, 99, cov_pct)} ({cov_pct}%)"
    else:
        row["Coverage"] = "N/A"

    if "outliers" in results:
        row["Outliers"] = rag_status(results["outliers"]["passed"])
    else:
        row["Outliers"] = "N/A"

    summary_rows.append(row)

summary_df = pl.DataFrame(summary_rows)
summary_df
```

All datasets pass index integrity. Coverage is high for market data (ETFs,
crypto, futures, FX). Firm characteristics show full coverage for the sampled
columns, though sparsity increases for less common characteristics.
US Equities has outlier flags (price spikes) that `02_preprocessing_pipeline`
addresses with domain filters and spike detection.

## Normality Testing

The **Jarque-Bera test** jointly tests whether skewness and kurtosis match
a normal distribution. Financial returns almost always reject normality
due to fat tails and asymmetric distributions.

```python
from ml4t.diagnostic.evaluation.distribution.tests import jarque_bera_test

quiet_ml4t_logging()

if "etfs" in diagnostic_results:
    etfs_returns = (
        etfs.sort(["symbol", "timestamp"])
        .with_columns(pl.col("close").pct_change().over("symbol").alias("returns"))
        .filter(pl.col("returns").is_not_null() & pl.col("returns").is_finite())
    )

    spy_returns = etfs_returns.filter(pl.col("symbol") == "SPY")["returns"].to_numpy()
    jb_result = jarque_bera_test(spy_returns)

    print("Jarque-Bera Normality Test (SPY Daily Returns)")
    print(f"  Statistic:       {jb_result.statistic:.1f}")
    print(f"  P-value:         {jb_result.p_value:.2e}")
    print(f"  Skewness:        {jb_result.skewness:.3f}")
    print(f"  Excess Kurtosis: {jb_result.excess_kurtosis:.3f}")
    print(f"  Normal:          {jb_result.is_normal}")
```

SPY daily returns strongly reject normality ($p \approx 0$), driven by
excess kurtosis (fat tails). This confirms that robust scaling and
winsorization - not standard z-scoring - are appropriate for financial
returns. See `02_preprocessing_pipeline` for winsorization in practice.

### Stationarity Quick Check

Prices are non-stationary (they trend); returns are typically stationary.
This validates using returns, not prices, as ML features and labels.
Stationarity testing is covered formally in Chapter 9.

```python
from ml4t.diagnostic.evaluation.stationarity import analyze_stationarity
from statsmodels.tools.sm_exceptions import InterpolationWarning

warnings.filterwarnings(
    "ignore",
    category=InterpolationWarning,
    module=r"ml4t\.diagnostic\.evaluation\.stationarity\.kpss_test",
)
quiet_ml4t_logging()

if "etfs" in diagnostic_results:
    spy_prices = etfs.filter(pl.col("symbol") == "SPY").sort("timestamp")["close"].to_numpy()

    price_stat = analyze_stationarity(spy_prices)
    return_stat = analyze_stationarity(spy_returns)

    print(f"SPY Prices  - Consensus: {price_stat.consensus}")
    print(f"SPY Returns - Consensus: {return_stat.consensus}")
```

Prices carry a unit root and returns do not, which is what the standard practice of
feeding models returns (or differences) rather than levels rests on. Chapter 9 covers
formal time series analysis including ADF, KPSS, and Phillips-Perron tests.

KPSS reports its p-value from a lookup table, and a statistic outside that table's
range gets the nearest tabulated bound rather than an interpolated value. The cell
above silences the resulting `InterpolationWarning` for this one module: the consensus
label turns on which side of the critical value the statistic falls, not on the
p-value's fourth digit, so the clipping changes nothing that is read here.

## Key Findings

Based on the diagnostic survey, the datasets fall into two groups:

**Ready for direct use** (no structural cleaning needed):
- **ETFs** - complete OHLCV coverage; outlier flags reflect Yahoo adjustment artifacts.
- **Crypto perpetuals and premium** - complete coverage, 8-hour-aligned timestamps.
- **CME futures** - session-aligned bars across every product, with no null closes.
- **Firm Characteristics** - column-level coverage is complete by construction;
  downstream models inherit the source paper's coverage rule.

**Require active preprocessing** (see `02_preprocessing_pipeline`):
1. **US Equities** - a penny-stock filter, extreme-return handling, and price-spike
   review. Every row the return threshold flags is an *upward* move, so corporate
   actions should be taken from `split_ratio` rather than inferred from return size;
   see the split analysis under **US Equities** above.
2. **FX Pairs** - almost every weekend timestamp is a Sunday reopen bar at or after
   the reopen hour and stays in the panel. Only the Saturday and Sunday-daytime bars
   are anomalies to filter before the spot-vs-forward analytics in Section 7.2.

The two lists above are qualitative on purpose. The cell below sizes them from the
stored diagnostics, so the figures move with the run instead of ageing inside a
sentence.

```python
us_diag = diagnostic_results.get("us_equities")
if us_diag is not None:
    penny_pct = 100 * us_diag["penny_stocks"] / us_diag["index"]["n_rows"]
    spike_bars = us_diag["outliers"]["flags"].get("price_spikes", 0)
    print("US equities")
    print(f"  rows below $1:              {us_diag['penny_stocks']:>10,}  ({penny_pct:.1f}%)")
    print(f"  rows with |return| > 1:     {us_diag['extreme_returns']:>10,}")
    print(f"  smallest such return:       {us_diag['extreme_return_min']:>+10.4f}")
    print(f"  price-spike bars:           {spike_bars:>10,}")

fx_diag = diagnostic_results.get("fx_pairs")
if fx_diag is not None:
    print("\nFX pairs")
    print(f"  weekend timestamps:         {fx_diag['weekend_data']:>10,}")
    print(f"  Sunday reopen, expected:    {fx_diag['reopen_bars']:>10,}")
    print(f"  Saturday or Sunday daytime: {fx_diag['anomalous_weekend']:>10,}")
```

The next notebook applies these cleaning steps and demonstrates split-aware
preprocessing to prevent information leakage.

## Key Takeaways

1. **Index integrity** is fundamental - time monotonicity and uniqueness must pass,
   and the check has to be capable of failing: sorting a frame before asking whether
   it is sorted returns True for every input
2. **Column coverage is not panel coverage** - a dataset can report complete columns
   while whole rows are absent, because a null count only sees rows that exist
3. **Domain violations** (negative prices/volume) indicate data quality issues
4. **Detect events from the column that records them** - `split_ratio` identifies
   splits directly, while a `|return| > 1` threshold is one-sided by construction and
   misses nearly every forward split
5. **Calendar alignment** matters for FX (weekends) and crypto (24/7)

**Next**: See `02_preprocessing_pipeline` for cleaning and split-aware preprocessing.
**Book**: Section 7.1 discusses why data quality determines model quality.
![notebook output](figures/p1_1.png)