Human-Centricity & Empathy

# ─── 1. Confidence-Gated HITL Decision System ────────────────
from dataclasses import dataclass, field
from typing import Optional, List
from datetime import datetime
from enum import Enum
import numpy as np

class DecisionType(Enum):
    AUTO_APPROVED = "auto_approved"
    AUTO_REJECTED = "auto_rejected"
    HUMAN_REVIEW = "human_review"
    HUMAN_REQUIRED = "human_required"

@dataclass
class Decision:
    item_id: str
    decision_type: DecisionType
    ai_prediction: Optional[int]
    ai_confidence: float
    human_decision: Optional[int] = None
    human_reviewer: Optional[str] = None
    review_timestamp: Optional[str] = None
    explanation: Optional[str] = None
    appeals: List[str] = field(default_factory=list)

    def human_overrode_ai(self) -> bool:
        return (self.human_decision is not None and
                self.ai_prediction is not None and
                self.human_decision != self.ai_prediction)

class HITLDecisionSystem:
    def __init__(self,
                 auto_approve_threshold: float = 0.92,
                 auto_reject_threshold: float = 0.92,
                 review_threshold: float = 0.70):
        self.auto_approve_threshold = auto_approve_threshold
        self.auto_reject_threshold = auto_reject_threshold
        self.review_threshold = review_threshold
        self.decisions: List[Decision] = []
        self.review_queue: List[Decision] = []

    def process(self, item_id: str, ai_pred: int, ai_confidence: float,
                explanation: str = "") -> Decision:
        if ai_pred == 1 and ai_confidence >= self.auto_approve_threshold:
            d = Decision(item_id=item_id, decision_type=DecisionType.AUTO_APPROVED,
                         ai_prediction=1, ai_confidence=ai_confidence,
                         explanation=explanation)
        elif ai_pred == 0 and ai_confidence >= self.auto_reject_threshold:
            d = Decision(item_id=item_id, decision_type=DecisionType.AUTO_REJECTED,
                         ai_prediction=0, ai_confidence=ai_confidence,
                         explanation=explanation)
        elif ai_confidence >= self.review_threshold:
            d = Decision(item_id=item_id, decision_type=DecisionType.HUMAN_REVIEW,
                         ai_prediction=ai_pred, ai_confidence=ai_confidence,
                         explanation=explanation)
            self.review_queue.append(d)
        else:
            d = Decision(item_id=item_id, decision_type=DecisionType.HUMAN_REQUIRED,
                         ai_prediction=None, ai_confidence=ai_confidence,
                         explanation="Low confidence — AI recommendation withheld to prevent anchoring bias")
            self.review_queue.append(d)

        self.decisions.append(d)
        return d

    def human_review(self, item_id: str, reviewer: str, decision: int, notes: str = ""):
        for d in self.review_queue:
            if d.item_id == item_id:
                d.human_decision = decision
                d.human_reviewer = reviewer
                d.review_timestamp = datetime.utcnow().isoformat() + "Z"
                self.review_queue.remove(d)
                return d
        raise ValueError(f"Item {item_id} not found in review queue")

    def governance_report(self):
        total = len(self.decisions)
        auto = sum(1 for d in self.decisions
                   if d.decision_type in (DecisionType.AUTO_APPROVED, DecisionType.AUTO_REJECTED))
        reviewed = sum(1 for d in self.decisions if d.human_decision is not None)
        overrides = sum(1 for d in self.decisions if d.human_overrode_ai())

        print(f"\nHITL Governance Report")
        print(f"{'─'*40}")
        print(f"  Total decisions:       {total}")
        print(f"  Automated decisions:   {auto} ({100*auto//max(total,1)}%)")
        print(f"  Queued for review:     {len(self.review_queue)}")
        print(f"  Reviewed by humans:    {reviewed}")
        print(f"  Human override rate:   {100*overrides//max(reviewed,1)}%")
        if overrides > 0.2 * reviewed:
            print(f"  ⚠ Override rate > 20% — model retraining recommended")

# Simulate loan application processing
system = HITLDecisionSystem(auto_approve_threshold=0.90, auto_reject_threshold=0.90)

applications = [
    ("APP-001", 1, 0.97, "Strong income, clean credit history, low DTI"),
    ("APP-002", 0, 0.95, "Multiple missed payments, high debt-to-income ratio"),
    ("APP-003", 1, 0.82, "Good income but short credit history"),
    ("APP-004", 0, 0.61, "Mixed signals: good income but recent job change"),
    ("APP-005", 1, 0.55, "New-to-credit, insufficient signals for automated decision"),
    ("APP-006", 0, 0.91, "Significant derogatory marks on credit report"),
]

print("Processing loan applications:")
for app_id, pred, conf, explanation in applications:
    d = system.process(app_id, pred, conf, explanation)
    status = d.decision_type.value.replace("_", " ").upper()
    print(f"  {app_id}: [{status}] conf={conf:.2f} {'→ AI: ' + str(pred) if pred is not None else '→ needs human'}")

