香川大学SLPからお届け！（Vol.77掲載）

著者：石上椋一

毎年年末に開催されている競馬の「有馬記念」というレースをご存知でしょうか。今回は、その有馬記念の順位を予測するAIを開発した話を紹介します。2021年12月には、開発したAIを使って第66回有馬記念の順位を予測してみました。結果がどうだったのかについては、記事の最後に書いています。

シェルスクリプトマガジン Vol.77は以下のリンク先でご購入できます。

図2　RankNetを実装するPythonコード

from tensorflow.keras import layers, Model, Input
from tensorflow.nn import leaky_relu

class RankNet(Model):
  def __init__(self):
    super().__init__()
    self.dense = [layers.Dense(16, activation=leaky_relu),
                  layers.Dense(8, activation=leaky_relu)]
    self.o = layers.Dense(1, activation='linear')
    self.oi_minus_oj = layers.Subtract()
  def call(self, inputs):
    xi, xj = inputs
    densei = self.dense[0](xi)
    densej = self.dense[0](xj)
    for dense in self.dense[1:]:
      densei = dense(densei)
      densej = dense(densej)
    oi = self.o(densei)
    oj = self.o(densej)
    oij = self.oi_minus_oj([oi, oj])
    output = layers.Activation('sigmoid')(oij)
    return output

from tensorflow.keras import layers, Model, Input

from tensorflow.nn import leaky_relu

class RankNet(Model):

def __init__(self):

super().__init__()

self.dense = [layers.Dense(16, activation=leaky_relu),

layers.Dense(8, activation=leaky_relu)]

self.o = layers.Dense(1, activation='linear')

self.oi_minus_oj = layers.Subtract()

def call(self, inputs):

xi, xj = inputs

densei = self.dense[0](xi)

densej = self.dense[0](xj)

for dense in self.dense[1:]:

densei = dense(densei)

densej = dense(densej)

oi = self.o(densei)

oj = self.o(densej)

oij = self.oi_minus_oj([oi, oj])

output = layers.Activation('sigmoid')(oij)

return output

図3　データの整理とラベル付けをするPythonコード

import pandas as pd
import numpy as np
from itertools import combinations

years = [2020,2019,2018,2017,2016,2015,2014,2013,2012,2011]
# ここで任意のCSVファイルを指定する
# 今回はCSVにデータを残しているため、CSVから読み込んでいる
df = pd.read_csv(CSVFile)
df["タイム指数2-3"] = df["タイム指数2"] - df["タイム指数3"]
index_num = 0
xi = xj = pij = pair_ids = pair_query_id = []
for year in years:
  one_year_Data = df[df['年数'] == year]
  index_list = [i for i in range(len(one_year_Data))]
  random.shuffle(index_list)
  for pair_id in combinations(index_list, 2):
    pair_query_id.append(year)
    pair_ids.append(pair_id)
    i = pair_id[0]
    j = pair_id[1]
    xi.append([one_year_Data.at[i+index_num,"タイム指数2"],
               one_year_Data.at[i+index_num,"タイム指数2-3"],
               one_year_Data.at[i+index_num,"上り"]])
    xj.append([one_year_Data.at[j+index_num,"タイム指数2"],
               one_year_Data.at[j+index_num,"タイム指数2-3"],
               one_year_Data.at[j+index_num,"上り"]])
    if one_year_Data.at[i+index_num,"順位"] == one_year_Data.at[j+index_num,"順位"] :
      pij_com = 0.5
    elif one_year_Data.at[i+index_num,"順位"] > one_year_Data.at[j+index_num,"順位"] :
      pij_com = 0
    else:
      pij_com = 1
    pij.append(pij_com)
  index_num += len(one_year_Data)
  index_list.clear()
xi = np.array(xi)
xj = np.array(xj)
pij = np.array(pij)
pair_query_id = np.array(pair_query_id)

import pandas as pd

import numpy as np

from itertools import combinations

years = [2020,2019,2018,2017,2016,2015,2014,2013,2012,2011]

# ここで任意のCSVファイルを指定する

# 今回はCSVにデータを残しているため、CSVから読み込んでいる

df = pd.read_csv(CSVFile)

df["タイム指数2-3"] = df["タイム指数2"] - df["タイム指数3"]

index_num = 0

xi = xj = pij = pair_ids = pair_query_id = []

for year in years:

one_year_Data = df[df['年数'] == year]

index_list = [i for i in range(len(one_year_Data))]

random.shuffle(index_list)

for pair_id in combinations(index_list, 2):

pair_query_id.append(year)

