import pandas as pd

df = pd.read_csv("customers.csv")

# Count duplicates before removal
dupes = df.duplicated(subset=["customer_id"], keep=False)
print(f"Duplicate groups: {dupes.sum()} rows in {df[dupes]['customer_id'].nunique()} groups")

# Strategy A: Keep first occurrence (oldest record)
df_deduped = df.drop_duplicates(subset=["customer_id"], keep="first")

# Strategy B: Keep last occurrence (most recently updated)
df_deduped = df.drop_duplicates(subset=["customer_id"], keep="last")

# Strategy C: Keep the most complete record (fewest nulls)
df["_null_count"] = df.isnull().sum(axis=1)
df_sorted = df.sort_values("_null_count")
df_deduped = df_sorted.drop_duplicates(subset=["customer_id"], keep="first")
df_deduped = df_deduped.drop(columns=["_null_count"])

# Define composite key: name + email domain + signup month
# This catches cases where the same person signed up twice with
# slightly different info
df["_composite_key"] = (
    df["first_name"].str.lower().str.strip() + "|" +
    df["last_name"].str.lower().str.strip() + "|" +
    df["email"].str.split("@").str[1].str.lower() + "|" +
    df["signup_date"].str[:7]  # Year-month
)

df_deduped = df.drop_duplicates(subset=["_composite_key"], keep="last")
df_deduped = df_deduped.drop(columns=["_composite_key"])

from difflib import SequenceMatcher

def similarity(a: str, b: str) -> float:
    """Calculate string similarity ratio (0.0 to 1.0)."""
    if not a or not b:
        return 0.0
    return SequenceMatcher(None, a.lower(), b.lower()).ratio()

def find_fuzzy_duplicates(df, name_col="full_name", threshold=0.85):
    """
    Find potential duplicate pairs using fuzzy name matching.

    WARNING: O(n²) complexity. For large datasets (>10K rows),
    use blocking (pre-filter by first letter, zip code, etc.)
    to reduce comparison space.
    """
    potential_dupes = []
    names = df[name_col].tolist()
    indices = df.index.tolist()

    for i in range(len(names)):
        for j in range(i + 1, len(names)):
            sim = similarity(str(names[i]), str(names[j]))
            if sim >= threshold:
                potential_dupes.append({
                    "index_a": indices[i],
                    "index_b": indices[j],
                    "name_a": names[i],
                    "name_b": names[j],
                    "similarity": round(sim, 3),
                })

    return pd.DataFrame(potential_dupes)

def deduplicate_with_blocking(df, block_col="zip_code", match_cols=None, threshold=0.80):
    """
    Scalable fuzzy deduplication using blocking.

    Blocking dramatically reduces comparisons:
    - 100K records with no blocking: 5 billion comparisons
    - 100K records with 1000 zip codes: ~5 million comparisons (1000x faster)

    Args:
        df: Input DataFrame
        block_col: Column to group records by (records only compared within same block)
        match_cols: Columns to use for similarity scoring
        threshold: Minimum average similarity to consider a match
    """
    if match_cols is None:
        match_cols = ["first_name", "last_name", "email"]

    duplicates_to_remove = set()

    for block_value, group in df.groupby(block_col):
        if len(group) < 2:
            continue

        indices = group.index.tolist()
        for i in range(len(indices)):
            if indices[i] in duplicates_to_remove:
                continue
            for j in range(i + 1, len(indices)):
                if indices[j] in duplicates_to_remove:
                    continue

                # Score similarity across multiple columns
                scores = []
                for col in match_cols:
                    val_a = str(group.loc[indices[i], col])
                    val_b = str(group.loc[indices[j], col])
                    scores.append(similarity(val_a, val_b))

                avg_score = sum(scores) / len(scores)
                if avg_score >= threshold:
                    # Keep the record with fewer nulls
                    nulls_i = group.loc[indices[i]].isnull().sum()
                    nulls_j = group.loc[indices[j]].isnull().sum()
                    remove_idx = indices[j] if nulls_i <= nulls_j else indices[i]
                    duplicates_to_remove.add(remove_idx)

    return df.drop(index=duplicates_to_remove)

