fairness_metrics.py

"""
Fairness Analysis for Recommendation Systems

Provides simple, interpretable fairness checks:
- Demographic parity: Do different demographic groups receive similar recommendation rates?
- Representation variance: Is representation across protected attributes evenly distributed?

These metrics are analysis-only and do NOT enforce constraints.
They are designed to surface potential biases for governance decisions.
"""

import numpy as np
from typing import List, Dict, Tuple
from collections import Counter


def demographic_parity(
    recommendations: List[Dict],
    demographic_attr: str,
    outcome_attr: str = "is_recommended"
) -> Dict[str, float]:
    """
    Calculate demographic parity: recommendation rate per demographic group.
    
    Demographic parity examines whether the system recommends at similar rates
    across different demographic groups (e.g., gender, age). Large differences
    may indicate bias requiring governance attention.
    
    Args:
        recommendations: List of records, each with demographic attributes and outcomes
                        e.g., [{"gender": "M", "is_recommended": True}, ...]
        demographic_attr: Attribute to group by (e.g., "gender", "category")
        outcome_attr: Binary outcome attribute (e.g., "is_recommended", "selected")
    
    Returns:
        Dictionary mapping demographic values to recommendation rates (0-1)
    """
    if not recommendations:
        return {}
    
    # Group by demographic attribute
    groups = {}
    
    for record in recommendations:
        group = record.get(demographic_attr, "UNKNOWN")
        
        if group not in groups:
            groups[group] = {"total": 0, "recommended": 0}
        
        groups[group]["total"] += 1
        
        # Count positive outcome
        if record.get(outcome_attr, False):
            groups[group]["recommended"] += 1
    
    # Calculate recommendation rate per group
    rates = {}
    for group, counts in groups.items():
        rate = counts["recommended"] / counts["total"] if counts["total"] > 0 else 0.0
        rates[group] = rate
    
    return rates


def parity_gap(rates: Dict[str, float]) -> float:
    """
    Calculate maximum parity gap across groups.
    
    Gap = max_rate - min_rate
    
    A gap close to 0 indicates fairness (similar rates across groups).
    A gap > 0.1 warrants investigation.
    
    Args:
        rates: Dictionary of group -> rate mappings
    
    Returns:
        Gap value (0-1)
    """
    if not rates or len(rates) < 2:
        return 0.0
    
    values = list(rates.values())
    return max(values) - min(values)


def representation_variance(
    rankings: List[List[Dict]],
    demographic_attr: str,
    top_k: int = 5
) -> Dict[str, float]:
    """
    Calculate variance in demographic representation in top-k recommendations.
    
    For each ranking, analyze what fraction of top-k recommendations are from
    each demographic group. Variance measures how consistent this distribution is.
    
    High variance suggests that demographic composition of top-k recommendations
    varies significantly across users—some users might see more homogeneous
    recommendations than others.
    
    Args:
        rankings: List of rankings, each ranking is a list of recommended items
                 e.g., [[{...}, {...}, ...], [{...}, {...}, ...]]
        demographic_attr: Demographic attribute (e.g., "category")
        top_k: Position cutoff
    
    Returns:
        Dictionary mapping demographic values to their representation distribution
    """
    if not rankings:
        return {}
    
    # For each demographic group, collect representation percentages across rankings
    group_representations = {}
    
    for ranking in rankings:
        top_k_items = ranking[:top_k]
        
        if not top_k_items:
            continue
        
        # Count representations in this ranking
        group_counts = Counter()
        for item in top_k_items:
            group = item.get(demographic_attr, "UNKNOWN")
            group_counts[group] += 1
        
        # Convert to percentages
        for group in group_counts:
            if group not in group_representations:
                group_representations[group] = []
            
            percentage = group_counts[group] / len(top_k_items)
            group_representations[group].append(percentage)
    
    # Calculate variance (standard deviation) for each group
    variances = {}
    for group, percentages in group_representations.items():
        if percentages:
            variance = np.std(percentages)
            variances[group] = variance
        else:
            variances[group] = 0.0
    
    return variances


