競馬予測AIの作成⑰(モデル作成処理の改修)

Python

はじめに

本プログラムは、以前に作成したモデル生成プログラムの可読性と保守性を向上させたプログラムです。以前はモデル生成に3つプログラムを用いていました。それぞれのファイルをメンテナンスするのは手間でした。下記を行うことで可読性と保守性を向上させました。

<主な変更点>

  1. 3つのプログラムを1つのプログラムに統合
  2. メソッド名、変数名の変更(名前から用途が判るよう変更)
  3. コメントのスタイルをreStructuredTextスタイルに変更

参考にした書籍

楽天ブックス
¥2,420 (2024/04/11 14:07時点 | 楽天市場調べ)
\楽天ポイント4倍セール!/
楽天市場
\商品券4%還元!/
Yahooショッピング
¥3,190 (2024/01/23 07:45時点 | Yahooショッピング調べ)
\楽天ポイント4倍セール!/
楽天市場
\商品券4%還元!/
Yahooショッピング

プログラムの目的

競馬のレース結果データを基に、XGBoostを用いて予測モデルを作成し、予測モデルの性能を評価することを目的としています。

プログラムの動作概要

プログラムは以下の順序で動作します。

  1. 指定された入力パスからレース結果データ(CSVファイル)を読み込みます。
  2. データを前処理し、特徴量とラベルを抽出します。
  3. XGBoostアルゴリズムを用いてデータセットに対する予測モデルを訓練します。
  4. 訓練したモデルの性能を評価し、結果を保存します。
  5. 予測結果と的中率の分析を行い、その結果を保存します。

プログラムの主な機能

  • レース結果データの読み込みと前処理を行います。
  • 特徴量の抽出とカテゴリデータのエンコードを行います。
  • XGBoostモデルの訓練と保存します。
  • モデル性能の評価と結果の保存します。
  • 的中率の分析と結果の保存します。

プログラムで生成されるデータ

プログラムを実行すると下記のデータが生成されます。

データ 内容
gbm.pkl 訓練済みモデル
oe_x.pkl エンコードモデル
ハイパーパラメータ.txt 訓練時に使用したハイパーパラメータ
予測結果.csv 予測タイムと実タイムの比較
実タイムと予測タイムの散布図.svg 予測精度の視覚的評価
評価結果.txt R^2、RMSE、MAEの性能指標
予測タイムの重要度分析.svg 特徴量の重要度

<サンプル:ハイパーパラメータ.txt>

{'max_depth': 4, 'min_child_weight': 3, 'eta': 0.006703103364541497, 'subsample': 0.7225888856542279, 'colsample_bytree': 0.9058934098517724, 'alpha': 0.3703100584659331, 'lambda': 4.336819617562275, 'gamma': 0.27590111837793646, 'objective': 'reg:squarederror', 'eval_metric': 'rmse', 'tree_method': 'hist', 'device': 'cuda'}

<サンプル:予測結果.csvの一部>

 

<サンプル:実タイムと予測タイムの散布図.svg>

<サンプル:評価結果.txt>

R2: 0.9500124656251072
MAE: 2.8013089195721443
RMSE: 4.168081045419233

<サンプル:予測タイムの重要度分析.svg>

プログラムのユーザインターフェース

ユーザインターフェースは用意していません。

プログラムの構成

以下の主なモジュールから構成されます。

モジュール 概要
ModelCreator モデルの訓練、評価、的中率分析を行います。

プログラムのフローチャート

<補足>

  • 茶色:最初に実行される箇所
    マルチプロセスで動作するので最初に実行される箇所は2箇所となります。
  • 緑色:他の関数やメソッドを呼び出さない
  • 灰色:通常の関数、メソッド
  • 関数名、メソッド名の前の数字はプログラム内の行数

プログラムの関数とメソッドの概要

