综合python案例，VAR介入概率能预测吗？

wen python案例 2026-08-06 1

本文目录导读：

核心认知：VAR模型能做什么，不能做什么
综合Python实战案例
最终结论

这是一个非常经典且深刻的金融计量问题,简单直接的回答是：VAR模型本身不能直接预测“介入概率”，但它是构建“介入概率”预测模型的核心基础设施。

要理解这一点,我们需要先厘清“介入”和“VAR”的关系，然后我会给出一个完整的Python实战案例。

核心认知：VAR模型能做什么，不能做什么

VAR（向量自回归）模型：它擅长预测连续型数值，比如明天的收益率是0.5%还是1.2%，它通过历史数据的线性关系，预测变量未来的具体数值。
介入概率（或干预概率）：这是一个二分类问题（介入/不介入）或生存分析问题（何时介入），它输出的是0到1之间的概率，明天央行降息的概率是70%”。

VAR不能直接输出概率。但是，我们可以用VAR做两件事来间接预测概率：

压力测试：用VAR预测未来一段时间的宏观指标（如CPI、GDP、失业率）的路径，如果预测出的数值跌破某条“政策红线”，我们就认为介入概率极高。
特征提取器：VAR模型可以生成比原始数据更平滑、更具前瞻性的“结构性因子”（如脉冲响应值、预测误差方差分解），把这些因子作为特征，喂给机器学习分类器（如逻辑回归、XGBoost）来预测介入概率。

综合Python实战案例

场景设定：假设我们要预测央行是否会在下个月进行“降准”（货币政策介入）。逻辑：

使用VAR模型预测未来3个月的 CPI（通胀）、PMI（制造业景气） 和 M2（货币供应）。
设定介入规则：如果未来预测的CPI持续低于0（通缩），同时PMI跌破荣枯线50，且M2增速未见起色，则判定“高概率介入”。
我们把VAR预测的残差和预测值作为特征，训练一个机器学习模型来学习历史数据中“政策介入”的模式。

我们将使用真实的宏观经济数据集（如 statsmodels 内置的或模拟数据）来完成。

第一步：导入库与数据准备

我们使用 pandas 和 numpy 构建一个模拟但逼真的宏观数据集，模拟CPI、PMI和M2的月度数据，并生成“干预”标签。

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from statsmodels.tsa.api import VAR
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, roc_auc_score
import warnings
warnings.filterwarnings('ignore')
# 设置随机种子
np.random.seed(42)
# 生成200个月的数据（约17年）
n_obs = 200
dates = pd.date_range(start='2008-01-01', periods=n_obs, freq='ME')
# 模拟真实宏观数据（带有趋势和周期）
cpi = np.cumsum(np.random.randn(n_obs) * 0.05 + 0.02)  # 累计波动，模拟通胀路径
# 修正CPI使其围绕2%波动，偶尔进入负值
cpi = 2 + cpi + np.sin(np.arange(n_obs)/12) * 0.5
pmi = 50 + np.cumsum(np.random.randn(n_obs) * 0.1) * 0.1 + np.cos(np.arange(n_obs)/10) * 1.5
pmi = np.clip(pmi, 42, 58)  # 限制在40-60之间
m2 = np.cumsum(np.random.randn(n_obs) * 0.2 + 0.5) / 100 + 1.2
# 模拟M2同比增速在8%-15%之间
m2 = 10 + m2 * 5
m2 = np.clip(m2, 6, 16)
# 构建DataFrame
df = pd.DataFrame({'CPI': cpi, 'PMI': pmi, 'M2_YoY': m2}, index=dates)
# 生成介入标签 (y)：手动设定规则——当上3个月CPI均值<1.5且PMI<50时，下个月介入=1
df['Intervention'] = 0
for i in range(3, len(df)):
    if df['CPI'][i-3:i].mean() < 1.5 and df['PMI'][i-3:i].mean() < 50:
        df.loc[df.index[i], 'Intervention'] = 1
print(df.head(10))
print(f"\n介入事件占总样本比例：{df['Intervention'].mean():.2%}")
# 绘制原始数据
fig, ax = plt.subplots(3, 1, figsize=(12, 9))
ax[0].plot(df.index, df['CPI'], color='red'); ax[0].set_title('CPI (%)')
ax[0].axhline(y=1.5, linestyle='--', alpha=0.5); ax[0].axhline(y=0, linestyle='--', alpha=0.3)
ax[1].plot(df.index, df['PMI'], color='blue'); ax[1].set_title('PMI (Index)')
ax[1].axhline(y=50, linestyle='--', alpha=0.5)
ax[2].plot(df.index, df['M2_YoY'], color='green'); ax[2].set_title('M2 Growth YoY (%)')
plt.tight_layout()
plt.show()

第二步：训练VAR模型并提取预测特征

这是核心步骤,我们不直接用VAR输出概率，而是用VAR获取预测值和残差，这些作为新特征，代表“市场预期与现实的偏差”和“未来潜在压力”。

