作者：老余捞鱼
原创不易，转载请标明出处及原作者。
写在前面的话：传统的验证指标（如 RMSE、MSE 或分位数损失）主要用于衡量预测中的平均误差幅度。然而，在训练用于股票市场预测的深度学习模型时，它们在提供对交易风险和回报的有意义的见解方面存在不足。在预测股票市场时，它们没有清楚地说明所涉及的时间或波动性，它们不适合交易者的需求。这就是为什么在深度学习模型中开发用于股票预测的自定义验证指标变得很重要的原因。
以下是本文容的快速摘要：
- 开发自定义验证指标的重要性 — 为什么标准指标达不到要求，以及创建适合股票市场预测的定制验证指标的重要性
- 实施指南 — 有关实施自定义验证指标的详细分步说明。我们将使用 NeuralForecast 来实现该指标
- 训练模型 — 配置模型、准备数据并使用指标训练模型以进行股票市场预测。我们将使用 NeuralForecast 中的 Temporal Fusion Transformer （TFT） 实现
一、为什么要使用自定义验证指标？
在机器学习中，不同的训练损失函数和验证指标具有独特的用途。这种区别有助于我们捕捉学习过程的细节和模型的实际使用情况。
1.1 培训损失的作用
考虑一个分类任务。在训练过程中，我们经常使用交叉熵损失函数。为什么？因为交叉熵损失在比较类的预测概率分布与实际分布时非常有效。这种比较对于模型有效地从数据中学习至关重要。
但是为什么我们不能使用 F1 分数或准确性之类的东西进行训练呢？原因在于对可区分性的需要。深度学习中的训练损失函数必须是可微分的，以便优化算法通过梯度下降来调整模型的权重。像 F1-Score 这样的指标虽然非常适合评估最终表现，但无法区分。
1.2 验证指标的重要性
在验证过程中，我们重点关注模型在实际场景中的表现，在这些场景中，F1 分数等指标变得很重要。F1 分数对于分类任务特别有用，因为它同时考虑了精确度和召回率，从而提供了模型性能的平衡视图，尤其是在处理不平衡数据集的情况下。此外，与交叉熵损失相比，F1 分数更易于人类阅读和直观，因为它直接反映了模型正确分类阳性实例的能力，同时最大限度地减少假阳性和假阴性。让我们看一个使用垃圾邮件分类任务的示例。假设我们有一个二元分类任务来检测电子邮件是垃圾邮件（正类）还是非垃圾邮件（负类）。为简单起见，假设在测试数据集上评估模型后，我们有以下混淆矩阵：
从混淆矩阵中，我们计算出：
- 误报 （FP）：5（电子邮件被错误地识别为垃圾邮件）
- 漏报 （FN）：10 （垃圾邮件被错误地识别为非垃圾邮件）
使用以下值：
因此，F1 分数为：
这个 84% 的 F1 分数为模型的准确性提供了一个清晰、易懂的衡量标准。它突出了模型预测中精确度和召回率之间的平衡。
1.3 股市预测呢？
拥有自定义验证指标也适用于股票市场预测。对于多个时间序列问题，使用 MSE、RMSE 和分位数损失作为验证指标是完全可以接受的。它们提供了一种可靠的方法来评估预测模型的整体准确性。
但是，对于日内交易，使用均方误差 （MSE） 和均方根误差 （RMSE） 等指标可能会有问题。它们的信息量较少，也更难解释交易决策。这些指标主要用于衡量预测中的平均误差幅度，但它们并不能清楚地了解交易的时间或波动性。日内交易者对其交易的风险/回报更感兴趣。因此，虽然 RMSE 或 MSE 可用于训练，但夏普比率或索蒂诺比率等验证指标更相关。
例如，RMSE 可以最小化预测股票价格和实际股票价格之间的差异，但与此指标相关的交易没有风险或回报。相比之下，夏普比率既考虑了回报又考虑了风险，提供了更直观的交易表现衡量标准。如果模型预测高回报但风险高，交易者可能更喜欢略低的回报和显着降低的风险，夏普比率会强调这一点。
通过NeuralForecast的时间融合转换器（TFT）模型，并在NeuralForecast框架中创建自定义验证指标，我们可以确保我们的模型符合交易者的需求。这种方法有助于选择具有业务代表性的验证指标，使模型在股票预测中更加实用。
二、用于训练和测试模型的配置
我们几乎没有对 TFT 模型进行特征工程，也没有进行降维、交叉验证或超参数优化。主要目标是展示用于股票预测的自定义验证指标。
2.1 使用的数据
为了保持模型的简单性，我们将只关注一个单变量时间序列：SPY（SPDR S&P 500 ETF Trust）的每日回报率。该模型使用 2005 年至 2024 年的历史数据进行训练，涵盖训练期和测试期。我们使用每天的开盘价和收盘价来计算每日回报。这为模型增加了一层真实感。日内交易者可能会在早上开仓，并在一天结束时平仓。为简单起见，我们在初始模型中省略了佣金和滑点等因素，但这些因素可以很容易地纳入。
预测回报使我们能够处理稳态数据。回报通常是静止的——或者至少是微弱的静止——与价格不同，价格是非静止的。此外，虽然预测价格可能与策略取决于未来价格的期权交易者有关，但对于日内交易者来说，重点主要是资产可能产生的潜在回报，而不是其价格。仅仅知道价格，而没有额外的背景信息，如历史价格，对交易者来说几乎没有价值。
此外，我们的模型还结合了每日外生变量，例如 VIX（波动率指数），即 5 年盈亏平衡通货膨胀率，以提高我们的模型性能。
2.2 配置
下面是用于训练和测试模型的 YAML 配置。
start_date :'2005-07-01'
end_date :'2024-05-23'
max_missing_data :0.02
min_nb_trades :60
train_test_split : [0.9, 0.10]
val_proportion_size :0.15
output :
-source :'yahoo'
data :
-'SPY'
historic_variables:
-source:'fred'
data:
-'T5YIE'
-'T10YIE'
-'T10Y3M'
-'DGS10'
-'DGS2'
-'DTB3'
-'DEXUSNZ'
-'VIXCLS'
-'T10Y2Y'
-'NASDAQCOM'
-'DCOILWTICO'
-source:'yahoo'
data:
-"GC=F"
-'MSFT'
-'GOOGL'
-'AAPL'
-'AMZN'
future_variables :
-'day'
-'month'
TFT_parameters:
h:1
input_size:64
max_steps:500
val_check_steps:1
batch_size:32
inference_windows_batch_size:-1
valid_batch_size:2000
learning_rate:0.0005
scaler_type :'robust'
random_seed:42
loss:'HuberMQLoss'
hidden_size:256
n_head:8
dropout:0.1
attn_dropout :0.1
gradient_clip_val:1
other_parameters :
confidence_level:0.6
quantiles : [ 0.05, 0.4, 0.5, 0.6, 0.95 ]
callbacks :
EarlyStopping :
monitor :'valid_loss'
patience :20
verbose :True
mode :'min'
ModelCheckPoint :
monitor :'valid_loss'
mode :'min'
save_top_k :1
verbose :True- 该模型在 2005-07-01 到 2024-05-23（start_date 年和 end_date）期间使用每日交易日进行了训练和测试。
- max_missing_data ：0.02 — 任何给定的外生变量允许的缺失数据点的最大百分比。如果超过该值，我们将放弃该功能。
- min_nb_trades ：60 — 此变量设置验证和测试期间所需的最小交易数量。由于我们使用分位数的概率，因此有时我们可能不会进行交易。有关更多信息，请参阅下面的部分，为什么要使用最小交易数量？
- train_test_split ：[0.9， 0.10] — 数据比例分为训练集 （90%） 和测试集 （10%）。
- val_proportion_size ：0.15 — 为验证目的分配的训练数据比例 （15%）。15% * 90% = 总数据集的 13.5%
- 输出： 为了简单起见，我们预测了一个单一的时间序列：SPY（标准普尔500 ETF）。
- historic_variables ： 过去的外生变量（每日基础数据），可能会提高模型的性能。为了简单起见，使用的变量数量保持最少，因为模型是在没有 GPU 的本地计算机上训练和测试的。我们专注于最有用的变量。我们肯定可以增加外生变量的数量，并应用特征工程来提高模型性能。
- 圣路易斯联邦储备委员会经济数据（fred）中的一些例子：VIXCLS – VIX，衡量市场波动性的指标，T10Y3M – 10年减去3个月国债固定期限，T5YIE – 5年盈亏平衡通货膨胀率。来自雅虎财经（yahoo）：GC=F — 黄金期货每日价格，MSFT — Microsoft股票每日价格。这两个来源都可以免费使用。
- future_variables ： 预测时已知的未来外生变量，例如日和月。例如，如果我们认为投资组合经理在每个月底重新平衡他们的投资组合，这可能会影响当时 SPY 的每日回报率。考虑月底（例如，第 1 个月和第 31 天）等变量可以捕捉这些影响，并有助于做出更好的预测。
- TFT_parameters ： 有关 TFT 参数和超参数的定义，请参阅此页面。对于大多数超参数，我们使用了 NeuralForecast 提供的默认值。
- h ：1 — 预测范围，即模型预测的提前多少步。我们将其设置为 1，因为我们想要表现得像日内交易者。在每天结束时，我们想知道我们是否应该为第二天做多、做空或不做仓。这使得日常决策成为可能。
- input_size ：64 — 用作模型输入的过去时间步长数，相当于大约 3 个月的交易日。
- val_check_steps ：1 — 训练期间计算验证指标的频率（以步长为单位）。这是非常小的。对于更大的数据集和更复杂的模型（例如更大的hidden_size），它可以增加到 10、50、100，因为计算成本会很高。
- valid_batch_size ：2000 —验证期间的批处理大小设置为 2000，以确保在计算验证度量时一次处理所有验证数据。这是因为自定义验证指标的构建方式使我们只需要在一个批处理中传递所有验证数据。稍后会详细介绍。
- scaler_type ：’robust’ — 用于在训练迭代期间分别对每个输入窗口的数据进行归一化 （TemporalNorm） 的缩放器类型。在本例中，使用鲁棒定标器，这是用于时间融合转换器 （TFT） 的默认定标器。它对异常值有效，异常值在经济和金融时间序列中很常见。
- loss ：’HuberMQLoss’ — 用于训练的损失函数是具有多分位数回归的 Huber 损失。这种方法对异常值很可靠。Quantiles 通过提供某些结果的概率而不仅仅是点预测，提供了更现实的视角，尤其是在时间序列数据中。此外，它还允许使用分位数推断概率。
- gradient_clip_val ：1 — 梯度剪裁的最大值，以防止梯度爆炸。我们可以通过使用 TensorBoard 跟踪梯度来确认最佳值。
- confidence_level ：0.6 — 这是我们进入交易的最低水平。负回报或正回报的概率必须超过此水平;否则，我们不会在该数据点（或日期）进行交易。这个置信度设置得很低，因为模型很简单，而且想法是构建一个自定义验证指标。
- 分位数 ：[ 0.05， 0.4， 0.5， 0.6， 0.95 ] —用于预测的不同分位数。
- EarlyStopping：PyTorch Lightning 回调，用于监控自定义验证指标，如果损失在 20 个 epoch 内没有改善（耐心），将停止训练，最小化监控值（mode：’min’）。
- ModelCheckPoint：其他 PyTorch Lightning 回调，使用最佳自定义验证指标（模式：’min’）保存模型，仅保留性能最高的模型 （save_top_k：1）。
2.3 为什么要使用最小交易数量？
在推理过程中，预测分位数而不是点预测使我们能够评估特定每日回报的概率。例如，对于第 80 个百分位数，如果回报率为 -0.1%，我们可以预期当天回报率为 -0.1% 或以下的可能性为 80%。
这种方法使我们能够根据特定的阈值来持有头寸。例如，如果我们设定了 60% 的置信水平，而第 40 个百分位数的回报率为正，那么我们当天就会做多，预计有 60% 的机会获得正回报。另一方面，如果第 60 个百分位的回报率为负，我们将做空，预计有 60% 的机会出现负回报。如果这两个条件都不满足，避免头寸有助于防止不满足预期概率的不必要交易。注意我们可以在推理过程中推断概率，因为模型是用分位数损失函数训练的：多分位数 Huber 损失。
三、准备数据
3.1 获取数据
让我们考虑一个使用 FRED 的 5 年盈亏平衡通货膨胀率的例子。此处概述的步骤适用于其他变量，包括外生变量和目标变量 （SPY）。
第一部分是从源获取数据。
import pandas as pd
import yaml
withopen('custom_validation_metric/config.yaml', "r") as yaml_file:
config = yaml.load(yaml_file, Loader=yaml.FullLoader)
request = f"https://fred.stlouisfed.org/graph/fredgraph.csv?id=T5YIE"
request += f"&cosd={config['start_date']}"
request += f"&coed={config['end_date']}"
T5YIE_raw = pd.read_csv(request, parse_dates=['DATE'])
T5YIE_raw.rename(
columns={
'DATE': "ds",
T5YIE_raw.columns[1]: "value_T5YIE",
},
inplace=True,
)
print(T5YIE_raw.head(10))
根据 NeuralForecast 数据输入中的规定，必须命名日期戳列ds，Y_df是一个包含三列的 DataFrame：每个时间序列都有一个唯一标识符，一个列带有日期戳，一列包含序列的值。
清理数据
我们删除空数据，只保留市场开放的日子。如上图所示，2005 年 7 月 4 日（独立日）有一个空值。
import pandas_market_calendars as mcal
from typing importOptional
defobtain_market_dates(start_date: str, end_date: str, market : Optional[str] = "NYSE") -> pd.DataFrame:
nyse = mcal.get_calendar(market)
market_open_dates = nyse.schedule(
start_date=start_date,
end_date=end_date,
)
return market_open_dates
market_dates = obtain_market_dates(config['start_date'],config['end_date'])
T5YIE_correct_dates = T5YIE_raw.loc[
T5YIE_raw['ds'].isin(market_dates.index)
]
defreplace_empty_data(df : pd.DataFrame) -> pd.DataFrame:
mask = df.isin(["", ".", None])
rows_to_remove = mask.any(axis=1)
return df.loc[~rows_to_remove]
T5YIE_cleaned = replace_empty_data(T5YIE_correct_dates)
print(T5YIE_cleaned.head(10))
3.2 处理丢失的数据
对于任何外生变量，如果缺失数据量超过 （2%），则该变量将从模型训练中排除。如果该变量有足够的数据，我们将用前一个值替换任何缺失的数据。max_missing_data
from typing importUnion, Tuple
import logging
defhandle_missing_data(
data: pd.DataFrame,
market_open_dates : pd.DataFrame,
) -> Tuple[Union[None,pd.DataFrame], Union[pd.DataFrame, None]]:
modified_data = data.copy()
market_open_dates["count"] = 0
market_open_dates.index = market_open_dates.index.strftime("%Y-%m-%d")
date_counts = data['ds'].value_counts()
market_open_dates["count"] = market_open_dates.index.map(
date_counts
).fillna(0)
missing_dates = market_open_dates.loc[
market_open_dates["count"] < 1
]
ifnot missing_dates.empty:
max_count = (
len(market_open_dates)
* config["max_missing_data"]
)
iflen(missing_dates) > max_count:
logging.warning(
f"For current asset T5YIE there are "
f"{len(missing_dates)} missing data which is than the maximum threshold of "
f"{config['max_missing_data'] * 100}%"
)
returnNone, None
else:
for date, row in missing_dates.iterrows():
modified_data = insert_missing_date(
modified_data, date, 'ds'
)
return modified_data, missing_dates
definsert_missing_date(
data: pd.DataFrame,
date: str,
date_column: str
) -> pd.DataFrame:
date = pd.to_datetime(date)
if date notin data[date_column].values:
prev_date = (
data[data[date_column] < date].iloc[-1]
ifnot data[data[date_column] < date].empty
else data.iloc[0]
)
new_row = prev_date.copy()
new_row[date_column] = date
data = (
pd.concat([data, new_row.to_frame().T], ignore_index=True)
.sort_values(by=date_column)
.reset_index(drop=True)
)
return data
T5YIE_processed, missing_dates = handle_missing_data(T5YIE_cleaned,market_dates)
T5YIE_processed['ds'] = pd.to_datetime(T5YIE_processed['ds'])
sample_date = pd.to_datetime(missing_dates.index[0])
previous_day_data = T5YIE_processed[T5YIE_processed['ds'] < sample_date].tail(1)
missing_day_data = T5YIE_processed[T5YIE_processed['ds'] == sample_date]
combined_data = pd.concat([previous_day_data, missing_day_data])
print(f'\n{combined_data}\n')
print(T5YIE_processed.head(10))10 月 10 日，我们有一个缺失值。尽管哥伦布日是美国的联邦假日，但市场仍然开放。因此，我们将 2005 年 10 月 10 日的缺失值替换为前一个交易日（2005 年 10 月 7 日）的值。
3.3 拆分训练集和测试集
我们将每个变量分为训练集和测试集。
train_proportion = config['train_test_split']
split_index = int(len(T5YIE_processed ) * train_proportion[0])
train_T5YIE= T5YIE_processed.iloc[:split_index]
test_T5YIE = T5YIE_processed.iloc[split_index:]3.4 特性工程
在本节中，我们将以 Microsoft 股票价格为例执行特性工程。这同样适用于其他外生变量。我们创建了两个新功能：每日回报率和高低比率。我们希望这些功能能够提高模型性能。
train_MSFT['high_low'] = train_MSFT['high'] / train_MSFT['low'] - 1
train_MSFT['return'] = train_MSFT['close'] / train_MSFT['open'] - 1
test_MSFT['high_low'] = test_MSFT['high'] / test_MSFT['low'] - 1
test_MSFT['return'] = test_MSFT['close'] / test_MSFT['open'] - 1
print(f"\n{train_MSFT[['high_low', 'return']].head(10)}")
plt.figure(figsize=(10, 5))
plt.plot(train_MSFT['ds'], train_MSFT['high_low'], label='High-Low MSFT', color='blue')
plt.xlabel('Date')
plt.ylabel('High-Low MSFT')
plt.legend()
plt.grid(True)
plt.show()
plt.plot(train_MSFT['ds'], train_MSFT['return'], label='Return MSFT', color='green')
plt.xlabel('Date')
plt.ylabel('Return MSFT')
plt.legend()
plt.grid(True)
plt.show()
我们将为输出变量 SPY 创建相同的特征，包括开盘价、最高价、最低价、收盘价以及最高价和最低价之间的差值。但是，目标列 SPY 每日返回值需要标识符。我们还需要一个系列。yunique_id
SPY_processed['y'] = SPY_processed['close'] / SPY_processed['open'] - 1
SPY_processed = SPY_processed.copy()
SPY_processed['unique_id'] = 'SPY'
train_SPY= SPY_processed.iloc[:split_index]
test_SPY = SPY_processed.iloc[split_index:]接下来，我们将提取日和月作为未来的外生变量。我们只需要对一个变量执行此操作，因为我们的所有变量都共享相同的日期。
from sklearn.preprocessing import MinMaxScaler
scaler = MinMaxScaler()
scalers = {}
columns_to_scale = []
train_FUTURE = pd.DataFrame()
test_FUTURE = pd.DataFrame()
if'day'in config['future_variables']:
columns_to_scale.append('day')
train_FUTURE['day'] = train_MSFT['ds'].dt.day
test_FUTURE['day'] = test_MSFT['ds'].dt.day
scalers['day'] = MinMaxScaler()
if'month'in config['future_variables']:
columns_to_scale.append('month')
train_FUTURE['month'] = train_MSFT['ds'].dt.month
test_FUTURE['month'] = test_MSFT['ds'].dt.month
scalers['month'] = MinMaxScaler()
if columns_to_scale:
train_FUTURE['ds'] = train_MSFT['ds']
test_FUTURE['ds'] = test_MSFT['ds']
for column in columns_to_scale:
data_reshaped_train = train_FUTURE[column].values.reshape(-1, 1)
data_reshaped_test = test_FUTURE[column].values.reshape(-1, 1)
train_FUTURE[[column]] = scalers[column].fit(data_reshaped_train)
train_FUTURE[[column]] = scalers[column].transform(data_reshaped_train)
test_FUTURE[[column]] = scalers[column].transform(data_reshaped_test)
print(f"\n{train_FUTURE.head(10)}\n")
使用训练数据对 and where 进行缩放，然后对训练集和测试集应用相同的缩放。对于 和 ，值均使用 在 0 和 1 之间缩放。daymonthdaymonthMinMaxScaler
3.5 使数据保持静止
为了提高模型的鲁棒性，我们确保数据是静止的。我们将使用具有 95% 置信水平的增强 Dickey-Fuller 检验来确定连续时间序列是否是平稳的。我们将测试训练数据的平稳性，并将必要的转换应用于训练集和测试集。
让我们以一个外生变量为例：黄金连续合约期货。我们将只检查黄金的收盘价，但相同的方法适用于其他特征。
last_date = train_GOLD['ds'].iloc[-1].strftime('%Y-%m-%d')
plt.figure(figsize=(10, 5))
plt.plot(train_GOLD['ds'], train_GOLD['close'], label='Close Price')
plt.title('Gold Close Price from {} to {}'.format(start_date.date(), last_date))
plt.xlabel('Date')
plt.ylabel('Close Price')
plt.legend()
plt.grid(True)
plt.show()
我们可以观察到时间序列的平均值随着时间的推移不是恒定的，这表明它不是平稳的。
from statsmodels.tsa.stattools import adfuller
adf_result = adfuller(train_GOLD['close'])
p_value = adf_result[1]
print("\nADF Test p-value:", p_value)
ADF 结果确认：p 值大于 5%，表示时间序列为非平稳。
first_diff_GOLD = train_GOLD.copy()
first_diff_GOLD.loc[:, 'close_diff'] = first_diff_GOLD['close'].diff()
first_diff_GOLD = first_diff_GOLD.dropna()
plt.figure(figsize=(10, 5))
plt.plot(first_diff_GOLD['ds'], first_diff_GOLD['close_diff'], label='First Difference of Close Price')
plt.title('First Difference of Gold Close Price')
plt.xlabel('Date')
plt.ylabel('First Difference')
plt.legend()
plt.grid(True)
plt.show()
通过应用第一次微分，我们观察到时间序列的平均值接近 0，并且方差随时间基本保持不变。
adf_result_diff = adfuller(first_diff_GOLD['close_diff'])
p_value_diff = adf_result_diff[1]
print("\nADF Test p-value after first differencing:", p_value_diff)
ADF p 值为 0，表示时间序列在 95% 的置信水平上保持平稳。
3.6 对数据进行规范化
在训练模型之前，我们使数据保持静止。此外，我们在训练迭代期间使用了定标器，参数设置为 。scaler_typerobust
3.7 在训练模型之前对数据进行分组
根据 NeuralForecast 文档中的规定，我们需要列出历史和未来的外生变量。
要向模型添加外生变量，首先在初始化期间将前一个数据帧中的每个变量的名称指定到相应的模型超参数。
defrename_columns(df, suffix):
return df.rename(columns=lambda col: f"{col}_{suffix}"if col notin ['y','unique_id'] else col)
current_vars = locals().copy()
train_dfs = []
test_dfs = []
for var_name, df in current_vars.items():
ifisinstance(df, pd.DataFrame):
if var_name.startswith('train_'):
suffix = var_name.split('_')[1]
if suffix != 'FUTURE':
df = df.drop(columns='ds')
df = rename_columns(df, suffix)
train_dfs.append(df.reset_index(drop=True))
elif var_name.startswith('test_'):
suffix = var_name.split('_')[1]
if suffix != 'FUTURE':
df = df.drop(columns=['ds'])
df = rename_columns(df, suffix)
test_dfs.append(df.reset_index(drop=True))
neuralforecast_train_df = pd.concat(train_dfs, axis=1).reset_index(drop=True)
neuralforecast_test_df = pd.concat(test_dfs, axis=1).reset_index(drop=True)
futr_list = config['future_variables'] if config['future_variables'] elseNone
hist_list = [col for col in neuralforecast_test_df.columns if col notin futr_list and col
notin ["ds","time","y","unique_id"]]四、为交易决策实施自定义验证指标
NeuralForecast 中的所有模型都包含该参数，该参数允许我们定义一个自定义验证指标，并进行一些修改。valid_loss
NeuralForecast 中的大多数损失函数都继承自该类，例如 Huberized Multi-Quantile Loss。这些函数与构造函数和可调用方法共享相同的实现。唯一需要的参数是 ，张量格式的实际值，以及张量格式的预测值。该方法将损失值以张量的形式返回。BasePointLoss__init____call____call__yy_hat__call__
验证损失作为实例通过参数传递或设置为等于参数。因此，我们需要将验证指标作为实例传递到 。
#Sample code from NeuralForecast
classTFT(BaseWindows):
def__init__(
self,
loss=MAE(),
valid_loss=None,
**other_parameters)
super(TFT, self).__init__(
loss=loss,
valid_loss=valid_loss,
**other_parameters
)
...
classBaseWindows(BaseModel):
def__init__(
self,
loss,
valid_loss,
**other_parameters)
if valid_loss isNone:
self.valid_loss = loss
else:
self.valid_loss = valid_loss
...现在，我们需要实现一个自定义验证指标，该指标采用相同的参数并返回相同的值类型（张量）。
from neuralforecast.losses.pytorch import BasePointLoss, level_to_outputs, quantiles_to_outputs
from typing importOptional, Union
import torch
classRiskReturn(BasePointLoss):
def__init__(
self, level=[80, 90], quantiles=None, delta: float = 1.0, horizon_weight=None, config_manager : Optional [ConfigManager] = None,
):
withopen('custom_validation_metric/config.yaml', "r") as yaml_file:
self._config = yaml.load(yaml_file, Loader=yaml.FullLoader)
qs, output_names = level_to_outputs(level)
qs = torch.Tensor(qs)
if quantiles isnotNone:
_, output_names = quantiles_to_outputs(quantiles)
qs = torch.Tensor(quantiles)
super(RiskReturn, self).__init__(
horizon_weight=horizon_weight,
outputsize_multiplier=len(qs),
output_names=output_names,
)
self.quantiles = torch.nn.Parameter(qs, requires_grad=False)
self.delta = delta
self._lower_quantile = 1
self._upper_quantile = 3
def__call__(
self,
y: torch.Tensor,
y_hat: torch.Tensor,
mask: Union[torch.Tensor, None] = None,
):
daily_returns = MetricCalculation.calculate_daily_returns(y, y_hat, lower_quantile=self._lower_quantile, upper_quantile=self._upper_quantile)
metrics = MetricCalculation.get_risk_rewards_metrics(daily_returns)
if metrics['nb_of_trades'] < self._config['min_nb_trades']:
return torch.tensor(float('inf'))
return -return_on_risk对于每个数据点，表示实际值，而表示五个分位数中每个分位数的预测值。yy_hat
我们返回否定的，因为我们的目标是最小化验证损失。另一方面，如果执行的交易数量低于要求 （60），我们将返回一个无限值，以忽略其对跨时期训练过程的影响。return_on_riskmetrics['nb_of_trades']min_nb_trades
classMetricCalculation:
def__int__(self):
self._daily_returns = torch.empty(0)
self._metrics = {}
defcalculate_daily_returns(self,
y : torch.Tensor,
y_hat : torch.Tensor,
lower_quantile :int,
upper_quantile : int) -> torch.Tensor:
y_hat = y_hat.squeeze(1)
y = y.squeeze(1)
low_predictions = y_hat[:, lower_quantile]
high_predictions = y_hat[:, upper_quantile]
positive_condition = low_predictions > 0
negative_condition = high_predictions < 0
daily_returns = torch.full_like(y, float('-inf'))
daily_returns[positive_condition] = y[positive_condition]
daily_returns[negative_condition] = -y[negative_condition]
valid_returns = daily_returns[daily_returns != float('-inf')]
self._daily_returns = valid_returns
return self._daily_returns
...现在，我们评估所有预测值。如果任何较低分位数（第 40 个百分位数）值大于 0，我们预计当天将获得正回报并持有多头头寸。如果任何高分位数（第 60 个百分位数）值小于 0，我们持有空头头寸。y_hat
...
defget_risk_rewards(self,daily_returns : torch.Tensor = None):
if daily_returns isNone:
daily_returns =self._daily_returns
self._metrics = {}
self._metrics["nb_of_trades"] = daily_returns.shape[0]
if self._metrics["nb_of_trades"] <= self._config['min_nb_trades']:
return self._set_zero_to_metrics(self._metrics["nb_of_trades"])
self._metrics['annualized_return'] = torch.prod(1 + daily_returns) ** (
252.0 / daily_returns.shape[0]) - 1
self._metrics['annualized_risk'] = daily_returns.std() * (252 ** 0.5)
self._metrics['return_on_risk'] = self._metrics['annualized_return'] / self._metrics['annualized_risk']
return self._metrics
def_set_zero_to_metrics(self, nb_of_trades) -> Dict:
self._metrics = {
'annualized_return': torch.tensor(0.0),
'annualized_risk': torch.tensor(0.0),
'return_on_risk': torch.tensor(0.0),
'nb_of_trades' : nb_of_trades
}
return self._metrics这是我们的自定义验证指标。它的计算方式是除以（年化标准差），或者简单地是风险回报率。虽然我们可以使用 Sortino 或 Sharpe 比率来获得更高的精度，但这个指标对于我们的用例来说已经足够了。越高越好。annualized_returnannualized_riskreturn_on_riskreturn_on_risk
如果交易数量小于所需的最低要求，我们返回 0。否则，该函数将返回计算出的指标。min_nb_tradesreturn_on_risk
五、训练和测试模型
我们使用自定义验证指标训练 TFT 模型。
第 1 步：训练过程
from neuralforecast.models import TFT
from neuralforecast import NeuralForecast
from neuralforecast.losses.pytorch import HuberMQLoss
from pandas.tseries.offsets import CustomBusinessDay
from pytorch_lightning.loggers.tensorboard import TensorBoardLogger
from pytorch_lightning.callbacks import ModelCheckpoint, EarlyStopping
from tensorboard.backend.event_processing import event_accumulator
import os
import numpy as np
defcreate_custom_trading_days(start_date: str, end_date: str, market: str = "NYSE") -> CustomBusinessDay:
market_dates = obtain_market_dates(start_date, end_date, market)
trading_days = pd.DatetimeIndex(market_dates.index)
all_dates = pd.date_range(start=start_date, end=end_date, freq='B')
return CustomBusinessDay(holidays=all_dates.difference(trading_days))
defcreate_callbacks():
callbacks_list = []
for callback_name, params in config['other_parameters']['callbacks'].items():
if callback_name == 'EarlyStopping':
early_stopping = EarlyStopping(
monitor=params['monitor'],
patience=params['patience'],
verbose=params['verbose'],
mode=params['mode']
)
callbacks_list.append(early_stopping)
if callback_name == 'ModelCheckPoint':
model_checkpoint = ModelCheckpoint(
monitor=params['monitor'],
mode=params['mode'],
save_top_k=params['save_top_k'],
verbose=params['verbose']
)
callbacks_list.append(model_checkpoint)
return callbacks_list
defsave_metrics_from_tensorboard(logger_dir):
metrics_dict = {}
os.makedirs(f'custom_validation_metric/tensorboard', exist_ok=True)
for event_file in os.listdir(logger_dir):
ifnot event_file.startswith('events.out.tfevents'):
continue
full_path = os.path.join(logger_dir, event_file)
ea = event_accumulator.EventAccumulator(full_path)
ea.Reload()
for tag in ea.Tags()['scalars']:
metrics_dict[tag] = ea.Scalars(tag)
for metric, scalars in metrics_dict.items():
plt.figure(figsize=(10, 5))
if metric == 'train_loss_step':
steps = [scalar.step for scalar in scalars]
else:
steps = list(range(len(scalars)))
if metric == 'valid_loss'or metric == 'ptl/val_loss':
values = [scalar.value for scalar in scalars]
steps, values = zip(*[(step, value) for step, value inzip(steps, values) ifnot np.isinf(value)])
else:
values = [scalar.value for scalar in scalars]
plt.plot(steps, values, label=metric)
plt.xlabel('Steps'if metric == 'train_loss_step'else'Epoch')
plt.ylabel('Value')
plt.title(metric)
plt.legend(loc='upper right')
plt.savefig(f"custom_validation_metric/tensorboard/{metric.replace('/', '_')}.png")
plt.close()
keys_to_remove = {'loss'}
hyper_params = config['TFT_parameters']
other_param = config['other_parameters']
hyper_to_keep = {key: value for key, value in hyper_params.items() if key notin keys_to_remove}
logger = TensorBoardLogger('custom_validation_metric')
logger_dir = logger.log_dir
callbacks = create_callbacks()
nf = NeuralForecast(models=[TFT(
loss=HuberMQLoss(quantiles=other_param['quantiles']),
valid_loss=RiskReturn(),
**hyper_to_keep,
futr_exog_list=futr_list,
hist_exog_list=hist_list,
logger=logger,
callbacks=callbacks,
enable_model_summary=True,
enable_checkpointing=True,
enable_progress_bar=True,
)],
freq=create_custom_trading_days( start_date=config['start_date'],
end_date = config['end_date'])
)
val_size = int(len(neuralforecast_train_df)*config['val_proportion_size'])
nf.fit(df=neuralforecast_train_df, val_size=val_size, use_init_models=True)
save_metrics_from_tensorboard(logger_dir)NeuralForecast 建立在 PyTorch Lightning 之上，PyTorch Lightning 是一个 PyTorch 包装器。因此，它允许我们传递其他参数，例如使用 PyTorch Lightning 定义的 and。loggercallbacks
我们使用指定分位数作为训练损失。通过参数传递。HuberMQLoss()quantileRiskReturn()validation_loss
该函数确保模型只考虑交易日。交易日与熊猫的营业频率或每日频率不同。该函数使用适当的参数值设置回调。该函数从 TensorBoard 事件日志中提取每个时期的训练和验证损失值，并将其保存为图像。create_custom_trading_days()BDcreate_callbacks()save_metrics_from_tensorboard()
defvalidation_step(self, batch, batch_idx):
if self.val_size == 0:
return np.nan
windows = self._create_windows(batch, step="val")
n_windows = len(windows["temporal"])
y_idx = batch["y_idx"]
...
for i inrange(n_batches):
...
valid_loss_batch = self._compute_valid_loss(
outsample_y=original_outsample_y,
output=output_batch,
outsample_mask=outsample_mask,
temporal_cols=batch["temporal_cols"],
y_idx=batch["y_idx"],
)
valid_losses.append(valid_loss_batch)
batch_sizes.append(len(output_batch))
valid_loss = torch.stack(valid_losses)
batch_sizes = torch.tensor(batch_sizes).to(valid_loss.device)
valid_loss = torch.sum(valid_loss * batch_sizes) / torch.sum(batch_sizes)在验证过程中，NeuralForecast 使用 PyTorch Lightning 中的方法。此方法计算每个批次的 ，然后根据批次大小进行加权平均。我们的自定义验证过程必须考虑整个批次的最小交易数量。目前的实现不允许这样做。validation_step()validation_lossmin_nb_tradesvalidation_step()
为了解决这个问题，我们可以创建自己的模型，类似于 NeuralForecast 中的 TFT，但具有自定义 。为简单起见，我们选择设置一个大 （2000），以便在 期间只处理一个批次。这解决了我们的问题。
如第一张图所示，训练损失在第 5 纪元附近迅速收敛。这表明模型快速学习了训练数据中的底层模式。该模型在第 19 纪元达到其峰值性能，如图第二张图所示。最优验证损失接近 -3，等于大约 3 的 a。return_on_risk
验证损失连续 20 个 epoch 没有改善，并且在回调中将参数设置为 20。因此，由于该准则，训练循环在第 39 纪元停止。这有助于在验证损失停止改善时终止训练循环，从而防止过拟合。patienceEarlyStoppingEarlyStopping
第 2 步：测试过程
让我们看看我们的模型在样本外数据上的表现如何。
import torch
import json
defpredict():
test_size = len(neuralforecast_test_df)
y_hat_test = pd.DataFrame()
current_train_data = neuralforecast_train_df.copy()
test_ftr = neuralforecast_test_df.reset_index(drop=True)
y_hat = nf.predict(current_train_data,futr_df=test_ftr)
y_hat_test = pd.concat([y_hat_test, y_hat.iloc[[-1]]])
for i inrange(test_size-1):
combined_data = pd.concat([current_train_data, neuralforecast_test_df.iloc[[i]]])
y_hat = nf.predict(combined_data,futr_df=test_ftr)
y_hat_test = pd.concat([y_hat_test, y_hat.iloc[[-1]]])
current_train_data = combined_data
y_hat_test.reset_index(drop=True, inplace=True)
all_columns_except_ds = [col for col in y_hat_test.columns if col notin'ds']
median_column = [col for col in y_hat_test.columns if'-median'in col][0]
quantile_cols = [col for col in y_hat_test.columns if col notin [median_column, 'ds']]
return y_hat_test, all_columns_except_ds, median_column, quantile_cols
y_hat_test, all_columns_except_ds, median_column, quantile_cols = predict()我们没有按原样使用 NeuralForecast 中的函数，因为 NeuralForecast 不支持一次对每个数据点进行 n 步预测。这个限制意味着我们必须一个接一个地做出预测（）。对于想要根据今天的数据了解第二天回报的交易者来说，这是必要的。predicty_hat_test
defcalculate_metric():
torch_target = torch.tensor(neuralforecast_test_df['y'].to_numpy(), dtype=torch.float32).unsqueeze(-1)
torch_predicted = torch.tensor(y_hat_test[all_columns_except_ds].to_numpy(), dtype=torch.float32)
metric_calc = MetricCalculation()
metric_calc.calculate_daily_returns(y=torch_target,
y_hat=torch_predicted,
lower_quantile=1,
upper_quantile=3)
metrics = metric_calc.get_risk_rewards(is_checking_nb_trades=False)
return torch_target,metrics
torch_target, metrics = calculate_metric()使用相同的前一个函数，我们计算测试期间执行的和 。get_risk_rewardsannualized_returnreturn_on_risknb_of_trades
defprint_metrics():
aggregate_metrics = {
"annualized_return": metrics['annualized_return'].item(),
"actual_annualized_return": actual_annualized_return,
"return_on_risk":metrics['return_on_risk'].item(),
"actual_return_on_risk": actual_return_on_risk,
"nb_of_trades": metrics["nb_of_trades"],
'first_last_ds' : (first_date, last_date)
}
print(f'\n{json.dumps(aggregate_metrics, indent=4)}\n')
first_date= (neuralforecast_test_df['ds'].iloc[0]).strftime("%Y-%m-%d")
last_date = (neuralforecast_test_df['ds'].iloc[-1]).strftime("%Y-%m-%d")
nb_days = len(neuralforecast_test_df)
spy_data = pd.read_csv('custom_validation_metric/SPY.csv')
first_value = spy_data.loc[spy_data['ds'] == first_date, 'open'].iloc[0]
last_value = spy_data.loc[spy_data['ds'] == last_date, 'close'].iloc[0]
actual_annualized_return = (last_value/first_value)** (252 / nb_days) - 1
std_daily_return = neuralforecast_test_df['y'].std()
actual_annualized_risk = std_daily_return * (252 ** 0.5)
actual_return_on_risk = actual_annualized_return/actual_annualized_risk
print_metrics()
为了进行比较，我们计算了同一时期的 and。这是通过实施买入并持有策略，然后计算年化每日标准差来完成的。
最后，我们打印指标。
defplot_predictions():
plt.figure(figsize=(10, 6))
plt.plot(neuralforecast_test_df['ds'], neuralforecast_test_df['y'], color='black', label='Actual')
plt.plot(y_hat_test['ds'], y_hat_test[median_column], label='Predicted', color='blue')
plt.fill_between(y_hat_test['ds'], y_hat_test[quantile_cols[0]], y_hat_test[quantile_cols[-1]], color='gray', alpha=0.5)
plt.xlabel('Date')
plt.ylabel('Output')
plt.title('Actual vs Predicted Values over time')
plt.legend()
plt.savefig(os.path.join(f'custom_validation_metric', 'actual_vs_predicted_test.png'))
plt.close()
plot_predictions()
我们生成了一个比较实际值（黑线）和预测中值（蓝线）的图。它还将 0.05 和 0.95 分位数之间的区域设置为灰色阴影，表示 90% 的预测区间。这种可视化效果使我们能够看到测试集中的预测值与实际值的对齐程度。
解释结果
我们的策略结果良好，年化回报率为28.66%，风险回报率为2.97。这一表现超过了标准普尔500指数的买入并持有策略，标准普尔500指数的年回报率为18.47%，同期的风险回报率为1.37。这些是我们的自定义验证指标的第一个良好结果。
但是，有一些重要的注意事项：
- 在测试期间，我们在 476 个交易日中只执行了 55 笔交易。
- 该图表显示，该模型没有很好地捕捉潜在的市场模式。它的行为更像是 n 天移动平均线。
- 0.05 和 0.95 分位数之间的范围比实际的 90% 置信区间更宽。预测分位数在第一年超过 0.03 和 -0.03，捕获了实际值的 90% 以上。
- 该模型在短短 20 个时期内迅速收敛，表明该模型可能缺乏复杂性。
这些观察结果表明，该模型过于简单，可能是由于特征不足或超参数值太小，例如较小的 .为了增加交易数量并提高性能，我们可以考虑开发一个多变量时间序列模型。额外的证券不应与SPY过于相关，例如相关系数在-0.5至0.5之间的债券或商品。这种方法可以帮助我们捕捉潜在的市场模式。hidden_size
六、总结
在本文中，我们看到 RMSE 和 MSE 等标准验证指标并不是深度学习模型中股票市场预测的最佳选择。为了解决这个问题，我们提供了一个分步指南，用于实施和训练具有自定义验证指标的深度学习模型，以实现更好的交易决策。即使这种方法有潜力，结果也不是决定性的。为了改进这种方法，使用多变量时间序列数据添加更多特征可能会有所帮助。
引用：
Philippe Ostiguy.Enhancing Deep Learning Model Evaluation for Stock Market Forecasting.medium.2024
本文内容仅仅是技术探讨和学习，并不构成任何投资建议。
转发请注明原作者和出处
Be First to Comment