関数、メソッド 行数 概要
run 10 入力ディレクトリからCSVファイルを読み込み、各ファイルについてモデル訓練、性能評価、的中率分析を実行します。
prepare_features_and_labels 23 レース結果データから特徴量とラベルを準備します。
encode_categorical_features 31 訓練データセットとテストデータセットのカテゴリ変数をエンコードし、エンコーダーを保存します。
create_dmatrix_for_xgboost 41 XGBoostが使用するDMatrix形式でデータセットを変換します。
calculate_win_accuracy 51 単勝の的中率を計算します。
calculate_place_accuracy 57 複勝の的中率を計算します。
calculate_quinella_accuracy 66 馬連の的中率を計算します
calculate_quinella_place_accuracy 72 ワイド(馬連複)の的中率を計算します。
calculate_trio_accuracy 78 三連複の的中率を計算します
split_data_into_train_test 84 特定の日付でデータセットを訓練セットとテストセットに分割します。特定日が範囲外の場合はNoneを返します。
split_train_data_into_train_validation 97 訓練データセットをさらに訓練セットと検証セットに分割します
plot_actual_predicted 103 テストセットの実ラベルと予測ラベルで散布図を作成し、保存します。
plot_feature_importance 113 XGBoostモデルの特徴量重要度をグラフ化し、保存します。
optimize_model_hyperparameters 123 Optunaを使用してXGBoostモデルのハイパーパラメータを最適化します。
predict_and_save_model 143 最適化されたパラメータを使用してモデルを訓練し、予測を行い、モデルとハイパーパラメータを保存します。
evaluate_model_performance 162 モデルの性能を評価し、R^2スコア、MAE、RMSEを計算し、保存します。
save_predictions_to_csv 174 テストセットの特徴量、実ラベル、予測ラベルを含むCSVファイルを保存します。
add_real_and_predicted_ranks 183 レース結果データに実際の順位と予測順位を追加します。
analyze_and_save_win_rates 195 的中率を計算し、CSVファイルに保存します。
train_and_save_model 218 レース結果データを用いて予測モデルを訓練し、モデルと予測結果を保存します。

プログラムの実行環境

本プログラムは、Pythonのバージョン3.12.2で実行できることを確認しています。

本プログラムを実行するために必要なパッケージをインストールします。
Pythonの仮想環境を起動し、pipコマンドで下記のパッケージをインストールしてください。

(venv)> pip install pandas 
(venv)> pip install scikit-learn 
(venv)> pip install xgboost 
(venv)> pip install optuna 
(venv)> pip install matplotlib 
(venv)> pip install japanize_matplotlib
(venv)> pip install setuptools

プログラムの実行

「C:¥tmp」でプログラムを実行する手順となります。「C:¥tmp」が無ければ事前に作成しておいてください。また「C:¥tmp」以外で実行する場合は、使用するフォルダに読み替えてください。

  1. 本プログラムを実行する前に、netkeibaからレース結果を取得しておく必要があります。
    レース結果の取得は「競馬予測AIの作成⑯(レース結果取得プログラムのバージョンアップ)」を参照してください。
  2. 取得したレース結果を整形するプログラム「prepare_raceResults.py」を実行します。「prepare_raceResults.py」は「競馬予測AIの作成⑭(動作安定向上)」から取得してください。
  3. 取得したレース結果を整形するプログラム「prepare_raceResults.py」を実行する前に整形したレース結果を出力するための下記フォルダを作成します。
    「C:¥tmp\data\raceresults\prepared_raceresults」
  4. 下記のコマンドで「prepare_raceResults.py」を実行します。
    (venv)$ python web_scraping\prepare_raceResults.py
  5. 「プログラムの全コード」を保存するフォルダ「make_model」を作成してください。
  6. 「make_model」フォルダを作成したら、「プログラムの全コード」を「create_evaluate_models.py」として保存してください。
  7. 下記のようなフォルダ・ファイル構成になっていることを確認してください。
    C:\tmp
    │
    ├─venv
    │
    ├─data
    │   └─raceresults
    │       │  2013_1_2013_12_raceresults.csv
    │       │  2014_1_2014_12_raceresults.csv
    │       │  2015_1_2015_12_raceresults.csv
    │       │  2016_1_2016_12_raceresults.csv
    │       │  2017_1_2017_12_raceresults.csv
    │       │  2018_1_2018_12_raceresults.csv
    │       │  2019_1_2019_12_raceresults.csv
    │       │  2020_1_2020_12_raceresults.csv
    │       │  2021_1_2021_12_raceresults.csv
    │       │  2022_1_2022_12_raceresults.csv
    │       │  2023_1_2023_12_raceresults.csv
    │       │  2024_1_2024_12_raceresults.csv
    │       │
    │       └─prepared_raceresults
    │               2013_2024_prepared_raceresults.csv
    │               2014_2024_prepared_raceresults.csv
    │               2015_2024_prepared_raceresults.csv
    │               2016_2024_prepared_raceresults.csv
    │               2017_2024_prepared_raceresults.csv
    │               2018_2024_prepared_raceresults.csv
    │               2019_2024_prepared_raceresults.csv
    │               2020_2024_prepared_raceresults.csv
    │               2021_2024_prepared_raceresults.csv
    │               2022_2024_prepared_raceresults.csv
    │               2023_2024_prepared_raceresults.csv
    │               2024_2024_prepared_raceresults.csv
    │
    ├─make_model
    │      create_evaluate_models.py
    │
    └─web_scraping
            get_raceResults.py
            prepare_raceResults.py
  8. 下記のコマンドを実行すると、モデルの作成と評価を開始します。
    実行完了までの時間は、取得したレース結果の年数と実行環境により変わりますが、30~120分ほどで完了します。
    (venv)$ python make_model\create_evaluate_models.py
  9. 実行完了すると「C:¥tmp\data\model\yyyyMMdd_hhmmss」フォルダが作成されます。「yyyyMMdd_hhmmss」は実行した日時です。

