手把手教你ai顾投：AI对股票回报的精准预测（运用 CFC SageMaker 算法的 LNN）

写在前面的话：学术研究者与行业从业者始终针对股票回报预测展开了广泛探究。为此，他们提出了多种从简单线性回归至复杂深度学习模型的机器学习模型。在本文中，我们主要利用液体神经网络（LNN）的性能，这是一种用以处理序列数据的新型神经网络架构。

LNN 属于连续时间循环神经网络（CT-RNN），其隐藏状态随时间演化的过程被建模为常微分方程（ODE）。LNN 基于液体时间常数（LTC）常微分方程，其隐藏状态的导数和时间常数均通过神经网络进行建模。

本文将重点阐述闭合形式的连续深度（CfC）网络。CfC 网络通过实现 LTC 常微分方程的近似闭式解，与其他 CT-RNN（包括 LTC）相比，能够达成更迅速的训练和推理性能，因后者需要运用数值求解器去求解常微分方程。

我们会运用由 Amazon sageMaker 实现的 CfC 网络进行概率时间序列预测，也即 CfC SageMaker 算法。我们将使用标准普尔 500 指数的已实现波动率以及不同的隐含波动率指数当作输入，以预测标准普尔 500 指数 30 天回报的条件均值和条件标准差。

我们将运用 2022 年 6 月 30 日至 2024 年 6 月 28 日的每日收盘价数据，这些数据会从雅虎财经进行下载。模型将基于截至 2023 年 9 月 8 日的数据展开训练，并运用训练好的模型预测截至 2024 年 6 月 28 日的后续数据。结果表明，CfC SageMaker 算法实现了 1.4%的平均绝对误差以及 95.8%的平均方向准确率。

1.模型输出即为标准普尔 500 指数的 30 天回报率，其计算方式如下：

对于每一天 t，其中 P（t）为标准普尔 500 指数在第 t 天的收盘价。我们将运用 30 天的预测长度，这意味着模型将输出随后 30 天的 30 天回报。假设我们运用重叠（或滚动）回报，从第 t 天至第 t 天 + 30 的预测 30 天回报即为输出序列中的最后一个回报。

2.输入方面，该模型使用标准普尔 500 指数过去 30 天的回报率以及下列波动率指标的过去值：

RVOL：标准普尔 500 指数的实际波动率，通过计算标准普尔 500 指数每日回报的 30 天滚动样本标准差得出。

• VIX指数：用于衡量标准普尔 500 期权的 30 天隐含波动率。
• VVIX指数：反映 VIX 的 30 天预期波动率。
• VXN指数：用于衡量纳斯达克 100 期权的 30 天隐含波动率。
• GVZ 指数：用于衡量 SPDR 黄金股票 ETF（GLD）期权 30 天隐含波动率。
• OVX指数：用于衡量美国石油基金（USO）期权的 30 天隐含波动率。

RVOL 属于向后看的指标，因其估计了过去 30 天的波动率，而 VIX、VVIX、VXN、GVZ 以及 OVX 均为前瞻性指标，因其反映了市场对未来 30 天波动率的预期。

我们将运用 30 天的上下文长度，也就是说，该模型将运用 30 天的回报率以及过去 30 天的波动率指标当作输入，以预测随后 30 天的 30 天回报率。

3.环境设置

首先导入所有依赖项并设置 SageMaker 环境。

import ioimport sagemakerimport pandas as pdimport numpy as npimport matplotlib.pyplot as pltimport yfinance as yffrom sklearn.metrics import root_mean_squared_error, mean_absolute_error, accuracy_score, f1_score
# SageMaker sessionsagemaker_session = sagemaker.Session()# SageMaker rolerole = sagemaker.get_execution_role()# S3 bucketbucket = sagemaker_session.default_bucket()# EC2 instanceinstance_type = "ml.m5.4xlarge"

其后，我们明确神经网络的上下文长度与预测长度。上下文长度即为用作输入的过去时间步长数量，而预测长度即为要预测的未来时间步长数量。我们将两者均设定为等同于 30 天，也就是说，我们运用输入与输出的前 30 个值去预测输出的后续 30 个值。

