こんにちは。コミュニケーションIT事業部 ITソリューション部の英です。

普段はWebアプリやスマホアプリの案件などを担当しています。あと、趣味でAIを勉強しています。

みなさん、株式投資はお好きですか？私は大好きです。
最近の株式市場は大荒れですね。

気になりますよね。未来の株価。
タイムトラベルの能力があれば、真っ先に株価を見に行くでしょう。
まあ、そんなことができたならば今ごろ会社員なんてやっておりません。

前置きはこれくらいにして、今回は過去の株価データを使って未来の株価を予測するモデルを作成してみましょう。

過去の株価の数値だけを参考値として予測するため、決して皆さまの資産運用の参考にはしないようにお願いいたします。
資産運用には世界情勢などの"不確かさ"が付き物です。そこが面白さでもあります。投資は自己責任です。

ちなみにAWSで時系列予測を扱う場合、アプローチの方法としては2つあります。

SageMakerを使用する (線形学習モデル)
Amazon Forecastを使用する (マネージドサービス)

今回の記事ではSageMakerで予測する手法を解説し、次回はAmazon Forecastで予測させます。
予測する題材はS&P500の指数にしましょう。著名な投資家ウォーレン・バフェットが推奨している米国の代表的な株価指数ですね。

ここから本題

STEP1：ライブラリの準備

今回使用するライブラリをまとめてインストールしましょう。
yfinanceはYahoo Financeから株価データを取得するためのライブラリです。

# ライブラリのインストール
pip install yfinance pandas numpy matplotlib boto3 sagemaker scikit-learn

# ライブラリのインポート
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import boto3
import sagemaker
import json
from sagemaker import image_uris
from sagemaker.tuner import HyperparameterTuner, ContinuousParameter, IntegerParameter, CategoricalParameter
from matplotlib.dates import DateFormatter, YearLocator

STEP2：データの取得

yfinanceでS&P500の過去10年間のデータを取得します。
GSPCはS&P500インデックスのティッカーシンボルです。

# データ取得
sp500 = yf.Ticker('^GSPC')
df = sp500.history(period='10y')
df.interpolate(method='linear', inplace=True)

STEP3：特徴量の計算

今回は移動平均線(50日、100日、200日)とボリンジャーバンド、終値を使用します。
すべて終値(Close)をもとに計算が可能な情報です。

# 移動平均やボリンジャーバンドなどの特徴量を計算
def calculate_features(df):
    df['MA_50'] = df['Close'].rolling(window=50).mean()
    df['MA_100'] = df['Close'].rolling(window=100).mean()
    df['MA_200'] = df['Close'].rolling(window=200).mean()
    df['BB_MID'] = df['Close'].rolling(window=20).mean()
    df['BB_STD'] = df['Close'].rolling(window=20).std()
    df['BB_UPPER'] = df['BB_MID'] + (df['BB_STD'] * 2)
    df['BB_LOWER'] = df['BB_MID'] - (df['BB_STD'] * 2)

calculate_features(df)
print("特徴量を計算しました: MA_50, MA_100, MA_200, BB_MID, BB_UPPER, BB_LOWER")

STEP4：ラベル作成

現在の日付から100日後の株価をFuture Closeに設定します。
これにより、モデルが予測するターゲットが決まります。

# 100日後の株価をラベルとして設定
df['Future Close'] = df['Close'].shift(-100)
print("ラベルとして100日後の株価を設定しました。")

STEP5：データのクリーニング

データの欠損値を補間します。
連続で0が続く場合、一発で補間することができないため、二段階で補間しています。

# データの補間を行い、欠損値がないことを確認
initial_missing_values = df.isnull().sum().sum()
df.interpolate(method='linear', inplace=True)
final_missing_values = df.isnull().sum().sum()
print(f"補間前の欠損値の数: {initial_missing_values}")
print(f"補間後の欠損値の数: {final_missing_values}")
if final_missing_values > 0:
    df.ffill(inplace=True)
    df.bfill(inplace=True)
    final_missing_values = df.isnull().sum().sum()
    print(f"再補間後の欠損値の数: {final_missing_values}")
    if final_missing_values > 0:
        raise ValueError("データに欠損値が含まれています。補間が正しく行われていることを確認してください。")

STEP6：データ分割

過去10年間のデータのうち直近の200日を除いて、学習データとします。
直近の200日から100日をハイパーパラメーターのチューニングに使用します。(100日後の株価(正解)を含むデータ)
直近の100日は予測に使用します。

# データをトレーニング、検証、およびテストデータに分割
train_data = df.iloc[:-200]  # 直近の200日を除いた過去データ
validation_data = df.iloc[-200:-100]  # 直近の200日から100日を除いたデータ
test_data = df.iloc[-100:].copy()  # 直近の100日
test_data['Future Close'] = np.nan  # Future Closeは空にする
print("データの分割が完了しました。")

