【Aidemy最終課題】テキストデータの感情分析による米国株の株価予測モデル作成

１．はじめに

当記事は、プログラミングスクールAidemy Premiumのデータ分析講座6か月コースの修了成果物として、作成したプログラムを公開したものです。当記事をAidemyに修了判定をしていただくことで、晴れて受講修了となります。（※2024年1月現在、無事、Aidemyに修了判定をしていただき、講座修了できました。）
成果物のテーマは、以下の通りに設定しました。
【@WSJ Marketsアカウントのツイートの感情分析を用いた米国株の株価予測モデル作成】
※実行環境：Google Colaboratory
※このブログはAidemy Premiumのカリキュラムの一環で、受講修了条件を満たすために公開しています。

２．自己紹介

私は金融機関に勤める入社7年目の社会人です。
所属企業においては、デジタル技術を活用して新しいビジネスを生み出そうという部門に属しており、私個人もデジタル分野には非常に興味がありました。
2023年度上期には、たまたま会社でデータサイエンティストを養成する大学の講座を聴講する機会があり、同講座を聴講したものの、難易度が高く消化しきれない部分もあったため、一度体系的にPythonを学ぶ必要性を感じ、プログラミングスクールに通うことを決意しました。

３．受講の費用について

金銭面の負担が懸念点であったものの（6か月コースで85万円ほど）、調べた結果、Aidemyのデータ分析講座は厚労省の専門実践教育訓練給付金を利用することが可能であることが判りました。
専門実践教育訓練給付金とは、厚生労働大臣の指定する講座を受講し修了した場合、修了時点までに実際に支払った受講料の最大70%が支給される制度です。Aidemyのデータ分析講座は、第四次産業革命スキル習得講座に該当したため、きちんと受講期間内に修了すれば受講料の70%が支給されます。
逆に、しっかりと修了をしなければ給付金は返ってこない制度です。
今回は同給付金制度を利用して、データ分析6か月コースを受講しました。（ライセンス期限：2023年8月21日～2024年2月19日）

４．成果物のテーマ設定にあたって

普段、所属企業にて株の自己ポジションの管理を行う業務を行っております。自己ポジションというのは個人・法人を問わず自己勘定で保有する有価証券のことです。保有株の株価が変動すれば、自己ポジションの時価総額も連動して増減するため、財務面での影響が生じます。そのため、ポジションを保有する金融機関では日々、市場リスクの計測を行い、相場急変時に備えたリスクヘッジ戦略を適時に策定しております。
今回の成果物のテーマを設定するにあたっては、自己ポジションを構成する米国株式の株価を機械学習モデルを用いて予測できれば、リスクヘッジ戦略の策定に活かせるのではないかと思い、S＆P500*の予測をテーマとすることにしました。
※S＆P500…アメリカを代表する500銘柄で構成された、アメリカ経済全体の動きを表す代表的な株価指数。

５．感情分析に用いるテキストデータ

S&P500の予測には、講座で学んだ内容を実践するために、テキストデータの感情分析を用いました。S&P500の価格と関連があるテキストデータを、講座で学んだ自然言語処理の手法により、形態素解析のうえネガポジ分析を行い、各日付の平均PN値を算出して終値に紐づけしました。
最初は、Kaggle上に公開されていたWall Street Journal articlesデータを用いてデータテーブルを作成したのですが、作成したデータテーブルを確認すると必要な日付のデータが飛んでいる等の事態が起こっており、原因は元データの日付表示が一部特殊であること等であると判明しました。そのまま当初のデータを用いても分析がうまく進まないと判断し、オープンデータではなく自分で実際にデータセットを取得する方針に途中から切り替えることにしました。
そのため、TwitterAPIを利用して、（＠WSJ Markets）アカウントからスクレイピングした投稿をテキストデータとして用いることにしました。

６．TwitterAPIの利用について

サーバーへの負担等を鑑み、2023年12月現在、TwitterAPIの利用には、有料化や取得できる投稿の数に制限等が課されております。
ここ数年で色々とTwitterAPIの利用ルールが変わってる最中であり、今回の分析を行ううえでインターネット上で情報収集した際もネット上で情報が錯綜しておりました。同様の分析を行う方は、最新の利用ルールを確認のうえ行ってください。

私が利用したTwitterAPIの基本情報は以下の通りです。
①プランは「Basic」で、1か月あたり10,000件まで投稿を取得可能。
②15分間あたり15リクエストしか行えないRate limitがある。

