Python 64.6%
TeX 23.3%
HTML 11.6%
Makefile 0.5%
1"""Quantile regression models for heterogeneous price effect analysis."""23import logging4from pathlib import Path5from typing import Dict, List, Optional, Sequence, Tuple67import numpy as np8import pandas as pd9import statsmodels.api as sm10from statsmodels.regression.quantile_regression import QuantReg1112logger = logging.getLogger(__name__)1314DEFAULT_TAUS: List[float] = [0.10, 0.25, 0.50, 0.75, 0.90]151617def _available_cols(df: pd.DataFrame, requested: List[str]) -> List[str]:18 """Return the subset of requested columns present in df."""19 present = [c for c in requested if c in df.columns]20 missing = set(requested) - set(present)21 if missing:22 logger.warning("Missing columns (skipped): %s", sorted(missing))23 return present242526def _prepare_data(27 df: pd.DataFrame,28 y_col: str,29 x_cols: List[str],30) -> Tuple[pd.Series, pd.DataFrame]:31 """Drop NaN rows, add constant, return (y, X)."""32 cols = _available_cols(df, x_cols)33 if not cols:34 raise ValueError("No valid feature columns found in the DataFrame.")35 subset = df[[y_col] + cols].dropna()36 if subset.empty:37 raise ValueError("No observations remaining after dropping NaN rows.")38 y = subset[y_col].astype(float)39 X = sm.add_constant(subset[cols].astype(float))40 return y, X414243# ---------------------------------------------------------------------------44# Single quantile estimation45# ---------------------------------------------------------------------------4647def estimate_quantile(48 df: pd.DataFrame,49 y_col: str,50 x_cols: List[str],51 tau: float,52 max_iter: int = 10_000,53) -> sm.regression.quantile_regression.QuantRegResults:54 """Single quantile regression at quantile tau."""55 if not 0 < tau < 1:56 raise ValueError(f"tau must be in (0, 1), got {tau}")57 y, X = _prepare_data(df, y_col, x_cols)58 model = QuantReg(y, X)59 results = model.fit(q=tau, max_iter=max_iter)60 logger.info(61 "Quantile regression (tau=%.2f): pseudo-R2=%.4f, n=%d",62 tau, results.prsquared, int(results.nobs),63 )64 return results656667# ---------------------------------------------------------------------------68# Quantile sequence69# ---------------------------------------------------------------------------7071def estimate_quantile_sequence(72 df: pd.DataFrame,73 y_col: str,74 x_cols: List[str],75 taus: Optional[List[float]] = None,76 max_iter: int = 10_000,77) -> Dict[float, sm.regression.quantile_regression.QuantRegResults]:78 """Estimate quantile regressions at multiple quantiles."""79 if taus is None:80 taus = DEFAULT_TAUS81 results: Dict[float, sm.regression.quantile_regression.QuantRegResults] = {}82 for tau in sorted(taus):83 try:84 res = estimate_quantile(df, y_col, x_cols, tau, max_iter=max_iter)85 results[tau] = res86 except Exception as exc:87 logger.error("Quantile tau=%.2f failed: %s", tau, exc)88 logger.info("Estimated %d / %d quantile regressions.", len(results), len(taus))89 return results909192# ---------------------------------------------------------------------------93# Coefficient comparison across quantiles94# ---------------------------------------------------------------------------9596def compare_quantile_coefficients(97 results_dict: Dict[float, sm.regression.quantile_regression.QuantRegResults],98) -> pd.DataFrame:99 """Extract coefficients and CIs across quantiles into a long DataFrame."""100 records = []101 for tau, res in sorted(results_dict.items()):102 ci = res.conf_int()103 for var in res.params.index:104 records.append({105 "tau": tau,106 "variable": var,107 "coefficient": res.params[var],108 "std_error": res.bse[var],109 "p_value": res.pvalues[var],110 "ci_lower": ci.loc[var, 0],111 "ci_upper": ci.loc[var, 1],112 })113 coef_df = pd.DataFrame(records)114 logger.info(115 "Coefficient comparison: %d variables x %d quantiles.",116 coef_df["variable"].nunique(), coef_df["tau"].nunique(),117 )118 return coef_df119120121# ---------------------------------------------------------------------------122# Plotting123# ---------------------------------------------------------------------------124125def plot_quantile_coefficients(126 coef_df: pd.DataFrame,127 variables: List[str],128 save_path: Optional[str] = None,129 figsize: Tuple[int, int] = (14, 4),130 ols_results: Optional[object] = None,131) -> None:132 """Plot coefficient evolution across quantiles with confidence bands."""133 try:134 import matplotlib.pyplot as plt135 except ImportError:136 logger.error("matplotlib is required for plotting.")137 return138139 n_vars = len(variables)140 if n_vars == 0:141 logger.warning("No variables specified for plotting.")142 return143144 n_cols = min(3, n_vars)145 n_rows = int(np.ceil(n_vars / n_cols))146 fig, axes = plt.subplots(n_rows, n_cols, figsize=(figsize[0], figsize[1] * n_rows))147 if n_vars == 1:148 axes = np.array([axes])149 axes = np.atleast_1d(axes).flatten()150151 for idx, var in enumerate(variables):152 ax = axes[idx]153 sub = coef_df[coef_df["variable"] == var].sort_values("tau")154 if sub.empty:155 ax.set_title(f"{var} (no data)")156 continue157158 taus = sub["tau"].values159 coefs = sub["coefficient"].values160 ci_lo = sub["ci_lower"].values161 ci_hi = sub["ci_upper"].values162163 ax.plot(taus, coefs, "o-", color="steelblue", linewidth=2, label="QR coef")164 ax.fill_between(taus, ci_lo, ci_hi, alpha=0.2, color="steelblue", label="95% CI")165166 if ols_results is not None and var in ols_results.params.index:167 ols_b = ols_results.params[var]168 ax.axhline(ols_b, color="firebrick", linestyle="--", linewidth=1, label="OLS")169170 ax.axhline(0, color="grey", linestyle=":", linewidth=0.5)171 ax.set_xlabel("Quantile (tau)")172 ax.set_ylabel("Coefficient")173 ax.set_title(var)174 ax.legend(fontsize=7)175176 # Hide unused axes177 for idx in range(n_vars, len(axes)):178 axes[idx].set_visible(False)179180 fig.tight_layout()181 if save_path:182 Path(save_path).parent.mkdir(parents=True, exist_ok=True)183 fig.savefig(save_path, dpi=200, bbox_inches="tight")184 logger.info("Quantile coefficient plot saved to %s", save_path)185 plt.close(fig)186187188# ---------------------------------------------------------------------------189# Segment effects summary190# ---------------------------------------------------------------------------191192def get_segment_effects(193 results_dict: Dict[float, sm.regression.quantile_regression.QuantRegResults],194) -> pd.DataFrame:195 """Summarize how effects differ across price segments (budget vs. luxury)."""196 coef_df = compare_quantile_coefficients(results_dict)197198 segments = {199 "budget (q10)": 0.10,200 "lower-mid (q25)": 0.25,201 "median (q50)": 0.50,202 "upper-mid (q75)": 0.75,203 "luxury (q90)": 0.90,204 }205206 available_taus = set(coef_df["tau"].unique())207 records = []208 variables = [v for v in coef_df["variable"].unique() if v != "const"]209210 for var in variables:211 var_data = coef_df[coef_df["variable"] == var]212 row: Dict[str, object] = {"variable": var}213214 for seg_name, tau in segments.items():215 if tau in available_taus:216 match = var_data[var_data["tau"] == tau]217 if not match.empty:218 b = match["coefficient"].values[0]219 p = match["p_value"].values[0]220 star = "***" if p < 0.001 else ("**" if p < 0.01 else ("*" if p < 0.05 else ""))221 row[seg_name] = f"{b:.4f}{star}"222 else:223 row[seg_name] = ""224 else:225 row[seg_name] = ""226227 # Heterogeneity indicator: range of coefficients228 coefs = var_data["coefficient"].values229 if len(coefs) >= 2:230 row["coef_range"] = round(float(np.max(coefs) - np.min(coefs)), 4)231 row["heterogeneity"] = (232 "high" if row["coef_range"] > 0.1233 else ("moderate" if row["coef_range"] > 0.03 else "low")234 )235 else:236 row["coef_range"] = np.nan237 row["heterogeneity"] = "n/a"238239 records.append(row)240241 summary = pd.DataFrame(records)242 logger.info("Segment effects summarised for %d variables.", len(summary))243 return summary244