AI Ethics, Responsible AI, and Career Roadmap

import shap
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split

# Create dataset with named features
np.random.seed(42)
X, y = make_classification(n_samples=500, n_features=6, n_informative=4, random_state=42)
feature_names = ["Income", "Age", "Credit_Score", "Debt_Ratio", "Employment_Years", "Num_Accounts"]

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Train model
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
print(f"Model accuracy: {model.score(X_test, y_test):.4f}")

# SHAP analysis
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_test)

# Global feature importance (mean absolute SHAP values)
print("\n=== Global Feature Importance (SHAP) ===")
mean_shap = np.abs(shap_values[1]).mean(axis=0)  # Class 1 (approved)
for name, importance in sorted(zip(feature_names, mean_shap), key=lambda x: -x[1]):
    bar = "#" * int(importance * 50)
    print(f"  {name:18s}: {importance:.4f} {bar}")

# Explain a single prediction
print("\n=== Single Prediction Explanation ===")
sample_idx = 0
prediction = model.predict(X_test[sample_idx:sample_idx+1])[0]
print(f"Prediction: {'Approved' if prediction == 1 else 'Rejected'}")
print(f"\nFeature contributions:")
for name, value, shap_val in zip(feature_names, X_test[sample_idx], shap_values[1][sample_idx]):
    direction = "+" if shap_val > 0 else "-"
    print(f"  {name:18s}: value={value:+.3f}, SHAP={direction}{abs(shap_val):.4f}")

from lime.lime_text import LimeTextExplainer
from sklearn.pipeline import Pipeline
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
import numpy as np

# Train a simple text classifier
texts = [
    "The product quality is excellent and delivery was fast",
    "Amazing service and great customer support",
    "The item arrived quickly and works perfectly",
    "Terrible quality and the package was damaged",
    "Worst experience ever, customer service was rude",
    "Product broke after one day, complete waste of money",
    "Love this product, highly recommend it",
    "Do not buy this, it is a scam"
]
labels = [1, 1, 1, 0, 0, 0, 1, 0]  # 1=positive, 0=negative

pipeline = Pipeline([
    ("tfidf", TfidfVectorizer()),
    ("clf", LogisticRegression())
])
pipeline.fit(texts, labels)

# LIME explanation
explainer = LimeTextExplainer(class_names=["Negative", "Positive"])

test_text = "The product quality is good but delivery was slow"
exp = explainer.explain_instance(
    test_text,
    pipeline.predict_proba,
    num_features=6
)

prediction = pipeline.predict([test_text])[0]
proba = pipeline.predict_proba([test_text])[0]

print(f"Text: '{test_text}'")
print(f"Prediction: {'Positive' if prediction == 1 else 'Negative'}")
print(f"Confidence: Negative={proba[0]:.3f}, Positive={proba[1]:.3f}")
print(f"\nLIME Explanation (top features):")
for feature, weight in exp.as_list():
    direction = "Positive" if weight > 0 else "Negative"
    print(f"  '{feature}': {weight:+.4f} (pushes toward {direction})")

import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from collections import Counter

np.random.seed(42)

# Simulate a loan approval dataset with gender bias
n = 2000
gender = np.random.choice([0, 1], n)  # 0=Female, 1=Male
income = np.random.normal(50000, 15000, n) + gender * 5000  # Males earn more (bias)
credit_score = np.random.normal(700, 50, n)
debt_ratio = np.random.uniform(0.1, 0.8, n)

# Biased approval: males get approved more often even with same qualifications
noise = np.random.normal(0, 0.5, n)
approval = (0.3 * (income / 50000) + 0.4 * (credit_score / 700) - 0.3 * debt_ratio + 0.1 * gender + noise > 0.5).astype(int)

X = np.column_stack([income, credit_score, debt_ratio])
y = approval

# Train model (WITHOUT gender as feature)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
gender_test = gender[len(X_train):]  # Gender for test set

model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

# Check for disparate impact
predictions = model.predict(X_test)

male_approved = predictions[gender_test == 1].mean()
female_approved = predictions[gender_test == 0].mean()