def link_records(df_a, df_b, match_fields, threshold=0.85):
    """
    Link records across two DataFrames from different source systems.

    Returns a mapping of df_a index -> df_b index for matched pairs.
    Unmatched records from either side are preserved (outer join behavior).

    Example:
        CRM has "John Smith, john@acme-corp.example.com"
        Billing has "J. Smith, jsmith@acme-corp.example.com"
        → These should link to the same entity
    """
    links = {}

    for idx_a, row_a in df_a.iterrows():
        best_match = None
        best_score = 0.0

        for idx_b, row_b in df_b.iterrows():
            scores = []
            for field_a, field_b, weight in match_fields:
                val_a = str(row_a.get(field_a, ""))
                val_b = str(row_b.get(field_b, ""))
                sim = similarity(val_a, val_b)
                scores.append(sim * weight)

            total_weight = sum(w for _, _, w in match_fields)
            weighted_score = sum(scores) / total_weight

            if weighted_score > best_score:
                best_score = weighted_score
                best_match = idx_b

        if best_score >= threshold:
            links[idx_a] = {"matched_index": best_match, "confidence": best_score}

    return links

# Usage:
# match_fields = [
#     ("email", "billing_email", 2.0),      # High weight: email is reliable
#     ("last_name", "surname", 1.5),         # Medium weight
#     ("first_name", "given_name", 1.0),     # Lower weight (more variable)
#     ("phone", "contact_number", 1.5),      # Medium weight
# ]
# links = link_records(crm_df, billing_df, match_fields)

# Simple: keep the most recently updated record
df_golden = df.sort_values("updated_at").drop_duplicates("customer_id", keep="last")

# Keep the record with the most non-null values
df["_completeness"] = df.notna().sum(axis=1)
df_golden = df.sort_values("_completeness", ascending=False).drop_duplicates("customer_id", keep="first")

def merge_duplicate_group(group):
    """
    For each field, pick the best non-null value from the group.

    Priority rules:
    - Timestamps: most recent non-null
    - Numeric: maximum (or average, depending on business rule)
    - Text: longest non-null (more complete)
    - Email: first valid non-null
    """
    merged = {}
    for col in group.columns:
        non_null = group[col].dropna()
        if non_null.empty:
            merged[col] = None
        elif col.endswith("_date") or col.endswith("_at"):
            merged[col] = non_null.max()  # Most recent
        elif group[col].dtype in ("float64", "int64"):
            merged[col] = non_null.max()  # Highest value
        else:
            # Longest string (usually most complete)
            merged[col] = max(non_null.astype(str), key=len)
    return pd.Series(merged)

df_golden = df.groupby("customer_id").apply(merge_duplicate_group).reset_index(drop=True)

import pandas as pd
import numpy as np


def missing_data_report(df: pd.DataFrame) -> pd.DataFrame:
    """
    Generate a comprehensive missing data report.

    Returns a DataFrame with one row per column showing:
    - Count and percentage of missing values
    - Data type (to inform imputation strategy)
    - Number of unique values (helps distinguish categorical vs continuous)
    """
    report = pd.DataFrame({
        "column": df.columns,
        "dtype": df.dtypes.values,
        "missing_count": df.isnull().sum().values,
        "missing_pct": (df.isnull().sum() / len(df) * 100).values,
        "unique_values": df.nunique().values,
        "sample_values": [df[col].dropna().head(3).tolist() for col in df.columns],
    })

    report = report.sort_values("missing_pct", ascending=False)
    report["missing_pct"] = report["missing_pct"].round(1)

    return report


def missing_pattern_matrix(df: pd.DataFrame) -> pd.DataFrame:
    """
    Show co-occurrence patterns of missing values.

    If columns A and B are always missing together, they likely have
    the same root cause (e.g., a form section that users skip).
    """
    missing_mask = df.isnull().astype(int)
    # Each unique pattern of missingness
    patterns = missing_mask.drop_duplicates()
    pattern_counts = missing_mask.groupby(list(df.columns)).size().reset_index(name="count")
    pattern_counts = pattern_counts.sort_values("count", ascending=False)
    return pattern_counts

