IBM AI Fairness 360

Tool ID: ibm_aif360 Latest Tested Version: 0.6.1 Documentation: AIF360 GitHub

Overview

IBM AI Fairness 360 (AIF360) is an open source toolkit containing over 70 fairness metrics and 10 bias mitigation algorithms. ATHENA integrates with AIF360 to ingest fairness metrics and detect bias amplification.

Prerequisites

pip install aif360

Supported Metrics

AIF360 Metric	ATHENA Metric Name	Value Range	Ideal Value
statistical_parity_difference()	statistical_parity_difference	-1 to 1	0
disparate_impact()	disparate_impact	0 to infinity	1
equal_opportunity_difference()	equal_opportunity	-1 to 1	0
average_odds_difference()	equalized_odds	-1 to 1	0
theil_index()	theil_index	0 to infinity	0
consistency()	consistency	0 to 1	1

Integration Code

Complete Example

from aif360.datasets import BinaryLabelDataset
from aif360.metrics import BinaryLabelDatasetMetric, ClassificationMetric
import requests
import json
from datetime import datetime, timezone

# Your ATHENA API configuration
ATHENA_API_KEY = "sk_live_xxxxx"
ATHENA_API_URL = "https://api.athenatrust.ai/v1"

def analyze_and_send_to_athena(
    dataset: BinaryLabelDataset,
    predictions,
    model_id: str,
    protected_attribute: str,
    privileged_groups: list,
    unprivileged_groups: list,
    aif360_version: str = "0.6.1"
):
    """
    Analyze a dataset with AIF360 and send results to ATHENA.
    """
    
    # Calculate dataset level metrics
    dataset_metric = BinaryLabelDatasetMetric(
        dataset,
        unprivileged_groups=unprivileged_groups,
        privileged_groups=privileged_groups
    )
    
    # Calculate classification metrics (requires predictions)
    classified_dataset = dataset.copy()
    classified_dataset.labels = predictions
    
    classification_metric = ClassificationMetric(
        dataset,
        classified_dataset,
        unprivileged_groups=unprivileged_groups,
        privileged_groups=privileged_groups
    )
    
    # Gather all metrics
    metrics = {
        "statistical_parity_difference": dataset_metric.statistical_parity_difference(),
        "disparate_impact": dataset_metric.disparate_impact(),
        "consistency": dataset_metric.consistency()[0],
        "equal_opportunity_difference": classification_metric.equal_opportunity_difference(),
        "average_odds_difference": classification_metric.average_odds_difference(),
        "theil_index": classification_metric.theil_index()
    }
    
    # Define thresholds (adjust based on your domain)
    thresholds = {
        "statistical_parity_difference": 0.1,
        "disparate_impact": 0.8,  # EEOC 4/5ths rule
        "equal_opportunity_difference": 0.1,
        "average_odds_difference": 0.1,
        "consistency": 0.9,
        "theil_index": 0.2
    }
    
    # Send each metric to ATHENA
    results = []
    for metric_name, metric_value in metrics.items():
        # Skip NaN values
        if metric_value is None or (isinstance(metric_value, float) and metric_value != metric_value):
            continue
            
        threshold = thresholds.get(metric_name)
        
        # Determine if threshold is violated
        passes_threshold = True
        if threshold is not None:
            if metric_name in ["disparate_impact", "consistency"]:
                passes_threshold = metric_value >= threshold
            else:
                passes_threshold = abs(metric_value) <= threshold
        
        # Normalize metric value to 0 to 1 range for ATHENA
        normalized_value = normalize_metric(metric_name, metric_value)
        
        payload = {
            "externalToolId": "ibm_aif360",
            "externalToolVersion": aif360_version,
            "modelId": model_id,
            "metricName": athena_metric_name(metric_name),
            "metricValue": normalized_value,
            "threshold": threshold,
            "passesThreshold": passes_threshold,
            "protectedAttribute": protected_attribute,
            "privilegedGroup": str(privileged_groups[0]),
            "unprivilegedGroup": str(unprivileged_groups[0]),
            "sampleSize": len(dataset.features),
            "rawPayload": {
                "original_value": metric_value,
                "metric_name": metric_name,
                "all_metrics": metrics
            },
            "signalTimestamp": datetime.now(timezone.utc).isoformat()
        }
        
        response = requests.post(
            f"{ATHENA_API_URL}/model-fairness-signals",
            headers={
                "Authorization": f"Bearer {ATHENA_API_KEY}",
                "Content-Type": "application/json"
            },
            json=payload
        )
        
        if response.status_code == 201:
            results.append({
                "metric": metric_name,
                "signalId": response.json().get("signalId"),
                "status": "success"
            })
        else:
            results.append({
                "metric": metric_name,
                "error": response.json(),
                "status": "failed"
            })
    
    return results