# 选择变量
data_endog = df[['CPI', 'PMI', 'M2_YoY']]
# 训练VAR模型（使用前150个数据）
train_size = 150
data_train = data_endog.iloc[:train_size]
data_test = data_endog.iloc[train_size:]
# 确定最优滞后阶数（简化选择，实际可用AIC）
model = VAR(data_train)
lag_order = model.select_order(maxlags=10)
optimal_lag = lag_order.aic  # 获取AIC最小阶数
print(f"\n最优滞后阶数: {optimal_lag}")
# 拟合VAR模型
var_model = model.fit(optimal_lag)
print(var_model.summary())
# 进行多步预测（预测未来6个月）
forecast_steps = 6
forecast = var_model.forecast(data_train.values[-optimal_lag:], steps=forecast_steps)
forecast_df = pd.DataFrame(forecast, columns=data_endog.columns, index=pd.date_range(start=data_endog.index[train_size], periods=forecast_steps, freq='ME'))
print("\nVAR预测的下6个月数据：")
print(forecast_df.head())

第三步：构建动态因子集（滚动窗口）

为了预测概率,我们不仅要当前数据，还要 “距离红线的距离” 和 “VAR预测的偏离度”。

# 创建特征框
features = []
labels = []
X_data = data_endog.copy()
# 滚动窗口构建特征
lookback = 6  # 回看6个月
for t in range(train_size, len(df)):
    # 当前窗口数据
    window_data = X_data.iloc[t-lookback:t]
    current = df.iloc[t]
    # 特征1：当前值
    cpi_now = current['CPI']
    pmi_now = current['PMI']
    m2_now = current['M2_YoY']
    # 特征2：过去3个月均值
    cpi_ma3 = window_data['CPI'].tail(3).mean()
    pmi_ma3 = window_data['PMI'].tail(3).mean()
    # 特征3：距离红线的距离（政策压力）
    cpi_distance = cpi_now - 0.0  # 距离通缩线
    pmi_distance = pmi_now - 50.0 # 距离荣枯线
    # 特征4：随时间变化的压力累积
    cpi_below_zero = (window_data['CPI'] < 0).sum()  # 过去6个月低于0的月数
    pmi_below_50 = (window_data['PMI'] < 50).sum()
    # 特征5：VAR预测的压力（类似脉冲响应）
    # 模拟：预测未来3个月的均值压力
    future_stress_cpi = forecast_df.loc[:forecast_df.index[train_size + 2], 'CPI'].mean() if t < len(df) - 6 else cpi_now
    # 组合特征
    feat = [
        cpi_now, pmi_now, m2_now,
        cpi_ma3, pmi_ma3,
        cpi_distance, pmi_distance,
        cpi_below_zero, pmi_below_50,
        future_stress_cpi,
        1 if cpi_now < 0 else 0,  # 是否处于通缩
        1 if pmi_now < 50 else 0  # 是否处于收缩
    ]
    features.append(feat)
    labels.append(df['Intervention'].iloc[t])
# 转换为DataFrame
X = pd.DataFrame(features, columns=[
    'cpi_now', 'pmi_now', 'm2_now',
    'cpi_ma3', 'pmi_ma3',
    'cpi_distance', 'pmi_distance',
    'cpi_below_zero', 'pmi_below_50',
    'future_stress_cpi',
    'is_deflation', 'is_contraction'
])
y = pd.Series(labels)
# 划分训练集与测试集（按时间）
split_idx = int(len(X) * 0.8)
X_train, X_test = X.iloc[:split_idx], X.iloc[split_idx:]
y_train, y_test = y.iloc[:split_idx], y.iloc[split_idx:]
print(f"\n特征形状: {X.shape}")
print(f"训练集介入比例: {y_train.mean():.2%}")
print(f"测试集介入比例: {y_test.mean():.2%}")

第四步：使用机器学习模型预测介入概率

这里使用 随机森林（Random Forest） 来处理非线性关系，并输出概率。

# 训练分类器
clf = RandomForestClassifier(
    n_estimators=200,
    max_depth=5,
    min_samples_leaf=2,
    random_state=42,
    class_weight='balanced'  # 平衡类别
)
clf.fit(X_train, y_train)
# 预测概率
y_pred_prob = clf.predict_proba(X_test)[:, 1]
y_pred = (y_pred_prob > 0.5).astype(int)
# 评估
print("\n===== 模型评估 =====")
print(f"准确率: {accuracy_score(y_test, y_pred):.3f}")
print(f"AUC Score: {roc_auc_score(y_test, y_pred_prob):.3f}")
# 特征重要性
importance = pd.Series(clf.feature_importances_, index=X.columns).sort_values(ascending=False)
print("\n特征重要性排序：")
print(importance.head(10))

第五步：结果可视化与解读

# 绘制实际介入与预测概率对比
plt.figure(figsize=(14, 6))
plt.plot(range(len(y_test)), y_test.values, 'o-', label='Actual Intervention', alpha=0.7)
plt.plot(range(len(y_test)), y_pred_prob, 's--', label='Predicted Probability', alpha=0.8)
plt.axhline(y=0.5, color='gray', linestyle=':')'Intervention Probability Prediction using VAR + ML')
plt.xlabel('Time (Test Period)')
plt.ylabel('Probability / Label')
plt.legend()
plt.grid(alpha=0.3)
plt.show()