def classify_missingness(df: pd.DataFrame, target_col: str) -> dict:
    """
    Heuristic classification of missingness mechanism.

    Tests whether missingness in target_col correlates with other columns.
    If correlations are found → likely MAR.
    If no correlations found → possibly MCAR (or MNAR, which can't be tested).

    Returns dict with correlations and suggested mechanism.
    """
    is_missing = df[target_col].isnull().astype(int)

    correlations = {}
    for col in df.columns:
        if col == target_col:
            continue
        if df[col].dtype in ("float64", "int64"):
            # Point-biserial correlation between missingness indicator and numeric column
            valid_mask = df[col].notna()
            if valid_mask.sum() > 10:
                corr = is_missing[valid_mask].corr(df[col][valid_mask])
                if abs(corr) > 0.1:
                    correlations[col] = round(corr, 3)

    if not correlations:
        mechanism = "MCAR (no correlations found) or MNAR (untestable)"
    elif max(abs(v) for v in correlations.values()) > 0.3:
        mechanism = "MAR (strong correlations with other variables)"
    else:
        mechanism = "Possibly MAR (weak correlations detected)"

    return {
        "target_column": target_col,
        "missing_pct": round(is_missing.mean() * 100, 1),
        "correlated_columns": correlations,
        "suggested_mechanism": mechanism,
    }

def drop_missing_rows(df: pd.DataFrame, columns: list[str], threshold: float = 0.5) -> pd.DataFrame:
    """
    Drop rows where critical columns are missing.

    Only use when:
    - Missing % is low (< 5%)
    - Missingness is MCAR
    - You have plenty of data remaining

    Args:
        df: Input DataFrame
        columns: Columns to check for missingness
        threshold: Drop row if more than this fraction of specified columns are null

    Returns:
        DataFrame with rows removed
    """
    null_fraction = df[columns].isnull().sum(axis=1) / len(columns)
    mask = null_fraction <= threshold
    dropped = (~mask).sum()
    print(f"Dropping {dropped} rows ({dropped/len(df):.1%}) with >{threshold:.0%} missing in {columns}")
    return df[mask].copy()

def impute_simple(df: pd.DataFrame, strategies: dict[str, str]) -> pd.DataFrame:
    """
    Apply simple imputation strategies per column.

    Strategies:
        - "mean": Replace with column mean (numeric, symmetric distributions)
        - "median": Replace with column median (numeric, skewed distributions)
        - "mode": Replace with most frequent value (categorical)
        - "zero": Replace with 0 (when absence = zero, e.g., count of purchases)
        - "unknown": Replace with "Unknown" string (categorical, explicit missing category)
        - "ffill": Forward fill (time series, carry last known value)
        - "bfill": Backward fill (time series, use next known value)

    Args:
        df: Input DataFrame
        strategies: Dict mapping column name -> strategy name

    Returns:
        DataFrame with nulls filled
    """
    df = df.copy()

    for col, strategy in strategies.items():
        if col not in df.columns:
            continue

        null_count = df[col].isnull().sum()
        if null_count == 0:
            continue

        if strategy == "mean":
            df[col] = df[col].fillna(df[col].mean())
        elif strategy == "median":
            df[col] = df[col].fillna(df[col].median())
        elif strategy == "mode":
            mode_val = df[col].mode().iloc[0] if not df[col].mode().empty else "Unknown"
            df[col] = df[col].fillna(mode_val)
        elif strategy == "zero":
            df[col] = df[col].fillna(0)
        elif strategy == "unknown":
            df[col] = df[col].fillna("Unknown")
        elif strategy == "ffill":
            df[col] = df[col].ffill()
        elif strategy == "bfill":
            df[col] = df[col].bfill()

        print(f"  {col}: filled {null_count} nulls with {strategy}")

    return df


# Example usage:
# strategies = {
#     "age": "median",           # Skewed, use median not mean
#     "income": "median",        # Skewed right
#     "gender": "mode",          # Categorical
#     "signup_source": "unknown", # Explicit missing category
#     "purchase_count": "zero",  # No purchases = 0 purchases
#     "temperature": "ffill",    # Time series: last known value
# }
# df_clean = impute_simple(df, strategies)

def impute_conditional(
    df: pd.DataFrame,
    target_col: str,
    group_cols: list[str],
    method: str = "median",
) -> pd.DataFrame:
    """
    Impute missing values using group-specific statistics.

    Instead of using the global median, uses the median within each group.
    This handles MAR data where missingness correlates with group membership.

    Example:
        Missing income → impute with median income for same age_group + education_level
        (Much better than global median because a 25-year-old with a PhD has different
        expected income than a 55-year-old with a high school diploma)

    Args:
        df: Input DataFrame
        target_col: Column to impute
        group_cols: Columns to group by for conditional statistics
        method: "median" or "mean"
    """
    df = df.copy()

    if method == "median":
        fill_values = df.groupby(group_cols)[target_col].transform("median")
    else:
        fill_values = df.groupby(group_cols)[target_col].transform("mean")

    # Fill with group statistic where available
    null_mask = df[target_col].isnull()
    df.loc[null_mask, target_col] = fill_values[null_mask]

    # Fall back to global statistic for groups that are entirely null
    still_null = df[target_col].isnull().sum()
    if still_null > 0:
        global_fill = df[target_col].median() if method == "median" else df[target_col].mean()
        df[target_col] = df[target_col].fillna(global_fill)
        print(f"  {still_null} values fell back to global {method}")

    return df

