【手撕代码-04】Python+金融数据,演示潜在结果框架的基本逻辑

import numpy as npimport pandas as pdimport akshare as ak# 获取数据HS_300 = ak.stock_zh_index_daily(symbol='sh000300')VIX = ak.index_option_300etf_qvix()# 合并为一个DataFramedata = pd.merge(HS_300, VIX, on='date').dropna()data = data[['date', 'close_x', 'close_y']]data.columns = ['date', 'HS300', 'VIX']data['HS300_ret'] = data['HS300'].pct_change()data["VIX"] = pd.to_numeric(data["VIX"], errors="coerce")data["VIX"] = data["VIX"].ffill()data['VIX_ret'] = data['VIX'].pct_change()data = data.dropna().reset_index()

np.random.seed(99)N = len(data)# 随机生成干预指示变量(50%的概率)data['D'] = np.random.binomial(1, 0.5, N)# 定义结果变量Y：当天的HS300收益率data['Y'] = data['HS300_ret']# 检查随机化是否成功（干预组与控制组在协变量上是否平衡）balance_check = data.groupby('D')[['HS300_ret', 'VIX_ret', 'VIX']].mean()print('随机化平衡性检查:')print(balance_check)

随机化平衡性检查:   HS300_ret   VIX_ret        VIXD                                0  -0.000004  0.001214  19.6622071   0.000341  0.004305  19.779037

from scipy.stats import ttest_indfor var in ['HS300_ret', 'VIX_ret', 'VIX']:    t, p = ttest_ind(data[data['D'] == 0][var], data[data['D'] == 1][var])    print(f'{var}: t={t:.3f}, p={p:.3f}')

HS300_ret: t=-0.572, p=0.568VIX_ret: t=-0.772, p=0.440VIX: t=-0.526, p=0.599

ate_obs = data[data['D']==1]['Y'].mean() - data[data['D']==0]['Y'].mean()print(f"观测到的干预组-控制组均值差(ATE)={ate_obs:.6f}")

观测到的干预组-控制组均值差(ATE)=0.000345

# 定义两个潜在结果（现实中不可同时观测）# 假设在没有干预（D=0）时的回报率就是实际的HS300_ret# 而如果有干预，回报率会受到一个固定处理效应tau的影响tau_true = 0.001  # 假设干预使得回报率每天增加0.1%# 潜在结果data['Y0'] = data['HS300_ret']data['Y1'] = data['Y0'] + tau_true# 观测结果由分配机制决定data['Y_obs'] = np.where(data['D'] == 1, data['Y1'], data['Y0'])# 平均因果效应（真实值）ate_true = data['Y1'].mean() - data['Y0'].mean()print(f"真实的ATE={ate_true:.6f}")# 观测到的均值差（估计值）ate_est = data[data['D']==1]['Y_obs'].mean() - data[data['D']==0]['Y_obs'].mean()print(f"估计的ATE={ate_est:.6f}")

# 制造非随机干预：当VIX变化大于0时，D=1（高波动）data['VIX_change'] = data['VIX'].diff()data['D_nonrandom'] = np.where(data['VIX_change']> 0, 1, 0)data = data.dropna()# 干预组treated = data[data['D_nonrandom'] == 1]# 控制组control = data[data['D_nonrandom'] == 0]# 观测到的均差值obs_diff = treated['HS300_ret'].mean() - control['HS300_ret'].mean()print(f"观测到的均差值（非随机干预）={obs_diff:.6f}")