def representation_metrics(
    rankings: List[List[Dict]],
    demographic_attr: str,
    top_k: int = 5
) -> Dict[str, float]:
    """
    Comprehensive representation analysis.
    
    Returns mean and max representation variance.
    
    Args:
        rankings: List of ranked recommendation lists
        demographic_attr: Demographic attribute
        top_k: Cutoff
    
    Returns:
        Dictionary with "mean_variance" and "max_variance"
    """
    variances = representation_variance(rankings, demographic_attr, top_k)
    
    if not variances:
        return {"mean_variance": 0.0, "max_variance": 0.0}
    
    values = list(variances.values())
    
    return {
        "mean_variance": np.mean(values),
        "max_variance": np.max(values),
        "variances_by_group": variances,
    }


def fairness_report(
    recommendations: List[Dict],
    rankings: List[List[Dict]],
    demographic_attrs: List[str] = ["gender", "category"],
    top_k: int = 5
) -> Dict:
    """
    Generate comprehensive fairness report.
    
    Analyzes demographic parity and representation variance across specified
    demographic attributes.
    
    Args:
        recommendations: Flattened list of all recommendations with demographics
        rankings: List of ranked lists (for representation variance)
        demographic_attrs: List of attributes to analyze
        top_k: Cutoff for representation analysis
    
    Returns:
        Dictionary with fairness metrics for each demographic attribute
    """
    report = {
        "demographic_parity": {},
        "parity_gaps": {},
        "representation_metrics": {},
    }
    
    for attr in demographic_attrs:
        # Demographic parity
        rates = demographic_parity(recommendations, attr)
        report["demographic_parity"][attr] = rates
        
        # Parity gap
        gap = parity_gap(rates)
        report["parity_gaps"][attr] = gap
        
        # Representation variance
        rep_metrics = representation_metrics(rankings, attr, top_k)
        report["representation_metrics"][attr] = rep_metrics
    
    return report


def fairness_summary(report: Dict) -> str:
    """
    Generate human-readable fairness summary.
    
    Args:
        report: Output from fairness_report()
    
    Returns:
        Formatted string with findings
    """
    lines = [
        "=" * 80,
        "FAIRNESS ANALYSIS REPORT",
        "=" * 80,
        "",
    ]
    
    # Demographic parity section
    lines.append("DEMOGRAPHIC PARITY (Recommendation Rates by Group)")
    lines.append("-" * 80)
    
    for attr, rates in report.get("demographic_parity", {}).items():
        lines.append(f"\n{attr.upper()}:")
        for group, rate in sorted(rates.items()):
            lines.append(f"  {group:20s}: {rate:.1%}")
        
        gap = report.get("parity_gaps", {}).get(attr, 0)
        gap_indicator = "✓ OK" if gap < 0.1 else "⚠ WARNING"
        lines.append(f"  Gap: {gap:.1%} {gap_indicator}")
    
    # Representation variance section
    lines.append("\n")
    lines.append("REPRESENTATION VARIANCE (Top-K Distribution Consistency)")
    lines.append("-" * 80)
    
    for attr, metrics in report.get("representation_metrics", {}).items():
        lines.append(f"\n{attr.upper()}:")
        lines.append(f"  Mean variance: {metrics.get('mean_variance', 0):.4f}")
        lines.append(f"  Max variance:  {metrics.get('max_variance', 0):.4f}")
        
        variances_by_group = metrics.get("variances_by_group", {})
        if variances_by_group:
            for group, var in sorted(variances_by_group.items()):
                lines.append(f"    {group}: {var:.4f}")
    
    lines.append("")
    lines.append("=" * 80)
    lines.append("Note: High values indicate potential fairness concerns.")
    lines.append("Analysis is for monitoring only; no constraints are enforced.")
    lines.append("=" * 80)
    
    return "\n".join(lines)