STEP7：ローカルとS3への保存

確認用にローカルに保存し、学習用にS3にアップロードしておきます。

# トレーニングデータの保存
csv_train_data = train_data[['Close', 'MA_50', 'MA_100', 'MA_200', 'BB_MID', 'BB_UPPER', 'BB_LOWER', 'Future Close']]
csv_train_data.to_csv('prepared_sp500_train_data.csv', index=False, header=False)

# 検証データの保存
csv_validation_data = validation_data[['Close', 'MA_50', 'MA_100', 'MA_200', 'BB_MID', 'BB_UPPER', 'BB_LOWER', 'Future Close']]
csv_validation_data.to_csv('prepared_sp500_validation_data.csv', index=False, header=False)

# テストデータの保存（Future Closeを含まない）
csv_test_data = test_data[['Close', 'MA_50', 'MA_100', 'MA_200', 'BB_MID', 'BB_UPPER', 'BB_LOWER']]
csv_test_data.to_csv('prepared_sp500_test_data.csv', index=False, header=False)

# S3へのアップロード
sagemaker_session = sagemaker.Session()
bucket_name = sagemaker_session.default_bucket()
s3_train_data = f's3://{bucket_name}/prepared_sp500_train_data.csv'
s3_validation_data = f's3://{bucket_name}/prepared_sp500_validation_data.csv'
s3_test_data = f's3://{bucket_name}/prepared_sp500_test_data.csv'
boto3.resource('s3').Bucket(bucket_name).Object('prepared_sp500_train_data.csv').upload_file('prepared_sp500_train_data.csv')
boto3.resource('s3').Bucket(bucket_name).Object('prepared_sp500_validation_data.csv').upload_file('prepared_sp500_validation_data.csv')
boto3.resource('s3').Bucket(bucket_name).Object('prepared_sp500_test_data.csv').upload_file('prepared_sp500_test_data.csv')

STEP8：学習データのプロット

過去10年分のS&P500のスコアをプロットします。

# データの可視化
plt.figure(figsize=(12, 6))

# 全期間のデータをプロット
plt.plot(df.index, df['Close'], label='Actual Close', color='blue', linewidth=2)

# 学習データと検証データのプロット
plt.plot(train_data.index, train_data['Close'], label='Train Data Close', color='lightblue', linewidth=1)
plt.plot(validation_data.index, validation_data['Close'], label='Validation Data Close', color='orange', linewidth=2)

# 移動平均とボリンジャーバンドをプロット
plt.plot(df.index, df['MA_50'], label='50-Day MA', color='orange', linewidth=1)
plt.plot(df.index, df['MA_100'], label='100-Day MA', color='green', linewidth=1)
plt.plot(df.index, df['MA_200'], label='200-Day MA', color='brown', linewidth=1)
plt.plot(df.index, df['BB_MID'], label='Bollinger Mid Band', color='red', linestyle='--', linewidth=1)
plt.plot(df.index, df['BB_UPPER'], label='Bollinger Upper Band', color='purple', linestyle='--', linewidth=1)
plt.plot(df.index, df['BB_LOWER'], label='Bollinger Lower Band', color='gray', linestyle='--', linewidth=1)

# x軸のメモリを1年ごとに設定
ax = plt.gca()
ax.xaxis.set_major_locator(YearLocator())
ax.xaxis.set_major_formatter(DateFormatter('%Y'))

plt.xlabel('Date')
plt.ylabel('Close Price')
plt.title('S&P 500 Close Price with MA and Bollinger Bands')
plt.legend()
plt.grid(True)
plt.show()
print("学習データをプロットしました。")

・青色：過去10年のデータ

・水色：トレーニングに使用したデータ(10年分-直近の200日)

・橙色：チューニングに使用するバリデーションデータ

STEP9：ハイパーパラメーターチューニングの設定

以下のように設定し、トレーニングを開始します。
今回使用するフレームワークはSageMakerのLinear Learnerです。予測タイプは回帰(regressor)を選択します。
→linear-learner

ランダムサーチ手法で線形学習モデル(Linear Learner)のハイパーパラメーターをチューニングしていきます。平均二乗誤差(validation:mse)を使って、予測値と実際の値のズレを評価します。
hyperparameter_rangesでハイパーパーラメーターの範囲を指定しています。
→HyperparameterTuner

線形学習モデルのハイパーパラメータは他にもたくさんありますが、チューニングが可能なパラメータは限られているので注意してください。
→ハイパーパラメーター一覧

# ランダムサーチの準備
role = sagemaker.get_execution_role()
container = image_uris.retrieve(region=sagemaker_session.boto_region_name, framework='linear-learner', version='1')
linear = sagemaker.estimator.Estimator(
    image_uri=container,
    role=role,
    instance_count=1,
    instance_type='ml.m4.xlarge',
    output_path=f's3://{bucket_name}/linear_learner_output',
    sagemaker_session=sagemaker_session
)

