# Project export: FairFlow
This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.
## Project metadata
- Hackathon: OpenAI Build Week
- Tagline: FairFlow is an open-source AI fairness auditing tool that detects algorithmic bias using SPD, Disparate Impact, Equalized Odds, SHAP, and AI-powered explanations for everyone.
- Devpost: https://devpost.com/software/citadel-ai
- GitHub: https://github.com/CheerathAniketh/EquiLens-AI
- Demo: https://equilens-ai.onrender.com/
- Video: https://www.youtube.com/embed/V2TeTDVrK6g?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 0 GitHub contributor(s) —
## Devpost submission (written by the team)
### Overview
Artificial intelligence is increasingly used to make high-impact decisions in hiring, lending, healthcare, education, and public services. While these systems improve efficiency, they can also inherit and amplify historical bias hidden within training data. Well-known examples, such as biased hiring algorithms and unfair criminal risk assessment systems, highlight the importance of auditing AI before deployment. Existing fairness toolkits are powerful but primarily target machine learning practitioners, often exposing statistical outputs that are difficult for non-technical users to interpret. EquiLens-AI was inspired by a simple question: What if anyone—an NGO worker, teacher, policymaker, or student—could audit AI systems for fairness without needing a data science degree? Our goal was to democratize AI fairness by combining explainable AI, fairness metrics, and natural-language explanations into one accessible platform. EquiLens-AI is an open-source AI fairness auditing platform that detects, explains, and helps mitigate algorithmic bias in datasets. Users simply upload a CSV dataset, select a target column and one or two sensitive attributes, and EquiLens-AI generates a complete fairness report. Features Upload CSV datasets Compute fairness metrics: Statistical Parity Difference (SPD) Disparate Impact (DI) Equalized Odds (EO) Statistical Parity Difference (SPD) Disparate Impact (DI) Equalized Odds (EO) SHAP-based explainability Intersectional fairness analysis Audience-aware AI explanations (NGO, Student, Policymaker) Interactive "What-if" simulator Downloadable PDF audit reports Compliance guidance (EEOC, GDPR, EU AI Act) Fairness Metrics Statistical Parity Difference $$ SPD = P(\hat{Y}=1|A=a)-P(\hat{Y}=1|A=b) $$ where $\hat{Y}$ is the predicted outcome $A$ is the sensitive attribute Disparate Impact $$ DI=\frac{P(\hat{Y}=1|A=a)}{P(\hat{Y}=1|A=b)} $$ Values below $$ DI < 0.8 $$ may indicate potential discrimination under the EEOC 4/5ths rule. Equalized Odds We compare both $$ TPR $$ and $$ FPR $$ between protected groups. A large difference indicates unequal model performance across demographics. Backend FastAPI Python 3.12 Pandas NumPy Scikit-learn SHAP ReportLab Frontend HTML CSS JavaScript Chart.js AI Layer Gemini 2.5 Flash Workflow Upload dataset Parse CSV Train Random Forest model Compute fairness metrics Generate SHAP explanations Perform intersectionality analysis Generate AI explanations Produce PDF audit report The backend exposes REST APIs that power an interactive dashboard, while Gemini converts statistical outputs into audience-specific explanations that anyone can understand. One of the biggest challenges was implementing fairness metrics that worked reliably across different datasets while supporting both numerical and categorical labels. Building intersectionality analysis was especially challenging because many subgroup combinations contain very few samples. We introduced minimum sample thresholds to prevent misleading conclusions. Another challenge was balancing explainability and performance. SHAP can be computationally expensive, so we optimized our workflow to keep response times practical while still providing meaningful explanations. Finally, translating statistical fairness metrics into plain language without oversimplifying the results required careful prompt engineering and fallback mechanisms when the AI API was unavailable. We're proud that EquiLens-AI delivers much more than a fairness calculator. Highlights include: End-to-end AI fairness auditing Real intersectionality analysis SHAP explainability Audience-specific AI explanations Interactive What-if simulator Professional PDF audit reports Automatic dataset column detection Open-source implementation Clean, intuitive dashboard for non-technical users Most importantly, we built a platform that lowers the barrier to responsible AI by making fairness auditing accessible beyond data scientists. This project taught us that responsible AI is as much about communication as it is about algorithms. While implementing fairness metrics required a solid understanding of machine learning, the real challenge was presenting those results in a way that users could actually understand and act upon. We also gained valuable experience with: FastAPI backend architecture Explainable AI using SHAP Fairness evaluation methodologies REST API design Interactive data visualization Prompt engineering for AI-assisted explanations Building production-ready ML applications Our roadmap includes several major improvements: Additional fairness metrics Bias mitigation algorithms Real-time monitoring for deployed models Image, text, and multimodal model support Team collaboration workspaces Automated compliance reporting Cloud deployment with authentication Mobile-responsive interface Model versioning and audit history API integrations with popular ML workflows Our long-term vision is to make EquiLens-AI a comprehensive fairness platform that enables organizations to build trustworthy AI systems from development through deployment.
## README (from the GitHub repository)
# EquiLens AI
### AI-powered bias detection for non-technical users
[](https://solution-challenge-h2pw.onrender.com/)
[](LICENSE)
[](https://python.org)
[](https://fastapi.tiangolo.com)
> "Amazon's hiring AI downgraded women's CVs. COMPAS flagged Black defendants at 2× the rate. These failures could have been caught. EquiLens catches them."
---
## The Problem
AI makes life-changing decisions about jobs, loans, and healthcare. When trained on biased historical data, these systems don't just repeat discrimination — they amplify it at scale, silently, with no accountability.
## The Solution
EquiLens gives any organization — NGO, school, small business — the ability to audit their data for bias before it causes harm. No data science degree required.
---
## How It Works
1. Upload your CSV dataset
2. Select your target column and sensitive attribute
3. Optionally select a second sensitive attribute for intersectional analysis
4. EquiLens computes SPD, Disparate Impact, and Equalized Odds
5. SHAP explains which features are driving bias
6. Gemini translates everything into plain language
7. Intersectionality heatmap reveals compounded disadvantage across identity combinations
8. What-if simulator lets you drop features and measure bias impact in real time
9. Download a full PDF audit report
---
## Real User Story
Priya runs an NGO in Pune distributing scholarships. She uploads her dataset, selects gender as the sensitive attribute and caste as the intersect. She discovers that lower-caste girls are approved at **8%** — far below the 34% rate for upper-caste boys. A Disparate Impact of **0.24**, well below the legal threshold of 0.8. Gemini explains this in plain language and suggests fixes. Priya downloads the audit report and shares it with her board. **All in under 5 minutes.**
---
## Tech Stack
| Layer | Technology |
|-------|-----------|
| Backend | FastAPI (Python 3.12) |
| Bias Metrics | SPD, Disparate Impact, Equalized Odds |
| Explainability | SHAP (TreeExplainer) |
| AI Layer | Gemini 2.5 Flash |
| Frontend | HTML / CSS / JS + Chart.js |
| PDF Export | ReportLab |
| Deployment | Render |
---
## Fairness Metrics
| Metric | Threshold | Meaning |
|--------|-----------|---------|
| Disparate Impact | < 0.8 = biased | Legal standard (EEOC 4/5ths rule) |
| Statistical Parity Difference | > 0.1 = biased | Outcome gap between groups |
| Equalized Odds | > 0.1 = biased | Error rate gap between groups |
---
## What Makes EquiLens Different
Every existing tool — IBM AI Fairness 360, Fairlearn, Aequitas — outputs p-values and confusion matrices that only data scientists can interpret. EquiLens translates those results into plain language tuned to who's reading:
| Audience | Output |
|----------|--------|
| NGO worker | Policy implication |
| Student | Learning-oriented explanation |
| Policy maker | Legal risk framing |
---
## Setup
```bash
git clone https://github.com/CheerathAniketh/EquiLens-AI
cd EquiLens-AI/backend
pip install -r requirements.txt
```
Create a `.env` file in the `backend/` folder:
```env
GEMINI_API_KEY=your_api_key_here
```
Run the server:
```bash
uvicorn main:app --reload
```
Open `http://127.0.0.1:8000` in your browser.
---
## What's Built
### Backend
- FastAPI server with CORS middleware and multi-user session management
- CSV upload and parsing via `/analyze` endpoint
- Bias metrics computed locally: SPD, Disparate Impact, Equalized Odds (real TPR/FPR per group)
- SHAP feature importance via `explainer.py`
- Model training and evaluation via `trainer.py` (RandomForest, ROC + calibration curves)
- Gemini 2.5 Flash for plain-language explanations with audience toggle
- Graceful fallback when Gemini quota is exhausted or API key is missing — dynamic, not hardcoded
- Fallback responses marked with `*` so developers know Gemini is offline
- Smart label decoding: encoded columns (0/1/2...) mapped to human-readable names
- String target column support (`yes/no`, `hired/rejected`, `>50K/<=50K`)
- Real intersectionality via `compute_intersectionality()` — every (col1 × col2) subgroup pair
- Cells with fewer than 10 samples excluded and marked null to avoid misleading statistics
- Real Equalized Odds via `compute_eod()` — true TPR/FPR difference per group
- `/whatif` endpoint — retrains model on reduced feature set, measures bias delta
- `/whatif/features` endpoint — returns available features from cached session
- PDF audit report via ReportLab — verdict banner, metric scorecards, SHAP bars, Gemini explanation, regulation compliance table
### Frontend
- Single-page app with sidebar navigation and landing page
- Overview: score cards (DI, SPD, severity), approval rate chart, group comparison table
- Fairness metrics: metric bars, calibration curve, ROC by group — labeled with real group names
- Explainability: SHAP feature importance bars with proxy variable detection
- Intersectionality: real heatmap from backend + subgroup table ranked worst → best
- Remediation: before/after radar charts, recommended steps
- Audit report: structured findings + copy-to-clipboard + PDF download
- Demo presets: Hiring / Credit / Healthcare with one click
- Auto-detects target and sensitive columns from CSV headers
- Drag-and-drop CSV upload
- Audience toggle (NGO / Student / Policy maker)
- Regulation compliance pills (EEOC, EU AI Act, GDPR)
- What-if simulator: feature checkboxes, before/after radar, delta cards, Gemini explanation
---
## What's Pending
- Audience toggle re-fetches explanation without re-uploading CSV
- Intersectionality: sample size tooltip on sparse cells
- Mobile responsive layout
- Loading skeletons instead of spinner
- Inline error messages instead of `alert()` popups
- Environment variable management for production (`.env` → Cloud Secrets)
---
## License
MIT
## Detected evidence (automated analysis)
Indexed codebase: 10 recognized source files, 145 KB.
- FastAPI (technology) — detected in the code
- HTML (language) — detected in the code
- Python (language) — detected in the code
- LangChain (technology) — claimed on Devpost, not found in the code
## Codebase structure (from repository index)
### Files (19 of 19)
```
.gitignore
backend/analyzer.py
backend/explainer.py
backend/gemini_client.py
backend/main.py
backend/pdf_exporter.py
backend/requirements.txt
backend/schemas.py
backend/trainer.py
backend/utils.py
DockerFile
frontend/index.html
notebooks/adult_preprocessing.ipynb
notebooks/Unbiased_Data.ipynb
README.md
sample_data/adult_processed.csv
sample_data/adult.csv
sample_data/hiring_sample.csv
sample_data/true_unbiased.csv
```
### Dependencies
- backend/requirements.txt: fastapi, google-genai, joblib, numpy, pandas, pydantic, python-dotenv, python-multipart, reportlab, scikit-learn, shap, uvicorn
### Recent commits (newest first)
- Again verifying
- Verification
- removed unwatned ss
- Added images
- New Frontend
- fix: loading state, stale results clear, error banner, column validation
- Updated README.md
- Revert "test direct push"
- test direct push
- Conscise Output
- Added synth data
- Optimized Gemini Response
- Added Landing Page
- fix: frontend path for Docker
- Updated README.md
- Updated Intersectionality
- added PDF Exporter
- Working Whatif
- Added Whatif
- Updated README.md
## Key source files (fetched from GitHub, selected and truncated for size)
### DockerFile
```
FROM python:3.11-slim
WORKDIR /app
COPY backend/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY backend/ .
COPY frontend/ ./frontend/
COPY sample_data/ ./sample_data/
EXPOSE 8080
RUN useradd --create-home --shell /bin/bash appuser \
&& chown -R appuser:appuser /app
USER appuser
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]
```
### backend/requirements.txt
```
fastapi
uvicorn
pandas
scikit-learn
shap
google-genai
python-multipart
pydantic
python-dotenv
joblib
numpy
reportlab
```
### backend/main.py
```python
import os
import threading
import uuid
from fastapi import FastAPI, UploadFile, File, HTTPException, Header
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from typing import List, Optional
from fastapi.responses import Response
from pdf_exporter import generate_audit_pdf
from analyzer import analyze_bias, compute_intersectionality, compute_eod
from trainer import train_and_evaluate, prepare_features
from explainer import get_shap_values
from gemini_client import explain_results, suggest_fixes, explain_whatif
app = FastAPI(title="EquiLens AI")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*", "X-Session-Id"],
)
# ─── Multi-user session cache ──────────────────────────────────────────────────
# Each browser gets a unique session ID so concurrent judges don't clash.
session_cache: dict = {}
session_lock = threading.Lock()
MAX_SESSIONS = 100
def _evict_oldest():
if len(session_cache) >= MAX_SESSIONS:
oldest = next(iter(session_cache))
del session_cache[oldest]
# ─── Pydantic models ──────────────────────────────────────────────────────────
class WhatIfRequest(BaseModel):
drop_features: List[str]
audience: str = "ngo"
# ─── /analyze ─────────────────────────────────────────────────────────────────
@app.post("/analyze")
async def analyze(
file: UploadFile = File(...),
target_col: str = "",
sensitive_col: str = "",
sensitive_col_2: str = "",
audience: str = "ngo",
x_session_id: Optional[str] = Header(default=None),
):
try:
df = pd.read_csv(file.file)
except (pd.errors.ParserError, pd.errors.EmptyDataError, UnicodeDecodeError):
raise HTTPException(
status_code=400,
detail="Could not parse the uploaded file. Please upload a valid, non-empty CSV."
)
if target_col not in df.columns:
raise HTTPException(
status_code=400,
detail=f"Column '{target_col}' not found in the uploaded CSV."
)
if sensitive_col not in df.columns:
raise HTTPException(
status_code=400,
detail=f"Column '{sensitive_col}' not found in the uploaded CSV."
)
with session_lock:
if x_session_id and x_session_id in session_cache:
session_id = x_session_id
else:
session_id = str(uuid.uuid4())
while session_id in session_cache:
session_id = str(uuid.uuid4())
_evict_oldest()
session_cache[session_id] = {
"df": df,
"target_col": target_col,
"sensitive_col": sensitive_col,
}
stats = analyze_bias(df, target_col, sensitive_col)
model, X_train, X_test, y_test, curves, y_pred, y_prob, sensitive_test = train_and_evaluate(
df, target_col, sensitive_col
)
_, y_all = prepare_features(df, target_col, sensitive_col)
y_train = y_all.loc[X_train.index]
with session_lock:
if session_id in session_cache:
session_cache[session_id].update({
"X_train": X_train,
"X_test": X_test,
"y_train": y_train,
"y_test": y_test,
})
shap_data = get_shap_values(model, X_train, X_test)
eod_data = compute_eod(y_test, y_pred, sensitive_test)
stats["eod"] = eod_data["eod"]
stats["eod_details"] = eod_data
with session_lock:
if session_id in session_cache:
session_cache[session_id]["original_eod"] = eod_data["eod"]
intersectionality = None
if sensitive_col_2 and sensitive_col_2 != sensitive_col and sensitive_col_2 in df.columns:
intersectionality = compute_intersectionality(df, target_col, sensitive_col, sensitive_col_2)
try:
explanation = explain_results(stats, shap_data, audience=audience)
except Exception as e:
explanation = f"Gemini unavailable: {str(e)}"
try:
fixes = suggest_fixes(stats, shap_data)
except Exception as e:
fixes = [
"Rebalance your dataset so all groups have equal representation.",
"Remove proxy features that correlate with the sensitive attribute.",
"Collect more representative data from underrepresented groups.",
]
return {
"session_id": session_id,
"stats": stats,
"shap": shap_data,
"explanation": explanation,
"fixes": fixes,
"curves": curves,
"intersectionality": intersectionality,
}
# ─── /whatif/features ─────────────────────────────────────────────────────────
@app.get("/whatif/features")
async def whatif_features(x_session_id: Optional[str] = Header(default=None)):
with session_lock:
if not x_session_id or x_session_id not in session_cache:
raise HTTPException(
status_code=400,
detail="No dataset in cache. Run /analyze first."
)
session = session_cache[x_session_id]
df = session["df"]
target_col = session["target_col"]
sensitive_col = session["sensitive_col"]
features = [c for c in df.columns if c not in [target_col, sensitive_col]]
return {"features": features}
# ─── /whatif ──────────────────────────────────────────────────────────────────
@app.post("/whatif")
async def whatif(
body: WhatIfRequest,
x_session_id: Optional[str] = Header(default=None),
):
with session_lock:
if not x_session_id or x_session_id not in session_cache:
raise HTTPException(
status_code=400,
detail="No dataset in cache. Run /analyze first to upload a CSV."
)
session = session_cache[x_session_id]
df = session["df"]
target_col = session["target_col"]
sensitive_col = session["sensitive_col"]
[truncated — 6059 more characters]
```
### backend/schemas.py
```python
from pydantic import BaseModel
from typing import Dict, List, Optional
class GroupStat(BaseModel):
count: int
positive_rate: float
class AnalyzeRequest(BaseModel):
target_col: str
sensitive_col: str
audience: str = "ngo"
class BiasReport(BaseModel):
sensitive_col: str
target_col: str
group_stats: Dict[str, GroupStat]
spd: float
di: float
bias_detected: bool
severity: str
top_features: List[str]
explanation: str
fixes: List[str]
```
### backend/utils.py
```python
import pandas as pd
def parse_and_clean(file):
df = pd.read_csv(file)
# drop rows with missing target
# fill missing numerics with median
# fill missing categoricals with mode
for col in df.columns:
if df[col].dtype == 'object':
df[col].fillna(df[col].mode()[0], inplace=True)
else:
df[col].fillna(df[col].median(), inplace=True)
return df
def bin_continuous(df, col, bins=4):
# converts age/income into quartile groups
# so SPD/DI can be calculated on them
df[col] = pd.qcut(df[col], q=bins,
labels=['Q1','Q2','Q3','Q4'],
duplicates='drop')
return df
def validate_columns(df, target_col, sensitive_col):
errors = []
if target_col not in df.columns:
errors.append(f"{target_col} not found")
if sensitive_col not in df.columns:
errors.append(f"{sensitive_col} not found")
if df[target_col].nunique() > 10:
errors.append("Target column must be binary or categorical")
return errors
```
### backend/explainer.py
```python
import shap
import pandas as pd
import numpy as np
# Known proxy features that correlate with protected attributes
PROXY_KEYWORDS = [
"gap", "zip", "cost", "insurance", "prestige",
"address", "neighborhood", "redline", "parental"
]
def get_shap_values(model, X_train, X_test):
explainer = shap.TreeExplainer(model)
X_sample = X_test.iloc[:min(100, len(X_test))]
shap_values = explainer.shap_values(X_sample)
if isinstance(shap_values, list):
shap_values = shap_values[1]
elif shap_values.ndim == 3:
shap_values = shap_values[:, :, 1]
importance = (
pd.DataFrame({
"feature": X_sample.columns,
"importance": np.abs(shap_values).mean(axis=0)
})
.sort_values("importance", ascending=False)
.reset_index(drop=True)
)
top = importance.head(5).to_dict("records")
# detect proxy variables from ACTUAL features in this dataset
proxies = [
f["feature"] for f in top
if any(kw in f["feature"].lower() for kw in PROXY_KEYWORDS)
]
return {
"top_features": top,
"proxy_features": proxies # real proxies from actual CSV columns
}
```
### backend/trainer.py
```python
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
from sklearn.metrics import roc_curve, auc
from sklearn.calibration import calibration_curve
import pandas as pd
import numpy as np
def prepare_features(df, target_col, sensitive_col):
df_encoded = df.copy()
le = LabelEncoder()
for col in df_encoded.select_dtypes(include='object').columns:
if col != sensitive_col: # ← DON'T encode sensitive col here
df_encoded[col] = le.fit_transform(df_encoded[col].astype(str))
# encode target separately
if df_encoded[target_col].dtype == object:
df_encoded[target_col] = le.fit_transform(df_encoded[target_col].astype(str))
X = df_encoded.drop(columns=[target_col])
y = df_encoded[target_col]
return X, y
def safe_float(x):
"""Convert to float, replacing nan/inf with None."""
try:
v = float(x)
if np.isnan(v) or np.isinf(v):
return None
return round(v, 3)
except:
return None
def get_roc_data(y_true, y_prob, sensitive_values, sensitive_test):
"""Compute ROC curves per group."""
groups = sorted(sensitive_test.unique())
roc_data = {}
for group in groups:
mask = sensitive_test == group
if mask.sum() < 10:
continue
if y_true[mask].sum() == 0: # no positive samples in this group
continue
fpr, tpr, _ = roc_curve(y_true[mask], y_prob[mask])
roc_auc = round(auc(fpr, tpr), 3)
# downsample to 20 points for frontend
idx = np.linspace(0, len(fpr) - 1, min(20, len(fpr)), dtype=int)
roc_data[str(group)] = {
"fpr": [safe_float(x) for x in fpr[idx]],
"tpr": [safe_float(x) for x in tpr[idx]],
"auc": roc_auc
}
return roc_data
def get_calibration_data(y_true, y_prob, sensitive_values, sensitive_test):
"""Compute calibration curves per group."""
groups = sorted(sensitive_test.unique())
calib_data = {}
for group in groups:
mask = sensitive_test == group
if mask.sum() < 10:
continue
fraction_of_positives, mean_predicted = calibration_curve(
y_true[mask], y_prob[mask], n_bins=10, strategy='uniform'
)
calib_data[str(group)] = {
"mean_predicted": [safe_float(x) for x in mean_predicted],
"fraction_positive": [safe_float(x) for x in fraction_of_positives],
}
return calib_data
def train_and_evaluate(df, target_col, sensitive_col):
# Keep original sensitive values BEFORE encoding (for curve labels)
sensitive_original = df[sensitive_col].astype(str) if sensitive_col in df.columns else None
X, y = prepare_features(df, target_col, sensitive_col)
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42)
# Drop sensitive col from training features to avoid leakage
X_train_model = X_train.drop(columns=[sensitive_col], errors='ignore')
X_test_model = X_test.drop(columns=[sensitive_col], errors='ignore')
model = RandomForestClassifier(
n_estimators=100,
max_depth=8,
random_state=42,
n_jobs=-1
)
model.fit(X_train_model, y_train)
# Compute real ROC + calibration curves using original string group names
y_prob = model.predict_proba(X_test_model)[:, 1]
curves = {}
if sensitive_original is not None:
sensitive_test = sensitive_original.iloc[y_test.index].reset_index(drop=True)
y_test_reset = y_test.reset_index(drop=True)
curves["roc"] = get_roc_data(y_test_reset, y_prob, sensitive_test.unique(), sensitive_test)
curves["calibration"] = get_calibration_data(y_test_reset, y_prob, sensitive_test.unique(), sensitive_test)
y_pred = model.predict(X_test_model)
sensitive_test_series = sensitive_original.iloc[y_test.index].reset_index(drop=True) if sensitive_original is not None else None
return model, X_train_model, X_test_model, y_test, curves, y_pred, y_prob, sensitive_test_series
```
### backend/analyzer.py
```python
import pandas as pd
# Known label mappings for common encoded columns
LABEL_MAPS = {
"sex": {0: "Female", 1: "Male", "0": "Female", "1": "Male"},
"gender": {0: "Female", 1: "Male", "0": "Female", "1": "Male"},
"race": {
0: "Amer-Indian-Eskimo", 1: "Black", 2: "Asian-Pac-Islander",
3: "Other", 4: "White",
"0": "Amer-Indian-Eskimo", "1": "Black", "2": "Asian-Pac-Islander",
"3": "Other", "4": "White",
},
"income": {0: "<=50K", 1: ">50K", "0": "<=50K", "1": ">50K"},
}
POSITIVE_VALUES = {
'yes', 'true', '1', 'hired', 'approved', 'positive',
'>50k', '>50k.', 'accept', 'accepted', 'pass', 'grant', 'granted'
}
POSITIVE_LABELS = {
"yes", "hired", "approved", "true", "1", "accept",
"accepted", "grant", "granted", ">50k", ">50k."
}
def decode_group_label(sensitive_col, value):
col = sensitive_col.lower()
mapping = LABEL_MAPS.get(col)
if mapping and value in mapping:
return mapping[value]
return str(value)
def _resolve_target(series):
"""Convert any target column to a binary int Series (0/1)."""
if series.dtype == object or str(series.dtype) == 'string':
return series.astype(str).str.strip().str.lower().apply(
lambda x: 1 if x in POSITIVE_VALUES else 0
)
return pd.to_numeric(series, errors='coerce').fillna(0).astype(int)
def get_group_stats(df, target_col, sensitive_col):
target = _resolve_target(df[target_col])
stats = {}
for group in df[sensitive_col].unique():
mask = df[sensitive_col] == group
label = decode_group_label(sensitive_col, group)
group_target = target[mask]
if len(group_target) == 0:
continue
stats[label] = {
"count": int(mask.sum()),
"positive_rate": round(float(group_target.mean()), 4)
}
return stats
def calculate_spd(group_stats):
rates = [
v.get("positive_rate")
for v in group_stats.values()
if v.get("positive_rate") is not None and pd.notna(v.get("positive_rate"))
]
if not rates:
return 0.0
return round(max(rates) - min(rates), 4)
def calculate_di(group_stats):
rates = [
v.get("positive_rate")
for v in group_stats.values()
if v.get("positive_rate") is not None and pd.notna(v.get("positive_rate"))
]
if not rates:
return 0.0
max_rate = max(rates)
return round(min(rates) / max_rate, 4) if max_rate > 0 else 0.0
def get_severity(spd, di):
if di < 0.6 or spd > 0.3:
return "high"
elif di < 0.8 or spd > 0.1:
return "medium"
return "low"
def analyze_bias(df, target_col, sensitive_col):
df = df.copy()
# Encode string target to binary
if pd.api.types.is_string_dtype(df[target_col]) or df[target_col].dtype == object:
unique_vals = df[target_col].dropna().unique()
pos_label = next(
(v for v in unique_vals if str(v).strip().lower() in POSITIVE_LABELS),
unique_vals[0]
)
df[target_col] = (
df[target_col].astype(str).str.strip().str.lower()
== str(pos_label).strip().lower()
).astype(int)
group_stats = get_group_stats(df, target_col, sensitive_col)
spd = calculate_spd(group_stats)
di = calculate_di(group_stats)
clean_stats = {
str(group): {
"count": int(vals["count"]),
"positive_rate": float(vals["positive_rate"])
}
for group, vals in group_stats.items()
}
return {
"group_stats": clean_stats,
"spd": float(spd),
"di": float(di),
"bias_detected": bool(spd > 0.1 or di < 0.8),
"severity": get_severity(spd, di)
}
def compute_intersectionality(df, target_col, sensitive_col, sensitive_col_2):
"""
Compute approval rates for every (col1 × col2) subgroup combination.
Skips if either sensitive column has more than 10 unique values.
"""
if df[sensitive_col].nunique() > 10 or df[sensitive_col_2].nunique() > 10:
return None
df = df.copy()
df['__target__'] = _resolve_target(df[target_col])
col1_vals = sorted(df[sensitive_col].unique(), key=lambda x: str(x))
col2_vals = sorted(df[sensitive_col_2].unique(), key=lambda x: str(x))
col1_labels = [decode_group_label(sensitive_col, v) for v in col1_vals]
col2_labels = [decode_group_label(sensitive_col_2, v) for v in col2_vals]
groups = {}
matrix = {}
for i, v1 in enumerate(col1_vals):
l1 = col1_labels[i]
matrix[l1] = {}
for j, v2 in enumerate(col2_vals):
l2 = col2_labels[j]
mask = (df[sensitive_col] == v1) & (df[sensitive_col_2] == v2)
count = int(mask.sum())
if count < 10:
rate = None
else:
rate = round(float(df.loc[mask, '__target__'].mean()), 4)
key = f"{l1} × {l2}"
groups[key] = {"count": count, "positive_rate": rate}
matrix[l1][l2] = rate
valid = {k: v for k, v in groups.items() if v["positive_rate"] is not None}
return {
"col1": sensitive_col,
"col2": sensitive_col_2,
"col1_values": col1_labels,
"col2_values": col2_labels,
"groups": groups,
"matrix": matrix,
"valid_count": len(valid),
}
def compute_eod(y_test, y_pred, sensitive_test):
"""
True Equalized Odds Difference:
Max difference in TPR and FPR across all group pairs.
"""
y_test = pd.Series(y_test).reset_index(drop=True)
y_pred = pd.Series(y_pred).reset_index(drop=True)
sensitive_test = pd.Series(sensitive_test).reset_index(drop=True)
group_metrics = {}
for group in sensitive_test.unique():
mask = sensitive_test == group
if mask.sum() < 10:
continue
yt = y_test[mask]
yp = y_pred[mask]
tp = int(((yt == 1) & (yp == 1)).sum())
fn = i
[truncated — 1254 more characters]
```
### backend/gemini_client.py
```python
import json
import os
import re
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
from dotenv import load_dotenv
from google import genai
load_dotenv()
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
GEMINI_TIMEOUT_SECONDS = 15
GEMINI_EXECUTOR = ThreadPoolExecutor(max_workers=4)
PERSONAS = {
"student": "a student learning about AI fairness for the first time",
"ngo": "an NGO worker with no technical background who needs to act on this",
"policy": "a policy maker concerned about legal and ethical risk",
"general": "a general audience with no technical background",
}
def _is_api_error(e: Exception) -> bool:
err = str(e)
return any(code in err for code in [
"429", "RESOURCE_EXHAUSTED",
"400", "INVALID_ARGUMENT",
"API_KEY_INVALID", "API Key not found",
"GEMINI_TIMEOUT", "TIMEOUT", "timed out"
])
def _generate_content_with_timeout(prompt: str, timeout_seconds: int = GEMINI_TIMEOUT_SECONDS):
future = GEMINI_EXECUTOR.submit(
client.models.generate_content,
model="gemini-2.5-flash",
contents=prompt,
)
try:
return future.result(timeout=timeout_seconds)
except FuturesTimeoutError as exc:
future.cancel()
raise TimeoutError("GEMINI_TIMEOUT") from exc
def _fallback_explanation(bias_report) -> str:
di = bias_report.get("di", 0)
spd = bias_report.get("spd", 0)
severity = bias_report.get("severity", "unknown")
groups = bias_report.get("group_stats", {})
group_lines = ""
if groups:
sorted_groups = sorted(groups.items(), key=lambda x: x[1]["positive_rate"])
least = sorted_groups[0]
most = sorted_groups[-1]
group_lines = (
f"The most favored group is '{most[0]}' with an approval rate of "
f"{most[1]['positive_rate']*100:.1f}%, while '{least[0]}' has only "
f"{least[1]['positive_rate']*100:.1f}%."
)
bias_line = (
"This level of disparity means equally qualified people are being treated differently "
"based on a protected attribute. In a hiring context, this could mean qualified candidates "
"are being rejected due to gender, race, or other protected characteristics. "
if di < 0.8 else
"The model appears to be treating groups fairly based on these metrics. "
)
return (
f"This dataset shows {'significant' if di < 0.8 else 'no significant'} bias. "
f"The Disparate Impact Ratio is {di:.3f} "
f"({'below' if di < 0.8 else 'above'} the legal threshold of 0.80), "
f"and the outcome gap between groups is {spd*100:.1f} percentage points. "
f"Severity: {severity.upper()}. {group_lines}\n\n"
f"{bias_line}\n\n"
f"To address this: first, review and remove any proxy features that indirectly encode "
f"the sensitive attribute. Second, retrain the model on a rebalanced dataset where all "
f"groups have equal representation in positive outcomes. *"
)
def _fallback_fixes(bias_report, shap_data) -> list:
features = shap_data.get("top_features", [])
proxy_keywords = ["gap", "zip", "cost", "insurance", "prestige", "address"]
proxies = [f["feature"] for f in features if any(p in f["feature"].lower() for p in proxy_keywords)]
fixes = []
if proxies:
fixes.append(f"Remove or replace proxy features: {', '.join(proxies)} — these correlate with the sensitive attribute and cause indirect discrimination.")
fixes.append("Rebalance your training dataset so all demographic groups have equal representation in positive outcomes.")
fixes.append("Apply post-processing threshold calibration to equalize false positive and false negative rates across groups.")
return fixes[:3]
def _parse_json(text: str):
cleaned = re.sub(r"```(?:json)?", "", text).strip().strip("`").strip()
return json.loads(cleaned)
def explain_results(bias_report, shap_data, audience="ngo"):
persona = PERSONAS.get(audience, PERSONAS["ngo"])
prompt = f"""
You are an AI fairness expert. Be extremely concise.
Bias analysis results:
- Groups: {bias_report['group_stats']}
- Disparate Impact: {bias_report['di']} (below 0.8 = biased)
- Outcome gap: {bias_report['spd']} (above 0.1 = biased)
- Severity: {bias_report['severity']}
- Top features driving decisions: {shap_data['top_features']}
Write exactly 3 sentences. No more.
Sentence 1: Which group is disadvantaged and by how much (use the actual numbers).
Sentence 2: Which feature(s) are causing it and why that's a problem.
Sentence 3: One specific fix.
Rules:
- No intros, no conclusions, no filler
- Use plain language suited for {persona}
- Name the actual columns, not generic terms
- Total response must be under 60 words
"""
try:
response = _generate_content_with_timeout(prompt)
return response.text
except Exception as e:
if _is_api_error(e):
return _fallback_explanation(bias_report)
raise e
def suggest_fixes(bias_report, shap_data):
prompt = f"""
Given this bias analysis: {bias_report}
And these influential features: {shap_data['top_features']}
Suggest exactly 3 specific actionable fixes.
Return a JSON array only. No markdown, no backticks, no explanation.
Example format: ["fix 1", "fix 2", "fix 3"]
"""
try:
response = _generate_content_with_timeout(prompt)
try:
return _parse_json(response.text)
except (ValueError, json.JSONDecodeError):
return _fallback_fixes(bias_report, shap_data)
except Exception as e:
if _is_api_error(e):
return _fallback_fixes(bias_report, shap_data)
raise e
def explain_whatif(original, modified, delta, dropped_features, audience="ngo"):
"""
Explain what changed in bias metrics after dropping specific features.
Returns a plain-language summary of the before/after comparison.
[truncated — 3310 more characters]
```
### backend/pdf_exporter.py
```python
"""
pdf_exporter.py — EquiLens AI audit report PDF generator
Drop this file into backend/ alongside main.py
"""
import io
from datetime import datetime
from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import mm
from reportlab.platypus import (
SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle,
HRFlowable, PageBreak
)
from reportlab.lib.enums import TA_LEFT, TA_CENTER, TA_RIGHT
# ── Brand colours ────────────────────────────────────────────────────────────
TEAL = colors.HexColor("#0d9488")
TEAL_LIGHT = colors.HexColor("#ccfbf1")
RED = colors.HexColor("#ef4444")
AMBER = colors.HexColor("#f59e0b")
GREEN = colors.HexColor("#22c55e")
GRAY_DARK = colors.HexColor("#1e293b")
GRAY_MID = colors.HexColor("#64748b")
GRAY_LIGHT = colors.HexColor("#f1f5f9")
WHITE = colors.white
# ── Style helpers ─────────────────────────────────────────────────────────────
def _styles():
base = getSampleStyleSheet()
custom = {}
custom["Title"] = ParagraphStyle(
"ELTitle", parent=base["Normal"],
fontSize=24, textColor=WHITE, leading=30,
fontName="Helvetica-Bold", alignment=TA_LEFT,
)
custom["Subtitle"] = ParagraphStyle(
"ELSubtitle", parent=base["Normal"],
fontSize=10, textColor=colors.HexColor("#b2f5ea"), leading=14,
fontName="Helvetica", alignment=TA_LEFT,
)
custom["H2"] = ParagraphStyle(
"ELH2", parent=base["Normal"],
fontSize=13, textColor=GRAY_DARK, leading=18,
fontName="Helvetica-Bold", spaceAfter=4,
)
custom["Body"] = ParagraphStyle(
"ELBody", parent=base["Normal"],
fontSize=9, textColor=GRAY_DARK, leading=14,
fontName="Helvetica",
)
custom["BodyGray"] = ParagraphStyle(
"ELBodyGray", parent=base["Normal"],
fontSize=8, textColor=GRAY_MID, leading=12,
fontName="Helvetica",
)
custom["Mono"] = ParagraphStyle(
"ELMono", parent=base["Normal"],
fontSize=8, textColor=GRAY_DARK, leading=12,
fontName="Courier",
)
custom["Footer"] = ParagraphStyle(
"ELFooter", parent=base["Normal"],
fontSize=7, textColor=GRAY_MID, leading=10,
fontName="Helvetica", alignment=TA_CENTER,
)
return custom
def _severity_color(severity: str):
s = (severity or "").lower()
if s == "high": return RED
if s == "medium": return AMBER
return GREEN
def _pill_table(label: str, value: str, color):
"""A small coloured badge rendered as a 1-row Table."""
t = Table([[Paragraph(f"{label}", ParagraphStyle(
"pill", fontSize=8, textColor=WHITE, fontName="Helvetica-Bold",
leading=10, alignment=TA_CENTER,
)), Paragraph(value, ParagraphStyle(
"pillv", fontSize=8, textColor=WHITE, fontName="Helvetica",
leading=10, alignment=TA_CENTER,
))]], colWidths=[28*mm, 22*mm])
t.setStyle(TableStyle([
("BACKGROUND", (0, 0), (0, 0), color),
("BACKGROUND", (1, 0), (1, 0), color),
("ROUNDEDCORNERS", [3]),
("ALIGN", (0, 0), (-1, -1), "CENTER"),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("TOPPADDING", (0, 0), (-1, -1), 3),
("BOTTOMPADDING", (0, 0), (-1, -1), 3),
]))
return t
# ── Page template (header/footer on every page) ───────────────────────────────
def _on_page(canvas, doc):
canvas.saveState()
w, h = A4
# Top teal stripe
canvas.setFillColor(TEAL)
canvas.rect(0, h - 12*mm, w, 12*mm, fill=1, stroke=0)
canvas.setFillColor(WHITE)
canvas.setFont("Helvetica-Bold", 8)
canvas.drawString(15*mm, h - 8*mm, "EquiLens AI | Bias Audit Report")
canvas.setFont("Helvetica", 7)
canvas.drawRightString(w - 15*mm, h - 8*mm,
datetime.now().strftime("%d %b %Y"))
# Bottom footer line
canvas.setStrokeColor(GRAY_LIGHT)
canvas.setLineWidth(0.5)
canvas.line(15*mm, 12*mm, w - 15*mm, 12*mm)
canvas.setFillColor(GRAY_MID)
canvas.setFont("Helvetica", 7)
canvas.drawCentredString(w / 2, 8*mm,
f"Page {doc.page} — Generated by EquiLens AI · SDG 10 · GDG Solution Challenge 2026")
canvas.restoreState()
# ── Public API ────────────────────────────────────────────────────────────────
def generate_audit_pdf(report_data: dict) -> bytes:
"""
Build the audit PDF and return raw bytes.
Expected keys in report_data:
dataset, target_col, sensitive_col, audience,
group_stats : {group_name: {count, positive_rate}}
spd, di, eod : float
severity : str ("High" | "Medium" | "Low")
bias_detected : bool
shap_features : [{feature, importance}] (optional)
explanation : str (optional, Gemini text)
intersectionality: {…} (optional)
whatif : {original, simulated, dropped_features} (optional)
remediation_steps: [str] (optional)
"""
buf = io.BytesIO()
report_data = report_data if isinstance(report_data, dict) else {}
def _to_float(value, default=0.0):
try:
return float(value)
except (TypeError, ValueError):
return default
doc = SimpleDocTemplate(
buf, pagesize=A4,
leftMargin=15*mm, rightMargin=15*mm,
topMargin=20*mm, bottomMargin=20*mm,
)
S = _styles()
story = []
# ── COVER BLOCK ──────────────────────────────────────────────────────────
cover_bg = Table(
[[Paragraph("EquiLens AI", S["Title"])],
[Paragraph("Algorithmic Bias Audit Report", S["Subtitle"])],
[Spacer(1, 4*mm)],
[Paragraph(
f"Dataset: {report_data.get('dataset', 'N/A')} | "
f"Target: {report_data.get('target_col', 'N/A')} | "
f"Sensitive: {r
[truncated — 14844 more characters]
```