为何SVR-GARCH模型预测条件波动率会输出恒定值?如何解决?
SVR-GARCH模型预测条件波动率出现恒定值的问题排查与解决
问题描述
参考Abdullah Karasan所著《Machine Learning for Financial Risk Management with Python: Algorithms for Modeling Risk》一书,实现SVR-GARCH模型预测条件波动率,但运行代码后,预测区间内的条件波动率预测值均为恒定重复值,尽管模型初始参数为随机设置。代码及输出如下:
原代码
import yfinance as yf from datetime import datetime, timedelta import pandas as pd import numpy as np from sklearn.svm import SVR from sklearn.model_selection import RandomizedSearchCV from scipy.stats import uniform as sp_rand from sklearn.preprocessing import StandardScaler # Select assets stock_name = ['AAPL'] end_date = datetime.today() start_date = end_date - timedelta(days = 365 * 25) # Download the prices prices = yf.download( stock_name, start = start_date, end = end_date, interval = '1d', )['Adj Close'] prices = prices.dropna() stock_name = ['Apple'] prices = prices.rename(stock_name[0], inplace = True) # Log returns returns = np.log(np.array(prices)[1:] / np.array(prices)[:-1]) # Forecasting horizon H = 146 returns_series = pd.Series(returns) realized_vol = returns_series.rolling(5).std() realized_vol = pd.DataFrame(realized_vol) realized_vol.reset_index(drop=True, inplace=True) returns_svm = pd.DataFrame(returns ** 2) X = pd.concat([realized_vol, returns_svm], axis=1, ignore_index=True) X = X[4:].copy() X = X.reset_index() X.drop('index', axis=1, inplace=True) realized_vol = realized_vol.dropna().reset_index() realized_vol.drop('index', axis=1, inplace=True) conditional_volatility = pd.DataFrame(index=prices.index[-H:], columns=['SVM Linear','SVM RBF','SVM Poly']) para_grid = {'gamma': sp_rand(0.1, 1), 'C': sp_rand(0.1, 10), 'epsilon': sp_rand(0.1, 1)} svr_lin = SVR(kernel='linear') clf = RandomizedSearchCV(svr_lin, para_grid) clf.fit(X[:-H], realized_vol.iloc[1:-(H-1)].values.reshape(-1,)) predict_svr_lin = clf.predict(X[-H:]) conditional_volatility['SVM Linear'] = predict_svr_lin svr_rbf = SVR(kernel='rbf') clf = RandomizedSearchCV(svr_rbf, para_grid) clf.fit(X[:-H], realized_vol.iloc[1:-(H-1)].values.reshape(-1,)) predict_svr_rbf = clf.predict(X[-H:]) conditional_volatility['SVM RBF'] = predict_svr_rbf svr_poly = SVR(kernel='poly') clf = RandomizedSearchCV(svr_poly, para_grid) clf.fit(X[:-H], realized_vol.iloc[1:-(H-1)].values.reshape(-1,)) predict_svr_poly = clf.predict(X[-H:]) conditional_volatility['SVM Poly'] = predict_svr_poly print(conditional_volatility)
原输出
[*********************100%%**********************] 1 of 1 completed SVM Linear SVM RBF SVM Poly Date 2024-01-09 0.168156 0.168156 0.138204 2024-01-10 0.168156 0.168156 0.138204 2024-01-11 0.168156 0.168156 0.138204 2024-01-12 0.168156 0.168156 0.138204 2024-01-16 0.168156 0.168156 0.138204 ... ... ... ... 2024-08-01 0.168156 0.168156 0.138204 2024-08-02 0.168156 0.168156 0.138204 2024-08-05 0.168156 0.168156 0.138204 2024-08-06 0.168156 0.168156 0.138204 2024-08-07 0.168156 0.168156 0.138204 [146 rows x 3 columns]
原因分析
- 数据对齐错误:原代码中X和目标变量
realized_vol的样本索引未正确匹配,导致模型学习到无意义的映射关系,最终输出常数。 - 特征未标准化:SVM对特征尺度极度敏感,未标准化的特征会让模型偏向于尺度较大的特征,甚至退化为常数预测。
- 参数搜索配置错误:线性SVM不需要
gamma参数,将其加入参数网格会干扰RandomizedSearchCV的搜索过程;同时部分参数范围设置不合理,导致模型无法找到有效参数。 - 预测逻辑错误(数据泄露):原代码直接使用
X[-H:]进行预测,但该部分包含了预测区间内的真实波动率数据(realized_vol),这是实际预测中无法获取的信息,且不符合GARCH模型递归预测的逻辑。
解决方法
1. 修正数据对齐
确保特征集X的每个样本对应t时刻的信息,目标变量是t+1时刻的波动率,保证样本一一对应。
2. 添加特征标准化
使用StandardScaler对训练集特征进行拟合,再转换训练集和测试集,避免数据泄露。
3. 优化参数搜索网格
为不同核函数的SVM设置对应的参数网格,移除无效参数(如线性SVM的gamma)。
4. 实现递归预测逻辑
由于预测H步时,后续时刻的realized_vol无法提前获取,需采用递归方式:用前一步的预测值作为下一步的输入特征。
修改后的完整代码
import yfinance as yf from datetime import datetime, timedelta import pandas as pd import numpy as np from sklearn.svm import SVR from sklearn.model_selection import RandomizedSearchCV from scipy.stats import uniform as sp_rand from sklearn.preprocessing import StandardScaler # 下载数据 stock_name = ['AAPL'] end_date = datetime.today() start_date = end_date - timedelta(days=365*25) prices = yf.download(stock_name, start=start_date, end=end_date, interval='1d')['Adj Close'] prices = prices.dropna().rename('Apple') # 计算对数收益率 returns = np.log(prices.shift(1)/prices).dropna() # 计算滚动实现波动率(5天窗口) realized_vol = returns.rolling(5).std().dropna() # 构建特征集:t时刻的realized_vol和t时刻的平方收益率,目标是t+1时刻的realized_vol X = pd.concat([realized_vol.shift(1), returns.shift(1)**2], axis=1).dropna() y = realized_vol.loc[X.index] # 划分训练集和预测起始点 H = 146 train_X = X.iloc[:-H] train_y = y.iloc[:-H] # 初始预测输入:训练集最后一个样本 current_X = X.iloc[-H:-H+1].values # 特征标准化 scaler = StandardScaler() train_X_scaled = scaler.fit_transform(train_X) current_X_scaled = scaler.transform(current_X) # 定义不同核的参数网格 param_grids = { 'linear': {'C': sp_rand(0.1, 10), 'epsilon': sp_rand(0.1, 1)}, 'rbf': {'C': sp_rand(0.1, 10), 'gamma': sp_rand(0.001, 1), 'epsilon': sp_rand(0.1, 1)}, 'poly': {'C': sp_rand(0.1, 10), 'gamma': sp_rand(0.001, 1), 'epsilon': sp_rand(0.1, 1), 'degree': [2,3]} } # 初始化预测结果DataFrame pred_index = prices.index[-H:] conditional_volatility = pd.DataFrame(index=pred_index, columns=['SVM Linear','SVM RBF','SVM Poly']) # 训练并递归预测线性SVM svr_lin = SVR(kernel='linear') clf_lin = RandomizedSearchCV(svr_lin, param_grids['linear'], n_iter=20, cv=3, random_state=42) clf_lin.fit(train_X_scaled, train_y.values) lin_preds = [] temp_X = current_X_scaled.copy() for _ in range(H): pred = clf_lin.predict(temp_X)[0] lin_preds.append(pred) # 更新下一个特征:用预测的波动率作为下一个realized_vol特征,平方收益率用最新的实际值(或递归预测,这里用实际值简化) next_return_sq = returns.loc[pred_index[_]]**2 if _ < len(returns) else lin_preds[-1]**2 temp_X = scaler.transform(np.array([[pred, next_return_sq]])) conditional_volatility['SVM Linear'] = lin_preds # 训练并递归预测RBF SVM svr_rbf = SVR(kernel='rbf') clf_rbf = RandomizedSearchCV(svr_rbf, param_grids['rbf'], n_iter=20, cv=3, random_state=42) clf_rbf.fit(train_X_scaled, train_y.values) rbf_preds = [] temp_X = current_X_scaled.copy() for _ in range(H): pred = clf_rbf.predict(temp_X)[0] rbf_preds.append(pred) next_return_sq = returns.loc[pred_index[_]]**2 if _ < len(returns) else rbf_preds[-1]**2 temp_X = scaler.transform(np.array([[pred, next_return_sq]])) conditional_volatility['SVM RBF'] = rbf_preds # 训练并递归预测Poly SVM svr_poly = SVR(kernel='poly') clf_poly = RandomizedSearchCV(svr_poly, param_grids['poly'], n_iter=20, cv=3, random_state=42) clf_poly.fit(train_X_scaled, train_y.values) poly_preds = [] temp_X = current_X_scaled.copy() for _ in range(H): pred = clf_poly.predict(temp_X)[0] poly_preds.append(pred) next_return_sq = returns.loc[pred_index[_]]**2 if _ < len(returns) else poly_preds[-1]**2 temp_X = scaler.transform(np.array([[pred, next_return_sq]])) conditional_volatility['SVM Poly'] = poly_preds print(conditional_volatility)
关键说明
- 递归预测中,平方收益率可以选择用实际值(若在预测时能获取当日收益率)或递归预测(用前一步的波动率平方代替),根据实际场景调整。
- 增加了
random_state保证参数搜索的可复现性,n_iter和cv参数可根据计算资源调整。
内容的提问来源于stack exchange,提问作者Barbab
相关产品推荐
相关产品推荐