def add_missing_indicators(df: pd.DataFrame, columns: list[str]) -> pd.DataFrame:
    """
    Add binary indicator columns for missingness BEFORE imputing.

    This preserves the information that a value was missing, which
    may be predictive in itself (e.g., people who skip income question
    tend to have different behavior patterns).

    Always add indicators BEFORE imputation so the model can learn
    from the missingness pattern.

    Args:
        df: Input DataFrame
        columns: Columns to create indicators for

    Returns:
        DataFrame with added {col}_missing columns
    """
    df = df.copy()
    for col in columns:
        if col in df.columns:
            indicator_col = f"{col}_was_missing"
            df[indicator_col] = df[col].isnull().astype(int)
    return df

def handle_categorical_missing(df: pd.DataFrame, cat_columns: list[str]) -> pd.DataFrame:
    """
    For categorical columns, treat missing as its own category.

    This is often the BEST strategy for categorical data because:
    1. "Unknown" is meaningful (customer chose not to answer)
    2. Imputing mode creates false certainty
    3. Models can learn that "Unknown" has its own behavior pattern

    For boolean-like columns, you might want "Unknown" as a third state
    rather than forcing True/False.
    """
    df = df.copy()
    for col in cat_columns:
        if col in df.columns:
            null_count = df[col].isnull().sum()
            if null_count > 0:
                df[col] = df[col].fillna("Unknown")
                print(f"  {col}: {null_count} nulls → 'Unknown' category")
    return df

# WRONG: Statistics computed on full dataset leak test info into training
df_imputed = df.fillna(df.mean())  # Uses ALL data for mean
train, test = train_test_split(df_imputed)

# CORRECT: Compute statistics on training set only, apply to both
train, test = train_test_split(df)
train_means = train.mean()
train_filled = train.fillna(train_means)
test_filled = test.fillna(train_means)  # Use TRAINING means on test

# WRONG: Inner join silently drops rows where join key is null
merged = orders.merge(customers, on="customer_id")  # Loses null customer_id orders

# CORRECT: Explicitly handle null keys before joining
orphan_orders = orders[orders["customer_id"].isnull()]
print(f"WARNING: {len(orphan_orders)} orders have no customer_id — excluded from join")
merged = orders.dropna(subset=["customer_id"]).merge(customers, on="customer_id")

# For income data: mean = $85,000, median = $52,000
# Mean is pulled up by high earners — bad imputation value for typical person

# WRONG for skewed data:
df["income"] = df["income"].fillna(df["income"].mean())  # $85K — too high for most

# CORRECT for skewed data:
df["income"] = df["income"].fillna(df["income"].median())  # $52K — better central tendency

def validate_imputation(df_before: pd.DataFrame, df_after: pd.DataFrame, columns: list[str]):
    """
    Compare distributions before and after imputation to catch problems.

    Red flags:
    - Mean shifted by more than 10%
    - Variance decreased by more than 20% (over-smoothing)
    - New values outside original range (impossible values created)
    """
    print("=" * 60)
    print("IMPUTATION VALIDATION REPORT")
    print("=" * 60)

    for col in columns:
        if col not in df_before.columns:
            continue

        before = df_before[col].dropna()
        after = df_after[col].dropna()

        if before.dtype in ("float64", "int64"):
            mean_shift = abs(after.mean() - before.mean()) / before.mean() * 100
            var_change = (after.var() - before.var()) / before.var() * 100

            status = "OK"
            if mean_shift > 10:
                status = "WARNING: Mean shifted >10%"
            if var_change < -20:
                status = "WARNING: Variance decreased >20%"

            print(f"\n  {col}:")
            print(f"    Mean:  {before.mean():.2f} → {after.mean():.2f} ({mean_shift:+.1f}%)")
            print(f"    Std:   {before.std():.2f} → {after.std():.2f}")
            print(f"    Range: [{before.min():.2f}, {before.max():.2f}] → [{after.min():.2f}, {after.max():.2f}]")
            print(f"    Status: {status}")