πŸŽ‰ 75% of content is free forever β€” Unlock Premium from $10/mo β†’
CW
Search courses…
πŸ’Ό Servicesℹ️ Aboutβœ‰οΈ ContactView Pricing Plansfrom $10

Probability: Bayes Theorem, Distributions, A/B Testing

Data Science Interview PremiumProbability⭐ Premium

Advertisement

UBER & AIRBNB INTERVIEW QUESTION

Probability: Bayes Theorem, Distributions, A/B Testing

Probability Theory & Applications

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
  1. If a ride is flagged as fraud, what's the actual probability it's fraudulent?
  2. How does this change if fraud rate increases to 1%?
  3. Design a two-stage verification system to reduce false positives
  4. 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:

Architecture Diagram
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:

Architecture Diagram
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:

ScenarioDistributionParameters
Count of events in fixed timePoissonΞ» (rate)
Time between eventsExponentialΞ² = 1/Ξ»
Ride durationNormal/Truncated NormalΞΌ, Οƒ
Rating (1-5 stars)Ordinal/Beta-BinomialΞ±, Ξ²
Surge pricing multiplierLognormalΞΌ, Οƒ
No-show probabilityBernoullip

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

Advertisement