linear.set_hyperparameters(
    predictor_type='regressor'
)

hyperparameter_ranges = {
    'learning_rate': ContinuousParameter(0.0001, 0.1),
    'mini_batch_size': IntegerParameter(16, 128),
    'l1': ContinuousParameter(0.0, 0.1),
    'wd': ContinuousParameter(0.0, 0.1)
}

objective_metric_name = 'validation:mse'
tuner = HyperparameterTuner(
    estimator=linear,
    objective_metric_name=objective_metric_name,
    hyperparameter_ranges=hyperparameter_ranges,
    max_jobs=20,
    max_parallel_jobs=3,
    objective_type='Minimize'
)

train_input = sagemaker.inputs.TrainingInput(
    s3_data=s3_train_data,
    content_type='text/csv'
)

validation_input = sagemaker.inputs.TrainingInput(
    s3_data=s3_validation_data,
    content_type='text/csv'
)

# チューニングジョブの開始
tuner.fit({'train': train_input, 'validation': validation_input})
tuner.wait()

# 最適なハイパーパラメータとモデル番号の出力
best_estimator = tuner.best_estimator()
best_hyperparameters = best_estimator.hyperparameters()
best_model_name = best_estimator.latest_training_job.name
print("最適なハイパーパラメータ:", best_hyperparameters)
print("最適なモデル番号:", best_model_name)

チューニングが始まると並列で3ジョブ、合計で20ジョブ実行されます。
その中でもっとも優れているものを最終的に使用します。

STEP10：トレーニング

STEP9で導き出したbest_estimatorで再学習します。

# 最適なハイパーパラメータを使用してモデルをトレーニング
best_estimator.fit({'train': train_input, 'validation': validation_input})

STEP11：モデルのデプロイ

トレーニングが完了したので、デプロイします。

# 学習済みのモデルをデプロイ
linear_predictor = best_estimator.deploy(initial_instance_count=1, instance_type='ml.m4.xlarge')
print("最適なハイパーパラメータを使用してモデルをデプロイしました。")

STEP12：予測結果のプロット

test_data(直近100日)をlinear_predictorに渡して、1～100日後の株価を予測させます。

# test_dataを使用して予測
test_predictions = []
for i in range(len(test_data)):
    features = test_data[['Close', 'MA_50', 'MA_100', 'MA_200', 'BB_MID', 'BB_UPPER', 'BB_LOWER']].iloc[i].values.tolist()
    print(f"Test features (day {i}): {features}")  # 特徴量を出力
    payload = {
        'instances': [{'features': features}]
    }
    result = linear_predictor.predict(json.dumps(payload), initial_args={'ContentType': 'application/json'})
    prediction = json.loads(result.decode('utf-8'))['predictions'][0]['score']
    print(f"Prediction for test day {i}: {prediction}")  # 予測結果を出力
    test_predictions.append(prediction)

# test_dataの予測結果のプロット
plt.figure(figsize=(14, 7))
plt.plot(df.index, df['Close'], label='10-Year Actual Close', color='blue')
plt.plot(test_data.index, test_data['Close'], label='Test Data Close', color='green')
# 100日後の位置にプロットするためにインデックスをシフト
test_pred_index = pd.date_range(start=test_data.index[-1], periods=101, freq='B')[1:]
plt.plot(test_pred_index, test_predictions, label='Predicted Close for Test Data', color='orange')
plt.xlabel('Date')
plt.ylabel('Close Price')
plt.title('S&P 500 Close Price and Predictions for Test Data')
plt.legend()
plt.grid(True)
plt.show()

以下のようにプロットされました。
内訳は以下のとおりです。
・青色：トレーニングデータ

・緑色：テストデータ(100～1日前のデータ)

・橙色：未来の100日間の予測データ

※注意：これは一つの予測に過ぎず、未来の株価を正確に予測するものではございません。占い感覚でご覧ください。

さいごに

今回はS&P500の過去データを使って未来の株価を予測しました。
結果としては、ポジティブ(？)な予測結果となっているかと思います。
過去10年間は上昇傾向にありましたから、その傾き(トレンド)はしっかり学習できているように見受けられます。
より高度な予測を行いたい場合は、VIX指数や雇用統計など株価指数以外の情報も特徴量に加えると良いかと思います。

これからもAWS×AI関連の検証記事をたくさん書いていきます。
↓ のスターを押していただけると嬉しいです。励みになります。
最後まで読んでいただき、ありがとうございました。

私たちは一緒に働いてくれる仲間を募集しています！

コミュニケーションIT事業部

執筆：英良治 (@hanabusa.ryoji)、レビュー：@kobayashi.hinami
（Shodoで執筆されました）

電通総研テックブログ

電通総研が運営する技術ブログ

yfinanceとAWSで未来の株価を予測する(前編)