The Mid-Series Capstone: The 'Production-Ready' Check

/capstone
    ├── release.py        # The Orchestrator (The Boss)
    ├── data.csv          # The Raw Data
    ├── model_card_template.md

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, precision_score
from dataclasses import dataclass
import json
import os

# --- 1. CONFIGURATION & CONTRACTS ---
@dataclass
class ReleaseConfig:
    min_accuracy: float = 0.80
    max_latency_ms: float = 50.0
    fairness_threshold: float = 0.9  # Female/Male acceptance ratio
    model_version: str = "1.0.0"

config = ReleaseConfig()

# --- 2. DATA LAYER (with Contracts) ---
def load_and_validate_data():
    print("Step 1: Loading Data...")
    # Simulating data loading
    df = pd.DataFrame({
        'income': np.random.randint(20000, 100000, 1000),
        'credit_score': np.random.randint(300, 850, 1000),
        'gender': np.random.choice(['M', 'F'], 1000),
        'loan_approved': np.random.randint(0, 2, 1000)
    })

    # DATA CONTRACT (Day 45)
    if df.isnull().any().any():
        raise ValueError("⛔ CONTRACT FAIL: Null values detected.")
    if (df['income'] < 0).any():
        raise ValueError("⛔ CONTRACT FAIL: Negative income detected.")

    print("✅ Data Contract Passed.")
    return df

# --- 3. TRAINING LAYER ---
def train_model(df):
    print("Step 2: Training Model...")
    X = df[['income', 'credit_score']]
    y = df['loan_approved']

    # Bias mitigation: remove sensitive attribute 'gender' from features
    X_train, X_test, y_train, y_test, gender_train, gender_test = train_test_split(
        X, y, df['gender'], test_size=0.2, random_state=42
    )

    clf = RandomForestClassifier(n_estimators=10, max_depth=5)
    clf.fit(X_train, y_train)

    return clf, X_test, y_test, gender_test

# --- 4. EVALUATION LAYER (Accuracy + Fairness) ---
def evaluate_model(clf, X_test, y_test, gender_test):
    print("Step 3: Evaluating Performance & Fairness...")
    y_pred = clf.predict(X_test)

    # Metrics
    acc = accuracy_score(y_test, y_pred)

    # Fairness Audit (Day 49)
    test_df = X_test.copy()
    test_df['gender'] = gender_test
    test_df['pred'] = y_pred

    acceptance_rate_m = test_df[test_df['gender'] == 'M']['pred'].mean()
    acceptance_rate_f = test_df[test_df['gender'] == 'F']['pred'].mean()

    # Avoid division by zero
    disparate_impact = acceptance_rate_f / acceptance_rate_m if acceptance_rate_m > 0 else 0

    metrics = {
        "accuracy": acc,
        "disparate_impact": disparate_impact
    }

    # THE GATE (Day 50 Logic)
    if acc < config.min_accuracy:
        raise RuntimeError(f"⛔ REJECT: Accuracy {acc:.2f} < {config.min_accuracy}")

    if disparate_impact < config.fairness_threshold:
        # In a real scenario, this might be a warning or hard block depending on policy
        print(f"⚠️ WARNING: Disparate Impact {disparate_impact:.2f} is below threshold.")

    print(f"✅ Evaluation Passed. Acc: {acc:.2f}, Fairness: {disparate_impact:.2f}")
    return metrics

# --- 5. DOCUMENTATION LAYER (Model Cards) ---
def generate_model_card(metrics):
    print("Step 4: Generating Model Card...")
    card_content = f"""
# Model Card: Loan Approver v{config.model_version}
## Performance
- Accuracy: {metrics['accuracy']:.2f}
- Fairness (F/M Ratio): {metrics['disparate_impact']:.2f}

## Usage
- Input: Income, Credit Score
- Intended Use: Automated screening for unsecured personal loans.
- Limitations: Not validated for incomes > $200k.

## Governance
- Owner: Responsible AI Team
- Status: DEPLOYED via Release Pipeline
    """
    with open("MODEL_CARD.md", "w") as f:
        f.write(card_content)
    print("✅ Model Card Generated: MODEL_CARD.md")

