Testing, Auditing & Data Hygiene

# pip install checklist pandas great_expectations evidently scikit-learn

# ─── 1. CheckList Behavioral Tests for Sentiment Model ───────
import checklist
from checklist.test_types import MFT, INV, DIR
from checklist.editor import Editor
from checklist.perturb import Perturb
from transformers import pipeline

sentiment_model = pipeline("sentiment-analysis",
    model="distilbert-base-uncased-finetuned-sst-2-english")

def predict_fn(texts):
    results = sentiment_model(texts)
    # Return probability of POSITIVE class
    return [r['score'] if r['label'] == 'POSITIVE' else 1 - r['score'] for r in results]

editor = Editor()

# ── MFT: Basic Positive and Negative ──
mft_data = [
    ("This product is amazing!", 1),
    ("Absolutely terrible experience.", 0),
    ("Works as described, nothing special.", None),  # neutral, skip
    ("I hate everything about this.", 0),
    ("Best purchase I've ever made!", 1),
    ("Complete waste of money.", 0),
]
test_mft = MFT(
    data=[d for d, _ in mft_data if _ is not None],
    labels=[l for _, l in mft_data if l is not None],
    name="Basic Positive/Negative",
    capability="Vocabulary",
    description="Verify model correctly classifies clearly positive/negative statements",
)

# ── INV: Name Invariance (gender, ethnicity should not matter) ──
base_texts = [
    "John Smith gave an excellent presentation.",
    "Mary Johnson submitted a great report.",
]
perturbed = []
for text in base_texts:
    # Perturb names: swap to different demographic names
    perturbed.extend([
        text.replace("John Smith", "Muhammad Ali").replace("Mary Johnson", "Fatima Hassan"),
        text.replace("John Smith", "Wei Chen").replace("Mary Johnson", "Yuki Tanaka"),
    ])

test_inv = INV(
    data=base_texts + perturbed,
    name="Name Invariance",
    capability="Fairness",
    description="Predictions should be identical when different demographic names are substituted",
    threshold=0.1,  # allow max 0.1 probability change
)

# ── DIR: Negation Test ──
positive_texts = [
    "This is a good product.",
    "The service was excellent.",
    "I really enjoyed this.",
]
negated = [t.replace("good", "not good").replace("excellent", "not excellent")
             .replace("enjoyed", "did not enjoy") for t in positive_texts]

test_dir = DIR(
    data=positive_texts + negated,
    labels=[1] * len(positive_texts) + [0] * len(negated),
    name="Negation Direction",
    capability="NLU",
    description="Adding negation should flip sentiment direction",
    threshold=0.5,
)

# Run tests
print("CheckList Behavioral Tests:")
for test in [test_mft, test_inv, test_dir]:
    try:
        test.run(predict_fn, n=len(test.data) if hasattr(test, 'data') else 10)
        test.summary(file=None)
    except Exception as e:
        print(f"  {test.name}: {e}")

# ─── 2. Sliced Evaluation ────────────────────────────────────
import pandas as pd
import numpy as np
from sklearn.metrics import accuracy_score, f1_score

# Simulate evaluation dataset with metadata
np.random.seed(42)
n = 500
eval_df = pd.DataFrame({
    'text': [f"Sample text {i}" for i in range(n)],
    'true_label': np.random.randint(0, 2, n),
    'pred_label': np.random.randint(0, 2, n),
    'confidence': np.random.uniform(0.5, 1.0, n),
    'gender': np.random.choice(['male', 'female', 'non-binary'], n, p=[0.48, 0.48, 0.04]),
    'age_group': np.random.choice(['18-34', '35-54', '55+'], n, p=[0.35, 0.40, 0.25]),
    'language': np.random.choice(['en', 'es', 'fr', 'other'], n, p=[0.70, 0.15, 0.10, 0.05]),
    'content_length': np.random.choice(['short', 'medium', 'long'], n),
})