# Simulate human reviews
system.human_review("APP-003", "jane.smith@bank.com", 1, "Approved — first-time buyer with strong employment")
system.human_review("APP-004", "jane.smith@bank.com", 0, "Rejected — job change within 3 months is a disqualifier per policy")
system.human_review("APP-005", "john.doe@bank.com", 1, "Approved — student loan will be paid off next month, reviewed bank statements")

system.governance_report()

# ─── 2. Active Learning Loop ──────────────────────────────────
from sklearn.datasets import make_classification
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
import numpy as np

X_all, y_all = make_classification(n_samples=3000, n_features=20, random_state=42)
X_labeled, X_pool = X_all[:100], X_all[100:]
y_labeled, y_pool = y_all[:100], y_all[100:]
X_test, y_test = X_all[2500:], y_all[2500:]

print("\nActive Learning vs Random Sampling:")
print(f"{'Round':<8} {'Labeled':<10} {'AL Accuracy':<15} {'Random Accuracy'}")
print("─" * 50)

al_X_labeled, al_y_labeled = X_labeled.copy(), y_labeled.copy()
rand_X_labeled, rand_y_labeled = X_labeled.copy(), y_labeled.copy()
remaining_pool_mask = np.ones(len(X_pool), dtype=bool)

for round_num in range(5):
    # Active Learning: query most uncertain samples
    al_model = RandomForestClassifier(n_estimators=50, random_state=42)
    al_model.fit(al_X_labeled, al_y_labeled)

    probs = al_model.predict_proba(X_pool[remaining_pool_mask])
    uncertainty = 1 - probs.max(axis=1)
    top_k_idx = np.argsort(uncertainty)[-20:]  # 20 most uncertain

    pool_indices = np.where(remaining_pool_mask)[0]
    query_indices = pool_indices[top_k_idx]
    al_X_labeled = np.vstack([al_X_labeled, X_pool[query_indices]])
    al_y_labeled = np.concatenate([al_y_labeled, y_pool[query_indices]])
    remaining_pool_mask[query_indices] = False

    al_acc = accuracy_score(y_test, al_model.predict(X_test))

    # Random sampling baseline
    rand_model = RandomForestClassifier(n_estimators=50, random_state=42)
    rand_model.fit(rand_X_labeled, rand_y_labeled)
    rand_idx = np.random.choice(len(X_pool), 20, replace=False)
    rand_X_labeled = np.vstack([rand_X_labeled, X_pool[rand_idx]])
    rand_y_labeled = np.concatenate([rand_y_labeled, y_pool[rand_idx]])
    rand_acc = accuracy_score(y_test, rand_model.predict(X_test))

    print(f"  {round_num+1:<6}  {len(al_X_labeled):<10} {al_acc:.4f}{'*' if al_acc>rand_acc else ' ':2}        {rand_acc:.4f}")

print("  (* AL beats random sampling with same number of labels)")

# ─── 3. Contestability API ───────────────────────────────────
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="AI Decision Contestability API")

class AppealRequest(BaseModel):
    decision_id: str
    applicant_id: str
    reason: str
    additional_evidence: str = ""

class AppealResponse(BaseModel):
    appeal_id: str
    status: str
    assigned_reviewer: str
    expected_response_days: int
    explanation_provided: str
    next_steps: str

@app.post("/decisions/{decision_id}/appeal", response_model=AppealResponse)
async def submit_appeal(decision_id: str, request: AppealRequest):
    """GDPR Art.22 contestability endpoint — required for all significant automated decisions."""
    return AppealResponse(
        appeal_id=f"APL-{decision_id}-001",
        status="RECEIVED",
        assigned_reviewer="appeals.team@bank.com",
        expected_response_days=10,
        explanation_provided="The automated decision considered: credit score (primary factor), debt-to-income ratio, and payment history. SHAP analysis available at /decisions/{decision_id}/explanation.",
        next_steps="A human reviewer will assess your appeal within 10 business days. You may submit additional supporting documents to the address above.",
    )

@app.get("/decisions/{decision_id}/explanation")
async def get_explanation(decision_id: str):
    """Return SHAP-based explanation for the decision."""
    return {
        "decision_id": decision_id,
        "outcome": "rejected",
        "top_factors": [
            {"factor": "payment_history", "direction": "negative", "impact": "high",
             "detail": "2 missed payments in the past 24 months reduced approval likelihood"},
            {"factor": "debt_to_income_ratio", "direction": "negative", "impact": "medium",
             "detail": "DTI of 47% exceeds preferred threshold of 43%"},
            {"factor": "credit_score", "direction": "neutral", "impact": "low",
             "detail": "Credit score of 680 is within acceptable range"},
        ],
        "what_would_change_decision": [
            "Reducing monthly debt obligations by €300 would bring DTI below threshold",
            "12 months of clean payment history would significantly improve outcome",
        ],
        "human_review_available": True,
        "appeal_url": f"/decisions/{decision_id}/appeal",
    }