プログラムの全コード

# 整形されたレース結果を基にモデルを作成し、そのモデルの評価結果を記録する
# 呼び出し元:
#   無し
#   事前にprepare_raceResult.pyを実行して整形されたレース結果を用意しておく
# 実行前の設定:
#   input_pathでレース結果のパスを設定する
#   output_pathでモデル出力先のパスを設定する
#   train_test_split_dateで訓練データとテストデータに分割するための日付を設定する
# 実行方法:
#   仮想環境で下記のコマンドを実行する
#   (venv) python create_evaluate_models.py
# 出力内容:
#   gbm.pkl(モデル),oe_x.pl(エンコードモデル),ハイパーパラメータ.txt,予測結果.csv
#   実タイムと予測タイムの散布図.svg,評価結果.txt(r2,rmse,mae),予測タイムの重要度分析.svg

import datetime
import os
import sys

import pandas as pd
import time
import numpy as np
from sklearn.preprocessing import OrdinalEncoder
from sklearn.model_selection import train_test_split
import xgboost as xgb
import optuna
from sklearn.metrics import r2_score, mean_absolute_error, mean_squared_error
import matplotlib.pyplot as plt
import japanize_matplotlib
import joblib
import csv


class ModelCreator:
    def __init__(self, input_path, output_path, train_test_split_date):
        self.input_path = input_path
        self.output_path = output_path
        self.train_test_split_date = train_test_split_date

    def run(self):
        '''
        入力ディレクトリ内の全てのCSVファイルに対してモデル訓練と評価を自動的に実行し、結果を出力ディレクトリに保存する。
        1. 入力ディレクトリからCSVファイル(レース結果データ)を読み込む。
        2. 読み込んだレース結果データ前処理を行い、訓練セットを準備する。
        3. XGBoostを用いて訓練を実施し、モデルを保存する。
        4. モデルの性能評価を行い、結果を保存する。
        5. 予測結果とともに、的中率の分析を行い、結果を保存する。
        :return: 無し
        '''
        current_date_time = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
        for file in os.listdir(self.input_path):
            if file.endswith(".csv"):
                race_results = pd.read_csv(os.path.join(self.input_path, file), encoding='cp932')
                race_results['date'] = pd.to_datetime(race_results['date'])
                file_split = file.split('_')
                dir_path = os.path.join(self.output_path, f'{current_date_time}/traindata_{file_split[0]}_{file_split[1]}/')
                os.makedirs(dir_path, exist_ok=True)

                self.train_and_save_model(race_results, dir_path, train_test_split_date)
                self.analyze_and_save_win_rates(dir_path)

    def prepare_features_and_labels(self, race_results):
        """
        レース結果のデータから特徴量とラベルを抽出する。
        'time'列をラベルとし、残りの列を特徴量として使用する。
        :param DataFrame race_results: レース結果のデータ。特徴量(出走馬の情報など)と予測対象(レースタイム)を含む。
        :return: 特徴量、ラベル
        :rtype: DataFrame, np.ndarray
        """
        features = race_results.drop(['time'], axis=1)  # 'time'列以外を特徴量として使用する
        labels = race_results['time'].values            # 'time'列をラベルとして使用する

        return features, labels

    def encode_categorical_features(self, all_train_features, test_features, model_save_dir_path):
        """
        OrdinalEncoderを使用して、訓練データセットとテストデータセットのカテゴリされた特徴量をエンコードし、
        その後エンコーダーを指定されたディレクトリに保存する。
        :param DataFrame all_train_features: 特徴量を含む訓練セット。
        :param DataFrame test_features: 特徴量を含むテストセット。
        :param str model_save_dir_path: エンコーダーを保存するディレクトリパス。
        :return: エンコードされた訓練セット、テストセット。
        :rtype: pd.DataFrame, pd.DataFrame
        """
        # 訓練データのラベルエンコード
        oe_x = OrdinalEncoder()
        encoded_train_features = oe_x.fit_transform(all_train_features) # Dataframeからndarrayに変わるとカラムがなくなり、
        encoded_train_features = pd.DataFrame(encoded_train_features)   # 特徴量分析で特徴量名がf0,f1,f2,,となるのでndarrayからDataframeに戻す
        encoded_train_features.columns = list(all_train_features.columns.values)  # カラムを付ける

        encoded_test_features = oe_x.fit_transform(test_features)   # Dataframeからndarrayに変わるとカラムがなくなり、
        encoded_test_features = pd.DataFrame(encoded_test_features) # 特徴量分析で特徴量名がf0,f1,f2,,となるのでndarrayからDataframeに戻す
        encoded_test_features.columns = list(test_features.columns.values) # カラムを付ける

        # エンコーダーを保存する
        encoder_save_path = model_save_dir_path + 'oe_x.pkl'
        joblib.dump(oe_x, encoder_save_path, compress=3)

        return encoded_train_features, encoded_test_features

    def create_dmatrix_for_xgboost(self, train_features, validation_features, encoded_test_features, train_lables, validation_labels, test_lables):
        """
        XGBoostが使用するDMatrix形式で、訓練セット、検証セット、テストセットを変換する。
        :param np.ndarray train_features: 特徴量を含む訓練セット。
        :param np.ndarray validation_features: 特徴量を含む検証セット。
        :param np.ndarray encoded_test_features: エンコード済みの特徴量を含むテストセット。
        :param np.ndarray train_lables: 正解ラベルを含む訓練用データ
        :param np.ndarray validation_labels: 正解ラベルを含む検証用データ
        :param nd.ndarray test_lables: 正解ラベルを含むテスト用データ
        :return: DMatrix形式に変換した訓練セット、検証セット、テストセット
        :rtype:xgb.DMatrix, xgb.DMatrix, xgb.DMatrix
        """
        train_dmatrix = xgb.DMatrix(train_features, label=train_lables)
        validation_dmatrix = xgb.DMatrix(validation_features, label=validation_labels)
        test_dmatrix = xgb.DMatrix(encoded_test_features, label=test_lables)

        return train_dmatrix, validation_dmatrix, test_dmatrix

    def calculate_win_accuracy(self, ranked_race_results):
        """
        単勝(英訳:win)の的中を確認する。
        実順位の1位と予測順位の1位を比較し、一致の場合は1を返し、不一致の場合は0を返す。
        :param ranked_race_results:
        :return:
        """
        return int(ranked_race_results.iloc[0]['rank_real'] == ranked_race_results.iloc[0]['rank_predict'])

    def calculate_place_accuracy(self, ranked_race_results):
        """
        複勝(英訳:place)の的中を確認する。
        8頭以上の出走レースでは、3着以内に予測順位1~3の有無を確認する。
        8頭未満が出走しているレースでは、2着以内に予測順位1~2の有無を確認する。
        予想の当たり1/外れ0を返す
        :param ranked_race_results:
        :return:
        """
        hores_num = len(ranked_race_results)
        if hores_num >= 8:
            return int(any(ranked_race_results.iloc[:3]['rank_predict'] <= 3))
        else:
            return int(any(ranked_race_results.iloc[:2]['rank_predict'] <= 2))

    def calculate_quinella_accuracy(self, ranked_race_results):
        """
        馬連(英訳:quinella)の的中を確認する。
        実順位の1位と2位、予測順位の1位,2位が一致するかを確認する。
        予測順位の1行目と2行目の合計が3(1位+2位)だった場合は当たり、それ以外は外れ。
        予想の当たり1/外れ0を返す。
        :param ranked_race_results:
        :return:
        """
        return int(sum(ranked_race_results.iloc[:2]['rank_predict']) == 3)

    def calculate_quinella_place_accuracy(self, ranked_race_results):
        """
        ワイド(英訳:quinella place)の的中を確認する。
        実順位の3着以内に予測順位1~3が2つ以上あるかを確認する。
        予想の当たり1/外れ0を返す。
        :param ranked_race_results:
        :return:
        """
        return int(sum(ranked_race_results.iloc[:3]['rank_predict'] <= 3) >= 2)

    def calculate_trio_accuracy(self, ranked_race_results):
        """
        三連複(英訳:trio)の的中を確認する。
        予測順位の1~3行目の合計が6(1位+2位+3位)だった場合は当たり、それ以外は外れ。
        予想の当たり1/外れ0を返す。
        :param ranked_race_results:
        :return:
        """
        return int(sum(ranked_race_results.iloc[:3]['rank_predict']) == 6)

    def split_data_into_train_test(self, features, labels, split_day):
        """
        指定した日付で、特徴量とラベルのレース結果を訓練セットとテストセットに分割する。
        指定した日付の前のレース結果データは訓練セット、それ以降のレース結果はテストセットに分類する。
        split_dayがfeatures['date']に含まれない場合はNoneを返す。
        :param DataFrame features: 特徴量('time'列が含まれていないレース結果)
        :param np.ndarray labels: ラベル('time'列のみ)
        :param str split_day: 訓練セットとテストセットで分割する基準日。YYYY-MM-DD形式の文字列。
        :return: 訓練用特徴量セット、テスト用特徴量セット、訓練用ラベルセット、テスト用ラベルセット
        :rtype: pd.DataFrame, pd.DataFrame, np.ndarray, np.ndarray
        """
        mday = pd.to_datetime(split_day)

        # split_dayがfeatures['date']の範囲内にない場合、Noneを返す
        if not ((features['date'] < mday).any() and (features['date'] >= mday).any()):
            return None, None, None, None

        train_index = features['date'] < mday   # 基準日より前のデータのインデックス(訓練セット)
        test_index = features['date'] >= mday   # 基準日以降のデータインデックス(テストセット)

        return features[train_index], features[test_index], labels[train_index], labels[test_index]

    def split_train_data_into_train_validation(self, encoded_train_features, all_train_labels):
        """
        エンコードした訓練セットを、さらに訓練セットと検証セットに分割する。
        モデルの性能を評価する際に使用する検証セットを用意する。
        :param DataFrame encoded_train_features: エンコードした訓練セット。
        :param np.ndarray all_train_labels: 訓練セットの正解ラベル。
        :return: 訓練セット、検証セット、訓練用正解ラベル、検証用正解ラベル
        :rtype: pd.DataFrame, pd.DataFrame, np.ndarray, np.ndarray
        """
        return train_test_split(encoded_train_features, all_train_labels, test_size=0.2, random_state=0)

    def plot_actual_predicted(self, test_labels, predicted_race_times, model_save_dir_path, r2):
        """
        テストセットの実際のラベルと予測されたラベルで散布図を作成する。
        R2スコアをグラフのタイトルに表示し、グラフをSVG形式で指定されたディレクトリに保存する。
        :param np.ndarray test_labels: テストセットの正解ラベル。
        :param np.ndarray predicted_race_times: モデルの予測結果。
        :param model_save_dir_path:  グラフを保存するディレクトリパス。
        :param float r2: モデルのR^2スコア。
        :return: 無し
        """
        plt.figure(figsize=(6, 6))
        plt.plot([test_labels.min(), test_labels.max()], [test_labels.min(), test_labels.max()], 'k--', lw=4)
        plt.scatter(test_labels, predicted_race_times, color='darkorange')
        plt.xlabel('実タイム')
        plt.ylabel('予測タイム')
        plt.title(f'実タイムと予測タイムの散布図 R2={r2:.4f}')
        plt.legend()
        plt.savefig(os.path.join(model_save_dir_path, '実タイムと予測タイムの散布図.svg'))
        plt.close()

    def plot_feature_importance(self, xgb_model, model_save_dir_path):
        """
        XGBoostによる特徴量の重要度をSVGファイルとして保存します。
        :param xgb.Booster xgb_model: 特徴量の重要度を表示するXGBoostのモデル。
        :param str model_save_dir_path: グラフを保存するディレクトリパス。
        :return: 無し
        """
        fig, ax = plt.subplots(figsize=(10, 8))
        xgb.plot_importance(xgb_model, ax=ax, importance_type='gain', title='予測タイムの重要度分析', xlabel='Gain',
                            ylabel='Features')
        plt.savefig(os.path.join(model_save_dir_path, '予測タイムの重要度分析.svg'))
        plt.close()

    def optimize_model_hyperparamters(self, trial, train_dmatrix, validation_dmatrix, test_dmatrix, test_labels):
        """
        Optunaを用いてXGBoostモデルのハイパーパラメータを最適化する。この関数はOptunaのtrialオブジェクトによって
        呼び出され、最適なパラメータを探索するために使用される。訓練セットと検証セットを用いてモデルを訓練し、
        検証セット上での性能を基にパラメータを評価する。
        :param optuna.trial.Trial trial: Optunaのオブジェクト。ハイパーパラメータの試行の結果を記録するために使用する。
        :param xgb.DMatrix train_dmatrix: DMatrix形式の訓練セット。
        :param xgb.DMatrix validation_dmatrix: DMatrix形式の検証セット。
        :param xgb.DMatrix test_dmatrix: DMatrix形式のテストセット。
        :param np.ndarray test_labels: テストセットの正解ラベル
        :return: テストセットに対する予測のRMSE(平均二乗誤差の平方根)
        :rtype: float
        """
        xgb_hyperparams = {
            'max_depth': trial.suggest_int('max_depth', 3, 10),
            'min_child_weight': trial.suggest_int('min_child_weight', 0, 10),
            'eta': trial.suggest_float('eta', 0.001, 0.1, log=True),
            'subsample': trial.suggest_float('subsample', 0.5, 1.0),
            'colsample_bytree': trial.suggest_float('colsample_bytree', 0.5, 1.0),
            'alpha': trial.suggest_float('alpha', 0.01, 10.0, log=True),
            'lambda': trial.suggest_float('lambda', 0.01, 10.0, log=True),
            'gamma': trial.suggest_float('gamma', 0.01, 10.0, log=True),
            'objective': 'reg:squarederror',
            'eval_metric': 'rmse',
            'tree_method': 'hist',  # GPUを使用しない場合は、コメントアウトする
            'device': 'cuda',  # GPUを使用しない場合は、コメントアウトする
        }
        # 訓練セットと検証セットでモデルを訓練する
        xgb_model = xgb.train(xgb_hyperparams, train_dmatrix, num_boost_round=10000, early_stopping_rounds=10, evals=[(validation_dmatrix, 'eval')])

        # テストセットに対して予測を行い、RMSEを計算する
        preds = xgb_model.predict(test_dmatrix)
        rmse = np.sqrt(mean_squared_error(test_labels, preds))

        return rmse

    def predict_and_save_model(self, study, train_dmatrix, validation_dmatrix, test_dmatrix, model_save_dir_path):
        """
        Optunaの最適化プロセスによって得られた最良のパラメータを使用してXGBoostモデルを訓練し、
        訓練されたモデルを使ってテストデータセットの予測を行う。その後、訓練されたモデルと
        使用されたハイパーパラメータを指定されたディレクトリに保存する
        :param optuna.trial.Trial trial study: Optunaのオブジェクト。
        :param xgb.DMatrix train_dmatrix: DMatrix形式の訓練セット。
        :param xgb.DMatrix validation_dmatrix: DMatrix形式の検証セット。
        :param xgb.DMatrix test_dmatrix: DMatrix形式のテストセット。
        :param str model_save_dir_path: モデルとハイパーパラメータを保存するディレクトリパス。
        :return: テストセットに対する予測値、モデル
        :rtype: np.ndarray, xgb.Booster
        """
        xgb_hyperparams = study.best_params
        xgb_hyperparams['objective'] = 'reg:squarederror'
        xgb_hyperparams['eval_metric'] = 'rmse'
        xgb_hyperparams['tree_method'] = 'hist'  # GPUを使用しない場合は、コメントアウトする
        xgb_hyperparams['device'] = 'cuda'  # GPUを使用しない場合は、コメントアウトする
        xgb_model = xgb.train(xgb_hyperparams, train_dmatrix, num_boost_round=10000, early_stopping_rounds=10, evals=[(validation_dmatrix, 'eval')])
        predicted_race_times = xgb_model.predict(test_dmatrix)
        joblib.dump(xgb_model, os.path.join(model_save_dir_path, 'gbm.pkl'), compress=3)

        # 学習、評価で設定していたハイパーパラメータを保存する
        with open(os.path.join(model_save_dir_path, 'ハイパーパラメータ.txt'), 'w') as f:
            f.write(str(xgb_hyperparams))

        return predicted_race_times, xgb_model

    def evaluate_model_perfomance(self, test_labels, predicted_race_times, model_save_dir_path):
        """
        テストセットに対するモデルの予測性能を評価し、R^2スコア、MAE(平均絶対誤差)、RMSE(平均二乗誤差の平方根)を計算する。
        計算した評価は指定したディレクトリにテキストファイルとして保存する。
        :param np.ndarray test_labels: テストセットの正解ラベル。
        :param np.ndarray predicted_race_times: モデルの予測結果。
        :param str model_save_dir_path:評価結果を保存するディレクトリパス。
        :return: R^2スコア。
        :rtype: float
        """
        r2 = r2_score(test_labels, predicted_race_times)
        mae = mean_absolute_error(test_labels, predicted_race_times)
        rmse = np.sqrt(mean_squared_error(test_labels, predicted_race_times))
        with open(os.path.join(model_save_dir_path, '評価結果.txt'), 'w') as f:
            f.write(f'R2: {r2}\nMAE: {mae}\nRMSE: {rmse}\n')

        return r2

    def save_predictons_to_csv(self, test_features, test_labels, predicted_race_times, model_save_dir_path):
        """
        テストセットの特徴量、実際のラベル、予測したラベルを含むCSVファイルを保存する。
        :param DataFrame test_features: 特徴量を含むテストセット。
        :param np.ndarray test_labels: テストセットの正解ラベル。
        :param np.ndarray predicted_race_times: モデルの予測結果。
        :param str model_save_dir_path: 予測結果を保存するディレクトリパス。
        :return: 無し
        """
        results = test_features.copy()
        results['ActualTime'] = test_labels
        results['PredictedTime'] = predicted_race_times
        results.to_csv(os.path.join(model_save_dir_path, '予測結果.csv'), encoding='cp932', index=False, errors='ignore')

    def add_real_and_predicted_ranks(self, race_results):
        """
        実際のタイムと予測したタイムに基づいて、レース結果データに実際の順位(rank_real)と予測した順位(rank_predict)を追加する。
        :param race_results: レース結果データ。
        :return: rank_realとrank_predict列が追加されたレース結果データ。
        :rtype: pd.DataFrame
        """
        race_results['rank_real'] = range(1, len(race_results.index) + 1)       # 実際の順位を割り当てる
        race_results.sort_values('PredictedTime', inplace=True)                 # 予測したタイムでソード
        race_results['rank_predict'] = range(1, len(race_results.index) + 1)    # 予測の順位を割り当てる
        race_results.sort_values('rank_real', inplace=True)                     # 実際の順位で再ソート

        return race_results

    def analyze_and_save_win_rates(self, model_save_dir_path):
        """
        予測結果に基づいて単勝、複勝、馬連、ワイド、三連複の的中率を計算し、CSVファイルに保存する。
        :param model_save_dir_path: 予測結果を含むCSVファイルが保存されるディレクトリパス。
        :return: 無し
        """
        race_results = pd.read_csv(f'{model_save_dir_path}予測結果.csv', encoding='cp932')

        # 各的中率の初期値を設定する
        num_hits_win = 0
        num_hits_place = 0
        num_hits_quinella = 0
        num_hits_quinella_place = 0
        num_hits_trio = 0
        total_race = 0

        # レース結果ごとに的中率を計算する
        for (race_name, location, round), group_race_result in race_results.groupby(['race_name', 'location', 'round']):
            race_result_assign_rank = self.add_real_and_predicted_ranks(group_race_result)
            num_hits_win += self.calculate_win_accuracy(race_result_assign_rank)
            num_hits_place += self.calculate_place_accuracy(race_result_assign_rank)
            num_hits_quinella += self.calculate_quinella_accuracy(race_result_assign_rank)
            num_hits_quinella_place += self.calculate_quinella_place_accuracy(race_result_assign_rank)
            num_hits_trio += self.calculate_trio_accuracy(race_result_assign_rank)
            total_race += 1

        current_date = datetime.datetime.now().strftime('%Y%m%d')

        # CSVファイルにヘッダーを書き込むかを判定する
        # CSVファイルが存在しないあるいはファイルサイズが0の場合、ヘッダーを書き込む
        csv_columns = ['作成日', '単勝', '複勝', '馬連', 'ワイド', '三連複', 'モデルパス']
        csv_file_path = f'{model_save_dir_path}/../{current_date}_的中率結果.csv'
        write_header = not os.path.exists(csv_file_path) or os.stat(csv_file_path).st_size == 0

        # CSVファイルにデータを追記する
        with open(csv_file_path, 'a', newline='', encoding='cp932') as csvfile:
            writer = csv.writer(csvfile)
            if write_header:
                writer.writerow(csv_columns)
            writer.writerow([current_date, f'{num_hits_win / total_race:.2%}', f'{num_hits_place / total_race:.2%}',
                             f'{num_hits_quinella / total_race:.2%}', f'{num_hits_quinella_place / total_race:.2%}',
                             f'{num_hits_trio / total_race:.2%}', model_save_dir_path])

    def train_and_save_model(self, race_results, model_save_dir_path, train_test_split_date):
        """
        レース結果のデータを用いて予測モデルを訓練し、モデルと予測結果を保存する。
        :param train_test_split_date: 訓練データとテストデータに分割するための日付。
        :param DataFrame race_results: レース結果のデータ。特徴量(出走馬の情報など)と予測対象(レースタイム)を含む。
        :param str model_save_dir_path: 生成した訓練モデル、訓練モデルの予測結果や特徴量の重要度、性能評価を保存するディレクトリパス。
        :return: 無し
        """
        # 特徴量とラベルを準備する
        features, labels = self.prepare_features_and_labels(race_results)

        # データを訓練セットとテストセットに分割する
        (all_train_features,
         test_features,
         all_train_labels,
         test_labels) = self.split_data_into_train_test(features, labels, train_test_split_date)

        # 訓練セットがない場合はメソッドを終了する
        if all_train_features is None:
            sys.exit()

        # 特徴量をカテゴリでエンコードする
        (encoded_train_features,
         encoded_test_features) = self.encode_categorical_features(all_train_features,
                                                                   test_features,
                                                                   model_save_dir_path)
        # 訓練データを訓練セットと検証セットに分割する
        (train_features,
         validation_features,
         train_labels,
         validation_labels) = self.split_train_data_into_train_validation(encoded_train_features, all_train_labels)

        # XGBoost用のDMatrixを作成する
        (train_dmatrix,
         validation_dmatrix,
         test_dmatrix) = self.create_dmatrix_for_xgboost(train_features,
                                                         validation_features,
                                                         encoded_test_features,
                                                         train_labels,
                                                         validation_labels,
                                                         test_labels)

        # Optunaを用いたハイパーパラメータの最適化する
        study = optuna.create_study()
        study.optimize(lambda trial: self.optimize_model_hyperparamters(trial, train_dmatrix,
                                                                        validation_dmatrix, test_dmatrix,
                                                                        test_labels), timeout=30)
        # 最適化したパラメータでモデルを訓練し、予測を実行する
        predicted_race_times, xgb_model = self.predict_and_save_model(study, train_dmatrix, validation_dmatrix,
                                                                      test_dmatrix, model_save_dir_path)

        # モデルの性能を評価する
        r2 = self.evaluate_model_perfomance(test_labels, predicted_race_times, model_save_dir_path)

        # 実タイムと予測タイムのプロット、特徴量の重要度のプロットを行う
        self.plot_actual_predicted(test_labels, predicted_race_times, model_save_dir_path, r2)
        self.plot_feature_importance(xgb_model, model_save_dir_path)

        # 予測結果をCSVファイルとして保存する
        self.save_predictons_to_csv(test_features, test_labels, predicted_race_times, model_save_dir_path)


if __name__ == "__main__":
    start_time = time.time()

    # レース結果とモデル出力先のパス
    input_path = './data/raceresults/prepared_raceresults/'
    output_path = './data/model/'

    # 訓練データとテストデータに分割するための日付を設定する
    train_test_split_date = '2023-2-1'

    # モデルを作成する
    creator = ModelCreator(input_path, output_path, train_test_split_date)
    creator.run()

    print(f'実行時間: {time.time() - start_time:.1f} seconds')

出力先の変更方法

プログラムを実行することで生成されるデータの出力先を変更する場合は、460行目の下記の箇所を変更してください。

 output_path = './data/model/'

訓練データとテストデータを分割する日付の変更方法

訓練データとテストデータの分割する日付を変更することで、訓練データとテストデータの割合を調整することができます。463行目の下記の箇所を変更してください。

 train_test_split_date = '2023-2-1'

以上です。

コメント

タイトルとURLをコピーしました