context_length = 30prediction_length = 30

我们还明确了 CfC 网络的所有其余超参数。需注意，我们运用参数小于 5k 的相对较小的模型。

hyperparameters = {    "context-length": context_length,    "prediction-length": prediction_length,    "sequence-stride": 1,    "hidden-size": 20,    "backbone-layers": 1,    "backbone-units": 40,    "backbone-activation": "lecun",    "backbone-dropout": 0,    "minimal": True,    "no-gate": True,    "use-mixed": False,    "use-ltc": False,    "batch-size": 32,    "lr": 0.0001,    "lr-decay": 0.9999,    "epochs": 800,}

4.数据准备

接下来，我们运用 Yahoo！Finance Python API 从 Yahoo！Finance 下载 2022 年 6 月 30 日至 2024 年 6 月 28 日的每日收盘价时间序列。

tickers = ["^SPX", "^VIX", "^VVIX", "^VXN", "^GVZ", "^OVX"]dataset = yf.download(" ".join(tickers), start="2022-06-30", end="2024-06-29")dataset = dataset.loc[:, dataset.columns.get_level_values(0) == "Close"]dataset.columns = dataset.columns.get_level_values(1)dataset.ffill(inplace=True)

随后，我们计算标准普尔 500 指数的 30 天回报率以及 30 天的实际波动率。

# calculate the returns
dataset["Return30"] = np.log(dataset["^SPX"]).diff(periods=30)
# calculate the realized volatility
dataset["RVOL"] = np.log(dataset["^SPX"]).diff(periods=1).rolling(window=30).std(ddof=1)
# drop the prices
dataset.drop(labels=["^SPX"], axis=1, inplace=True)
# drop the missing values
dataset.dropna(inplace=True)
# move the returns to the first column
dataset = dataset[["Return30"] + dataset.columns.drop("Return30").tolist()]

该数据集涵盖 502 个每日观测值，在去除因计算收益和已实现波动率而产生的缺失值后，这些观测值减少至 472 个。

2022-08-12 至 2024-06-28 的 30 天回报、30 天实际波动率以及波动率指数。

我们继续依照 CfC SageMaker 算法预期的格式重命名列，其中输出名称应以 开头，输入名称应以 开头。

dataset.columns = ["y"] + [f"x{i}" for i in range(dataset.shape[1] - 1)]

请注意，算法的代码始终在输入中包含输出的过去值，所以在为模型准备数据时无需添加输出的滞后值。

5.测试

为了验证模型，我们将数据拆分为训练集与测试集。训练集包含前 70%的数据，而测试集包含最后 30%的数据。需注意，数据由算法在内部进行缩放，无需事先缩放数据。

 define the size of the test settest_size = int(0.3 * len(dataset))# extract the training datatraining_dataset = dataset.iloc[:- test_size - context_length - prediction_length - 1]# extract the test datatest_dataset = dataset.iloc[- test_size - context_length - prediction_length - 1:]

现在将训练数据保存至 S3 中，构建 SageMaker 估算器并运行训练作业。

# upload the training data to S3training_data = sagemaker_session.upload_string_as_file_body(    body=training_dataset.to_csv(index=False),    bucket=bucket,    key="training_data.csv")
# create the estimatorestimator = sagemaker.algorithm.AlgorithmEstimator(    algorithm_arn=algo_arn,    role=role,    instance_count=1,    instance_type=instance_type,    input_mode="File",    sagemaker_session=sagemaker_session,    hyperparameters=hyperparameters)
# run the training jobestimator.fit({"training": training_data})

训练作业完成后，我们将模型部署至可用于推理的实时端点。

serializer = sagemaker.serializers.CSVSerializer(content_type="text/csv")

deserializer = sagemaker.base_deserializers.PandasDeserializer(accept="text/csv")

predictor = estimator.deploy(
    initial_instance_count=1,
    instance_type=instance_type,)