print("=== Bias Analysis ===")
print(f"Male approval rate:   {male_approved:.3f} ({male_approved*100:.1f}%)")
print(f"Female approval rate: {female_approved:.3f} ({female_approved*100:.1f}%)")
print(f"Disparate impact ratio: {female_approved/male_approved:.3f}")
print(f"80% rule satisfied? {female_approved/male_approved >= 0.8}")
print()

# Even though gender is NOT a model feature, income correlates with gender
# so the model learns gender-biased patterns indirectly (proxy discrimination)
print("=== Root Cause ===")
print(f"Avg income (Male):   ${income[gender==1].mean():,.0f}")
print(f"Avg income (Female): ${income[gender==0].mean():,.0f}")
print(f"Income is a proxy for gender -> model has proxy bias")
print()
print("=== Mitigation Strategies ===")
print("1. Reweight training samples to balance outcomes across groups")
print("2. Use fairness-aware algorithms (adversarial debiasing)")
print("3. Post-processing: adjust thresholds per group for equal opportunity")
print("4. Remove or decorrelate proxy features")

import numpy as np
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

class LogisticRegressionScratch:
    """Logistic Regression implemented from scratch -- common ML interview question."""
    
    def __init__(self, learning_rate=0.01, n_iterations=1000):
        self.lr = learning_rate
        self.n_iter = n_iterations
        self.weights = None
        self.bias = None
        self.losses = []
    
    def sigmoid(self, z):
        return 1 / (1 + np.exp(-np.clip(z, -500, 500)))
    
    def fit(self, X, y):
        n_samples, n_features = X.shape
        self.weights = np.zeros(n_features)
        self.bias = 0
        
        for i in range(self.n_iter):
            # Forward pass
            z = X @ self.weights + self.bias
            predictions = self.sigmoid(z)
            
            # Compute loss (binary cross-entropy)
            loss = -np.mean(y * np.log(predictions + 1e-8) + 
                           (1 - y) * np.log(1 - predictions + 1e-8))
            self.losses.append(loss)
            
            # Compute gradients
            dw = (1 / n_samples) * X.T @ (predictions - y)
            db = (1 / n_samples) * np.sum(predictions - y)
            
            # Update parameters
            self.weights -= self.lr * dw
            self.bias -= self.lr * db
        
        return self
    
    def predict_proba(self, X):
        z = X @ self.weights + self.bias
        return self.sigmoid(z)
    
    def predict(self, X, threshold=0.5):
        return (self.predict_proba(X) >= threshold).astype(int)