②のRate limitを踏まえ、1回のコード実行で100件ずつ投稿を取得し、100件取得したら1度CSV出力が行われるようコーディングしました。最終的には、取得制限がかかった3000件弱まで投稿を取得できました。出力したCSVの1列目には投稿がされた日付に加え、時刻の情報も付いていましたが、今回の分析に当たってはExcelのワークシート関数を用いて時刻情報を取り除き、取得できた合計30件のCSVファイルを結合してテキストデータとしました（合計投稿数2984件）。
上述したKaggleデータとともにこの部分が非常に泥臭く、データ分析の業務における「前処理」の工程に該当するのだと思います。機械学習のモデルにデータを投入して実際に分析を行うよりも、整ったデータセットを用意する方がよほど時間を使っていました。実際のデータ分析の現場においてもそうなのではないかと思われます。

７．実際に作成したコード

（１）テキストデータ（Twitter投稿）の取得

実際に私が作成したコードはtweepyモジュールのインポートから開始しました。

# tweepyのインストール
!pip install tweepy --upgrade

import csv
import tweepy
import pandas as pd

#API_KEY等は個人情報のため、コード実行後に消した
API_KEY = ""
API_KEY_SECRET = ""
ACCESS_TOKEN = ""
ACCESS_TOKEN_SECRET = ""
BEARER_TOKEN = ""