创建端点后，我们能够生成测试集预测。鉴于我们运用滚动（或重叠）回报，我们仅对每个预测序列的最后一个元素感兴趣（回想一下，我们将预测长度设定为 30 天，与回报的范围相同）。

# create a list for storing the predictionspredictions = []
# loop across the datesfor t in range(context_length, len(test_dataset) - prediction_length + 1):
    # extract the inputs    payload = test_dataset.iloc[t - context_length: t]
    # invoke the endpoint    response = sagemaker_session.sagemaker_runtime_client.invoke_endpoint(        EndpointName=predictor.endpoint_name,        ContentType="text/csv",        Body=payload.to_csv(index=False)    )
    # deserialize the endpoint response    response = deserializer.deserialize(response["Body"], content_type="text/csv")
    # extract the predicted 30-day return    prediction = response.iloc[-1:]
    # extract the date corresponding to the predicted 30-day return    prediction.index = [test_dataset.index[t + prediction_length - 1]]
    # save the prediction    predictions.append(prediction)
# cast the predictions to data framepredictions = pd.concat(predictions)
# add the actual valuespredictions["y"] = test_dataset["y"]

测试集（从 2023-12-04 至 2024-06-28）的实际和预测 30 天回报率。

我们运用以下指标评估测试集预测：

• RMSE：返回值预测值的均方根误差。
• MAE：返回预测值的平均绝对误差。
• 准确性：预测回报迹象的准确性。
• F1：预测回报迹象的 F1 分数。

# calculate the model performance metricsmetrics = pd.DataFrame(    columns=["Metric", "Value"],    data=[        {"Metric": "RMSE", "Value": root_mean_squared_error(y_true=predictions["y"], y_pred=predictions["y_mean"])},        {"Metric": "MAE", "Value": mean_absolute_error(y_true=predictions["y"], y_pred=predictions["y_mean"])},        {"Metric": "Accuracy", "Value": accuracy_score(y_true=predictions["y"] > 0, y_pred=predictions["y_mean"] > 0)},        {"Metric": "F1", "Value": f1_score(y_true=predictions["y"] > 0, y_pred=predictions["y_mean"] > 0)},    ])

测试集（从 2023-12-04 至 2024-06-28）预测的 30 天回报的性能指标。

现在我们能够删除模型与端点。

predictor.delete_model()predictor.delete_endpoint(delete_endpoint_config=True)

6.预测

我们运用所有可用数据重新训练模型，并生成样本外预测，也就是说，我们预测当前日期（2024-06-28）之后 30（工作日）的 30 天回报。

data = sagemaker_session.upload_string_as_file_body(
body=dataset.to_csv(index=False),
bucket=bucket,
key="dataset.csv"
)
estimator = sagemaker.algorithm.AlgorithmEstimator(
algorithm_arn=algo_arn,
role=role,
instance_count=1,
instance_type=instance_type,
input_mode="File",
sagemaker_session=sagemaker_session,
hyperparameters=hyperparameters
)
estimator.fit({"training": data})

鉴于我们仅需要一个预测的 30 天序列，我们运用批量变换以生成预测。

inputs = sagemaker_session.upload_string_as_file_body(
body=dataset.iloc[- context_length:].to_csv(index=False),
bucket=bucket,
key="inputs.csv"
)
transformer = estimator.transformer(
instance_count=1,
instance_type=instance_type,
)
transformer.transform(
data=inputs,
content_type="text/csv",
)

批量转换作业完成后，我们能够从 S3 加载预测。

forecasts = sagemaker_session.read_s3_file(    bucket=bucket,    key_prefix=f"{transformer.latest_transform_job.name}/inputs.csv.out")
forecasts = pd.read_csv(io.StringIO(forecasts), dtype=float).dropna()
forecasts.index = pd.date_range(    start=dataset.index[-1] + pd.Timedelta(days=1),    periods=prediction_length,    freq="B")

30 天样本外预测回报（从 2024-07-01 至 2024-08-09）。

这个预测股票回报的测试到这里就结束了，最后期望关注我的朋友们早日实现财富自由！

本文内容仅仅是技术探讨和学习，并不构成任何投资建议。