pair_ids.append(pair_id)

i = pair_id[0]

j = pair_id[1]

xi.append([one_year_Data.at[i+index_num,"タイム指数2"],

one_year_Data.at[i+index_num,"タイム指数2-3"],

one_year_Data.at[i+index_num,"上り"]])

xj.append([one_year_Data.at[j+index_num,"タイム指数2"],

one_year_Data.at[j+index_num,"タイム指数2-3"],

one_year_Data.at[j+index_num,"上り"]])

if one_year_Data.at[i+index_num,"順位"] == one_year_Data.at[j+index_num,"順位"] :

pij_com = 0.5

elif one_year_Data.at[i+index_num,"順位"] > one_year_Data.at[j+index_num,"順位"] :

pij_com = 0

else:

pij_com = 1

pij.append(pij_com)

index_num += len(one_year_Data)

index_list.clear()

xi = np.array(xi)

xj = np.array(xj)

pij = np.array(pij)

pair_query_id = np.array(pair_query_id)

図4　学習用データと評価用データを仕分けるPythonコード

from sklearn.model_selection import train_test_split

xi_train, xi_test, xj_train, xj_test, pij_train, pij_test, \
pair_id_train, pair_id_test = train_test_split(
  xi, xj, pij, pair_ids, test_size=0.2, stratify=pair_query_id)

from sklearn.model_selection import train_test_split

xi_train, xi_test, xj_train, xj_test, pij_train, pij_test, \

pair_id_train, pair_id_test = train_test_split(

xi, xj, pij, pair_ids, test_size=0.2, stratify=pair_query_id)

図5　コンパイルと学習を実施するPythonコード

ranknet = RankNet()
# コンパイル
ranknet.compile(optimizer='sgd', loss='binary_crossentropy',metrics=['accuracy'])
# 学習
ranknet.fit([xi_train, xj_train], pij_train,
            epochs=85, batch_size=4,
            validation_data=([xi_test, xj_test], pij_test))

ranknet = RankNet()

# コンパイル

ranknet.compile(optimizer='sgd', loss='binary_crossentropy',metrics=['accuracy'])

# 学習

ranknet.fit([xi_train, xj_train], pij_train,

epochs=85, batch_size=4,

validation_data=([xi_test, xj_test], pij_test))

図8　ResNet50を使った学習モデルを実装するPythonコード

from tensorflow.keras.applications.resnet50 import ResNet50
from keras.layers import Dense, Dropout, Input, Flatten

# ResNetの準備
input_tensor = Input(shape=(64, 64, 3))
resnet50 = ResNet50(include_top=False, weights='imagenet',
                    input_tensor=input_tensor)
# FC層の準備
fc_model = Sequential()
fc_model.add(Flatten(input_shape=resnet50.output_shape[1:]))
fc_model.add(Dense(512, activation='relu'))
fc_model.add(Dropout(0.3))
fc_model.add(Dense(2, activation='sigmoid'))
# モデルの準備
resnet50_model = Model(resnet50.input, fc_model(resnet50.output))
# ResNet50の一部の重みを固定
for layer in resnet50_model.layers[:100]:
  layer.trainable = False

from tensorflow.keras.applications.resnet50 import ResNet50

from keras.layers import Dense, Dropout, Input, Flatten

# ResNetの準備

input_tensor = Input(shape=(64, 64, 3))

resnet50 = ResNet50(include_top=False, weights='imagenet',

input_tensor=input_tensor)

# FC層の準備

fc_model = Sequential()

fc_model.add(Flatten(input_shape=resnet50.output_shape[1:]))

fc_model.add(Dense(512, activation='relu'))

fc_model.add(Dropout(0.3))

fc_model.add(Dense(2, activation='sigmoid'))

# モデルの準備

resnet50_model = Model(resnet50.input, fc_model(resnet50.output))

# ResNet50の一部の重みを固定

for layer in resnet50_model.layers[:100]:

layer.trainable = False

図9　コンパイルと学習を実施するPythonコード

# コンパイル
resnet50_model.compile(optimizer='sgd',
                       loss='categorical_crossentropy',
                       metrics=['accuracy'])
# 学習
history = resnet50_model.fit(train_generator, 
                             batch_size=4, 
                             epochs=50, verbose=1,
                             validation_data=val_generator)

# コンパイル

resnet50_model.compile(optimizer='sgd',

loss='categorical_crossentropy',

metrics=['accuracy'])

# 学習

history = resnet50_model.fit(train_generator,

batch_size=4,

epochs=50, verbose=1,

validation_data=val_generator)