client = tweepy.Client(BEARER_TOKEN)
api = tweepy.Client(None, API_KEY, API_KEY_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
person_list = ["WSJMarkets"]

# 取得するアカウントを指定
person_name = person_list[0]
user=api.get_user(username=person_name,
                    user_fields=['description','protected','name','username','public_metrics','profile_image_url']
                    ,user_auth=True)
tweets = api.get_users_tweets(user.data.id, max_results=100,exclude=['retweets', ],\
                              tweet_fields=['created_at', ],until_id = None, user_auth=True,)

# ファイルを保存
with open(f'tweet_{person_name}_1.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    for tweet in tweets.data:
      writer.writerow([tweet.created_at, tweet.text, tweet.id])

import time
from google.colab import files
# 対象アカウントの決定
person_list = ["WSJMarkets"]
person_name = person_list[0]
user=api.get_user(username=person_name,
                    user_fields=['description','protected','name','username','public_metrics','profile_image_url']
                    ,user_auth=True)

# 次のuntil_idを取り出す関数の定義
def road_csv(df):
  next_id = df.iloc[-1, 2]
  return (next_id)

# ツイートを取得する関数を定義
def get_tweets(next_id):
  tweets = api.get_users_tweets(user.data.id, max_results=100,exclude=['retweets', ],\
                              tweet_fields=['created_at', ],until_id = next_id, user_auth=True,)
  return tweets

# ツイートをCSVに書き込む関数の定義
def save_csv(tweets, i):
  with open(f'tweet_{person_name}_{i+1}.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    for tweet in tweets.data:
      writer.writerow([tweet.created_at, tweet.text, tweet.id])
  return

# 連続で過去のツイートを取得する
#range()箇所は、CSVファイルが作成完了するたびに(1,5)→(5,9)→（9,13）と入力し直して実行し、最終的に制限がかかった（30,31）まで繰り返した。
for i in range(30,31):
  df_tweet = pd.read_csv(f"tweet_{person_name}_{i}.csv")
  next_id = road_csv(df_tweet)
  tweets = get_tweets(next_id)
  save = save_csv(tweets, i)
  # files.download(f"tweet_WSJMarkets_{i}.csv")
  time.sleep(15*60)

上記のコードと結合作業により、元となるテキストデータを作成することができました。作成したテキストデータの一部は以下の通りです。headlinesカラムには同日に投稿された複数の記事が入っております。カラム名は後の工程の株価データと揃えて入力しました。3列目には投稿IDが入ってます。

（２）データテーブルの作成

次に、headlinesカラムに格納された記事の単語を形態素解析し、各日付の平均PN値を算出します。今回のテキストは英語であり、高村らの単語感情極性対応表を用いて各単語の対応付けを行いました。

!pip install polyglot
!pip install pycld2
!pip install pyicu
!pip install morfessor
!polyglot download embeddings2.en
!polyglot download pos2.en

from pandas.core.tools.datetimes import to_datetime
from datetime import datetime as dt
from polyglot.text import Text
from sklearn.model_selection import train_test_split
from sklearn.neighbors import KNeighborsClassifier   # k近傍法(k-NN)
from sklearn.linear_model import LogisticRegression  # ロジスティック回帰
from sklearn.svm import LinearSVC                    # 線形SVM
from sklearn.svm import SVC                          # 非線形SVM
from sklearn.tree import DecisionTreeClassifier      # 決定木
from sklearn.ensemble import RandomForestClassifier  # ランダムフォレスト
import pandas as pd
import numpy as np
import glob
import pytz
import pycld2
import regex
import matplotlib.pyplot as plt

pn_df = pd.read_csv("/content/drive/MyDrive/テスト_20231222金/pn_en.dic.txt",\
                    sep=":",
                    encoding="utf-8",
                    names = ("Word","POS","PN"))

word_list = list(pn_df["Word"])
pn_list = list(pn_df["PN"])
pn_dict = dict(zip(word_list, pn_list))

pn_dict

# 形態素解析を行い、文書中からBaseFormを取り出す関数を定義
def get_diclist(text):

  diclist = [] # {"BaseForm":取得した単語}を格納するリスト


  # 入力言語が英語であれば形態素解析を実施
  is_reliable, text_bytesFound, details = pycld2.detect(text)
  if "en" == details[0][1]:

    # 形態素解析を実施
    tokens = Text(text)
    parsed = tokens.pos_tags

    # BaseFormを抽出
    for i in parsed:
      d = {"BaseForm": i[0]}
      diclist.append(d)

  # 対象テキストが英語でない場合は"None"を入力
  else:
    diclist.append({"BaseForm":"0"})

  return diclist

# BaseFormごとのPN値を求める関数を定義
def add_pnvalue(diclist_old, pn_dict):
  diclist_new = [] # {"BaseForm":word, "PN値":pn_value}を格納する
  for word in diclist_old:
    base = word["BaseForm"]
    if base in pn_dict:
      pn = float(pn_dict[base])
    else:
      pn = "notfound"
    word["PN"] = pn
    diclist_new.append(word)
  return diclist_new

# 1ツイートのPN値の平均値を求める
def get_mean(diclist_new):
  pn_list = []
  for word in diclist_new:
    pn = word["PN"]
    if pn != "notfound":
      pn_list.append(pn)
  if len(pn_list) > 0:
    pnmean = np.mean(pn_list)
  else:
    pnmean = 0
  return pnmean

# 日付けを文字列からdatetimeオブジェクト（datetime.datetime）に変換する関数を定義
def change_date(dates):
  dates = pd.to_datetime(dates)
  # dates_r = dates.dt.tz_convert("America/New_York")
  return dates

RE_BAD_CHARS = regex.compile(r"[\p{Cc}\p{Cs}]+")

def remove_bad_chars(text):
    return RE_BAD_CHARS.sub("", text)

df_list = [] # 各データフレームを格納するリスト


# df_list中の各DFを一つに統合する
df_all=pd.read_csv("/content/drive/MyDrive/テスト_20231222金/tweet_WSJMarkets_1~30.csv")
df_all["date"]=change_date(df_all["date"])
df_all=df_all.groupby(["date"])["headlines"].apply(list)

# df_allの1tweetごとのPN値を求める
pn_mean_list = []

for i in range(0,len(df_all.axes[0])):
  news = df_all.iloc[i]
  news=" ".join([str(n) for n in news])
  diclist = get_diclist(remove_bad_chars(news))
  diclist_new = add_pnvalue(diclist, pn_dict)
  pnmean = get_mean(diclist_new)
  pn_mean_list.append(pnmean)
df_all=pd.DataFrame({"date":df_all.index,"headlines":df_all.values})
df_all["pn"] = pn_mean_list           # df_allにツイートのPN値を追加
df_all = df_all.sort_values(by="date", ascending=True) # 日付けで昇順に並び替える
df_all

ここまでのコードを実行し、以下のデータテーブルが作成できました。

そして、データテーブルに各日付のS&P500の終値を紐づけます。
終値のデータセットは、Kaggle上で公開されていた以下のオープンデータを用いました。
・S&P 500 Stocks (daily updated)

# 株価データの読み込み
df_chart = pd.read_csv("/content/drive/MyDrive/テスト_20231222金/sp500_index.csv")


df_chart["date"]=pd.to_datetime(df_chart["Date"])
df_chart=df_chart.drop(["Date"],axis=1)
# df_chart = df_chart.sort_values(by="date", ascending=True) # 日付けで昇順に並び替える

# 株価データとPN値を内部結合する
df_table = pd.merge(df_all,df_chart,on="date",how="inner")

# Dateをカラムとして再設定
df_table.reset_index(inplace=True)
df_table

S&P500のカラムを追加できました。

（３）機械学習モデルの作成

作成したデータテーブルを訓練用データと検証用データに分け、各機械学習モデルを作成していきます。

x = df_table.values[:,3] # pn値を説明変数とする
y = df_table.values[:,4] # 株価を教師データ

# データを学習用と検証に8:2に分割
x_train, x_test, y_train, y_test = train_test_split(x, y, test_size=0.2, random_state=0, shuffle=False)

# pn値データを標準化する
x_train = (x_train - x_train.mean()) / x_train.std()
x_test  = (x_test - x_test.mean())  / x_test.std()

# 訓練データとテストデータの作成
df_train = pd.DataFrame({"pn":x_train, "Price":y_train}, columns=["pn", "Price"],\
                        index=df_table["date"][:len(x_train)])

df_test  = pd.DataFrame({"pn":x_test, "Price":y_test}, columns=["pn", "Price"],\
                        index=df_table["date"][len(x_train):])

x_test
x_train

"""
訓練用データの加工
"""
exchange_dates = [] # 日付けを格納するリストを用意

pn_rates = []
pn_rates_diff = [] # 1日ごとのPN値の差分を格納するリストを用意

exchange_rates = []
exchange_rates_diff = [] # 1日ごとの株価を格納するリストを用意

prev_pn   = df_train["pn"][0]           # PN値カラムの最初の値を指定
prev_exch =df_train["Price"][0] # 株価カラムの最初の値を指定

# 訓練データの個数分、PN値と株価の変化を算出し、exchange_dates, pn_rates_diff, exchange_rates_diffに追加していく
for i in range(len(x_train)):
  # 日にちごとのPN値、株価を順番に抽出する
  time     = df_train.index[i]
  pn_val   = df_train["pn"][i]
  exch_val = df_train["Price"][i]

  # 上記で抽出した日にちごとのPN値と株価の値それぞれの差分、及び対応する日付けをそれぞれのリストに追加する
  exchange_dates.append(time)
  pn_rates_diff.append(pn_val - prev_pn)            # 1日前のPN値と現在のPN値の差分を求める
  exchange_rates_diff.append(exch_val - prev_exch)  # 1日前の株価と現在の株価の差分を求める

  # 次の差分を求めるために、prev_pnとprev_exchを更新する
  prev_pn   = pn_val
  prev_exch = exch_val

# 横軸を時間として、PN値、株価のそれぞれをグラフで可視化する
fig = plt.figure(figsize=(6,8))

ax1 = fig.add_subplot(2,1,1)
ax2 = fig.add_subplot(2,1,2)

x_val    = exchange_dates # 横軸の時間を設定
y_val    = pn_rates_diff  # 縦のPN値の差分を設定
y_val_ex = exchange_rates_diff

# ax1のグラフの作成（PN値の差分の時系列）
ax1.set_title("PN Value (diff)")
ax1.plot(x_val, y_val)
ax1.grid(True)
plt.setp(ax1.get_xticklabels(), rotation=30)

# ax2のグラフの作成（株価の差分の時系列）
ax2.set_title("Price Valus(diff)")
ax2.plot(x_val, y_val_ex)
ax2.grid(True)
plt.setp(ax2.get_xticklabels(), rotation=30)

plt.show()

PN値と株価の、それぞれの前日との差分を時系列で表したグラフがmatplotlibで出力できました。

そして、機械学習モデルを作成し、各モデルの予測結果を評価します。

"""
訓練用データの加工（続き）
"""
# 次に、3日間ごとにデータをまとめていく
INPUT_LEN = 3
data_len  = len(pn_rates_diff) # pn_rates_diffのデータの個数

tr_input_mat = [] # 学習データを格納するリスト
tr_angle_mat = [] # 教師データを格納するリスト

# データを1つずつずらして、3日間ごとにまとめていく
# 3日間ごとの株価とPN値の変化に対応した株価の上下（上がったら1、下がったら0）を作成する
for i in range(INPUT_LEN, data_len):
  tmp_arr = [] # 3日間ごとの値を格納するリスト
               # PN値も株価も同じリストに追加していく

  for j in range(INPUT_LEN):
    tmp_arr.append(exchange_rates_diff[i - INPUT_LEN + j]) # (1,2,3), (4,5,6)のようなまとまりが作られる
    tmp_arr.append(pn_rates_diff[i - INPUT_LEN + j])       #i=3の時、(1,2,3), i=4の時、(2,3,4)
  tr_input_mat.append(tmp_arr) # [[e1,p1,e2,p2,e3,p3],[e2,p2,e3,p3,e4,p4],,, ]のようなリスト形式になる

  # 前日との株価の変化を格納したリストexchange_rates_diffについて、プラスなら１，マイナスなら0を出力する
  if exchange_rates_diff[i] >= 0:
    tr_angle_mat.append(1)
  else:
    tr_angle_mat.append(0)

# 訓練用データ
train_feature_arr = np.array(tr_input_mat) #作成した学習データ
train_label_arr   = np.array(tr_angle_mat) #作成した教師データ

"""
検証用データの加工
"""
prev_pn_test = df_test["pn"][0]
prev_exch_test =df_test["Price"][0]

exchange_dates_test = [] # 日付けを格納するリスト

pn_rates_test      = []
pn_rates_diff_test = []

exchange_rates_test      = []
exchange_rates_diff_test = []

for i in range(len(x_test)):
  time = df_test.index[i]
  pn_val = df_test["pn"][i]
  exch_val =df_test["Price"][i]

  exchange_dates_test.append(time)
  pn_rates_diff_test.append(pn_val - prev_pn_test)
  exchange_rates_diff_test.append(exch_val - prev_exch_test)

  prev_pn_test = pn_val
  prev_exch_test = exch_val

data_len = len(pn_rates_diff_test)
test_input_mat = []
test_angle_mat = []

for i in range(INPUT_LEN, data_len):
  test_arr = []

  for j in range(INPUT_LEN):
    test_arr.append(exchange_rates_diff_test[i - INPUT_LEN + j])
    test_arr.append(pn_rates_diff_test[i - INPUT_LEN + j])
  test_input_mat.append(test_arr)

  if exchange_rates_diff[i] >= 0:
    test_angle_mat.append(1)
  else:
    test_angle_mat.append(0)

# 検証用データ
test_feature_arr = np.array(test_input_mat)
test_label_arr   = np.array(test_angle_mat)

test_feature_arr

models = [KNeighborsClassifier(), LogisticRegression(random_state=42), LinearSVC(C=0.01,max_iter=10000, random_state=42),\
          SVC(random_state=42), DecisionTreeClassifier(random_state=42), RandomForestClassifier(random_state=42)]

num = [0,1,2,3,4,5]

model_diclist = dict(zip(num, models))

accuracy = {} # 正答率を格納する辞書型リスト

for i, model in enumerate(models):
  model.fit(train_feature_arr, train_label_arr)
  score = model.score(test_feature_arr, test_label_arr)
  model_name = model.__class__.__name__
  accuracy[model_name] = score


  print("--Method:", model_name, "--")
  print("Test score:{}".format(score))
  print()

# 正答率の高い順にモデルを並び替える
accuracy    = dict(sorted(accuracy.items(), key=lambda x:x[1], reverse=True))
keys_list   = list(accuracy.keys())
values_list = list(accuracy.values())

# 棒グラフを出力
fig, ax = plt.subplots()
bars = ax.bar(keys_list, values_list)

# 棒グラフ上部に正答率を表示する
for bar in bars:
  height = bar.get_height()
  ax.text(bar.get_x() + bar.get_width()/2., height, float(round(height, 2)), ha="center", va="bottom")

# 正答率を可視化
plt.xlabel("model_number")
plt.xticks(rotation = 80)
plt.ylabel("accuracy")
plt.title("accuracy of each model_rev1")
plt.ylim(0,1)
plt.show()

実行結果は以下の通りです。

今回は、Aidemy講座に則り、６つのモデルを使用しましたが、実行の結果、Linear SVCが最も高い正答率（0.69）を示しました。

８．終わりに

パラメータの調整を行う等し、作成した機械学習モデルの正答率を向上させるチャレンジもできればなお良かったです。今後、学習を続けて応用も行えるようにしたいと思います。
今回のAidemyの受講を通して、Pythonの基礎を体系的に身につけることができました。Aidemyの講座は、不明点等が生じた場合はチューターに技術カウンセリングにのっていただくことができ、効率的に学習を進めることができました。
所属企業においては、データベースマーケティングの部署やマーケット部門等で、一部の社員がPythonを活用して業務の効率化・高度化を行っているようです。
私も、今回の受講で獲得した知識をベースに、引き続きデータ分析分野の自主学習を続けていくことで、社内においても希少な専門性を身につけたデジタル人材に成長していきたいと思います。

９．参考にしたサイト・文献等

・「SNS情報と高頻度取引データを用いた株価リターンの実証分析」～Twitterと高頻度取引との関係～
・Twitter APIのKeyやSecretの取得・確認手順※2023年10月最新
・Twitter API有料(Basic)、無料プラン利用開始手順※2023年9月最新
・単語感情極性対応表
・S&P 500 Stocks (daily updated)