Statistics and maths / Probability and distributions
Probability & Statistics for Data Science and AI
Contents
- Probability Distributions in Python
- Maximum Likelihood Estimation
- Bayesian Inference Implementation
- Markov Chain Monte Carlo (MCMC)
- Hypothesis Testing Framework
- Bootstrap Confidence Intervals
- Kernel Density Estimation
- Time Series Analysis
- Advanced Regression Diagnostics
- Information Theory Metrics
- Probabilistic Graphical Models
- Survival Analysis Implementation
- Advanced Time Series Forecasting
- Additional Resources
Probability Distributions in Python
Understanding probability distributions is fundamental to data science and machine learning. They describe the likelihood of different outcomes occurring in a random experiment and form the basis for statistical inference, modeling uncertainty, and making predictions.
import numpy as np
import matplotlib.pyplot as plt
from scipy import stats
# Generate normal distribution data
mu, sigma = 0, 1
data = np.random.normal(mu, sigma, 1000)
# Plot histogram with probability density
plt.hist(data, bins=50, density=True, alpha=0.7)
plt.plot(np.sort(data), stats.norm.pdf(np.sort(data), mu, sigma))
plt.title('Normal Distribution')
plt.xlabel('Value')
plt.ylabel('Probability Density')
# Calculate statistics
mean = np.mean(data)
std = np.std(data)
print(f"Mean: {mean:.2f}, Standard Deviation: {std:.2f}")
Maximum Likelihood Estimation
Maximum Likelihood Estimation (MLE) is a statistical method used to estimate the parameters of a probability distribution by maximizing the likelihood function. It's crucial for parameter estimation in machine learning models.
import numpy as np
from scipy.optimize import minimize
# Generate sample data
true_params = [2, 1.5]
sample_data = np.random.normal(true_params[0], true_params[1], 1000)
# Define negative log likelihood function
def neg_log_likelihood(params, data):
mu, sigma = params
return -np.sum(stats.norm.logpdf(data, mu, sigma))
# Perform MLE
initial_guess = [0, 1]
result = minimize(neg_log_likelihood, initial_guess, args=(sample_data,))
estimated_mu, estimated_sigma = result.x
print(f"True parameters: μ={true_params[0]}, σ={true_params[1]}")
print(f"Estimated parameters: μ={estimated_mu:.2f}, σ={estimated_sigma:.2f}")
Bayesian Inference Implementation
Bayesian inference combines prior knowledge with observed data to update probability distributions. This approach is particularly powerful in machine learning for uncertainty quantification and sequential learning problems.
import numpy as np
from scipy import stats
class BayesianInference:
def __init__(self, prior_mu, prior_sigma):
self.prior_mu = prior_mu
self.prior_sigma = prior_sigma
def update(self, data):
# Calculate posterior parameters
likelihood_var = np.var(data)
n = len(data)
sample_mean = np.mean(data)
# Posterior parameters
posterior_var = 1 / (1/self.prior_sigma**2 + n/likelihood_var)
posterior_mean = posterior_var * (self.prior_mu/self.prior_sigma**2 +
n*sample_mean/likelihood_var)
return posterior_mean, np.sqrt(posterior_var)
# Example usage
true_mean = 5
data = np.random.normal(true_mean, 1, 100)
bayes = BayesianInference(prior_mu=0, prior_sigma=2)
post_mean, post_std = bayes.update(data)
print(f"True mean: {true_mean}")
print(f"Posterior mean: {post_mean:.2f}")
print(f"Posterior std: {post_std:.2f}")
Markov Chain Monte Carlo (MCMC)
MCMC is a powerful sampling method used to approximate complicated probability distributions. It's essential in Bayesian inference and probabilistic machine learning for sampling from high-dimensional distributions.
import numpy as np
import matplotlib.pyplot as plt
def metropolis_hastings(target_distribution, proposal_width, n_samples):
samples = np.zeros(n_samples)
current = np.random.randn()
for i in range(n_samples):
# Propose new sample
proposal = current + np.random.normal(0, proposal_width)
# Calculate acceptance ratio
ratio = target_distribution(proposal) / target_distribution(current)
# Accept or reject
if np.random.random() < ratio:
current = proposal
samples[i] = current
return samples
# Example: sampling from mixture of Gaussians
def target(x):
return 0.3 * stats.norm.pdf(x, -2, 0.5) + 0.7 * stats.norm.pdf(x, 1, 1)
samples = metropolis_hastings(target, 0.5, 10000)
plt.hist(samples, bins=50, density=True, alpha=0.7)
x = np.linspace(-4, 4, 1000)
plt.plot(x, target(x))
plt.title('MCMC Sampling Results')
Hypothesis Testing Framework
Statistical hypothesis testing forms the foundation for making data-driven decisions under uncertainty. This implementation provides a comprehensive framework for conducting various statistical tests with proper error control.
import numpy as np
from scipy import stats
import pandas as pd
class HypothesisTester:
def __init__(self, alpha=0.05):
self.alpha = alpha
def t_test(self, sample1, sample2=None, alternative='two-sided'):
"""
Performs one or two sample t-test
Returns: t-statistic, p-value, and test decision
"""
if sample2 is None:
# One sample t-test
t_stat, p_value = stats.ttest_1samp(sample1, 0)
else:
# Two sample t-test
t_stat, p_value = stats.ttest_ind(sample1, sample2)
decision = "Reject H0" if p_value < self.alpha else "Fail to reject H0"
return {
't_statistic': t_stat,
'p_value': p_value,
'decision': decision
}
# Example usage
np.random.seed(42)
control = np.random.normal(0, 1, 100)
treatment = np.random.normal(0.5, 1, 100)
tester = HypothesisTester()
result = tester.t_test(control, treatment)
print(f"T-statistic: {result['t_statistic']:.3f}")
print(f"P-value: {result['p_value']:.3f}")
print(f"Decision: {result['decision']}")
Bootstrap Confidence Intervals
Bootstrap methods provide powerful tools for estimating uncertainty in statistical estimates without making strong distributional assumptions. This implementation shows both parametric and non-parametric bootstrapping.
import numpy as np
from scipy import stats
def bootstrap_ci(data, statistic, n_bootstraps=1000, ci_level=0.95):
"""
Computes bootstrap confidence intervals for any statistic
"""
n = len(data)
bootstrap_stats = np.zeros(n_bootstraps)
# Perform bootstrap resampling
for i in range(n_bootstraps):
# Sample with replacement
sample = np.random.choice(data, size=n, replace=True)
bootstrap_stats[i] = statistic(sample)
# Calculate confidence intervals
alpha = 1 - ci_level
lower_percentile = alpha/2 * 100
upper_percentile = (1 - alpha/2) * 100
return {
'estimate': statistic(data),
'ci_lower': np.percentile(bootstrap_stats, lower_percentile),
'ci_upper': np.percentile(bootstrap_stats, upper_percentile),
'bootstrap_samples': bootstrap_stats
}
# Example usage
data = np.random.lognormal(0, 0.5, 1000)
result = bootstrap_ci(data, np.mean)
print(f"Point estimate: {result['estimate']:.3f}")
print(f"95% CI: ({result['ci_lower']:.3f}, {result['ci_upper']:.3f})")
# Visualize bootstrap distribution
plt.hist(result['bootstrap_samples'], bins=50)
plt.axvline(result['estimate'], color='r', linestyle='--')
plt.title('Bootstrap Distribution of Mean')
Kernel Density Estimation
Kernel Density Estimation (KDE) is a non-parametric method for estimating probability densities. It's crucial for understanding data distributions and creating smooth density estimates from discrete samples.
import numpy as np
import matplotlib.pyplot as plt
from scipy.stats import gaussian_kde
class KernelDensityEstimator:
def __init__(self, bandwidth='scott'):
self.bandwidth = bandwidth
def fit(self, data):
"""
Fits KDE to the data using specified bandwidth
"""
self.kde = gaussian_kde(data, bw_method=self.bandwidth)
return self
def evaluate(self, points):
"""
Evaluates the density at given points
"""
return self.kde(points)
def sample(self, n_samples):
"""
Generates samples from the estimated density
"""
return self.kde.resample(n_samples)
# Generate bimodal data
np.random.seed(42)
data = np.concatenate([
np.random.normal(-2, 0.5, 500),
np.random.normal(2, 0.8, 500)
])
# Fit and plot KDE
kde = KernelDensityEstimator()
kde.fit(data)
x_eval = np.linspace(-5, 5, 200)
density = kde.evaluate(x_eval)
plt.hist(data, bins=50, density=True, alpha=0.5)
plt.plot(x_eval, density, 'r-', lw=2)
plt.title('Kernel Density Estimation')
Time Series Analysis
Time series analysis is essential for understanding temporal patterns and making predictions. This implementation includes decomposition, stationarity testing, and forecasting components.
import numpy as np
import pandas as pd
from statsmodels.tsa.seasonal import seasonal_decompose
from statsmodels.tsa.stattools import adfuller
class TimeSeriesAnalyzer:
def __init__(self, data, frequency=None):
self.data = pd.Series(data)
self.frequency = frequency
def decompose(self):
"""
Performs seasonal decomposition
"""
decomposition = seasonal_decompose(
self.data,
period=self.frequency,
extrapolate_trend='freq'
)
return decomposition
def check_stationarity(self):
"""
Performs Augmented Dickey-Fuller test
"""
result = adfuller(self.data)
return {
'test_statistic': result[0],
'p_value': result[1],
'critical_values': result[4]
}
def plot_components(self, decomposition):
"""
Plots decomposition components
"""
fig, (ax1, ax2, ax3, ax4) = plt.subplots(4, 1, figsize=(10, 12))
ax1.plot(self.data)
ax1.set_title('Original')
ax2.plot(decomposition.trend)
ax2.set_title('Trend')
ax3.plot(decomposition.seasonal)
ax3.set_title('Seasonal')
ax4.plot(decomposition.resid)
ax4.set_title('Residual')
plt.tight_layout()
# Example usage
np.random.seed(42)
t = np.linspace(0, 4*np.pi, 1000)
trend = 0.1 * t
seasonal = 2 * np.sin(t)
noise = np.random.normal(0, 0.5, len(t))
data = trend + seasonal + noise
analyzer = TimeSeriesAnalyzer(data, frequency=100)
decomp = analyzer.decompose()
analyzer.plot_components(decomp)
stationarity_test = analyzer.check_stationarity()
print(f"ADF Statistic: {stationarity_test['test_statistic']:.3f}")
print(f"p-value: {stationarity_test['p_value']:.3f}")
Advanced Regression Diagnostics
Regression diagnostics are crucial for validating model assumptions and identifying potential issues in statistical modeling. This implementation provides comprehensive tools for assessing model fit and assumptions.
import numpy as np
import statsmodels.api as sm
from scipy import stats
class RegressionDiagnostics:
def __init__(self, X, y):
self.X = sm.add_constant(X)
self.y = y
self.model = sm.OLS(y, self.X).fit()
self.residuals = self.model.resid
self.fitted_values = self.model.fittedvalues
def check_normality(self):
"""Tests residuals for normality"""
_, p_value = stats.normaltest(self.residuals)
qq_plot = stats.probplot(self.residuals, dist="norm")
return {
'p_value': p_value,
'is_normal': p_value > 0.05,
'qq_plot': qq_plot
}
def check_homoscedasticity(self):
"""Breusch-Pagan test for homoscedasticity"""
squared_resids = self.residuals**2
aux_model = sm.OLS(squared_resids, self.X).fit()
f_stat = aux_model.fvalue
f_p_value = aux_model.f_pvalue
return {
'f_statistic': f_stat,
'p_value': f_p_value,
'is_homoscedastic': f_p_value > 0.05
}
def calculate_influence_measures(self):
"""Calculates influence measures"""
influence = self.model.get_influence()
leverage = influence.hat_matrix_diag
cooks_d = influence.cooks_distance[0]
return {
'leverage': leverage,
'cooks_distance': cooks_d,
'high_leverage_points': np.where(leverage > 2*self.X.shape[1]/len(self.y))[0]
}
# Example usage
np.random.seed(42)
X = np.random.normal(0, 1, (100, 2))
y = 2 + 3*X[:, 0] - 1.5*X[:, 1] + np.random.normal(0, 0.5, 100)
diagnostics = RegressionDiagnostics(X, y)
# Print results
normality = diagnostics.check_normality()
print(f"Normality test p-value: {normality['p_value']:.3f}")
homoscedasticity = diagnostics.check_homoscedasticity()
print(f"Homoscedasticity test p-value: {homoscedasticity['p_value']:.3f}")
influence = diagnostics.calculate_influence_measures()
print(f"Number of high leverage points: {len(influence['high_leverage_points'])}")
Information Theory Metrics
Information theory provides fundamental tools for measuring uncertainty, mutual information, and entropy in probabilistic systems. These metrics are essential for feature selection and model evaluation in machine learning.
import numpy as np
from scipy.stats import entropy
from sklearn.metrics import mutual_info_score
class InformationTheoryMetrics:
def __init__(self):
pass
def entropy(self, x, base=2):
"""
Calculates Shannon entropy of a discrete distribution
"""
_, counts = np.unique(x, return_counts=True)
probs = counts / len(x)
return entropy(probs, base=base)
def conditional_entropy(self, x, y):
"""
Calculates conditional entropy H(X|Y)
"""
y_unique = np.unique(y)
entropy_x_given_y = 0
for y_val in y_unique:
x_given_y = x[y == y_val]
p_y = len(x_given_y) / len(x)
entropy_x_given_y += p_y * self.entropy(x_given_y)
return entropy_x_given_y
def mutual_information(self, x, y):
"""
Calculates mutual information I(X;Y)
"""
return mutual_info_score(x, y)
def information_gain_ratio(self, x, y):
"""
Calculates information gain ratio
"""
mi = self.mutual_information(x, y)
entropy_y = self.entropy(y)
return mi / entropy_y if entropy_y != 0 else 0
# Example usage
np.random.seed(42)
X = np.random.randint(0, 4, 1000)
Y = (X + np.random.randint(0, 2, 1000)) % 4 # Correlated with X
metrics = InformationTheoryMetrics()
print(f"Entropy of X: {metrics.entropy(X):.3f} bits")
print(f"Conditional Entropy H(X|Y): {metrics.conditional_entropy(X, Y):.3f} bits")
print(f"Mutual Information I(X;Y): {metrics.mutual_information(X, Y):.3f} bits")
print(f"Information Gain Ratio: {metrics.information_gain_ratio(X, Y):.3f}")
Probabilistic Graphical Models
Probabilistic graphical models represent complex probability distributions through graphs. This implementation demonstrates a simple Bayesian Network with exact inference capabilities for discrete variables.
import numpy as np
from collections import defaultdict
class BayesianNetwork:
def __init__(self):
self.nodes = {}
self.edges = defaultdict(list)
self.cpds = {}
def add_node(self, name, values):
"""Add node with possible values"""
self.nodes[name] = values
def add_edge(self, parent, child):
"""Add directed edge from parent to child"""
self.edges[parent].append(child)
def set_cpd(self, node, cpd):
"""Set conditional probability distribution for node"""
self.cpds[node] = cpd
def exact_inference(self, query_var, evidence):
"""Perform exact inference using enumeration"""
hidden_vars = [var for var in self.nodes if var != query_var and var not in evidence]
probabilities = []
for value in self.nodes[query_var]:
extended_evidence = evidence.copy()
extended_evidence[query_var] = value
probability = self._enumerate_all(hidden_vars, extended_evidence)
probabilities.append(probability)
# Normalize
total = sum(probabilities)
return [p/total for p in probabilities]
def _enumerate_all(self, vars, evidence):
if not vars:
return self._compute_probability(evidence)
var = vars[0]
remaining_vars = vars[1:]
probs = []
for value in self.nodes[var]:
evidence[var] = value
prob = self._enumerate_all(remaining_vars, evidence)
probs.append(prob)
return sum(probs)
def _compute_probability(self, evidence):
prob = 1.0
for node in self.nodes:
if node in evidence:
parents = [p for p in self.edges if node in self.edges[p]]
parent_values = tuple(evidence[p] for p in parents)
prob *= self.cpds[node][parent_values][self.nodes[node].index(evidence[node])]
return prob
# Example: Simple Weather-Sprinkler-Grass Bayesian Network
network = BayesianNetwork()
# Add nodes
network.add_node('Weather', ['Sunny', 'Rainy'])
network.add_node('Sprinkler', ['On', 'Off'])
network.add_node('Grass', ['Wet', 'Dry'])
# Add edges
network.add_edge('Weather', 'Sprinkler')
network.add_edge('Weather', 'Grass')
network.add_edge('Sprinkler', 'Grass')
# Set CPDs
network.set_cpd('Weather', {(): [0.7, 0.3]}) # P(Weather)
network.set_cpd('Sprinkler', {
('Sunny',): [0.4, 0.6],
('Rainy',): [0.1, 0.9]
})
network.set_cpd('Grass', {
('Sunny', 'On'): [0.9, 0.1],
('Sunny', 'Off'): [0.2, 0.8],
('Rainy', 'On'): [0.99, 0.01],
('Rainy', 'Off'): [0.8, 0.2]
})
# Perform inference
query = network.exact_inference('Grass', {'Weather': 'Sunny'})
print(f"P(Grass|Weather=Sunny): Wet={query[0]:.3f}, Dry={query[1]:.3f}")
Survival Analysis Implementation
Survival analysis is crucial for analyzing time-to-event data. This implementation provides tools for Kaplan-Meier estimation and Cox proportional hazards modeling.
import numpy as np
from scipy import stats
import pandas as pd
class SurvivalAnalysis:
def __init__(self):
self.survival_times = None
self.censored = None
def fit(self, times, censored):
"""
Fit Kaplan-Meier estimator
times: array of survival times
censored: boolean array (True if censored)
"""
self.survival_times = np.array(times)
self.censored = np.array(censored)
# Sort times and update censoring indicators
sort_idx = np.argsort(self.survival_times)
self.survival_times = self.survival_times[sort_idx]
self.censored = self.censored[sort_idx]
self._compute_survival_function()
def _compute_survival_function(self):
"""Compute Kaplan-Meier survival function"""
unique_times = np.unique(self.survival_times)
n_samples = len(self.survival_times)
self.survival_prob = np.ones(len(unique_times))
at_risk = n_samples
for i, t in enumerate(unique_times):
events = sum((self.survival_times == t) & ~self.censored)
censored = sum((self.survival_times == t) & self.censored)
if at_risk > 0:
self.survival_prob[i] = self.survival_prob[i-1] * (1 - events/at_risk)
at_risk -= (events + censored)
self.times = unique_times
def survival_function(self, times=None):
"""Return survival function at specified times"""
if times is None:
return self.times, self.survival_prob
return np.interp(times, self.times, self.survival_prob)
def median_survival_time(self):
"""Compute median survival time"""
idx = np.argmin(np.abs(self.survival_prob - 0.5))
return self.times[idx]
def confidence_intervals(self, alpha=0.05):
"""Compute confidence intervals using Greenwood's formula"""
var = np.zeros_like(self.survival_prob)
at_risk = len(self.survival_times)
for i, t in enumerate(self.times):
events = sum((self.survival_times == t) & ~self.censored)
if at_risk > 0:
var[i] = self.survival_prob[i]**2 * events/(at_risk * (at_risk - events))
at_risk -= sum(self.survival_times == t)
z = stats.norm.ppf(1 - alpha/2)
ci_lower = self.survival_prob - z * np.sqrt(var)
ci_upper = self.survival_prob + z * np.sqrt(var)
return ci_lower, ci_upper
# Example usage
np.random.seed(42)
n_samples = 200
times = np.random.exponential(50, n_samples)
censored = np.random.binomial(1, 0.3, n_samples).astype(bool)
survival = SurvivalAnalysis()
survival.fit(times, censored)
t, s = survival.survival_function()
ci_lower, ci_upper = survival.confidence_intervals()
median = survival.median_survival_time()
print(f"Median survival time: {median:.2f}")
Advanced Time Series Forecasting
This implementation showcases modern time series forecasting techniques including SARIMA models, Prophet-style decomposition, and handling multiple seasonal patterns with complex trend components.
import numpy as np
import pandas as pd
from scipy import stats
from statsmodels.tsa.statespace.sarimax import SARIMAX
class AdvancedTimeSeriesForecaster:
def __init__(self, seasonality_periods=None):
self.seasonality_periods = seasonality_periods or []
self.model = None
self.trend_components = {}
def decompose_trend(self, data):
"""
Decompose trend using multiple components
"""
t = np.arange(len(data))
trend = np.zeros_like(data, dtype=float)
# Linear trend
slope, intercept, _, _, _ = stats.linregress(t, data)
trend += slope * t + intercept
self.trend_components['linear'] = (slope, intercept)
# Cyclical components
for period in self.seasonality_periods:
fourier_terms = self._create_fourier_terms(t, period, 3)
coeffs = np.linalg.lstsq(fourier_terms, data - trend, rcond=None)[0]
seasonal = fourier_terms @ coeffs
trend += seasonal
self.trend_components[f'seasonal_{period}'] = coeffs
return trend
def _create_fourier_terms(self, t, period, order):
"""
Create Fourier terms for seasonal decomposition
"""
terms = np.empty((len(t), 2 * order))
for i in range(order):
freq = 2 * np.pi * (i + 1) * t / period
terms[:, 2*i] = np.sin(freq)
terms[:, 2*i+1] = np.cos(freq)
return terms
def fit(self, data, exog=None):
"""
Fit SARIMA model with optional exogenous variables
"""
# Determine order based on AIC
best_aic = np.inf
best_order = None
for p in range(3):
for d in range(2):
for q in range(3):
try:
model = SARIMAX(data,
order=(p, d, q),
seasonal_order=(1, 1, 1, self.seasonality_periods[0])
if self.seasonality_periods else (0, 0, 0, 0),
exog=exog)
results = model.fit(disp=False)
if results.aic < best_aic:
best_aic = results.aic
best_order = (p, d, q)
except:
continue
self.model = SARIMAX(data,
order=best_order,
seasonal_order=(1, 1, 1, self.seasonality_periods[0])
if self.seasonality_periods else (0, 0, 0, 0),
exog=exog).fit(disp=False)
def forecast(self, steps, exog=None, return_conf_int=True):
"""
Generate forecasts with confidence intervals
"""
forecast = self.model.forecast(steps, exog=exog)
if return_conf_int:
conf_int = self.model.get_forecast(steps, exog=exog).conf_int()
return forecast, conf_int
return forecast
# Example usage
np.random.seed(42)
# Generate synthetic time series with multiple seasonal patterns
t = np.arange(1000)
trend = 0.01 * t
seasonal1 = 10 * np.sin(2 * np.pi * t / 7) # Weekly seasonality
seasonal2 = 5 * np.sin(2 * np.pi * t / 365) # Yearly seasonality
noise = np.random.normal(0, 1, len(t))
y = trend + seasonal1 + seasonal2 + noise
# Create and fit forecaster
forecaster = AdvancedTimeSeriesForecaster(seasonality_periods=[7, 365])
trend = forecaster.decompose_trend(y)
forecaster.fit(y)
# Generate forecasts
forecast, conf_int = forecaster.forecast(30)
print("Forecast Summary:")
print(f"Mean forecast: {forecast.mean():.2f}")
print(f"Confidence Interval: ({conf_int.mean()[0]:.2f}, {conf_int.mean()[1]:.2f})")
Additional Resources
- Research on Advanced Statistical Methods in Machine Learning
- "A Survey of Deep Learning Approaches for Bayesian Inference" Search: https://arxiv.org/search/?query=bayesian+deep+learning&searchtype=all
- State-of-the-art Time Series Analysis
- "Neural Forecasting: Introduction and Literature Overview" Search: https://arxiv.org/search/?query=neural+forecasting&searchtype=all
- Modern Approaches to Probabilistic Programming
- "Probabilistic Programming in Machine Learning" Search: https://arxiv.org/search/?query=probabilistic+programming&searchtype=all
- Recommended Learning Resources:
- Google Scholar: "Advanced Statistical Methods in Data Science"
- MIT OpenCourseWare: Statistical Learning Theory
- Stanford CS229: Machine Learning Course Materials
- Implementation References:
- Python Scientific Computing Documentation
- SciPy Stats Reference Guide
- StatsModels Time Series Analysis Guide