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 best_lag = min(result.keys(), key=lambda k: result[k][0]["ssr_ftest"][1])37 f_stat = result[best_lag][0]["ssr_ftest"][0]38 p_val = result[best_lag][0]["ssr_ftest"][1]39 f_mat.loc[caused, causing] = f_stat40 p_mat.loc[caused, causing] = p_val41 except Exception:42 pass4344 return f_mat, p_mat454647def estimate_var(48 data: pd.DataFrame,49 max_lags: int = 10,50 ic: str = "aic",51) -> dict:52 """Estimate VAR model with optimal lag selection.5354 Returns dict with 'model', 'optimal_lag', 'coefficients', 'irf', 'summary'.55 """56 clean = data.dropna()57 if len(clean) < max_lags + 10:58 return {"model": None, "optimal_lag": 0, "coefficients": {}, "irf": {}}5960 model = VAR(clean)61 result = model.fit(maxlags=max_lags, ic=ic)6263 coefficients = {}64 params = result.params65 if isinstance(params, pd.DataFrame):66 for col in params.columns:67 coefficients[col] = params[col]68 else:69 for col_idx, col in enumerate(data.columns):70 coefficients[col] = pd.Series(params[:, col_idx])7172 irf_data = {}73 try:74 irf_result = result.irf(periods=20)75 for i, shock in enumerate(data.columns):76 for j, resp in enumerate(data.columns):77 irf_data[(shock, resp)] = irf_result.irfs[:, j, i]78 except Exception:79 pass8081 return {82 "model": result,83 "optimal_lag": result.k_ar,84 "coefficients": coefficients,85 "irf": irf_data,86 "summary": str(result.summary()),87 }888990def compute_impulse_responses(var_result: dict, periods: int = 20) -> dict:91 """Extract impulse response functions from VAR result."""92 if var_result.get("model") is None:93 return {}94 irf = var_result["model"].irf(periods=periods)95 data = {}96 names = var_result["model"].names97 for i, shock in enumerate(names):98 for j, resp in enumerate(names):99 data[(shock, resp)] = irf.irfs[:, j, i]100 return data101102103def lead_lag_regression(104 y: pd.Series,105 x: pd.Series,106 leads: range = range(-5, 6),107) -> pd.DataFrame:108 """Cross-correlation and lead-lag OLS regressions.109110 For each lag k: regress y_t on x_{t-k}.111 """112 results = []113 for k in leads:114 x_shifted = x.shift(k)115 valid = pd.DataFrame({"y": y, "x": x_shifted}).dropna()116 if len(valid) < 5:117 results.append({"lag": k, "coefficient": np.nan, "t_stat": np.nan, "r_squared": np.nan})118 continue119120 x_vals = valid["x"].values121 y_vals = valid["y"].values122 x_dm = x_vals - x_vals.mean()123 y_dm = y_vals - y_vals.mean()124 var_x = (x_dm ** 2).sum()125 if var_x == 0:126 results.append({"lag": k, "coefficient": np.nan, "t_stat": np.nan, "r_squared": np.nan})127 continue128129 beta = (x_dm * y_dm).sum() / var_x130 residuals = y_dm - beta * x_dm131 se_beta = np.sqrt((residuals ** 2).sum() / ((len(valid) - 2) * var_x))132 t_stat = beta / se_beta if se_beta > 0 else np.nan133 ss_res = (residuals ** 2).sum()134 ss_tot = (y_dm ** 2).sum()135 r_sq = 1 - ss_res / ss_tot if ss_tot > 0 else np.nan136137 results.append({"lag": k, "coefficient": beta, "t_stat": t_stat, "r_squared": r_sq})138139 return pd.DataFrame(results)140141142def hasbrouck_information_shares(143 prices: pd.DataFrame,144 max_lags: int = 5,145) -> pd.DataFrame:146 """Hasbrouck (1995) information shares via variance decomposition.147148 Uses VAR on returns and computes forecast error variance decomposition149 as a proxy for information shares when series may not be cointegrated.150 """151 returns = prices.pct_change().dropna()152 clean = returns.dropna()153 if len(clean) < max_lags + 10:154 return pd.DataFrame({"market": prices.columns, "info_share": np.nan})155156 try:157 model = VAR(clean)158 result = model.fit(maxlags=max_lags, ic="aic")159 fevd = result.fevd(periods=10)160 decomp = fevd.decomp161 shares = decomp[-1]162 info_shares = pd.DataFrame(163 shares,164 index=prices.columns,165 columns=prices.columns,166 )167 diag_shares = pd.Series(168 {col: info_shares.loc[col, col] for col in prices.columns},169 name="info_share",170 )171 return diag_shares.to_frame().reset_index().rename(columns={"index": "market"})172 except Exception:173 return pd.DataFrame({"market": prices.columns.tolist(), "info_share": [np.nan] * len(prices.columns)})174