The Interview Question
βΉοΈ
Question: You're designing a fraud detection system for Uber. You have:
- Base fraud rate: 0.1% of all rides
- Fraud detection model accuracy: 95% true positive rate, 99.9% true negative rate
- A flagged ride needs manual review
- If a ride is flagged as fraud, what's the actual probability it's fraudulent?
- How does this change if fraud rate increases to 1%?
- Design a two-stage verification system to reduce false positives
- What probability distributions are relevant for modeling ride cancellations?
Detailed Answer
1. Bayes Theorem Application
Bayes theorem allows us to update our beliefs given new evidence. It's fundamental to understanding conditional probability.
Mathematical Formula:
P(A|B) = P(B|A) Γ P(A) / P(B)
where:
P(A|B) = posterior probability (probability of A given B)
P(B|A) = likelihood (probability of B given A)
P(A) = prior probability (initial probability of A)
P(B) = marginal probability (total probability of B)
# Problem Setup
p_fraud = 0.001 # Base fraud rate (prior)
p_flagged_given_fraud = 0.95 # True positive rate (sensitivity)
p_flagged_given_legit = 0.001 # False positive rate (1 - specificity)
# Apply Bayes Theorem
# P(fraud | flagged) = P(flagged | fraud) Γ P(fraud) / P(flagged)
p_flagged = (p_flagged_given_fraud * p_fraud +
p_flagged_given_legit * (1 - p_fraud))
p_fraud_given_flagged = (p_flagged_given_fraud * p_fraud) / p_flagged
print(f"Base fraud rate: {p_fraud:.3%}")
print(f"Probability of being flagged given fraud: {p_flagged_given_fraud:.1%}")
print(f"Probability of being flagged given legitimate: {p_flagged_given_legit:.1%}")
print(f"\nResult: P(fraud | flagged) = {p_fraud_given_flagged:.2%}")
# Output:
# Base fraud rate: 0.100%
# Probability of being flagged given fraud: 95.0%
# Probability of being flagged given legitimate: 0.1%
#
# Result: P(fraud | flagged) = 48.72%
β οΈ
Counter-intuitive Result: Even with a highly accurate model (95% sensitivity, 99.9% specificity), a flagged ride only has ~49% chance of being actually fraudulent! This is due to the low base rate.
2. Impact of Changing Base Rate
# Analyze how base rate affects posterior probability
base_rates = [0.001, 0.005, 0.01, 0.02, 0.05, 0.10]
print("Base Rate vs P(fraud | flagged):")
print("-" * 45)
for p_fraud in base_rates:
p_flagged = (p_flagged_given_fraud * p_fraud +
p_flagged_given_legit * (1 - p_fraud))
p_fraud_given_flagged = (p_flagged_given_fraud * p_fraud) / p_flagged
print(f"Base rate: {p_fraud:.1%} β P(fraud | flagged): {p_fraud_given_flagged:.2%}")
# Output:
# Base Rate vs P(fraud | flagged):
# ---------------------------------------------
# Base rate: 0.1% β P(fraud | flagged): 48.72%
# Base rate: 0.5% β P(fraud | flagged): 82.64%
# Base rate: 1.0% β P(fraud | flagged): 90.48%
# Base rate: 2.0% β P(fraud | flagged): 95.02%
# Base rate: 5.0% β P(fraud | flagged): 98.01%
# Base rate: 10.0% β P(fraud | flagged): 99.01%
Mathematical Insight:
As P(A) β 1, P(A|B) β 1 regardless of P(B|A)
As P(A) β 0, P(A|B) β 0 regardless of P(B|A)
3. Two-Stage Verification System
import numpy as np
from dataclasses import dataclass
from typing import Tuple
@dataclass
class FraudDetectionSystem:
"""Two-stage fraud detection system"""
# Stage 1: Initial model
stage1_tpr: float = 0.95 # True positive rate
stage1_fpr: float = 0.001 # False positive rate
# Stage 2: Refined model (higher precision)
stage2_tpr: float = 0.90 # True positive rate for uncertain cases
stage2_fpr: float = 0.0005 # False positive rate for uncertain cases
def calculate_stage1_posterior(self, prior_fraud: float) -> float:
"""Calculate P(fraud | flagged) after stage 1"""
p_flagged = (self.stage1_tpr * prior_fraud +
self.stage1_fpr * (1 - prior_fraud))
return (self.stage1_tpr * prior_fraud) / p_flagged
def calculate_stage2_posterior(self, stage1_posterior: float) -> float:
"""Calculate P(fraud | stage2_flagged) after stage 2"""
p_flagged_stage2 = (self.stage2_tpr * stage1_posterior +
self.stage2_fpr * (1 - stage1_posterior))
return (self.stage2_tpr * stage1_posterior) / p_flagged_stage2
def two_stage_detection(self, prior_fraud: float) -> dict:
"""Run complete two-stage detection"""
stage1_posterior = self.calculate_stage1_posterior(prior_fraud)
stage2_posterior = self.calculate_stage2_posterior(stage1_posterior)
return {
'prior': prior_fraud,
'stage1_posterior': stage1_posterior,
'stage2_posterior': stage2_posterior,
'stage1_improvement': stage1_posterior / prior_fraud,
'stage2_improvement': stage2_posterior / stage1_posterior
}
# Analyze the system
system = FraudDetectionSystem()
results = system.two_stage_detection(prior_fraud=0.001)
print("Two-Stage Fraud Detection System Analysis")
print("=" * 50)
print(f"Prior fraud rate: {results['prior']:.3%}")
print(f"After Stage 1: {results['stage1_posterior']:.2%}")
print(f"After Stage 2: {results['stage2_posterior']:.2%}")
print(f"\nStage 1 improvement factor: {results['stage1_improvement']:.1f}x")
print(f"Stage 2 improvement factor: {results['stage2_improvement']:.1f}x")
4. Probability Distributions for Ride Modeling
import numpy as np
from scipy import stats
import matplotlib.pyplot as plt
# Distribution 1: Poisson for ride cancellations per hour
# Lambda = average cancellations per hour
lambda_cancellations = 3.5
# P(exactly k cancellations in an hour)
k_values = np.arange(0, 11)
poisson_probs = stats.poisson.pmf(k_values, lambda_cancellations)
print("Poisson Distribution: Ride Cancellations per Hour")
print(f"Lambda (mean): {lambda_cancellations}")
print(f"\nProbabilities:")
for k, p in zip(k_values, poisson_probs):
print(f"P(X = {k}): {p:.4f}")
# Distribution 2: Exponential for time between ride requests
# Rate parameter (requests per minute)
rate_requests = 0.5 # 1 request every 2 minutes on average
# Mean time between requests
mean_time = 1 / rate_requests
print(f"\nExponential Distribution: Time Between Requests")
print(f"Rate: {rate_requests} requests/minute")
print(f"Mean time between requests: {mean_time} minutes")
# Probability of waiting more than t minutes
for t in [1, 2, 3, 5, 10]:
prob_wait = stats.expon.sf(t, scale=mean_time)
print(f"P(wait > {t} min): {prob_wait:.4f}")
# Distribution 3: Normal for ride duration (with bounds)
mean_duration = 25 # minutes
std_duration = 8 # minutes
# Truncated normal (duration can't be negative)
lower_bound = 5 # minimum ride duration
upper_bound = 60 # maximum ride duration
# Generate samples
np.random.seed(42)
ride_durations = stats.truncnorm.rvs(
(lower_bound - mean_duration) / std_duration,
(upper_bound - mean_duration) / std_duration,
loc=mean_duration,
scale=std_duration,
size=10000
)
print(f"\nTruncated Normal Distribution: Ride Duration")
print(f"Mean: {np.mean(ride_durations):.2f} minutes")
print(f"Std: {np.std(ride_durations):.2f} minutes")
print(f"Min: {np.min(ride_durations):.2f} minutes")
print(f"Max: {np.max(ride_durations):.2f} minutes")
Distribution Selection Guide:
| Scenario | Distribution | Parameters |
|---|---|---|
| Count of events in fixed time | Poisson | Ξ» (rate) |
| Time between events | Exponential | Ξ² = 1/Ξ» |
| Ride duration | Normal/Truncated Normal | ΞΌ, Ο |
| Rating (1-5 stars) | Ordinal/Beta-Binomial | Ξ±, Ξ² |
| Surge pricing multiplier | Lognormal | ΞΌ, Ο |
| No-show probability | Bernoulli | p |
5. Bayesian A/B Testing
import numpy as np
from scipy import stats
class BayesianABTest:
"""Bayesian A/B testing with Beta-Binomial model"""
def __init__(self, prior_alpha=1, prior_beta=1):
"""Initialize with Beta prior"""
self.prior_alpha = prior_alpha
self.prior_beta = prior_beta
def update_belief(self, successes, trials):
"""Update Beta distribution with observed data"""
alpha = self.prior_alpha + successes
beta = self.prior_beta + (trials - successes)
return alpha, beta
def calculate_probability_b_beats_a(self, a_data, b_data, n_samples=100000):
"""Calculate P(B > A) using Monte Carlo sampling"""
a_alpha, a_beta = self.update_belief(a_data['successes'], a_data['trials'])
b_alpha, b_beta = self.update_belief(b_data['successes'], b_data['trials'])
# Sample from posterior distributions
a_samples = np.random.beta(a_alpha, a_beta, n_samples)
b_samples = np.random.beta(b_alpha, b_beta, n_samples)
# Calculate P(B > A)
prob_b_wins = np.mean(b_samples > a_samples)
# Calculate expected lift
lift = (b_samples - a_samples) / a_samples
expected_lift = np.mean(lift)
return {
'prob_b_wins': prob_b_wins,
'expected_lift': expected_lift,
'lift_ci': np.percentile(lift, [2.5, 97.5]),
'a_posterior': (a_alpha, a_beta),
'b_posterior': (b_alpha, b_beta)
}
# Example: Testing two pricing strategies
ab_test = BayesianABTest()
# Control (Strategy A): 1000 rides, 150 conversions
a_data = {'successes': 150, 'trials': 1000}
# Treatment (Strategy B): 1000 rides, 180 conversions
b_data = {'successes': 180, 'trials': 1000}
results = ab_test.calculate_probability_b_beats_a(a_data, b_data)
print("Bayesian A/B Test Results")
print("=" * 40)
print(f"P(B > A): {results['prob_b_wins']:.2%}")
print(f"Expected lift: {results['expected_lift']:.2%}")
print(f"95% CI for lift: ({results['lift_ci'][0]:.2%}, {results['lift_ci'][1]:.2%})")
print(f"\nRecommendation:")
if results['prob_b_wins'] > 0.95:
print("Strong evidence for B - implement with confidence")
elif results['prob_b_wins'] > 0.90:
print("Good evidence for B - consider implementing")
elif results['prob_b_wins'] > 0.75:
print("Weak evidence for B - continue testing")
else:
print("No clear winner - continue testing or stick with A")
6. Real-World Application: Surge Pricing Probability
import numpy as np
from scipy import stats
class SurgePricingModel:
"""Model surge pricing using supply-demand dynamics"""
def __init__(self):
self.base_price = 10.0 # Base fare in dollars
def calculate_surge_probability(self,
demand_rate: float,
supply_rate: float,
demand_std: float = 2.0,
supply_std: float = 1.5) -> dict:
"""
Calculate probability of surge pricing given demand/supply
Surge occurs when demand > supply * threshold
"""
# Model demand and supply as normal distributions
demand_dist = stats.norm(demand_rate, demand_std)
supply_dist = stats.norm(supply_rate, supply_std)
# Monte Carlo simulation
n_simulations = 100000
demand_samples = demand_dist.rvs(n_simulations)
supply_samples = supply_dist.rvs(n_simulations)
# Surge threshold (demand > 1.5x supply)
surge_threshold = 1.5
surge_occurs = demand_samples > (supply_samples * surge_threshold)
# Calculate surge multiplier when surge occurs
ratios = demand_samples / supply_samples
surge_ratios = ratios[surge_occurs]
# Map ratio to surge multiplier
surge_multipliers = np.where(
surge_ratios > 3.0, 3.0,
np.where(surge_ratios > 2.5, 2.5,
np.where(surge_ratios > 2.0, 2.0,
np.where(surge_ratios > 1.5, 1.5, 1.0)))
)
return {
'surge_probability': np.mean(surge_occurs),
'expected_multiplier': np.mean(surge_multipliers),
'avg_surge_multiplier': np.mean(surge_multipliers[surge_multipliers > 1.0]),
'price_at_surge': self.base_price * np.mean(surge_multipliers[surge_multipliers > 1.0])
}
def calculate_expected_revenue(self, demand_rate: float, supply_rate: float) -> float:
"""Calculate expected revenue per hour"""
surge_probs = self.calculate_surge_probability(demand_rate, supply_rate)
# Base revenue (no surge)
base_revenue_per_ride = self.base_price
rides_per_hour = min(demand_rate, supply_rate)
# Surge revenue
surge_multiplier = surge_probs['expected_multiplier']
expected_revenue_per_ride = base_revenue_per_ride * surge_multiplier
return expected_revenue_per_ride * rides_per_hour
# Analyze different scenarios
model = SurgePricingModel()
scenarios = [
("Low demand, High supply", 5, 10),
("Balanced", 8, 8),
("High demand, Low supply", 12, 5),
("Peak hours", 15, 6),
("Event surge", 20, 4),
]
print("Surge Pricing Analysis")
print("=" * 60)
for name, demand, supply in scenarios:
results = model.calculate_surge_probability(demand, supply)
revenue = model.calculate_expected_revenue(demand, supply)
print(f"\n{name}:")
print(f" Demand: {demand}, Supply: {supply}")
print(f" Surge probability: {results['surge_probability']:.1%}")
print(f" Expected multiplier: {results['expected_multiplier']:.2f}x")
print(f" Expected revenue/hour: ${revenue:.2f}")
π‘
Pro Tip: In surge pricing models, consider the elasticity of demand. Higher prices may reduce demand, creating a feedback loop that needs to be modeled carefully.
7. Common Follow-Up Questions
Follow-up 1: How would you handle rare events with imbalanced data?
# Rare event modeling techniques
from imblearn.over_sampling import SMOTE
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import precision_recall_curve
# 1. Adjusted priors for rare events
def adjusted_prior_sampling(X, y, target_ratio=0.3):
"""Oversample minority class to achieve target ratio"""
minority_indices = np.where(y == 1)[0]
majority_indices = np.where(y == 0)[0]
# Calculate oversampling needed
n_minority = len(minority_indices)
n_majority = len(majority_indices)
target_minority = int(n_majority * target_ratio / (1 - target_ratio))
# Oversample minority
oversampled_minority = np.random.choice(
minority_indices,
size=target_minority - n_minority,
replace=True
)
# Combine
new_indices = np.concatenate([majority_indices, minority_indices, oversampled_minority])
return X[new_indices], y[new_indices]
# 2. Cost-sensitive learning
def cost_sensitive_loss(y_true, y_pred, cost_fp=1, cost_fn=10):
"""Custom loss function that penalizes false negatives more"""
tp = np.sum((y_pred == 1) & (y_true == 1))
fp = np.sum((y_pred == 1) & (y_true == 0))
fn = np.sum((y_pred == 0) & (y_true == 1))
tn = np.sum((y_pred == 0) & (y_true == 0))
total_cost = fp * cost_fp + fn * cost_fn
return total_cost / len(y_true)
Follow-up 2: How do you validate a probabilistic model?
# Calibration checking
from sklearn.calibration import calibration_curve
import matplotlib.pyplot as plt
def check_calibration(y_true, y_prob, n_bins=10):
"""Check if predicted probabilities match actual frequencies"""
prob_true, prob_pred = calibration_curve(y_true, y_prob, n_bins=n_bins)
# Plot calibration curve
plt.figure(figsize=(8, 6))
plt.plot([0, 1], [0, 1], 'k--', label='Perfectly calibrated')
plt.plot(prob_pred, prob_true, 's-', label='Model')
plt.xlabel('Mean predicted probability')
plt.ylabel('Fraction of positives')
plt.title('Calibration Curve')
plt.legend()
plt.grid(True)
plt.show()
# Brier score (lower is better)
brier_score = np.mean((y_prob - y_true) ** 2)
print(f"Brier Score: {brier_score:.4f}")
return brier_score
# Expected Calibration Error (ECE)
def expected_calibration_error(y_true, y_prob, n_bins=10):
"""Calculate Expected Calibration Error"""
bin_boundaries = np.linspace(0, 1, n_bins + 1)
ece = 0.0
for i in range(n_bins):
mask = (y_prob >= bin_boundaries[i]) & (y_prob < bin_boundaries[i + 1])
if mask.sum() > 0:
bin_accuracy = y_true[mask].mean()
bin_confidence = y_prob[mask].mean()
bin_weight = mask.sum() / len(y_true)
ece += bin_weight * abs(bin_accuracy - bin_confidence)
return ece
Company-Specific Tips
βΉοΈ
Uber Tips:
- Uber heavily tests on Bayesian methods for pricing and fraud
- Understand conjugate priors (Beta-Binomial, Normal-Normal)
- Know how to calculate expected value under uncertainty
- Practice Monte Carlo simulation problems
Airbnb Tips:
- Airbnb focuses on probability for pricing and availability
- Understand survival analysis for booking probability
- Know how to model supply and demand dynamics
- Be comfortable with Markov chains for user behavior
Quiz Section
Related Topics
- Bayesian Statistics β Deeper dive into Bayesian methods
- A/B Testing Design β Setting up proper experiments
- Probability Distributions β Statistical foundations
- Monte Carlo Methods β Simulation techniques