Python 56.2%
TeX 43.7%
1"""Information shares, Granger causality, and VAR-based price discovery."""23from __future__ import annotations45import numpy as np6import pandas as pd7from scipy import stats8from statsmodels.tsa.api import VAR9from statsmodels.tsa.stattools import grangercausalitytests, adfuller101112def granger_causality_matrix(13 data: pd.DataFrame,14 max_lag: int = 10,15) -> tuple[pd.DataFrame, pd.DataFrame]:16 """Pairwise Granger causality F-tests.1718 Returns (f_stats, p_values) DataFrames indexed by caused, columns by causing.19 """20 cols = data.columns.tolist()21 f_mat = pd.DataFrame(np.nan, index=cols, columns=cols)22 p_mat = pd.DataFrame(np.nan, index=cols, columns=cols)2324 clean = data.dropna()25 if len(clean) < max_lag + 5:26 return f_mat, p_mat2728 for caused in cols:29 for causing in cols:30 if caused == causing:31 continue32 try:33 result = grangercausalitytests(34 clean[[caused, causing]].values, maxlag=max_lag, verbose=False35 )36 # Lag chosen by the AIC of the unrestricted bivariate regression37 # (revision 2026-10-17). The earlier code selected the lag with38 # the smallest p-value, a search that inflates rejection rates.39 best_lag = min(result.keys(), key=lambda k: result[k][1][1].aic)40 f_stat = result[best_lag][0]["ssr_ftest"][0]41 p_val = result[best_lag][0]["ssr_ftest"][1]42 f_mat.loc[caused, causing] = f_stat43 p_mat.loc[caused, causing] = p_val44 except Exception:45 pass4647 return f_mat, p_mat484950def estimate_var(51 data: pd.DataFrame,52 max_lags: int = 10,53 ic: str = "aic",54) -> dict:55 """Estimate VAR model with optimal lag selection.5657 Returns dict with 'model', 'optimal_lag', 'coefficients', 'irf', 'summary'.58 """59 clean = data.dropna()60 if len(clean) < max_lags + 10:61 return {"model": None, "optimal_lag": 0, "coefficients": {}, "irf": {}}6263 model = VAR(clean)64 result = model.fit(maxlags=max_lags, ic=ic)6566 coefficients = {}67 params = result.params68 if isinstance(params, pd.DataFrame):69 for col in params.columns:70 coefficients[col] = params[col]71 else:72 for col_idx, col in enumerate(data.columns):73 coefficients[col] = pd.Series(params[:, col_idx])7475 irf_data = {}76 try:77 irf_result = result.irf(periods=20)78 for i, shock in enumerate(data.columns):79 for j, resp in enumerate(data.columns):80 irf_data[(shock, resp)] = irf_result.irfs[:, j, i]81 except Exception:82 pass8384 return {85 "model": result,86 "optimal_lag": result.k_ar,87 "coefficients": coefficients,88 "irf": irf_data,89 "summary": str(result.summary()),90 }919293def compute_impulse_responses(var_result: dict, periods: int = 20) -> dict:94 """Extract impulse response functions from VAR result."""95 if var_result.get("model") is None:96 return {}97 irf = var_result["model"].irf(periods=periods)98 data = {}99 names = var_result["model"].names100 for i, shock in enumerate(names):101 for j, resp in enumerate(names):102 data[(shock, resp)] = irf.irfs[:, j, i]103 return data104105106def lead_lag_regression(107 y: pd.Series,108 x: pd.Series,109 leads: range = range(-5, 6),110) -> pd.DataFrame:111 """Cross-correlation and lead-lag OLS regressions.112113 For each lag k: regress y_t on x_{t-k}.114 """115 results = []116 for k in leads:117 x_shifted = x.shift(k)118 valid = pd.DataFrame({"y": y, "x": x_shifted}).dropna()119 if len(valid) < 5:120 results.append({"lag": k, "coefficient": np.nan, "t_stat": np.nan, "r_squared": np.nan})121 continue122123 x_vals = valid["x"].values124 y_vals = valid["y"].values125 x_dm = x_vals - x_vals.mean()126 y_dm = y_vals - y_vals.mean()127 var_x = (x_dm ** 2).sum()128 if var_x == 0:129 results.append({"lag": k, "coefficient": np.nan, "t_stat": np.nan, "r_squared": np.nan})130 continue131132 beta = (x_dm * y_dm).sum() / var_x133 residuals = y_dm - beta * x_dm134 se_beta = np.sqrt((residuals ** 2).sum() / ((len(valid) - 2) * var_x))135 t_stat = beta / se_beta if se_beta > 0 else np.nan136 ss_res = (residuals ** 2).sum()137 ss_tot = (y_dm ** 2).sum()138 r_sq = 1 - ss_res / ss_tot if ss_tot > 0 else np.nan139140 results.append({"lag": k, "coefficient": beta, "t_stat": t_stat, "r_squared": r_sq})141142 return pd.DataFrame(results)143144145def hasbrouck_information_shares(146 prices: pd.DataFrame,147 max_lags: int = 5,148) -> pd.DataFrame:149 """Hasbrouck (1995) information shares via variance decomposition.150151 Uses VAR on returns and computes forecast error variance decomposition152 as a proxy for information shares when series may not be cointegrated.153 """154 returns = prices.pct_change().dropna()155 clean = returns.dropna()156 if len(clean) < max_lags + 10:157 return pd.DataFrame({"market": prices.columns, "info_share": np.nan})158159 try:160 model = VAR(clean)161 result = model.fit(maxlags=max_lags, ic="aic")162 fevd = result.fevd(periods=10)163 decomp = fevd.decomp164 shares = decomp[-1]165 info_shares = pd.DataFrame(166 shares,167 index=prices.columns,168 columns=prices.columns,169 )170 diag_shares = pd.Series(171 {col: info_shares.loc[col, col] for col in prices.columns},172 name="info_share",173 )174 return diag_shares.to_frame().reset_index().rename(columns={"index": "market"})175 except Exception:176 return pd.DataFrame({"market": prices.columns.tolist(), "info_share": [np.nan] * len(prices.columns)})177