# Test it
X, y = make_classification(n_samples=1000, n_features=10, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Our implementation
model = LogisticRegressionScratch(learning_rate=0.1, n_iterations=500)
model.fit(X_train, y_train)
our_acc = accuracy_score(y_test, model.predict(X_test))

# Compare with sklearn
from sklearn.linear_model import LogisticRegression
sklearn_model = LogisticRegression(max_iter=500)
sklearn_model.fit(X_train, y_train)
sklearn_acc = accuracy_score(y_test, sklearn_model.predict(X_test))

print("=== Logistic Regression from Scratch ===")
print(f"Our accuracy:     {our_acc:.4f}")
print(f"Sklearn accuracy: {sklearn_acc:.4f}")
print(f"Loss (start):     {model.losses[0]:.4f}")
print(f"Loss (end):       {model.losses[-1]:.4f}")
print(f"Convergence:      {model.losses[0] - model.losses[-1]:.4f} reduction")

class RecommendationSystemDesign:
    """ML System Design interview practice -- Recommendation System."""
    
    def __init__(self):
        self.components = {
            "1_data": {
                "title": "Data Sources",
                "items": [
                    "User profiles (age, location, preferences)",
                    "Item catalog (category, price, description)",
                    "Interaction logs (views, clicks, purchases, ratings)",
                    "Contextual data (time of day, device, session length)"
                ]
            },
            "2_features": {
                "title": "Feature Engineering",
                "items": [
                    "User features: purchase history embedding, avg rating, activity frequency",
                    "Item features: category embedding, popularity score, recency",
                    "Interaction features: user-item affinity, co-purchase patterns",
                    "Contextual: time decay, seasonal trends"
                ]
            },
            "3_model": {
                "title": "Model Architecture",
                "items": [
                    "Candidate generation: ANN (approximate nearest neighbor) on embeddings",
                    "Ranking model: Deep learning (Wide & Deep or Two-Tower)",
                    "Diversity: MMR (Maximal Marginal Relevance) reranking",
                    "Cold start: content-based fallback for new users/items"
                ]
            },
            "4_serving": {
                "title": "Serving Infrastructure",
                "items": [
                    "Candidate generation: precomputed, refreshed hourly",
                    "Ranking: real-time inference (< 50ms latency)",
                    "Caching: top-K recommendations cached per user",
                    "A/B testing: split traffic to compare model versions"
                ]
            },
            "5_monitoring": {
                "title": "Monitoring & Metrics",
                "items": [
                    "Online metrics: CTR, conversion rate, revenue per user",
                    "Offline metrics: NDCG, MAP, recall@K",
                    "Freshness: recommendation staleness alerts",
                    "Fairness: exposure distribution across item categories"
                ]
            }
        }
    
    def present(self):
        print("=== ML System Design: Recommendation System ===")
        print()
        for key in sorted(self.components.keys()):
            comp = self.components[key]
            print(f"{comp['title']}:")
            for item in comp["items"]:
                print(f"  - {item}")
            print()

design = RecommendationSystemDesign()
design.present()
print("Key discussion points in interview:")
print("  - How to handle cold start for new users?")
print("  - How to balance relevance and diversity?")
print("  - How to prevent filter bubbles?")
print("  - How to measure long-term user satisfaction vs short-term CTR?")

import numpy as np

def compute_fairness_metrics(y_true, y_pred, group_membership):
    """
    Compute fairness metrics for binary classification.
    group_membership: 0 or 1 for each sample (e.g., gender, race).
    """
    metrics = {}
    
    for group_val, group_name in [(0, "Group A"), (1, "Group B")]:
        mask = group_membership == group_val
        y_t = y_true[mask]
        y_p = y_pred[mask]
        
        tp = ((y_p == 1) & (y_t == 1)).sum()
        fp = ((y_p == 1) & (y_t == 0)).sum()
        fn = ((y_p == 0) & (y_t == 1)).sum()
        tn = ((y_p == 0) & (y_t == 0)).sum()
        
        approval_rate = y_p.mean()
        tpr = tp / (tp + fn) if (tp + fn) > 0 else 0  # True Positive Rate
        fpr = fp / (fp + tn) if (fp + tn) > 0 else 0  # False Positive Rate
        
        metrics[group_name] = {
            "n": mask.sum(),
            "approval_rate": round(approval_rate, 3),
            "tpr": round(tpr, 3),
            "fpr": round(fpr, 3)
        }
    
    # Fairness checks
    a, b = metrics["Group A"], metrics["Group B"]
    metrics["fairness"] = {
        "demographic_parity_diff": round(abs(a["approval_rate"] - b["approval_rate"]), 3),
        "equal_opportunity_diff": round(abs(a["tpr"] - b["tpr"]), 3),
        "equalized_odds_diff": round(abs(a["tpr"] - b["tpr"]) + abs(a["fpr"] - b["fpr"]), 3)
    }
    
    return metrics

# Simulate predictions with some group disparity
np.random.seed(42)
n = 1000
group = np.random.choice([0, 1], n)
y_true = np.random.choice([0, 1], n, p=[0.4, 0.6])

# Biased predictions: Group B gets approved more often
bias = 0.1 * group
y_pred = ((np.random.random(n) + bias) > 0.5).astype(int)

results = compute_fairness_metrics(y_true, y_pred, group)

print("=== Fairness Metrics Report ===")
for group_name in ["Group A", "Group B"]:
    m = results[group_name]
    print(f"\n{group_name} (n={m['n']}):")
    print(f"  Approval rate: {m['approval_rate']}")
    print(f"  True Positive Rate: {m['tpr']}")
    print(f"  False Positive Rate: {m['fpr']}")

print(f"\nFairness Checks:")
f = results["fairness"]
for metric, value in f.items():
    status = "FAIR" if value < 0.1 else "UNFAIR"
    print(f"  {metric}: {value} [{status}]")