src/preprocessing/clean.py

"""Data cleaning and schema validation for raw CSV inputs."""

import pandas as pd

from src.utils.logger import get_logger

logger = get_logger(__name__)


def clean_orders(orders: pd.DataFrame) -> pd.DataFrame:
    """Validate and clean the orders fact table.

    Operations performed:
    - Drop rows with null ``order_date`` or ``customer_id`` (mandatory keys).
    - Coerce date columns to proper dtypes.
    - Clip ``total_value`` at 2× the 99th percentile (winsorisation).
    - Enforce minimum sensible values for ``quantity`` and ``price``.

    Args:
        orders: Raw orders DataFrame.

    Returns:
        Cleaned DataFrame.
    """
    df = orders.copy()
    initial = len(df)

    df["order_date"] = pd.to_datetime(df["order_date"], errors="coerce")
    df["contract_date"] = pd.to_datetime(df["contract_date"], errors="coerce")
    df = df.dropna(subset=["order_date", "customer_id"])

    # Clip extreme monetary values (2× P99)
    p99 = df["total_value"].quantile(0.99)
    df["total_value"] = df["total_value"].clip(upper=p99 * 2.0)

    df["quantity"] = df["quantity"].clip(lower=1)
    df["price"] = df["price"].clip(lower=0.01)

    df["customer_id"] = df["customer_id"].astype(int)
    df["order_id"] = df["order_id"].astype(int)

    removed = initial - len(df)
    if removed:
        logger.warning(f"clean_orders: removed {removed} invalid rows")
    else:
        logger.info(f"clean_orders: {len(df):,} rows, no rows removed")

    return df.reset_index(drop=True)


def clean_customers(customers: pd.DataFrame) -> pd.DataFrame:
    """Validate and coerce the customer dimension table.

    Args:
        customers: Raw customers DataFrame.

    Returns:
        Cleaned DataFrame.
    """
    df = customers.copy()

    df["registration_date"] = pd.to_datetime(df["registration_date"], errors="coerce")
    df["birth_date"] = pd.to_datetime(df["birth_date"], errors="coerce")
    df["last_profile_update"] = pd.to_datetime(df["last_profile_update"], errors="coerce")
    df["customer_id"] = df["customer_id"].astype(int)

    missing_reg = df["registration_date"].isna().sum()
    if missing_reg:
        logger.warning(f"clean_customers: {missing_reg} missing registration_date – dropping")
        df = df.dropna(subset=["registration_date"])

    logger.info(f"clean_customers: {len(df):,} customers")
    return df.reset_index(drop=True)