# Introduce deliberate performance disparity for minority groups
mask_non_binary = eval_df['gender'] == 'non-binary'
eval_df.loc[mask_non_binary, 'pred_label'] = np.random.choice(
    [0, 1], mask_non_binary.sum(), p=[0.6, 0.4])  # worse for non-binary

print("\nSliced Evaluation Report:")
print(f"{'Slice':<25} {'N':>5} {'Accuracy':>10} {'F1':>8} {'Flag'}")
print("─" * 55)

# Overall
overall_acc = accuracy_score(eval_df['true_label'], eval_df['pred_label'])
overall_f1 = f1_score(eval_df['true_label'], eval_df['pred_label'])
print(f"  {'OVERALL':<23} {n:>5} {overall_acc:>10.4f} {overall_f1:>8.4f}")

# By subgroup
for col in ['gender', 'age_group', 'language', 'content_length']:
    for group in eval_df[col].unique():
        mask = eval_df[col] == group
        slice_df = eval_df[mask]
        if len(slice_df) < 10:
            continue
        acc = accuracy_score(slice_df['true_label'], slice_df['pred_label'])
        f1 = f1_score(slice_df['true_label'], slice_df['pred_label'], zero_division=0)
        flag = " ⚠ UNDERPERFORMING" if acc < overall_acc - 0.10 else ""
        print(f"  {col}={group:<18} {len(slice_df):>5} {acc:>10.4f} {f1:>8.4f}{flag}")

# ─── 3. Data Quality Audit ────────────────────────────────────
class DataQualityAuditor:
    def __init__(self, df: pd.DataFrame, dataset_name: str):
        self.df = df
        self.name = dataset_name
        self.issues = []

    def check_completeness(self, critical_columns: list, threshold: float = 0.95):
        for col in critical_columns:
            completeness = 1 - self.df[col].isna().mean()
            if completeness < threshold:
                self.issues.append({
                    'dimension': 'Completeness',
                    'severity': 'HIGH' if completeness < 0.8 else 'MEDIUM',
                    'column': col,
                    'value': f"{100*completeness:.1f}%",
                    'message': f"Column '{col}' is only {100*completeness:.1f}% complete (threshold: {100*threshold:.0f}%)"
                })
        return self

    def check_label_balance(self, label_col: str, min_ratio: float = 0.1):
        counts = self.df[label_col].value_counts(normalize=True)
        for label, ratio in counts.items():
            if ratio < min_ratio:
                self.issues.append({
                    'dimension': 'Representativeness',
                    'severity': 'HIGH',
                    'column': label_col,
                    'value': f"{100*ratio:.1f}%",
                    'message': f"Class '{label}' has only {100*ratio:.1f}% representation (min: {100*min_ratio:.0f}%)"
                })
        return self

    def check_duplicates(self, key_columns: list):
        dupes = self.df.duplicated(subset=key_columns).sum()
        if dupes > 0:
            self.issues.append({
                'dimension': 'Consistency',
                'severity': 'MEDIUM',
                'column': str(key_columns),
                'value': str(dupes),
                'message': f"{dupes} duplicate rows found on columns {key_columns}"
            })
        return self

    def check_demographic_representation(self, demo_col: str, expected_pop: dict):
        actual = self.df[demo_col].value_counts(normalize=True).to_dict()
        for group, expected_pct in expected_pop.items():
            actual_pct = actual.get(group, 0)
            deviation = abs(actual_pct - expected_pct / 100)
            if deviation > 0.10:  # >10% deviation
                self.issues.append({
                    'dimension': 'Representativeness',
                    'severity': 'HIGH' if deviation > 0.20 else 'MEDIUM',
                    'column': demo_col,
                    'value': f"actual={100*actual_pct:.0f}%, expected={expected_pct:.0f}%",
                    'message': f"Group '{group}': {100*actual_pct:.0f}% in dataset vs {expected_pct:.0f}% in population"
                })
        return self

    def report(self):
        print(f"\nData Quality Audit: {self.name}")
        print(f"{'='*60}")
        print(f"  Dataset shape: {self.df.shape}")
        if not self.issues:
            print("  ✓ No data quality issues found")
        else:
            high = [i for i in self.issues if i['severity'] == 'HIGH']
            medium = [i for i in self.issues if i['severity'] == 'MEDIUM']
            print(f"  Issues: {len(high)} HIGH, {len(medium)} MEDIUM")
            for issue in sorted(self.issues, key=lambda x: x['severity']):
                icon = "✗" if issue['severity'] == 'HIGH' else "~"
                print(f"\n  [{issue['severity']:6}] {icon} [{issue['dimension']}]")
                print(f"    {issue['message']}")
        return len([i for i in self.issues if i['severity'] == 'HIGH']) == 0

