loader.py

"""
Multi-source, multi-rail data loader.

Sources: synthetic Parquet | UCI San Diego CSV | IEEE-CIS CSV | real_org CSV.
IEEE-CIS columns (V*, C*, D*, card1-card6) are OPAQUE — never mapped to canonical schema.
Only harmonised fields: amount_usd, timestamp_utc, is_fraud, DeviceType, ProductCD.
"""
from __future__ import annotations

from pathlib import Path
from typing import Optional

import polars as pl


# ── Source adapters ───────────────────────────────────────────────────────────

def load_synthetic(path: Path) -> pl.DataFrame:
    """Load synthetic Parquet (already in canonical schema)."""
    df = pl.read_parquet(path)
    return df.with_columns(pl.lit("synthetic").alias("data_source"))


def load_uci_san_diego(path: Path) -> pl.DataFrame:
    """
    UCI creditcard.csv (Kaggle).
    Columns: Time, V1-V28 (PCA), Amount, Class
    No merchant, geo, or device columns available.
    """
    df = pl.read_csv(path)
    df = df.rename({"Class": "is_fraud", "Amount": "amount_usd"})
    # Time is seconds from first transaction — convert to relative offset only
    df = df.with_columns([
        pl.col("amount_usd").cast(pl.Float64),
        pl.col("is_fraud").cast(pl.Int8),
        pl.lit("uc_san_diego").alias("data_source"),
        pl.lit("cards").alias("rail"),
        pl.lit("card_not_present").alias("channel"),
        pl.lit("IN").alias("merchant_country"),
        pl.lit("USD").alias("currency_code"),
        (pl.col("amount_usd") * 84.0).alias("amount_inr"),  # approx USD→INR
    ])
    # Keep V1-V28 as-is in source namespace
    return df


def load_ieee_cis(
    identity_path: Path,
    transaction_path: Path,
) -> pl.DataFrame:
    """
    IEEE-CIS Fraud Detection dataset.
    Columns: TransactionID, isFraud, TransactionDT, TransactionAmt,
             ProductCD, card1-card6, addr1-addr2, dist1-dist2,
             P_emaildomain, R_emaildomain, C1-C14, D1-D15, M1-M9,
             V1-V339, DeviceType, DeviceInfo (from identity)
    Opaque columns (V*, C*, D*, card*) kept in ieee_cis namespace.
    Only harmonise: amount_usd, timestamp_utc relative, is_fraud.
    """
    txn = pl.read_csv(transaction_path)
    idt = pl.read_csv(identity_path, ignore_errors=True)
    df = txn.join(idt, on="TransactionID", how="left")

    df = df.rename({
        "isFraud": "is_fraud",
        "TransactionAmt": "amount_usd",
    })
    df = df.with_columns([
        pl.col("amount_usd").cast(pl.Float64),
        pl.col("is_fraud").cast(pl.Int8),
        (pl.col("amount_usd") * 84.0).alias("amount_inr"),
        pl.lit("ieee_cis").alias("data_source"),
        pl.lit("cards").alias("rail"),
        pl.lit("USD").alias("currency_code"),
        pl.lit("IN").alias("merchant_country"),
        # ProductCD → channel proxy
        pl.when(pl.col("ProductCD") == "W").then(pl.lit("ecommerce"))
          .when(pl.col("ProductCD") == "S").then(pl.lit("card_present"))
          .otherwise(pl.lit("card_not_present")).alias("channel"),
    ])
    return df


def load_real_org(path: Path, column_map: Optional[dict] = None) -> pl.DataFrame:
    """
    Real org transactions (anonymised CSV or Parquet).
    column_map: maps org-specific column names → canonical names.
    """
    if path.suffix == ".parquet":
        df = pl.read_parquet(path)
    else:
        df = pl.read_csv(path, ignore_errors=True)

    if column_map:
        df = df.rename(column_map)

    return df.with_columns(pl.lit("real_org").alias("data_source"))


# ── Multi-source combiner ─────────────────────────────────────────────────────

COMMON_COLUMNS = [
    "transaction_id", "data_source", "rail", "channel",
    "amount_inr", "amount_usd", "currency_code",
    "is_fraud", "merchant_country",
]


def combine(frames: list[pl.DataFrame]) -> pl.DataFrame:
    """
    Combine frames from multiple sources.
    Only common columns kept — source-specific feature columns (V*, C*) dropped here.
    For full IEEE-CIS feature use, pass the frame directly to feature engineering.
    """
    result_frames = []
    for df in frames:
        available = [c for c in COMMON_COLUMNS if c in df.columns]
        result_frames.append(df.select(available))
    return pl.concat(result_frames, how="diagonal_relaxed")


# ── Entrypoint ────────────────────────────────────────────────────────────────

def load_all(
    synthetic_path: Optional[Path] = None,
    uci_path: Optional[Path] = None,
    ieee_txn_path: Optional[Path] = None,
    ieee_idt_path: Optional[Path] = None,
    real_org_path: Optional[Path] = None,
    real_org_column_map: Optional[dict] = None,
) -> pl.DataFrame:
    """Load all available sources and combine into one DataFrame."""
    frames: list[pl.DataFrame] = []

    if synthetic_path and synthetic_path.exists():
        frames.append(load_synthetic(synthetic_path))

    if uci_path and uci_path.exists():
        frames.append(load_uci_san_diego(uci_path))

    if ieee_txn_path and ieee_idt_path and ieee_txn_path.exists() and ieee_idt_path.exists():
        frames.append(load_ieee_cis(ieee_idt_path, ieee_txn_path))

    if real_org_path and real_org_path.exists():
        frames.append(load_real_org(real_org_path, real_org_column_map))

    if not frames:
        raise FileNotFoundError("No data sources found. Run generator first.")

    return combine(frames)