時間序列計量經(jīng)濟(jì)學(xué)中的主要因果推斷方法總結(jié)
作者：Kyle Jones
經(jīng)濟(jì)時間序列中的因果關(guān)系研究已超越了傳統(tǒng)的相關(guān)性分析范疇。本文系統(tǒng)性地探討了時間序列經(jīng)濟(jì)數(shù)據(jù)中因果關(guān)系的識別與量化方法，涵蓋從經(jīng)典的格蘭杰因果檢驗到現(xiàn)代因果推斷技術(shù)的全方位論述。
經(jīng)濟(jì)時間序列中的因果關(guān)系研究已超越了傳統(tǒng)的相關(guān)性分析范疇。本文系統(tǒng)性地探討了時間序列經(jīng)濟(jì)數(shù)據(jù)中因果關(guān)系的識別與量化方法，涵蓋從經(jīng)典的格蘭杰因果檢驗到現(xiàn)代因果推斷技術(shù)的全方位論述。
格蘭杰因果檢驗
格蘭杰因果檢驗是評估時間序列預(yù)測能力的基礎(chǔ)性計量工具，用于檢驗一個時間序列對另一個時間序列的預(yù)測貢獻(xiàn)。
import numpy as np
import pandas as pd
import statsmodels.api as sm
from statsmodels.tsa.stattools import grangercausalitytests
from statsmodels.tsa.api import VAR
class GrangerAnalysis:
def __init__(self, data):
self.data = data
def test_granger_causality(self, variable1, variable2, max_lags=12):
"""執(zhí)行雙變量格蘭杰因果檢驗"""
data = pd.concat([self.data[variable1], self.data[variable2]], axis=1)
results = grangercausalitytests(data, maxlag=max_lags)
causality_results = pd.DataFrame(
index=range(1, max_lags + 1),
columns=['F-statistic', 'p-value']
)
for lag in range(1, max_lags + 1):
causality_results.loc[lag] = [
results[lag][0]['ssr_ftest'][0],
results[lag][0]['ssr_ftest'][1]
]
return causality_results
def plot_causality_results(self, results):
"""繪制各滯后階的顯著性檢驗結(jié)果"""
import matplotlib.pyplot as plt
plt.figure(figsize=(10, 6))
plt.plot(results.index, results['p-value'], marker='o')
plt.axhline(y=0.05, color='r', linestyle='--', label='5% significance')
plt.xlabel('Lag Order')
plt.ylabel('p-value')
plt.title('Granger Causality Test Results')
plt.legend()
plt.show()結(jié)構(gòu)向量自回歸模型（SVAR）
結(jié)構(gòu)向量自回歸模型通過引入經(jīng)濟(jì)理論支持的結(jié)構(gòu)性約束，對傳統(tǒng)向量自回歸（VAR）框架進(jìn)行了擴(kuò)展。
from statsmodels.tsa.api import VAR
class SVARModel:
def __init__(self, data):
self.data = data
self.var_model = None
self.svar_results = None
def fit(self, lags=1, A=None, B=None):
"""基于短期和長期約束擬合SVAR模型"""
self.var_model = VAR(self.data)
var_results = self.var_model.fit(lags)
if A is None:
A = np.eye(len(self.data.columns))
if B is None:
B = np.eye(len(self.data.columns))
self.svar_results = var_results.svar(A=A, B=B)
return self.svar_results
def impulse_response(self, periods=20):
"""估計脈沖響應(yīng)函數(shù)"""
return self.svar_results.irf(periods=periods)
def forecast_error_variance_decomposition(self, periods=20):
"""進(jìn)行預(yù)測誤差方差分解"""
return self.svar_results.fevd(periods=periods)局部投影法的因果分析
局部投影法為脈沖響應(yīng)估計提供了一種穩(wěn)健的非參數(shù)化方法，擺脫了傳統(tǒng)VAR模型的參數(shù)假設(shè)限制。
import statsmodels.api as sm
class LocalProjections:
def __init__(self, data):
self.data = data
def estimate_impulse_response(self, dependent_var, shock_var, controls=None, horizons=20):
"""采用局部投影法估計動態(tài)脈沖響應(yīng)"""
responses = []
confidence_intervals = []
for h in range(horizons + 1):
y = self.data[dependent_var].shift(-h)
X = self.data[[shock_var]]
if controls is not None:
X = pd.concat([X, self.data[controls]], axis=1)
X = sm.add_constant(X)
valid_idx = y.notna()
y = y[valid_idx]
X = X[valid_idx]
model = sm.OLS(y, X)
results = model.fit(cov_type='HAC', cov_kwds={'maxlags': h})
responses.append(results.params[shock_var])
confidence_intervals.append(results.conf_int().loc[shock_var])
return np.array(responses), np.array(confidence_intervals)合成控制法
合成控制法通過構(gòu)建最優(yōu)權(quán)重組合的對照組，為反事實因果分析提供了系統(tǒng)性的方法論框架。
from scipy.optimize import minimize
class SyntheticControl:
def __init__(self, data, treatment_unit, control_units, treatment_period, outcome_var):
self.data = data
self.treatment_unit = treatment_unit
self.control_units = control_units
self.treatment_period = treatment_period
self.outcome_var = outcome_var
def construct_synthetic_control(self):
"""構(gòu)建最優(yōu)權(quán)重的合成控制單元"""
pre_treatment = self.data[self.data.index < self.treatment_period]
def objective(weights):
synthetic = np.sum([
w * pre_treatment[self.outcome_var][pre_treatment.unit == u]
for w, u in zip(weights, self.control_units)
], axis=0)
treated = pre_treatment[self.outcome_var][
pre_treatment.unit == self.treatment_unit
]
return np.mean((treated - synthetic) ** 2)
constraints = [
{'type': 'eq', 'fun': lambda x: np.sum(x) - 1},
{'type': 'ineq', 'fun': lambda x: x}
]
result = minimize(
objective,
x0=np.ones(len(self.control_units)) / len(self.control_units),
constraints=constraints
)
return result.x多期雙重差分法
雙重差分法（DiD）是面板數(shù)據(jù)處理效應(yīng)分析的核心計量方法，特別適用于政策評估研究。
import statsmodels.api as sm
class DynamicDiD:
def __init__(self, data):
self.data = data
def estimate_dynamic_effects(self, outcome_var, treatment_var, unit_fe=True, time_fe=True):
"""估計動態(tài)處理效應(yīng)參數(shù)"""
leads_lags = range(-4, 5)
for t in leads_lags:
self.data[f'treat_t{t}'] = self.data[treatment_var].shift(-t)
formula = f"{outcome_var} ~ " + " + ".join([f"treat_t{t}" for t in leads_lags])
if unit_fe:
formula += " + EntityEffects"
if time_fe:
formula += " + TimeEffects"
model = sm.PanelOLS.from_formula(formula, data=self.data)
return model.fit(cov_type='clustered', cluster_entity=True)時間序列工具變量方法
工具變量方法通過引入滿足相關(guān)性和外生性條件的工具變量，為解決內(nèi)生性問題提供了可靠的計量框架。
class TSInstrumentalVariables:
def __init__(self, data):
self.data = data
def estimate_iv(self, dependent_var, endogenous_var, instrument_var, controls=None):
"""實施兩階段最小二乘估計"""
X_first = sm.add_constant(self.data[instrument_var])
if controls is not None:
X_first = pd.concat([X_first, self.data[controls]], axis=1)
first_stage = sm.OLS(self.data[endogenous_var], X_first).fit()
fitted_values = first_stage.predict()
X_second = sm.add_constant(fitted_values)
if controls is not None:
X_second = pd.concat([X_second, self.data[controls]], axis=1)
second_stage = sm.OLS(self.data[dependent_var], X_second).fit()
return first_stage, second_stage現(xiàn)代因果推斷方法
現(xiàn)代因果推斷方法整合了計量經(jīng)濟(jì)學(xué)的理論基礎(chǔ)與機(jī)器學(xué)習(xí)的算法優(yōu)勢，提供了更具穩(wěn)健性和可解釋性的分析框架。
from sklearn.ensemble import RandomForestRegressor
class ModernCausalInference:
def __init__(self, data):
self.data = data
def double_machine_learning(self, y, d, x, ml_model=None):
"""實現(xiàn)雙機(jī)器學(xué)習(xí)的因果參數(shù)估計"""
if ml_model is None:
ml_model = RandomForestRegressor(n_estimators=100)
ml_model.fit(self.data[x], self.data[d])
d_hat = ml_model.predict(self.data[x])
ml_model.fit(self.data[x], self.data[y])
y_hat = ml_model.predict(self.data[x])
treatment_effect = sm.OLS(
self.data[y] - y_hat,
self.data[d] - d_hat
).fit()
return treatment_effect總結(jié)
本文系統(tǒng)性地探討了時間序列計量經(jīng)濟(jì)學(xué)中的因果推斷方法體系。從經(jīng)典的格蘭杰因果檢驗入手，詳細(xì)闡述了結(jié)構(gòu)向量自回歸模型（SVAR）、局部投影法等基礎(chǔ)方法框架。文章深入探討了合成控制法在構(gòu)建反事實分析中的應(yīng)用，以及多期雙重差分法在面板數(shù)據(jù)處理效應(yīng)評估中的實現(xiàn)。在此基礎(chǔ)上，介紹了時間序列工具變量方法對內(nèi)生性問題的處理思路，并探討了包括雙機(jī)器學(xué)習(xí)在內(nèi)的現(xiàn)代因果推斷方法。通過結(jié)合傳統(tǒng)計量方法與現(xiàn)代機(jī)器學(xué)習(xí)技術(shù)，構(gòu)建了一個完整的時序因果推斷方法論體系，為相關(guān)領(lǐng)域的實證研究提供了系統(tǒng)的方法論指導(dǎo)。
責(zé)任編輯：華軒
來源：
DeepHub IMBA