# Create test dataset with quality issues
test_data = pd.DataFrame({
    'text': [f"Sample {i}" for i in range(200)],
    'label': np.random.choice([0, 1, 2], 200, p=[0.85, 0.10, 0.05]),  # severe imbalance
    'age': np.where(np.random.rand(200) < 0.15, np.nan, np.random.randint(18, 80, 200)),  # 15% missing
    'gender': np.random.choice(['male', 'female'], 200, p=[0.92, 0.08]),  # underrepresents women
    'country': ['US'] * 150 + ['UK'] * 30 + ['FR'] * 20,
})
test_data = pd.concat([test_data, test_data.iloc[:5]])  # add duplicates

auditor = DataQualityAuditor(test_data, "Loan Application Training Data v2.1")
passed = (auditor
    .check_completeness(['text', 'label', 'age'], threshold=0.90)
    .check_label_balance('label', min_ratio=0.15)
    .check_duplicates(['text', 'label'])
    .check_demographic_representation('gender', {'male': 50, 'female': 50})
    .report())
print(f"\n  Overall: {'PASS' if passed else 'FAIL — block training until issues resolved'}")

# ─── 4. Production Drift Monitoring ──────────────────────────
from scipy.stats import ks_2samp, chi2_contingency

def monitor_model_health(ref_scores, prod_scores, ref_labels=None, prod_preds=None):
    """Monitor model health: confidence distribution + performance drift."""
    print("\nProduction Health Monitor")
    print("─" * 45)

    # Confidence distribution shift (KS test)
    ks_stat, p_val = ks_2samp(ref_scores, prod_scores)
    drift = "⚠ DRIFT DETECTED" if p_val < 0.05 else "✓ STABLE"
    print(f"  Confidence distribution: KS={ks_stat:.4f}, p={p_val:.4f} {drift}")

    # Mean confidence shift
    ref_mean, prod_mean = np.mean(ref_scores), np.mean(prod_scores)
    delta = abs(ref_mean - prod_mean)
    flag = " ⚠" if delta > 0.05 else ""
    print(f"  Mean confidence: ref={ref_mean:.4f}, prod={prod_mean:.4f}, Δ={delta:.4f}{flag}")

    # Prediction distribution (label drift)
    if prod_preds is not None:
        pos_rate_ref = 0.48  # expected from training
        pos_rate_prod = np.mean(prod_preds)
        flag = " ⚠ LABEL DRIFT" if abs(pos_rate_prod - pos_rate_ref) > 0.10 else ""
        print(f"  Positive prediction rate: ref={pos_rate_ref:.3f}, prod={pos_rate_prod:.3f}{flag}")

    return {'ks_stat': ks_stat, 'p_value': p_val, 'drift_detected': p_val < 0.05}

# Simulate reference (training) vs production distributions
ref_scores = np.random.beta(8, 3, 1000)  # training distribution
prod_scores_ok = np.random.beta(7.8, 3.1, 500)  # similar to training (OK)
prod_scores_drifted = np.random.beta(5, 5, 500)  # shifted distribution (drift!)

print("\nScenario A: Production distribution similar to training")
monitor_model_health(ref_scores, prod_scores_ok)

print("\nScenario B: Production distribution shifted (concept drift!)")
monitor_model_health(ref_scores, prod_scores_drifted, prod_preds=np.random.randint(0,2,500))