File size: 6,488 Bytes
d491dc1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 | import numpy as np
from scipy.stats import norm
from scipy.optimize import minimize
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import HistGradientBoostingRegressor
class TobitRegressor:
"""
Type I Right-Censored Heteroscedastic Tobit Regression Model.
Estimates the latent demand distribution where the variance is modeled dynamically
as log(sigma_i) = Z_i * gamma to resolve heteroscedasticity bias.
"""
def __init__(self):
self.beta = None
self.gamma = None
self.fitted = False
def _neg_log_likelihood(self, params, X, y, censored):
"""
Calculates the negative log-likelihood of the Heteroscedastic Tobit Model.
params: contains [beta_0, beta_1, ..., beta_k, gamma_0, gamma_1, ..., gamma_k]
"""
n_features = X.shape[1]
beta = params[:n_features]
gamma = params[n_features:]
# Linear prediction of latent variable mean (mu_i) and scale (sigma_i)
mu = np.dot(X, beta)
# Using exponential link function to guarantee positive standard deviations
sigma = np.exp(np.dot(X, gamma))
# Clip to prevent numerical overflow/underflow
sigma = np.clip(sigma, 1e-4, 1e4)
# Uncensored observations (observed sales < stockout limit)
uncens = ~censored
y_uncens = y[uncens]
mu_uncens = mu[uncens]
sigma_uncens = sigma[uncens]
# Log-likelihood for uncensored data (normal distribution pdf with dynamic sigma_i)
ll_uncens = -0.5 * np.sum(np.log(2 * np.pi * sigma_uncens**2)) - \
np.sum(((y_uncens - mu_uncens) / sigma_uncens)**2) / 2.0
# Censored observations (observed sales >= stockout limit)
cens = censored
y_cens = y[cens]
mu_cens = mu[cens]
sigma_cens = sigma[cens]
# Probability of being censored: P(Y* >= y) = 1 - Phi((y - mu) / sigma) = Phi((mu - y) / sigma)
# Numerical stability via log survival function (logsf)
z = (y_cens - mu_cens) / sigma_cens
ll_cens = np.sum(norm.logsf(z))
return -(ll_uncens + ll_cens)
def fit(self, X, y, censored):
"""
Fit the Heteroscedastic Tobit model parameters.
"""
# Add intercept column to X
X_const = np.column_stack([np.ones(X.shape[0]), X])
n_features = X_const.shape[1]
# Initialize beta via OLS
ols = LinearRegression(fit_intercept=False).fit(X_const, y)
init_beta = ols.coef_
# Initialize gamma (log standard error coefficients)
residuals = y - ols.predict(X_const)
init_sigma = np.std(residuals) if np.std(residuals) > 0 else 1.0
# Standard OLS intercept initialization for log(sigma)
init_gamma = np.zeros(n_features)
init_gamma[0] = np.log(init_sigma)
init_params = np.append(init_beta, init_gamma)
# Optimize MLE log-likelihood
res = minimize(
self._neg_log_likelihood,
init_params,
args=(X_const, y, censored),
method='L-BFGS-B'
)
if not res.success:
# Fallback to homoscedastic OLS structure if convergence fails
self.beta = init_beta
self.gamma = init_gamma
else:
self.beta = res.x[:n_features]
self.gamma = res.x[n_features:]
self.fitted = True
return self
def predict_latent(self, X):
"""
Predict expected latent value E[Y* | X]
"""
if not self.fitted:
raise ValueError("Model not fitted yet.")
X_const = np.column_stack([np.ones(X.shape[0]), X])
return np.dot(X_const, self.beta)
def get_dynamic_sigma(self, X):
"""
Predict standard error sigma_i for each observation dynamically.
"""
if not self.fitted:
raise ValueError("Model not fitted yet.")
X_const = np.column_stack([np.ones(X.shape[0]), X])
sigma = np.exp(np.dot(X_const, self.gamma))
return np.clip(sigma, 1e-4, 1e4)
def impute_demand(self, X, y_obs, censored):
"""
Imputes latent demand for censored days using heteroscedastic Inverse Mills Ratio:
E[Y* | Y* >= y_obs, X] = X*beta + sigma_i * lambda(z_i)
"""
if not self.fitted:
raise ValueError("Model not fitted yet.")
y_pred_latent = self.predict_latent(X)
sigmas = self.get_dynamic_sigma(X)
y_imputed = np.copy(y_obs).astype(float)
if np.any(censored):
# Compute individual z_i using dynamic sigma_i
z = (y_obs[censored] - y_pred_latent[censored]) / sigmas[censored]
z_clipped = np.clip(z, -5.0, 5.0)
# Inverse Mills Ratio
imr = norm.pdf(z_clipped) / (norm.sf(z_clipped) + 1e-9)
y_imputed[censored] = y_pred_latent[censored] + sigmas[censored] * imr
y_imputed[censored] = np.maximum(y_imputed[censored], y_obs[censored])
return y_imputed
class CensoredDemandForecaster:
"""
Two-Stage Heteroscedastic Demand Forecaster.
"""
def __init__(self):
self.tobit = TobitRegressor()
self.forecaster = HistGradientBoostingRegressor(loss='absolute_error')
self.fitted = False
def fit(self, X, y_obs, censored):
self.tobit.fit(X, y_obs, censored)
y_imputed = self.tobit.impute_demand(X, y_obs, censored)
self.forecaster.fit(X, y_imputed)
self.fitted = True
return self
def predict(self, X):
if not self.fitted:
raise ValueError("Model not fitted yet.")
return self.forecaster.predict(X)
def predict_with_intervals(self, X, confidence_level=0.90):
"""
Uses the estimated heteroscedastic standard deviation (sigma_i) to return
dynamic confidence intervals tailored to each day's variance.
"""
point_forecast = self.predict(X)
sigmas = self.tobit.get_dynamic_sigma(X)
alpha = 1 - confidence_level
z_score = norm.ppf(1 - alpha / 2)
# Dynamic margin per observation
margin = z_score * sigmas
lower_bound = np.maximum(0, point_forecast - margin)
upper_bound = point_forecast + margin
return point_forecast, lower_bound, upper_bound
|