はじめに
競馬の着順予測をする際、出走する競走馬の両親の情報も加味して予測を行いたい。
競馬の順位予測の精度がなかなか向上しません。
学習データの期間、評価データの期間を調整しても、精度はあまり変わりませんでした。順位予測モデルを作成する際の特徴量(現状の特徴量は距離、天候、馬場状態など)を増やすことで精度を向上させることができないか。どんな特徴量を追加するのがよいのか、競走馬の能力はどういったことに影響するのかを調べてみました。
競走馬の能力に影響するのは下記3点であることを知りました。
- 影響1.遺伝子
競走馬が持つ遺伝子は、両親から受け継いでいるものが多いため、両親が持つ遺伝子によって競走馬の能力が影響されると考えられる。
例えば、父親が持つ遺伝子によって、筋肉量や体格が影響する場合がある。
- 影響2.血統
競走馬の祖先の中に、競走馬として優れた成績を残した馬がいる場合、その血統に優れた能力があると考えられる。
種牡馬や繁殖牝馬にも、競走馬として優れた成績を残した馬が多いことがあり、その血統を持つ馬にも、競走馬として優れた能力が期待される。
- 影響3.環境要因
競走馬が持つ能力は、遺伝子だけでなく、環境要因によっても影響されます。競走馬の成長過程での栄養状態や運動量、
調教方法などが、競走馬の能力に影響を与えることが知られている。
学習データに「影響1.遺伝子」=「両親の情報」を追加することにしました。
Webスクレイピングでレース結果を取得する際、「両親の情報」も追加する処理を含めました。
下記のようにC列に父親の名前、D列に母親の名前を追加したレース結果を取得することができるようになりました。
更新したプログラムの概要
更新したプログラム：get_raceResults.py
netkeibaから過去のレース情報を取得するプログラムです。
このプログラムを更新して、両親の情報も取得するようにしました。
主な更新内容は、両親の情報を取得する関数「get_parents_name」の追加です。
その他の更新として、各特徴量の取得処理をそれぞれ関数化、コメントの修正、不要な処理の削除を行いました。
追加した主な関数：get_parents_name
# レースに出場する出走馬の両親の情報を取得する
# 引数:レース情報のhtml:tag
# 戻値:各出走馬の父親、母親の名前:list
def get_parents_name(soup):
race_table_data = soup.find(class_='race_table_01 nk_tb_common') # 検索結果のテーブルを取得する
b_ml_list = [] # 各出走馬の父親の名前を保持するリスト
b_fml_list = [] # 各出走馬の母親の名前を保持するリスト
for element in race_table_data.find_all('a'):
url = element.get('href') # リンクページを取得する
# リンクページがjavascriptの場合は、次のリンクページの処理に移る
if 'javascript' in url:
continue
# リンクページの絶対URLを作成する
link_url = urllib.parse.urljoin('https://db.netkeiba.com', url)
# 出走馬の情報のみを抽出する
# 出走馬の情報のURLは'https://db.netkeiba.com/horse/yyyyXXXXXXX'となる
# 出走馬の情報以外のリンクページを除外するための単語リストを用意する
word_list = ['jockey', 'result', 'sum', 'list', 'movie',
'premium', 'horse_training', 'horse_comment']
# 出走馬の情報以外のリンクページだった場合、countを+1する
count = 0
for word in word_list:
if word in link_url:
count += 1
# 一致している単語が0のリンクページのURLのリストを作成する
if count == 0:
# 血統情報を取得する
res = requests.get(link_url)
soup_blood_table = BeautifulSoup(res.content, 'lxml')
# 父親の名前を取得する
elements_b_ml = soup_blood_table.find_all(class_='b_ml')
if len(elements_b_ml) == 0:
print(link_url)
return 'failed_to_get_b_ml'
element_b_ml = elements_b_ml[0].text.replace('\n', '')
b_ml_list.append(element_b_ml)
# 母親の名前を取得する
elements_b_fml = soup_blood_table.find_all(class_='b_fml')
if len(elements_b_ml) == 0:
print(link_url)
return 'failed_to_get_b_fml'
element_b_fml = elements_b_fml[1].text.replace('\n', '')
b_fml_list.append(element_b_fml)
return b_ml_list, b_fml_list更新したプログラム
プログラムの全容は下記となります。
# netkeibaから過去のレース情報を取得する
# 呼び出し元:無し
# 取得したデータは、./data/raceresults/に開始年_開始月_終了年_終了月_raceresults.csvを出力する
# 取得する期間の設定方法
# 例) 開始を2023年1月、終了を2023年12年の場合
# start_year = str(2023)
# end_year = str(2023)
# start_month = str(1)
# end_month = str(12)
# end_monthに設定する月は未来でも設定可能である
# seleniumの動作で描画待ちの秒数を手動で設定しているので、実行環境に応じて調整する必要がある
# 調整箇所
# 1. main関数内のbrowser.implicitly_waitの数値
# 2. search関数内のtime.sleepの数値
# 3. click_next関数内のtime.sleepの数値
# ライブラリの読み込み
import pandas as pd
import urllib
import requests
import re
import time
import os
from webdriver_manager.chrome import ChromeDriverManager
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import Select
from bs4 import BeautifulSoup
# プログレスバーを表示するためのライブラリを読み込む
from tqdm import tqdm
# スクレイピングする期間を指定する
start_year = str(2021)
end_year = str(2021)
start_month = str(1)
end_month = str(12)
# メイン処理
# 引数:無し
# 戻値:無し
def main():
# プログラムの開始時刻を取得する
start_time = time.time()
# chrome起動時のエラーを消す
chrome_options = webdriver.ChromeOptions()
chrome_options.add_experimental_option('excludeSwitches', ['enable-logging'])
# webdriver_managerで最適なchromeのバージョンをインストールして設定する
browser = webdriver.Chrome(ChromeDriverManager().install(), options=chrome_options)
# 競馬データベースを開く
browser.get('https://db.netkeiba.com/?pid=race_search_detail')
browser.implicitly_wait(20) # 指定した要素が見つかるまでの待ち時間を20秒と設定する
search_race(browser) # 検索条件を設定して検索する
link_list = [] # リンクページのURLリスト
no_click_next = 0 # 0:次のページ無し、1:次のページ有り
count_click_next = 0 # 0:検索結果の1ページ目、1以上:検索結果の2ページ目以上
while no_click_next == 0:
# レース結果のリンクページのURLを取得する
link_list = make_raceurl_list(browser, link_list)
# 「次」をクリックし、「次」の有無とクリックした回数を返す
no_click_next, count_click_next = click_next(browser, count_click_next)
df_race_result = pd.DataFrame()
# レース結果のリンクページにアクセスして、レース結果を取得する
# 日本レース以外のレース結果は取得しない
# 日本以外のレースのリンクページには、/race/2022H1a00108/のようにアルファベットが含まれている
# isdecimalで文字列が数字のみかを判定している
print('レース結果の詳細を取得します')
for url in tqdm(link_list):
if url.split('/')[-2].isdecimal():
detail_raceresult = get_detail_raceresult(url)
if type(detail_raceresult) == str:
print('レース情報の取得に失敗しました')
else:
df_race_result = pd.concat([df_race_result, detail_raceresult], axis=0)
# CSVにレース結果を保存する
# レース結果のCSVを保存するフォルダが無ければ作成する
os.makedirs('./data/raceresults', exist_ok=True)
file_name = start_year + '_' + start_month + '_' + end_year + '_' + end_month + '_raceresults.csv'
file_path = './data/raceresults/' + file_name
df_race_result.to_csv(file_path, encoding='cp932', header=False, index=False, errors="ignore")
# プログラムの実行時間を表示する
run_time = time.time() - start_time
print('実行時間:{:.2f}秒'.format(run_time))
# 検索条件を設定して検索する
# 引数:webdriver
# 戻値:無し
def search_race(browser):
print('検索条件を設定します')
# 競争種別で「芝」と「ダート」にチェックを入れる
elem_check_track_1 = browser.find_element(By.ID, value='check_track_1')
elem_check_track_2 = browser.find_element(By.ID, value='check_track_2')
elem_check_track_1.click()
elem_check_track_2.click()
# 期間を2021年から2022年に設定する
elem_start_year = browser.find_element(By.NAME, value='start_year')
elem_start_year_select = Select(elem_start_year)
elem_start_year_select.select_by_value(start_year)
# 月を指定する場合は、select_by_valueで月数を指定する
elem_start_month = browser.find_element(By.NAME, value='start_mon')
elem_start_month_select = Select(elem_start_month)
elem_start_month_select.select_by_value(start_month)
elem_end_year = browser.find_element(By.NAME, value='end_year')
elem_end_year_select = Select(elem_end_year)
elem_end_year_select.select_by_value(end_year)
# 月を指定する場合は、select_by_valueで月数を指定する
elem_end_month = browser.find_element(By.NAME, value='end_mon')
elem_end_month_select = Select(elem_end_month)
elem_end_month_select.select_by_value(end_month)
# 画面を下にスクロールする
browser.execute_script('window.scrollTo(0, 400);')
# 表示件数を100件にする
elem_list = browser.find_element(By.NAME, value='list')
elem_list_select = Select(elem_list)
elem_list_select.select_by_value('100')
# 検索をクリック(submit)する
elem_search = browser.find_element(By.CLASS_NAME, value='search_detail_submit')
time.sleep(4)
elem_search.submit()
# 取得したHTMLからレース結果のURLを抽出し、URLリストに追加してURLリストを作成する
# 引数:webdriver、リンクページのURLリスト
# 戻値:リンクページのURLリスト
def make_raceurl_list(browser, link_list):
print('取得したHTMLからレース結果のURLを抽出します')
html = browser.page_source.encode('utf-8') # UTF-8でHTMLを取得する
soup = BeautifulSoup(html, 'html.parser') # 検索結果をbeautifulSoupで読み込む
table_data = soup.find(class_='nk_tb_common') # 検索結果のテーブルを取得する
for element in table_data.find_all('a'):
url = element.get('href') # リンクページを取得する
# リンクページがjavascriptの場合は、次のリンクページの処理に移る
if 'javascript' in url:
continue
# リンクページの絶対URLを作成する
link_url = urllib.parse.urljoin('https://db.netkeiba.com', url)
# レース結果のみを抽出する
# レース結果のURLは'https://db.netkeiba.com/rase/yyyymmddXXXX'となる
# 馬情報のURLは'https://db.netkeiba.com/horse/yyyymmddXXXX'
# 騎手情報のURLは'https://db.netkeiba.com/jockey/result/recent/'
# レース結果以外のリンクページを除外するための単語リストを用意する
word_list = ['horse', 'jockey', 'result', 'sum', 'list', 'movie']
tmp_list = link_url.split('/') # リンクページのURLを'/'で分割する
and_list = set(word_list) & set(tmp_list) # word_listとtmp_listを比較し、一致している単語を抽出する
# 一致している単語が0のリンクページのURLのリストを作成する
if len(and_list) == 0:
link_list.append(link_url)
print('URLを抽出しました')
return link_list
# 画面下にスクロールして「次」をクリックする
# 引数:webdriver
# 戻値:「次」の有無(1/0)、「次」をクリックした回数
def click_next(browser, count_click_next):
# 画面を下にスクロールする
browser.execute_script('window.scrollTo(0, 2500);')
time.sleep(3)
# 次をクリックする
# 検索1ページ目のxpathは2ページ以降とは異なるため、count_click_nextで
# 検索1ページ目なのかを判定している
if count_click_next == 0:
xpath = '//*[@id="contents_liquid"]/div[2]/ul[1]/li[14]/a/span/span'
elem_search = browser.find_element(By.XPATH, value=xpath)
elem_search.click()
no_click_next = 0
else:
# 検索最後のページで次をクリックしようとすると例外処理が発生する
# exceptで例外処理を取得し、no_click_nextに1を代入する
try:
xpath = '//*[@id="contents_liquid"]/div[2]/ul[1]/li[14]/a/span/span'
elem_search = browser.find_element(By.XPATH, value=xpath)
time.sleep(2)
elem_search.click()
no_click_next = 0
except:
print('次のページ無し')
no_click_next = 1
count_click_next += 1 # ページ数を判別するためのフラグに1を加算する
return no_click_next, count_click_next
# レース結果の詳細を取得する
# 引数:取得するレース結果のURL
# 戻値:取得して整形したレース結果の情報
def get_detail_raceresult(url):
res = requests.get(url) # 指定したURLからデータを取得する
soup = BeautifulSoup(res.content, 'lxml') # content形式で取得したデータをhtml形式で分割する
# 各出走馬の両親の名前を取得する
parents_name = get_parents_name(soup)
if type(parents_name) == tuple:
b_ml_list = parents_name[0]
b_fml_list = parents_name[1]
else:
return 'failed_to_get_info'
# レース名を取得する
race_name = get_race_name(soup)
# ラウンド数を取得する
race_round = get_race_round(soup)
# 開催日を取得する
race_date = get_race_date(soup)
# 開催場所を取得する
race_place = get_race_place(soup)
# レースの距離を取得する
race_distance = get_race_distance(soup)
# 開催場所の天候を取得する
race_weather = get_race_weather(soup)
# 開催場所の馬場状態を取得する
race_condition = get_race_condition(soup)
# tableデータを抽出する
tables = soup.find('table', attrs={'class': 'race_table_01'})
tables = tables.find_all('tr')
# レース情報/結果を取得する
df_raceresults = pd.DataFrame()
for table in tables[1:]:
raceresults_row = table.text.split('\n')
df_raceresults_row = pd.Series(raceresults_row)
df_raceresults = pd.concat([df_raceresults, df_raceresults_row], axis=1)
# 学習に必要な情報のみを抽出する
# 着順:要素No.1、馬名:要素No.5、性齢:要素No.7、斤量:要素No.8、騎手:要素No.10、
# タイム:要素No.12、上り:要素No.21、単勝:要素No.23、人気:要素No.24、馬体重:要素No.25の情報を抽出する
df_raceresults_extracted_columns = pd.DataFrame()
# レース結果のテーブルが異なることがあるので、必要なカラム列の番号をそれぞれで用意する
columns_list1 = [1, 5, 7, 8, 10, 12, 21, 23, 24, 25]
columns_list2 = [1, 5, 7, 8, 10, 12, 18, 20, 21, 22]
# テーブルの状態次第で必要なカラム列の番号を変える
# 17列は「通過」の情報が記録されているが、テーブルによって'**'となっている場合となっていない場合の2種類のテーブルが存在している
# 17列が'**'となっている場合は、カラム列はcolumns_list1を使用する
df_speed_index = df_raceresults.iloc[17]
if df_speed_index.values[0] == '**':
for i in columns_list1:
df_extracted_columns = df_raceresults.iloc[i]
df_raceresults_extracted_columns = pd.concat([df_raceresults_extracted_columns,
df_extracted_columns], axis=1)
else:
for i in columns_list2:
df_extracted_columns = df_raceresults.iloc[i]
df_raceresults_extracted_columns = pd.concat([df_raceresults_extracted_columns,
df_extracted_columns], axis=1)
# カラム名を設定する
tmp = ['着順', '馬名', '性齢', '斤量', '騎手', 'タイム', '上り', '単勝', '人気', '馬体重']
df_columns = pd.Series(tmp)
df_raceresults_extracted_columns.columns = df_columns # index名を設定する
# 整形したレース情報を保存するデータフレームを用意する
df_reshaped_raceresults = pd.DataFrame()
try:
# 距離、天候、馬場、状態、開催日、レース名、開催場所、ラウンド、父親、母親の列を追加する
df_raceresults_extracted_columns['距離'] = race_distance
df_raceresults_extracted_columns['天候'] = race_weather[1]
df_raceresults_extracted_columns['馬場'] = race_condition[0]
df_raceresults_extracted_columns['状態'] = race_condition[1]
df_raceresults_extracted_columns['開催日'] = race_date
df_raceresults_extracted_columns['レース名'] = race_name
df_raceresults_extracted_columns['開催場所'] = race_place
df_raceresults_extracted_columns['ラウンド'] = race_round
df_raceresults_extracted_columns['父親'] = b_ml_list
df_raceresults_extracted_columns['母親'] = b_fml_list
# 説明変数として使用しない列を削除する
df_raceresults_extracted_columns.drop(['着順', '上り'], axis=1, inplace=True)
# 目的変数にするタイムを1列に変更する
df_reshaped_raceresults = df_raceresults_extracted_columns.reindex(columns=['タイム', '馬名', '父親', '母親',
'性齢', '斤量', '騎手', '単勝',
'人気', '馬体重', '距離', '天候',
'馬場', '状態', '開催日', 'レース名',
'開催場所', 'ラウンド'])
except:
print('レース情報を取得できませんでした')
return df_reshaped_raceresults
# レースに出場する出走馬の両親の情報を取得する
# 引数:レース情報のhtml:tag
# 戻値:各出走馬の父親、母親の名前:list
def get_parents_name(soup):
race_table_data = soup.find(class_='race_table_01 nk_tb_common') # 検索結果のテーブルを取得する
b_ml_list = [] # 各出走馬の父親の名前を保持するリスト
b_fml_list = [] # 各出走馬の母親の名前を保持するリスト
for element in race_table_data.find_all('a'):
url = element.get('href') # リンクページを取得する
# リンクページがjavascriptの場合は、次のリンクページの処理に移る
if 'javascript' in url:
continue
# リンクページの絶対URLを作成する
link_url = urllib.parse.urljoin('https://db.netkeiba.com', url)
# 出走馬の情報のみを抽出する
# 出走馬の情報のURLは'https://db.netkeiba.com/horse/yyyyXXXXXXX'となる
# 出走馬の情報以外のリンクページを除外するための単語リストを用意する
word_list = ['jockey', 'result', 'sum', 'list', 'movie',
'premium', 'horse_training', 'horse_comment']
# 出走馬の情報以外のリンクページだった場合、countを+1する
count = 0
for word in word_list:
if word in link_url:
count += 1
# 一致している単語が0のリンクページのURLのリストを作成する
if count == 0:
# 血統情報を取得する
res = requests.get(link_url)
soup_blood_table = BeautifulSoup(res.content, 'lxml')
# 父親の名前を取得する
elements_b_ml = soup_blood_table.find_all(class_='b_ml')
if len(elements_b_ml) == 0:
print(link_url)
return 'failed_to_get_b_ml'
element_b_ml = elements_b_ml[0].text.replace('\n', '')
b_ml_list.append(element_b_ml)
# 母親の名前を取得する
elements_b_fml = soup_blood_table.find_all(class_='b_fml')
if len(elements_b_ml) == 0:
print(link_url)
return 'failed_to_get_b_fml'
element_b_fml = elements_b_fml[1].text.replace('\n', '')
b_fml_list.append(element_b_fml)
return b_ml_list, b_fml_list
# レース名を取得する
# 引数:レース情報のhtml:tag
# 戻値:レース名:str
def get_race_name(soup):
race_name = soup.find_all('h1')
race_name = race_name[1].text
return race_name
# ラウンド数を取得する
# 引数:レース情報のhtml:tag
# 戻値:レースのラウンド:str
def get_race_round(soup):
race_round = soup.find('dl', attrs={'class': 'racedata fc'})
race_round = race_round.find_all('dt')
race_round = race_round[0].text
race_round = race_round.replace('\n', '')
return race_round
# 開催日を取得する
# 引数:レース情報のhtml:tag
# 戻値:開催日:str
def get_race_date(soup):
# 開催日を取得する
race_base_info = soup.find('p', attrs={'class': 'smalltxt'})
rase_base_info_text = race_base_info.text.replace(u'\xa0', u' ')
words = rase_base_info_text.split(' ')
race_date = words[0]
return race_date
# 開催場所を取得する
# 引数:レース情報のhtml:tag
# 戻値:開催場所:str
def get_race_place(soup):
race_base_info = soup.find('p', attrs={'class': 'smalltxt'})
rase_base_info_text = race_base_info.text.replace(u'\xa0', u' ')
words = rase_base_info_text.split(' ')
race_place = words[1]
return race_place
# レースの距離を取得する
# 引数:レース情報のhtml:tag
# 戻値:レースの距離:int
def get_race_distance(soup):
race_info = soup.find('diary_snap_cut')
race_info_text = race_info.text.replace(u'\xa0', u' ') # ノーブレークスペースを削除
race_info_text = race_info.text.replace(u'\xa5', u' ') # Webページの文字参照を削除
words = race_info_text.split('/')
words[0] = re.sub(r'2周', '', words[0]) # レース情報の'2周'を空文字に置き換える
race_distance = int(re.sub(r'\D', '', words[0])) # レース距離だけを取り出してint型で保存する
return race_distance
# 開催場所の天候を取得する
# 引数:レース情報のhtml:tag
# 戻値:開催場の天候:list
def get_race_weather(soup):
race_info = soup.find('diary_snap_cut')
race_info_text = race_info.text.replace(u'\xa0', u' ') # ノーブレークスペースを削除
race_info_text = race_info.text.replace(u'\xa5', u' ') # Webページの文字参照を削除
words = race_info_text.split('/')
race_weather = words[1].split(':')
return race_weather
# 開催場所の馬場状態を取得する
# 引数:レース情報のhtml:tag
# 戻値:開催場所の馬場状態:list
def get_race_condition(soup):
race_info = soup.find('diary_snap_cut')
race_info_text = race_info.text.replace(u'\xa0', u' ') # ノーブレークスペースを削除
race_info_text = race_info.text.replace(u'\xa5', u' ') # Webページの文字参照を削除
words = race_info_text.split('/')
race_condition = words[2].split(':')
return race_condition
if __name__ == "__main__":
main()実行方法
このプログラムは単体で実行することができます。
実行方法はREADME.mdの1.過去のレース情報取得を参照してください。
実行結果
下記のようなCSVファイルが出力されます。
出力ファイル：2022_1_2022_12_raceresults.csv
実行時の注意点
実行中は、CPUとメモリをそれほど使用しませんが、1年分を取得するのにおおよそ30時間かかりました。
実行中に端末の電源を落とさないようにしてください。
次回試したいこと
「影響1.遺伝子」＝「両親の情報」を特徴量に加えた学習データで順位予測モデルを作成してみます。
また、「影響2.血統」＝「3世代分の情報」と「影響3.環境要因」＝「調教師」も特徴量に加えれるようにしてみます。
コメント