def normalize_metric(metric_name: str, value: float) -> float:
    """Normalize AIF360 metrics to 0 to 1 range for ATHENA."""
    if metric_name == "disparate_impact":
        # 0 to 2 range, 1 is ideal
        return min(1.0, max(0.0, value / 2))
    elif metric_name in ["statistical_parity_difference", "equal_opportunity_difference", "average_odds_difference"]:
        # -1 to 1 range, 0 is ideal
        return (value + 1) / 2
    elif metric_name == "consistency":
        # Already 0 to 1, higher is better
        return value
    elif metric_name == "theil_index":
        # 0 to infinity, 0 is ideal
        return min(1.0, value)
    return min(1.0, max(0.0, value))


def athena_metric_name(aif360_name: str) -> str:
    """Map AIF360 metric names to ATHENA standard names."""
    mapping = {
        "statistical_parity_difference": "statistical_parity_difference",
        "disparate_impact": "disparate_impact",
        "equal_opportunity_difference": "equal_opportunity",
        "average_odds_difference": "equalized_odds",
        "consistency": "consistency",
        "theil_index": "theil_index"
    }
    return mapping.get(aif360_name, "custom")


# Usage Example
if __name__ == "__main__":
    from aif360.datasets import GermanDataset
    from sklearn.linear_model import LogisticRegression
    
    # Load dataset
    dataset = GermanDataset()
    
    # Train a simple model
    X = dataset.features
    y = dataset.labels.ravel()
    model = LogisticRegression(max_iter=1000)
    model.fit(X, y)
    predictions = model.predict(X)
    
    # Analyze and send to ATHENA
    results = analyze_and_send_to_athena(
        dataset=dataset,
        predictions=predictions,
        model_id="german_credit_v1",
        protected_attribute="age",
        privileged_groups=[{"age": 1}],
        unprivileged_groups=[{"age": 0}]
    )
    
    print(json.dumps(results, indent=2))

Automated Pipeline Integration

With MLflow

import mlflow

def log_fairness_to_athena(run_id: str, dataset, predictions, model_id: str):
    """Log fairness metrics to both MLflow and ATHENA."""
    
    # Get AIF360 metrics
    results = analyze_and_send_to_athena(
        dataset=dataset,
        predictions=predictions,
        model_id=model_id,
        protected_attribute="gender",
        privileged_groups=[{"gender": 1}],
        unprivileged_groups=[{"gender": 0}]
    )
    
    # Also log to MLflow for tracking
    with mlflow.start_run(run_id=run_id):
        for result in results:
            if result["status"] == "success":
                mlflow.log_metric(f"athena_{result['metric']}", result.get("signalId"))
    
    return results

With GitHub Actions

name: Fairness Audit

on:
  schedule:
    - cron: '0 6 * * *'  # Daily at 6 AM
  workflow_dispatch:

jobs:
  fairness_audit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: pip install aif360 requests pandas scikit-learn
      
      - name: Run fairness analysis
        env:
          ATHENA_API_KEY: ${{ secrets.ATHENA_API_KEY }}
        run: python scripts/fairness_audit.py

Interpreting Results in ATHENA

After sending AIF360 metrics, check the ATHENA dashboard:

1. External Bias Signals: View all ingested metrics

2. Amplification Alerts: See if human behavior is amplifying bias

3. Subgroup Chart: Compare AIF360 metrics with ATHENA trust patterns

Example Alert

{
  "alertId": "alert_xyz789",
  "severity": "high",
  "message": "Bias amplification detected: AIF360 shows 15% statistical parity difference for gender=female, and ATHENA detects 42% overtrust rate for this group. Humans are amplifying model bias.",
  "recommendation": "Review approval decisions for female applicants. Consider bias mitigation in model or human training."
}

Troubleshooting

Common Issues

Error	Cause	Solution
FAIRNESS_INVALID_METRIC	Unsupported metric name	Use standardized names from table above
SIGNAL_IN_FUTURE	Timestamp ahead of server time	Use UTC timestamps
SIGNAL_TOO_OLD	Signal more than 90 days old	Send recent metrics only
NaN values	Metric calculation failed	Check for division by zero in AIF360

Next Steps

• Fairlearn Integration

• AWS Clarify Integration

• Google Vertex Integration