# --- 6. DEPLOYMENT LAYER ---
def deploy_system(clf):
    print("Step 5: Simulating Deployment...")
    # In reality, this would 'docker build' and 'kubectl apply'
    # Here, we simulate saving the artifact
    import pickle
    with open("model_v1.pkl", "wb") as f:
        pickle.dump(clf, f)
    print("🚀 SYSTEM DEPLOYED. Endpoint active at http://api.internal/loan/v1")

# --- 7. LLM EXPLANATION LAYER (Day 48 Integration) ---
def generate_explanation(decision: str, input_data: dict, model_factors: dict) -> str:
    """
    Generate human-readable explanations using a LOCAL LLM.
    This synthesizes Day 48 (Local LLMs) into the production pipeline.

    In production: Use Ollama running on localhost (air-gapped for PII safety)
    """
    import requests

    OLLAMA_URL = "http://localhost:11434/api/generate"

    prompt = f"""You are a loan decision explainer. Generate a brief, professional
explanation for this lending decision.

DECISION: {decision}
APPLICANT DATA: Income=${input_data.get('income', 'N/A')}, Credit Score={input_data.get('credit_score', 'N/A')}
KEY FACTORS: {model_factors}

Requirements:
1. Be factual and cite the specific data points
2. Never mention protected attributes (gender, race, etc.)
3. Use plain language suitable for the applicant
4. Keep the explanation under 100 words

Generate the explanation:"""

    try:
        response = requests.post(OLLAMA_URL, json={
            "model": "llama3",
            "prompt": prompt,
            "stream": False,
            "format": "json",  # Structured output (Day 48 enhancement)
            "options": {"temperature": 0.3}
        }, timeout=30)

        if response.status_code == 200:
            result = response.json()
            return result.get('response', 'Explanation generation unavailable.')
        else:
            return f"[Fallback] Decision: {decision}. Please contact support for details."

    except requests.exceptions.ConnectionError:
        # Graceful degradation if Ollama is not running
        print("⚠️ Local LLM not available. Using template explanation.")
        return f"Your application was {decision.lower()} based on income and credit score analysis."

def predict_with_explanation(clf, input_data: dict) -> dict:
    """
    Full inference pipeline: Predict + Explain (Day 41-50 Synthesis)
    """
    import numpy as np

    X = np.array([[input_data['income'], input_data['credit_score']]])
    prediction = clf.predict(X)[0]
    proba = clf.predict_proba(X)[0]

    decision = "APPROVED" if prediction == 1 else "DECLINED"
    confidence = max(proba)

    # Feature importance for explanation
    model_factors = {
        "income_weight": f"{clf.feature_importances_[0]:.2%}",
        "credit_weight": f"{clf.feature_importances_[1]:.2%}",
        "confidence": f"{confidence:.2%}"
    }

    # Generate local LLM explanation (NEVER sends PII externally)
    explanation = generate_explanation(decision, input_data, model_factors)

    return {
        "decision": decision,
        "confidence": confidence,
        "explanation": explanation,
        "audit_log": {
            "model_version": config.model_version,
            "factors": model_factors
        }
    }

# --- MAIN ORCHESTRATOR ---
if __name__ == "__main__":
    try:
        print("🟢 STARTING RELEASE PIPELINE")
        df = load_and_validate_data()
        model, X_test, y_test, gender_test = train_model(df)
        metrics = evaluate_model(model, X_test, y_test, gender_test)
        generate_model_card(metrics)
        deploy_system(model)

        # --- NEW: Demo the full inference + explanation flow ---
        print("\n--- DEMO: Full Inference with Explanation ---")
        sample_applicant = {"income": 55000, "credit_score": 720}
        result = predict_with_explanation(model, sample_applicant)

        print(f"Decision: {result['decision']}")
        print(f"Confidence: {result['confidence']:.2%}")
        print(f"Explanation: {result['explanation']}")

        print("\n✅ PIPELINE SUCCESS")
    except Exception as e:
        print(f"\n🔴 PIPELINE FAILED: {e}")
        exit(1)