はじめに
JRAの競馬レースの着順を予測するモデルを作成していました。
今回は、そのプログラムのを改修しました。
改修のポイントは下記の3点です。
- コードを役割ごとにモジュール化
- 順位予測モデルをXGBoostにLightGBM、CatBoostを追加
- XGBoost、LightGBM、CatBoostを用いた5種類のアンサンブルモデルを追加
プログラムの概要
機械学習を用いて競馬レースの着順を予測するモデルを作成し、その性能を評価するプログラムです。複数のモデルとアンサンブル手法を使用して予測精度の向上を図っています。
プログラムの主な機能
主な機能は下記となります。
| 機能 | 説明 |
| 機械学習による学習と予測モデル作成 | – XGBoost, LightGBM, CatBoostの3種類で予測モデルを作成。 – 5種類のアンサンブル手法による予測モデルの作成（平均/重み付き/スタッキング/投票/ブレンディング） |
| モデルの評価と性能分析 | – モデルごとの各種的中率の計算（単勝、複勝、馬連、ワイド、三連複） – 性能指標の計算（R2スコア、RMSE、MAE） |
| 結果の可視化と保存 | – 実タイムと予測タイムの散布図、重要度分析グラフの作成 – 評価結果のCSV出力 |
予測モデル作成時の最適化
各モデルを作成する際、最適なハイパーパラメータを探索するために、Optunaを使用しています。最適なハイパーパラメータは、各モデルのディレクトリにテキストで保存しています。
アンサンブル手法では、下記の5つのアンサンブル手法を実装しています。
| アンサンブル | 手法 |
| 平均アンサンブル | – 各モデルの予測結果の単純平均を計算 – すべてのモデルに同じ重みを適用 |
| 重み付けアンサンブル | – 各モデルのR2スコアに基づいて重みを計算 – R2スコアが高いモデルの予測により大きな重みを付与 |
| スタッキングアンサンブル | – ベースモデルの予測を特徴量として使用 – メタモデル（XGBoost）で最終予測 |
| 投票アンサンブル | – 各モデルの予測を平均化 – モデルの寄与度を計算 |
| ブレンディングアンサンブル | – 元の特徴量とベースモデルの予測を結合 – 元の特徴量とベースモデルの予測を結合 |
プログラム実行により出力されるファイル
プログラムを実行すると予測モデル以外にも、その予測モデルを評価するための下記の情報を出力します。
| ファイル名 | 説明 |
| evaluation_results.csv | 評価指標（R2, RMSE, MAE） |
| 予測結果.csv | テストデータに対する予測結果 |
| 実タイムと予測タイムの散布図.svg | 予測精度の可視化グラフ |
| *_hyperparameters.txt | 最適化されたパラメータ |
| 予測タイムの重要度分析_*.svg | 各モデルの特徴量重要度 |
| *_feature_importance.svg | 各アンサンブルの特徴量重要度 |
＜「evaluation_results.csv」のサンプル＞
＜「予測結果.csv」のサンプル＞
R列が実タイムで、S列に予測タイムになっています。
＜「実タイムと予測タイムの散布図.svg」のサンプル＞
＜「予測タイムの重要度分析_*.svg」のサンプル＞
事前準備
Pythonの実行環境
実行環境はPython 3.9～3.11を用意してください。
Python3.10が一番安定して動作します。
Python3.12以上、3.8以下の場合、機械学習ライブラリが想定通りに動作しない可能性があります。
実行に必要なパッケージ
実行に必要なパッケージは「requirements_keiba_ai.txt」に登録してあります。
実行するPythonの実行環境で「requirements_keiba_ai.txt」を用いてパッケージをインストールしてください。
pip install -r requirements_keiba_ai.txt＜主なパッケージ一覧＞
| パッケージ名 | バージョン | 用途 |
| pandas | 2.2.2 | データ処理とCSV操作 |
| numpy | 1.26.4 | 数値計算とデータ処理 |
| scikit-learn | 1.5.1 | エンコーディングと評価指標計算 |
| xgboost | 2.1.1 | XGBoostモデルの学習と予測 |
| lightgbm | 4.5.0 | LightGBMモデルの学習と予測 |
| catboost | 1.2.7 | CatBoostモデルの学習と予測 |
| optuna | 3.6.1 | ハイパーパラメータの最適化 |
| matplotlib | 3.9.1 | グラフ作成と可視化 |
| japanize-matplotlib | 1.1.3 | グラフの日本語表示対応 |
| joblib | 1.4.2 | モデルの保存と読み込み |
学習に必要なデータの用意
モデルの学習に必要なデータを下記のプログラムで事前に取得しておく必要があります。
競馬予測AIの作成㉑(Netkeibaのデータベースからレース結果取得プログラムの改修) – リラックスした生活を過ごすために
設定の編集
train_model.pyの下記の箇所を適宜編集します。
# 設定値
INPUT_PATH = './data/raceresults/prepared_raceresults/'
OUTPUT_PATH = './data/model/'
TRAIN_TEST_SPLIT_DATE = '2023-12-1'＜設定パラメータの解説＞
- INPUT_PATH
– 学習データ（レース結果のCSVファイル）が格納されているフォルダのパス
– 前処理済みのデータが必要
– 基本的に変更は不要 - OUTPUT_PATH
– 学習したモデルや評価結果を保存するフォルダのパス
– モデルファイル、評価指標、グラフなどが保存される
– 基本的に変更は不要 - TRAIN_TEST_SPLIT_DATE
– データを学習用とテスト用に分ける基準日
– この日付より前が学習用、以降がテスト用として使用される
– YYYY-MM-D形式で指定
学習に必要なデータ
学習に必要なデータを含むCSVファイルは前述の「IPNUT_PATH」に配置しておく必要があります。
＜CSVファイルの例＞
＜必要なデータのカラム＞
タイム、馬名、馬名の父、馬名の母、馬齢、騎手体重、騎手名、オッズ、人気、馬体重、レース距離、天候、馬場、馬場状態、レース日、レース名、レース場、レース番号
必要なデータは、下記リンクのプログラムでNetkeibaから取得できます。
競馬予測AIの作成㉑(Netkeibaのデータベースからレース結果取得プログラムの改修) – リラックスした生活を過ごすために
実行手順
用意したPythonの実行環境で下記のコマンドを実行してください。
python train_model.py実行時間は、実行環境と用意したデータによって変わります。
2013～2024年のレース結果を基に予測モデルを作成する場合、おおよそ3時間ほどで完了しました。
プログラムの解説
プログラムの動作概要のフロー
各処理の概要
| 処理名 | 処理内容 | 入力 | 出力 | ポイント |
| 入力データ | 予測に使用するデータの読み込みと検証 | レース結果CSV | 生データフレーム | – データの存在確認 – エンコーディング確認 – 必須カラムの確認 |
| データ前処理 | レース結果データの前処理と変換 | 生データフレーム | – 特徴量データ – ラベルデータ – エンコードデータ |
– 特徴量とラベルの分離 – 学習/テストデータの分割 – カテゴリ変数のエンコード |
| モデル学習 | 3種類の機械学習モデルの学習 | – 前処理済み特徴量 – ラベルデータ |
各モデルの学習済みモデル | – ハイパーパラメータの最適化 – GPU活用による高速化 – モデルの保存 |
| アンサンブル学習 | 複数の予測結果の統合 | – 基本モデルの予測結果 – 学習済みモデル |
アンサンブル予測結果 | – 5種類のアンサンブル手法 – 重み付けによる最適化 – メタモデルの活用 |
| 評価処理 | モデルの性能評価と可視化 | – 予測結果 – 実際の結果 |
– 評価指標 – 可視化グラフ – 的中率 |
– R2スコア計算 – RMSE/MAE計算 – 散布図作成 |
| 出力結果 | 結果の保存と整理 | – 評価結果 – 予測結果 – モデルファイル |
– 学習済みモデル – 評価レポート – 予測結果CSV |
– 結果の一元管理 – レポート生成 – スコア計算 |
プログラムの動作フロー
プログラムのクラスとメソッドの解説
処理関連ごとのクラスとメソッドの一覧を下記にまとめました。
入力データ関連
ファイル名：train_model.py
| クラス/関数 | 目的 | 機能説明 |
| check_input_files() | 入力データの検証 | – 入力ディレクトリの存在確認 – CSVファイルの存在確認 – カレントディレクトリの確認 |
| main() | プログラムのエントリーポイント | – ModelCreatorの初期化と実行 – フォルダリストの生成 – スコア計算の実行 |
データ前処理関連
ファイル名：data_processor.py
| クラス/メソッド | 目的 | 機能説明 |
| DataProcessor | データの前処理を担当 | レース結果データの変換と分割を管理 |
| prepare_features_and_labels() | 特徴量とラベルの分離 | – ‘time’列をラベルとして分離 – 残りの列を特徴量として使用 |
| encode_categorical_features() | カテゴリ変数の変換 | – OrdinalEncoderによる変換 – エンコーダーの保存 – カラム情報の維持 |
| split_data_into_train_test() | データの分割 | – 日付基準でのデータ分割 – 学習用/テスト用データの作成 |
モデル学習関連
ファイル名：model_trainer.py
| クラス/メソッド | 目的 | 機能説明 |
| ModelTrainer | モデル学習の管理 | 3種類のモデルの学習と最適化を制御 |
| create_dmatrix_for_xgboost() | XGBoost用データ作成 | – 学習/検証/テストデータの変換 – DMatrix形式への変換 |
| create_dataset_for_lightgbm() | LightGBM用データ作成 | – 学習/検証/テストデータの変換 – Dataset形式への変換 |
| optimize_xgboost_hyperparamters() | XGBoostの最適化 | – ハイパーパラメータの最適化 – RMSEによる評価 |
| optimize_lightgbm_hyperparamters() | LightGBMの最適化 | – ハイパーパラメータの最適化 – 検証スコアの計算 |
| optimize_catboost_hyperparameters() | CatBoostの最適化 | – ハイパーパラメータの最適化 – RMSEによる評価 |
| predict_and_save_xgboost_model() | XGBoostモデルの保存 | – 最適化モデルの学習 – 予測の実行 – モデルの保存 |
| predict_and_save_lightgbm_model() | LightGBMモデルの保存 | – 最適化モデルの学習 – 予測の実行 – モデルの保存 |
| predict_and_save_catboost_model() | CatBoostモデルの保存 | – 最適化モデルの学習 – 予測の実行 – モデルの保存 |
| train_and_save_model() | モデルの学習と保存 | – パラメータ最適化 – モデルの学習 – 結果の保存 |
アンサンブル学習関連
ファイル名：ensemble_methods.py
| クラス/メソッド | 目的 | 機能説明 |
| EnsembleMethods | アンサンブル学習の管理 | 5種類のアンサンブル手法を実装 |
| weighted_ensemble_predict() | 重み付きアンサンブル | – R2スコアによる重み付け – 予測の統合 |
| average_ensemble_predict() | 平均アンサンブル | – 予測値の平均化 – 特徴量重要度の計算 |
| stacking_ensemble_predict() | スタッキングアンサンブル | – メタモデルの学習 – 予測の統合 |
| voting_ensemble_predict() | 投票アンサンブル | – 予測値の集計 – 寄与度の計算 |
| blending_ensemble_predict() | ブレンディングアンサンブル | – 特徴量の結合 – メタモデルによる予測 |
| plot_weighted_ensemble_feature_importance() | 重み付き重要度の可視化 | – 重み付き特徴量重要度の計算 – グラフの作成と保存 |
| plot_average_ensemble_feature_importance() | 平均重要度の可視化 | – 平均特徴量重要度の計算 – グラフの作成と保存 |
| plot_stacking_feature_importance() | スタッキング重要度の可視化 | – メタモデルの重要度計算 – グラフの作成と保存 |
| plot_voting_ensemble_feature_importance() | 投票重要度の可視化 | – 投票モデルの重要度計算 – グラフの作成と保存 |
| plot_blending_feature_importance() | ブレンディング重要度の可視化 | – 結合特徴量の重要度計算 – グラフの作成と保存 |
| save_model_contributions() | モデル寄与度の保存 | – 各モデルの寄与度計算 – 結果のファイル保存 |
評価処理関連
ファイル名：evaluator.py
| クラス/メソッド | 目的 | 機能説明 |
| Evaluator | モデル評価の管理 | 予測性能の評価と可視化を実行 |
| evaluate_model_performance() | 性能評価の実行 | – R2スコアの計算 – RMSE/MAEの計算 |
| plot_actual_predicted() | 予測結果の可視化 | – 散布図の作成 – R2スコアの表示 |
| plot_feature_importance() | 特徴量重要度の可視化 | – 重要度の計算 – グラフの作成 – モデル別の可視化 |
ファイル名：utils.py
| クラス/メソッド | 目的 | 機能説明 |
| add_real_and_predicted_ranks() | 順位情報の追加 | – 実際の順位付け – 予測順位の計算 – データのソート |
| calculate_win_accuracy() | 単勝的中の計算 | – 1着予測の確認 – 的中判定の実行 |
| calculate_place_accuracy() | 複勝的中の計算 | – 着順条件の確認 – 的中判定の実行 |
| calculate_quinella_accuracy() | 馬連的中の計算 | – 1,2着の組合せ確認 – 的中判定の実行 |
| calculate_quinella_place_accuracy() | ワイド的中の計算 | – 着順組合せの確認 – 的中判定の実行 |
| calculate_trio_accuracy() | 三連複的中の計算 | – 1,2,3着の組合せ確認 – 的中判定の実行 |
| save_predictions_to_csv() | 予測結果の保存 | – データの結合 – CSVファイルの作成 |
| analyze_and_save_win_rates() | 的中率の分析 | – 各種的中率の計算 – 結果の保存 |
出力結果関連
ファイル名：score_calculator.py
| クラス/メソッド | 目的 | 機能説明 |
| ScoreCalculator | スコア計算の管理 | 評価指標の統合とスコア化 |
| calculate_model_score() | 総合スコアの計算 | – 評価指標の正規化 – 重み付けスコアの計算 |
| add_scores_to_csv() | スコアの保存 | – CSVファイルへの追記 – 結果の永続化 |
| calculate_row_score() | 行単位のスコア計算 | – 評価指標の変換 – スコアの計算 – 欠損値の処理 |
ファイル名：generate_folder_list.py
| クラス/メソッド | 目的 | 機能説明 |
| generate_folder_list() | モデルフォルダの管理 | – YYYYMMdd_HHMMSS形式のフォルダ検索 – フォルダリストの生成 – JSONファイルへの保存 |
| main() | プログラムのエントリーポイント | – コマンドライン引数の処理 – フォルダリスト生成の実行 |
全体管理
ファイル名：model_creator.py
| クラス/メソッド | 目的 | 機能説明 |
| ModelCreator | システム全体の制御 | 全処理フローの管理と実行 |
| run() | 処理の実行制御 | – 各コンポーネントの連携 – 処理順序の制御 |
各プログラムのポイント
システム全体の性能と信頼性に大きく影響する重要な箇所について、下記に記載しました。
特に、データ分割、アンサンブル学習、ハイパーパラメータ最適化は予測精度を左右する箇所です。
- データ分割処理（data_processor.py）
・日付を基準にデータを学習用とテスト用に分割def split_data_into_train_test(self, features, labels, split_day): 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]
・指定された分割日が適切な範囲にない場合はNoneを返す
・バイナリマスクを使用して効率的にデータを分割 - アンサンブル学習の重み付け処理（ensemble_methods.py）
・各モデルのR2スコアに基づいて重みを計算def weighted_ensemble_predict(self, predictions, models, r2_scores, ensemble_save_dir_path): # R2スコアに基づいて重みを計算 total_r2 = sum(r2_scores.values()) weights = {model: score / total_r2 for model, score in r2_scores.items()} # 重み付けアンサンブル予測 weighted_ensemble_predictions = np.zeros_like(list(predictions.values())[0]) for model, pred in predictions.items(): weighted_ensemble_predictions += weights[model] * pred # アンサンブルモデルの特徴量重要度（重み付け平均） self.plot_weighted_ensemble_feature_importance(models, weights, ensemble_save_dir_path) # 重みの保存 with open(os.path.join(ensemble_save_dir_path, 'ensemble_weights.txt'), 'w') as f: for model, weight in weights.items(): f.write(f"{model}: {weight:.4f}\n") return weighted_ensemble_predictions
・重みに基づいて予測結果を統合
・特徴量重要度の計算と可視化
・重み情報の保存 - ハイパーパラメータ最適化（model_trainer.py）
・Optunaを使用したハイパーパラメータの自動最適化def optimize_xgboost_hyperparamters(self, trial, train_dmatrix, validation_dmatrix, test_dmatrix, test_labels): 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', 'device': 'cuda', } xgb_model = xgb.train(xgb_hyperparams, train_dmatrix, num_boost_round=10000, early_stopping_rounds=10, evals=[(validation_dmatrix, 'eval')]) preds = xgb_model.predict(test_dmatrix) rmse = np.sqrt(mean_squared_error(test_labels, preds)) return rmse
・GPU対応による学習の高速化
・early stoppingによる過学習防止
・RMSEを評価指標として使用 - 的中率の計算（utils.py）
・レース出走頭数に応じて判定基準を変更def calculate_place_accuracy(ranked_race_results): horses_num = len(ranked_race_results) if horses_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))
・8頭以上：3着以内の予測
・8頭未満：2着以内の予測
・的中は1、外れは0で返却 - 統合評価スコアの計算（score_calculator.py）
・複数の評価指標を統合して単一のスコアを算出def calculate_model_score(win_rate, r2, rmse, mae): try: # RMSEとMAEは小さいほど良いので逆数を計算 normalized_rmse = 1 / (1 + rmse) normalized_mae = 1 / (1 + mae) # 重み付けスコアを計算 score = (0.4 * win_rate + 0.3 * r2 + 0.2 * normalized_rmse + 0.1 * normalized_mae) return score except Exception as e: print(f"スコア計算中にエラーが発生しました: {e}") return 0.0
・各指標に重み付けを適用（単勝的中率を重視）
・RMSEとMAEは正規化して使用
・エラー発生時は0.0を返却 - メインの実行制御（train_model.py）
・入力ファイルの存在確認def main(): if not check_input_files(): sys.exit(1) start_time = time.time() creator = ModelCreator(INPUT_PATH, OUTPUT_PATH, TRAIN_TEST_SPLIT_DATE) current_date_time = datetime.datetime.now().strftime('%Y%m%d_%H%M%S') creator.run(current_date_time) target_path = OUTPUT_PATH output_file = f"{OUTPUT_PATH}folder_list.json" generate_folder_list(target_path, output_file) input_csv_path = f"{target_path}{current_date_time}/的中率結果.csv" output_csv_path = f"{target_path}{current_date_time}/的中率結果_with_score.csv" ScoreCalculator.add_scores_to_csv(input_csv_path, output_csv_path)
・ModelCreatorによる一連の処理実行
・実行結果のフォルダリスト生成
・的中率結果へのスコア付与
・処理時間の計測と表示
プログラムのコード
train_model.py
# ==============================================================================
# 競馬レース予測モデル作成・評価プログラム
# ==============================================================================
# プログラムの概要:
# 機械学習を用いて競馬レースの着順を予測するモデルを作成し、
# その性能を評価するプログラム。複数のモデルとアンサンブル手法を
# 使用して予測精度の向上を図る。
#
# プログラムの主な機能:
# 1. 複数の機械学習モデルの作成と学習
# - XGBoost, LightGBM, CatBoostの3種類の基本モデル
# - 5種類のアンサンブル手法による統合
# 2. モデルの評価と性能分析
# - 各種的中率の計算（単勝、複勝、馬連、ワイド、三連複）
# - 性能指標の計算（R2スコア、RMSE、MAE）
# 3. 結果の可視化と保存
# - 散布図、重要度分析グラフの作成
# - 評価結果のCSV出力
#
# ==============================================================================
# 実行手順
# ==============================================================================
# 1. 必要なライブラリのインストール:
# pip install pandas numpy scikit-learn xgboost lightgbm catboost optuna
# pip install matplotlib seaborn japanize_matplotlib joblib
#
# 2. 設定値の確認:
# INPUT_PATH = './data/raceresults/prepared_raceresults/'
# OUTPUT_PATH = './data/model/'
# TRAIN_TEST_SPLIT_DATE = '2023-12-1'
#
# 3. 入力データの準備:
# - prepare_raceResults.pyで前処理済みのデータが
# INPUT_PATHに存在することを確認
# - データ形式：CSV（エンコーディング：cp932）
#
# 4. 実行方法:
# python train_model.py
#
# 5. 処理の流れ:
# 1) ModelCreatorインスタンスを作成して入力パスと出力パスを設定
# 2) レース結果データを読み込み、学習データとテストデータに分割
# 3) 特徴量エンジニアリングを実行してモデルの入力データを作成
# 4) 各モデル（XGBoost、LightGBM、CatBoost）でトレーニングを実行
# 5) テストデータで予測を行い、的中率を計算して結果をCSVに保存
# 6) アンサンブル学習による予測と評価を実行
# 7) モデル保存フォルダの一覧をJSONファイルで管理
# 8) 的中率結果にスコアを付与して新しいCSVファイルを作成
#
# ==============================================================================
# 出力データ
# ==============================================================================
# 1. モデルごとのファイル（各モデルのディレクトリに保存）:
# - evaluation_results.csv：評価指標（R2, RMSE, MAE）
# - [モデル名]_hyperparameters.txt：最適化されたパラメータ
# - [モデル名]_model.pkl：学習済みモデル
# - 実タイムと予測タイムの散布図.svg：予測精度の可視化
# - 予測タイムの重要度分析_[モデル名].svg：特徴量の重要度
# - 予測結果.csv：テストデータに対する予測結果
#
# 2. 総合評価ファイル:
# - 的中率結果.csv：各モデルの的中率
# - 的中率結果_with_score.csv：スコア付きの的中率
# - folder_list.json：学習結果フォルダの一覧
#
# ==============================================================================
# プログラム間の関係
# ==============================================================================
# 1. 本プログラムが呼び出すモジュール:
# - ModelCreator (model_creator.py)
# └─ データ処理からモデル評価までの全体フローを管理
# - generate_folder_list (generate_folder_list.py)
# └─ モデルを保存したフォルダの一覧をJSON形式で管理
# - ScoreCalculator (score_calculator.py)
# └─ 的中率結果にスコアを付与して評価を行う
#
# 2. 本プログラムを呼び出すプログラム:
# - なし（実行起点となるプログラム）
#
# ==============================================================================
# 注意事項
# ==============================================================================
# 1. 実行環境:
# - Python 3.x
# - GPU環境での実行を推奨（処理速度向上のため）
# - 十分なメモリ容量（8GB以上推奨）
#
# 2. データ要件:
# - 入力CSVファイルは必須カラムを全て含むこと
# - 日付データは正しいフォーマットであること
# - 欠損値は事前に適切に処理されていること
#
# 3. 実行時の注意:
# - 処理時間は数時間程度かかる可能性あり
# - 大量のディスク容量が必要（数GB程度）
# - GPU使用時はメモリ使用量に注意
#
# 4. 結果の確認:
# - start_server_and_open_url.batで結果を確認可能
# - ブラウザで自動的にreport2.htmlが開く
#
import time
import datetime
import os
import sys
from model_creator import ModelCreator
from generate_folder_list import generate_folder_list
from score_calculator import ScoreCalculator
# 設定値
INPUT_PATH = './data/raceresults/prepared_raceresults/'
OUTPUT_PATH = './data/model/'
TRAIN_TEST_SPLIT_DATE = '2024-09-1'
def check_input_files():
"""
入力パスにCSVファイルが存在するかを確認する
:return: CSVファイルが存在する場合はTrue、存在しない場合はFalse
"""
# 現在のディレクトリを取得して表示
current_directory = os.getcwd()
print(f"現在のディレクトリ: {current_directory}")
# 入力ディレクトリの存在確認
if not os.path.exists(INPUT_PATH):
print(f"エラー: 入力ディレクトリが存在しません: {INPUT_PATH}")
return False
# CSVファイルの存在確認
csv_files = [f for f in os.listdir(INPUT_PATH) if f.endswith('.csv')]
if not csv_files:
print(f"エラー: {INPUT_PATH} にレース結果のCSVファイルが存在しません。")
return False
return True
def main():
# 入力ファイルの存在確認
if not check_input_files():
sys.exit(1)
start_time = time.time()
# ModelCreator インスタンスの作成
creator = ModelCreator(INPUT_PATH, OUTPUT_PATH, TRAIN_TEST_SPLIT_DATE)
# モデルの作成と評価の実行
current_date_time = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
creator.run(current_date_time)
# 実行時間の表示
# print(f'実行時間: {time.time() - start_time:.1f} seconds')
run_time = time.time() - start_time
print(f'\n処理開始時刻: {datetime.datetime.fromtimestamp(start_time).strftime("%Y-%m-%d %H:%M:%S")}')
print(f'処理終了時刻: {datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
print('実行時間: {:.2f}秒 ({:.2f}分)'.format(run_time, run_time / 60))
# モデルを作成したフォルダのリストを作成
target_path = OUTPUT_PATH
output_file = f"{OUTPUT_PATH}folder_list.json"
generate_folder_list(target_path, output_file)
# 的中率結果にスコアを追加
input_csv_path = f"{target_path}{current_date_time}/的中率結果.csv"
output_csv_path = f"{target_path}{current_date_time}/的中率結果_with_score.csv"
ScoreCalculator.add_scores_to_csv(input_csv_path, output_csv_path)
if __name__ == "__main__":
main()
model_trainer.py
# ==============================================================================
# 機械学習モデルのトレーニングと最適化プログラム
# ==============================================================================
# プログラムの概要:
# 競馬レース予測のための機械学習モデル（XGBoost, LightGBM, CatBoost）の
# トレーニングと最適化を行うプログラム。Optunaを使用して各モデルの
# ハイパーパラメータを最適化し、学習済みモデルを保存する。
#
# プログラムの主な機能:
# 1. モデルのトレーニング
# - XGBoostモデルのトレーニング
# - LightGBMモデルのトレーニング
# - CatBoostモデルのトレーニング
# 2. ハイパーパラメータの最適化
# - Optunaによる自動パラメータチューニング
# - モデルごとの最適パラメータ探索
# 3. モデルの評価と保存
# - 最適化されたモデルの保存
# - 予測実行と評価
#
# ==============================================================================
# 実行手順
# ==============================================================================
# 1. 必要なライブラリのインストール:
# pip install numpy xgboost lightgbm catboost optuna joblib
#
# 2. 設定値の確認:
# - Optuna最適化設定
# - 試行回数: 10回
# - タイムアウト: 1800秒（30分）
# - GPU設定（必要に応じて）
# - tree_method: 'gpu_hist'
# - predictor: 'gpu_predictor'
#
# 3. 入力データの準備:
# - DataProcessorで前処理された特徴量とラベル
# - トレーニングデータとテストデータの分割済みデータ
#
# 4. 実行方法:
# - model_creator.pyから呼び出される（直接実行は想定していない）
#
# 5. 処理の流れ:
# 1) データをモデル固有の形式に変換（DMatrix, Dataset等）
# 2) ハイパーパラメータの最適化をOptunaで実行
# 3) 最適化されたパラメータでモデルを学習
# 4) モデルの保存と予測の実行
#
# ==============================================================================
# 出力データ
# ==============================================================================
# 1. モデルファイル:
# - xgboost: xgb_model.pkl
# - lightgbm: lightgbm_model.pkl
# - catboost: catboost_model.cbm
#
# 2. 設定ファイル:
# - xgb_hyperparameters.txt
# - lightgbm_hyperparameters.txt
# - catboost_hyperparameters.txt
#
# ==============================================================================
# プログラム間の関係
# ==============================================================================
# 1. 呼び出し元:
# - model_creator.py
# └─ train_and_save_model()メソッドを通じて呼び出し
#
# 2. データの受け渡し:
# - DataProcessor: 前処理済みデータを受け取る
# - EnsembleMethods: 学習済みモデルを提供
# - Evaluator: モデル評価用のデータを提供
#
# ==============================================================================
# 注意事項
# ==============================================================================
# 1. 実行環境:
# - GPU環境での実行を推奨
# - 必要メモリ量: モデルごとに2-6GB程度
# - 十分なディスク容量（モデル保存用）
#
# 2. 最適化設定:
# - 試行回数とタイムアウト値は要調整
# - GPU使用時はメモリ使用量に注意
# - 早期終了条件の設定確認
#
# 3. モデル固有の注意:
# - XGBoost: DMatrix形式への変換が必要
# - LightGBM: Dataset形式への変換が必要
# - CatBoost: カテゴリカル変数の指定方法に注意
#
# 4. 保存データ:
# - モデルファイルは圧縮形式で保存
# - ハイパーパラメータは可読形式で保存
# - 既存のファイルは上書きされる
#
import numpy as np
import xgboost as xgb
import lightgbm as lgb
import catboost as cb
import optuna
import joblib
import os
from catboost import Pool
from sklearn.metrics import mean_squared_error
from sklearn.model_selection import train_test_split
class ModelTrainer:
def __init__(self):
pass
def create_dmatrix_for_xgboost(self, train_features, validation_features, encoded_test_features, train_labels,
validation_labels, test_labels):
"""
XGBoostモデルのトレーニングに使用するDMatrix形式のデータセットを作成する。
:param DataFrame train_features: トレーニング用の特徴量データセット。
:param DataFrame validation_features: 検証用の特徴量データセット。
:param DataFrame encoded_test_features: テスト用の特徴量データセット。
:param np.ndarray train_labels: トレーニング用のラベルデータ。
:param np.ndarray validation_labels: 検証用のラベルデータ。
:param np.ndarray test_labels: テスト用のラベルデータ。
:return: トレーニング用、検証用、テスト用のDMatrixオブジェクト。
:rtype: tuple(xgb.DMatrix, xgb.DMatrix, xgb.DMatrix)
"""
train_dmatrix = xgb.DMatrix(train_features, label=train_labels)
validation_dmatrix = xgb.DMatrix(validation_features, label=validation_labels)
test_dmatrix = xgb.DMatrix(encoded_test_features, label=test_labels)
return train_dmatrix, validation_dmatrix, test_dmatrix
def create_dataset_for_lightgbm(self, train_features, validation_features, test_features, train_labels,
validation_labels, test_labels):
"""
LightGBMモデルのトレーニングに使用するDataset形式のデータセットを作成する。
:param DataFrame train_features: トレーニング用の特徴量データセット。
:param DataFrame validation_features: 検証用の特徴量データセット。
:param DataFrame test_features: テスト用の特徴量データセット。
:param np.ndarray train_labels: トレーニング用のラベルデータ。
:param np.ndarray validation_labels: 検証用のラベルデータ。
:param np.ndarray test_labels: テスト用のラベルデータ。
:return: トレーニング用、検証用、テスト用のDatasetオブジェクト。
:rtype: tuple(lgb.Dataset, lgb.Dataset, lgb.Dataset)
"""
train_dataset = lgb.Dataset(train_features, label=train_labels)
validation_dataset = lgb.Dataset(validation_features, label=validation_labels)
test_dataset = lgb.Dataset(test_features, label=test_labels, reference=train_dataset)
return train_dataset, validation_dataset, test_dataset
def optimize_xgboost_hyperparamters(self, trial, train_dmatrix, validation_dmatrix, test_dmatrix, test_labels):
"""
Optunaを使用してXGBoostモデルのハイパーパラメータを最適化する。
:param 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',
'device': 'cuda',
}
xgb_model = xgb.train(xgb_hyperparams, train_dmatrix, num_boost_round=10000, early_stopping_rounds=10,
evals=[(validation_dmatrix, 'eval')])
preds = xgb_model.predict(test_dmatrix)
rmse = np.sqrt(mean_squared_error(test_labels, preds))
return rmse
def optimize_lightgbm_hyperparamters(self, trial, train_dataset, validation_dataset):
"""
Optunaを使用してLightGBMモデルのハイパーパラメータを最適化する。
:param Trial trial: Optunaのトライアルオブジェクト。
:param lgb.Dataset train_dataset: トレーニング用のデータセット。
:param lgb.Dataset validation_dataset: 検証用のデータセット。
:return: 検証データに対するRMSEスコア。
:rtype: float
"""
lgb_hyperparams = {
'objective': 'regression',
'metric': 'rmse',
'boosting_type': 'gbdt',
'max_depth': trial.suggest_int('max_depth', 3, 10),
'num_leaves': trial.suggest_int('num_leaves', 20, 150),
'learning_rate': trial.suggest_float('learning_rate', 0.001, 0.1, log=True),
'feature_fraction': trial.suggest_float('feature_fraction', 0.5, 1.0),
'bagging_fraction': trial.suggest_float('bagging_fraction', 0.5, 1.0),
'bagging_freq': trial.suggest_int('bagging_freq', 1, 10),
'lambda_l1': trial.suggest_float('lambda_l1', 0.0001, 10.0, log=True),
'lambda_l2': trial.suggest_float('lambda_l2', 0.0001, 10.0, log=True),
'min_child_weight': trial.suggest_float('min_child_weight', 0.001, 10.0, log=True),
'device_type': 'gpu',
'tree_method': 'gpu_hist'
}
lgb_model = lgb.train(
lgb_hyperparams,
train_dataset,
valid_sets=[validation_dataset],
callbacks=[lgb.early_stopping(stopping_rounds=10)]
)
val_rmse = lgb_model.best_score['valid_0']['rmse']
return val_rmse
def optimize_catboost_hyperparameters(self, trial, train_pool, validation_pool):
"""
Optunaを使用してCatBoostモデルのハイパーパラメータを最適化する。
:param Trial trial: Optunaのトライアルオブジェクト。
:param Pool train_pool: トレーニング用のデータプール。
:param Pool validation_pool: 検証用のデータプール。
:return: 検証データに対するRMSEスコア。
:rtype: float
"""
cb_hyperparams = {
'iterations': trial.suggest_int('iterations', 100, 1000),
'depth': trial.suggest_int('depth', 4, 10),
'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3, log=True),
'l2_leaf_reg': trial.suggest_float('l2_leaf_reg', 1e-8, 10.0, log=True),
'border_count': trial.suggest_int('border_count', 32, 255),
'bagging_temperature': trial.suggest_float('bagging_temperature', 0, 1),
'random_strength': trial.suggest_float('random_strength', 1e-8, 10, log=True),
'od_type': 'Iter',
'od_wait': 50,
'verbose': False,
'task_type': 'GPU'
}
cb_model = cb.CatBoostRegressor(**cb_hyperparams)
cb_model.fit(train_pool, eval_set=validation_pool, early_stopping_rounds=50, verbose=False)
return cb_model.get_best_score()['validation']['RMSE']
def predict_and_save_xgboost_model(self, study, train_dmatrix, validation_dmatrix, test_dmatrix,
model_save_dir_path):
"""
最適化されたハイパーパラメータを使用してXGBoostモデルを学習し、予測を行い、モデルを保存する。
:param Study 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: tuple(np.ndarray, xgb.Booster)
"""
xgb_hyperparams = study.best_params
xgb_hyperparams['objective'] = 'reg:squarederror'
xgb_hyperparams['eval_metric'] = 'rmse'
xgb_hyperparams['tree_method'] = 'hist'
xgb_hyperparams['device'] = 'cuda'
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, 'xgb_model.pkl'), compress=3)
with open(os.path.join(model_save_dir_path, 'xgb_hyperparameters.txt'), 'w') as f:
f.write(str(xgb_hyperparams))
return predicted_race_times, xgb_model
def predict_and_save_lightgbm_model(self, study, train_dataset, validation_dataset, test_features,
model_save_dir_path):
"""
最適化されたハイパーパラメータを使用してLightGBMモデルを学習し、予測を行い、モデルを保存する。
:param Study study: 完了したOptunaの学習履歴。
:param lgb.Dataset train_dataset: トレーニング用のデータセット。
:param lgb.Dataset validation_dataset: 検証用のデータセット。
:param DataFrame test_features: テスト用の特徴量データセット。
:param str model_save_dir_path: モデルと設定を保存するディレクトリパス。
:return: 予測結果と学習済みモデル。
:rtype: tuple(np.ndarray, lgb.Booster)
"""
lgb_hyperparams = study.best_params
lgb_hyperparams['objective'] = 'regression'
lgb_hyperparams['metric'] = 'rmse'
lgb_hyperparams['boosting_type'] = 'gbdt'
lgb_model = lgb.train(
lgb_hyperparams,
train_dataset,
valid_sets=[validation_dataset],
callbacks=[lgb.early_stopping(stopping_rounds=10)]
)
predicted_race_times = lgb_model.predict(test_features)
joblib.dump(lgb_model, os.path.join(model_save_dir_path, 'lightgbm_model.pkl'), compress=3)
with open(os.path.join(model_save_dir_path, 'lgb_hyperparameters.txt'), 'w') as f:
f.write(str(lgb_hyperparams))
return predicted_race_times, lgb_model
def predict_and_save_catboost_model(self, study, train_pool, validation_pool, test_pool, model_save_dir_path):
"""
最適化されたハイパーパラメータを使用してCatBoostモデルを学習し、予測を行い、モデルを保存する。
:param Study study: 完了したOptunaの学習履歴。
:param Pool train_pool: トレーニング用のデータプール。
:param Pool validation_pool: 検証用のデータプール。
:param Pool test_pool: テスト用のデータプール。
:param str model_save_dir_path: モデルと設定を保存するディレクトリパス。
:return: 予測結果と学習済みモデル。
:rtype: tuple(np.ndarray, CatBoostRegressor)
"""
cb_hyperparams = study.best_params
cb_hyperparams['iterations'] = 1000
cb_hyperparams['od_type'] = 'Iter'
cb_hyperparams['od_wait'] = 50
cb_hyperparams['verbose'] = False
cb_hyperparams['task_type'] = 'GPU'
cb_model = cb.CatBoostRegressor(**cb_hyperparams)
cb_model.fit(train_pool, eval_set=validation_pool, early_stopping_rounds=50, verbose=False)
predicted_race_times = cb_model.predict(test_pool)
cb_model.save_model(os.path.join(model_save_dir_path, 'catboost_model.cbm'))
with open(os.path.join(model_save_dir_path, 'catboost_hyperparameters.txt'), 'w') as f:
f.write(str(cb_hyperparams))
return predicted_race_times, cb_model
def train_and_save_model(self, encoded_train_features, encoded_test_features, all_train_labels, test_labels,
model_save_dir_path, model_type):
train_features, validation_features, train_labels, validation_labels = train_test_split(
encoded_train_features, all_train_labels, test_size=0.2, random_state=0)
"""
指定されたモデルタイプに応じて、モデルのトレーニング、ハイパーパラメータの最適化、予測、保存を行う。
:param DataFrame encoded_train_features: エンコードされたトレーニング用特徴量。
:param DataFrame encoded_test_features: エンコードされたテスト用特徴量。
:param np.ndarray all_train_labels: トレーニング用のラベル。
:param np.ndarray test_labels: テスト用のラベル。
:param str model_save_dir_path: モデルと設定を保存するディレクトリパス。
:param str model_type: モデルの種類（'xgboost', 'lightgbm', 'catboost'のいずれか）。
:return: 予測結果と学習済みモデル。
:rtype: tuple(np.ndarray, object)
"""
if model_type == 'xgboost':
train_data, validation_data, test_data = self.create_dmatrix_for_xgboost(
train_features, validation_features, encoded_test_features, train_labels, validation_labels,
test_labels)
optimize_func = lambda trial: self.optimize_xgboost_hyperparamters(
trial, train_data, validation_data, test_data, test_labels)
predict_and_save_func = lambda study: self.predict_and_save_xgboost_model(
study, train_data, validation_data, test_data, model_save_dir_path)
elif model_type == 'lightgbm':
train_data, validation_data, test_data = self.create_dataset_for_lightgbm(
train_features, validation_features, encoded_test_features, train_labels, validation_labels,
test_labels)
optimize_func = lambda trial: self.optimize_lightgbm_hyperparamters(trial, train_data, validation_data)
predict_and_save_func = lambda study: self.predict_and_save_lightgbm_model(
study, train_data, validation_data, encoded_test_features, model_save_dir_path)
else: # CatBoost
cat_features = [col for col in encoded_train_features.columns if
encoded_train_features[col].dtype == 'object']
train_data = Pool(train_features, train_labels, cat_features=cat_features)
validation_data = Pool(validation_features, validation_labels, cat_features=cat_features)
test_data = Pool(encoded_test_features, cat_features=cat_features)
optimize_func = lambda trial: self.optimize_catboost_hyperparameters(trial, train_data, validation_data)
predict_and_save_func = lambda study: self.predict_and_save_catboost_model(
study, train_data, validation_data, test_data, model_save_dir_path)
study = optuna.create_study(direction='minimize')
study.optimize(optimize_func, n_trials=10, timeout=1800)
predicted_race_times, model = predict_and_save_func(study)
return predicted_race_times, model
model_creator.py
# ==============================================================================
# モデル作成・学習・評価の統合管理プログラム
# ==============================================================================
# プログラムの概要:
# 競馬予測システムの中核となるプログラム。データの前処理から
# モデルの作成、学習、評価、アンサンブル学習までの一連の処理を
# 統合的に管理する。各種コンポーネントを連携させ、予測モデルの
# 作成から性能評価までを一貫して実行する。
#
# プログラムの主な機能:
# 1. 全体処理の統括
# - 各コンポーネントの初期化と実行制御
# - データフローの管理
# - 結果の統合管理
# 2. モデル作成と学習の制御
# - 基本モデル（XGBoost/LightGBM/CatBoost）の作成
# - アンサンブル手法の適用
# - パラメータ最適化の制御
# 3. 評価結果の管理
# - 各モデルの性能評価
# - 結果の保存と整理
#
# ==============================================================================
# 実行手順
# ==============================================================================
# 1. 必要なライブラリのインストール:
# pip install pandas scikit-learn xgboost lightgbm catboost optuna
#
# 2. 設定値の確認:
# - 入力パス設定
# - レース結果データの場所
# - モデル保存先ディレクトリ
# - 学習/テストデータの分割日設定
# - GPU使用設定
#
# 3. 入力データの準備:
# - レース結果のCSVファイル
# 必須カラム:
# - date: レース日付
# - time: レースタイム
# - その他の特徴量
#
# 4. 実行方法:
# - train_model.pyから呼び出される（直接実行は想定していない）
#
# 5. 処理の流れ:
# 1) 入力データの読み込みと日付の変換
# 2) DataProcessorによる前処理
# 3) ModelTrainerによる各モデルの学習
# 4) EnsembleMethodsによるアンサンブル学習
# 5) Evaluatorによる評価実行
# 6) 結果の保存
#
# ==============================================================================
# 出力データ
# ==============================================================================
# 1. モデルディレクトリ構造:
# traindata_YYYY_YYYY_[timestamp]/
# ├── xgboost/
# ├── lightgbm/
# ├── catboost/
# ├── average_ensemble/
# ├── weighted_ensemble/
# ├── stacking_ensemble/
# ├── voting_ensemble/
# └── blending_ensemble/
#
# 2. 各モデルディレクトリ内:
# - evaluation_results.csv
# - [model]_hyperparameters.txt
# - [model]_model.pkl
# - 実タイムと予測タイムの散布図.svg
# - 予測タイムの重要度分析.svg
# - 予測結果.csv
#
# ==============================================================================
# プログラム間の関係
# ==============================================================================
# 1. 呼び出し元:
# - train_model.py
# └─ モデル作成プロセス全体の実行
#
# 2. 呼び出すコンポーネント:
# - DataProcessor: データの前処理
# - ModelTrainer: モデルの学習
# - EnsembleMethods: アンサンブル学習
# - Evaluator: 性能評価
#
# ==============================================================================
# 注意事項
# ==============================================================================
# 1. 実行環境:
# - GPU環境での実行を推奨
# - 十分なディスク容量（数GB以上）
# - 大容量メモリ（8GB以上推奨）
#
# 2. データ要件:
# - 入力CSVは適切にフォーマットされていること
# - 日付データは正しい形式であること
# - 必須カラムが全て存在すること
#
# 3. 処理時間:
# - 全体の処理に数時間かかる可能性
# - GPUメモリの使用量に注意
# - モデルごとの学習時間は変動
#
# 4. エラー処理:
# - データ分割で適切なデータサイズを確保
# - モデル保存時の十分な空き容量確認
# - GPU使用時のメモリ管理
#
# 5. 拡張性:
# - 新しいモデルの追加が可能
# - アンサンブル手法の追加が可能
# - 評価指標の追加が可能
#
import datetime
import os
import utils
import pandas as pd
from model_trainer import ModelTrainer
from ensemble_methods import EnsembleMethods
from data_processor import DataProcessor
from evaluator import Evaluator
class ModelCreator:
"""
機械学習モデルの作成、訓練、評価を行うクラス。
"""
def __init__(self, input_path, output_path, train_test_split_date):
"""
:param str input_path: 入力データのパス
:param str output_path: 出力（モデルや結果）を保存するパス
:param str train_test_split_date: 訓練データとテストデータを分割する日付
"""
self.input_path = input_path
self.output_path = output_path
self.train_test_split_date = train_test_split_date
self.model_trainer = ModelTrainer()
self.ensemble_methods = EnsembleMethods()
self.data_processor = DataProcessor()
self.evaluator = Evaluator()
def run(self, current_date_time):
"""
モデルの作成、訓練、評価、およびアンサンブル手法の適用を実行するメインメソッド。
このメソッドは以下の手順を実行する：
1. 入力データの読み込み
2. データの前処理
3. 複数のモデル（XGBoost, LightGBM, CatBoost）の訓練と評価
4. 各種アンサンブル手法（平均、重み付け、スタッキング、ボーティング、ブレンディング）の適用
5. 結果の保存と可視化
:return: None
"""
# 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"):
# 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)
# 特徴量とラベルの準備
features, labels = self.data_processor.prepare_features_and_labels(race_results)
(all_train_features,
test_features,
all_train_labels,
test_labels) = self.data_processor.split_data_into_train_test(features,
labels,
self.train_test_split_date)
if all_train_features is None:
continue
# カテゴリカル変数のエンコーディング
(encoded_train_features,
encoded_test_features) = self.data_processor.encode_categorical_features(all_train_features,
test_features,
dir_path)
predictions = {}
models = {}
r2_scores = {}
# 各モデル（XGBoost, LightGBM, CatBoost）の訓練と評価
for model_type in ['xgboost', 'lightgbm', 'catboost']:
model_dir = os.path.join(dir_path, model_type)
os.makedirs(model_dir, exist_ok=True)
(predicted_race_times, model) = self.model_trainer.train_and_save_model(encoded_train_features,
encoded_test_features,
all_train_labels,
test_labels,
model_dir, model_type)
# モデルの評価
predictions[model_type] = predicted_race_times
models[model_type] = model
r2_scores[model_type] = self.evaluator.evaluate_model_performance(test_labels,
predicted_race_times,
model_dir)
self.evaluator.plot_actual_predicted(test_labels,
predicted_race_times,
model_dir,
r2_scores[model_type])
self.evaluator.plot_feature_importance(model, model_type, model_dir)
# 予測結果の保存
utils.save_predictions_to_csv(test_features, test_labels, predicted_race_times, model_dir)
# 予測結果から的中率を計算
utils.analyze_and_save_win_rates(model_dir, self.train_test_split_date)
# 単純平均アンサンブル予測の実行
average_ensemble_dir = os.path.join(dir_path, 'average_ensemble')
os.makedirs(average_ensemble_dir, exist_ok=True)
average_ensemble_predictions = self.ensemble_methods.average_ensemble_predict(predictions, models,
test_features,
test_labels,
average_ensemble_dir)
# 単純平均アンサンブルモデルの評価
r2 = self.evaluator.evaluate_model_performance(test_labels,
average_ensemble_predictions,
average_ensemble_dir)
self.evaluator.plot_actual_predicted(test_labels,
average_ensemble_predictions,
average_ensemble_dir,
r2)
# 予測結果の保存
utils.save_predictions_to_csv(test_features,
test_labels,
average_ensemble_predictions,
average_ensemble_dir)
# 予測結果から的中率を計算
utils.analyze_and_save_win_rates(average_ensemble_dir, self.train_test_split_date)
# 重み付けアンサンブル予測の実行
weighted_ensemble_dir = os.path.join(dir_path, 'weighted_ensemble')
os.makedirs(weighted_ensemble_dir, exist_ok=True)
weighted_ensemble_predictions = self.ensemble_methods.weighted_ensemble_predict(predictions, models,
r2_scores,
weighted_ensemble_dir)
# 重み付けアンサンブルモデルの評価
r2 = self.evaluator.evaluate_model_performance(test_labels,
weighted_ensemble_predictions,
weighted_ensemble_dir)
self.evaluator.plot_actual_predicted(test_labels,
weighted_ensemble_predictions,
weighted_ensemble_dir,
r2)
# 予測結果の保存
utils.save_predictions_to_csv(test_features,
test_labels,
weighted_ensemble_predictions,
weighted_ensemble_dir)
# 予測結果から的中率を計算
utils.analyze_and_save_win_rates(weighted_ensemble_dir, self.train_test_split_date)
# スタッキングアンサンブル予測の実行
stacking_ensemble_dir = os.path.join(dir_path, 'stacking_ensemble')
os.makedirs(stacking_ensemble_dir, exist_ok=True)
stacking_ensemble_predictions = self.ensemble_methods.stacking_ensemble_predict(encoded_train_features,
encoded_test_features,
all_train_labels,
test_labels, models,
stacking_ensemble_dir)
# スタッキングアンサンブルモデルの評価
r2 = self.evaluator.evaluate_model_performance(test_labels,
stacking_ensemble_predictions,
stacking_ensemble_dir)
self.evaluator.plot_actual_predicted(test_labels,
stacking_ensemble_predictions,
stacking_ensemble_dir,
r2)
# 予測結果の保存
utils.save_predictions_to_csv(test_features,
test_labels,
stacking_ensemble_predictions,
stacking_ensemble_dir)
# 予測結果から的中率を計算
utils.analyze_and_save_win_rates(stacking_ensemble_dir, self.train_test_split_date)
# ボルティングアンサンブル予測の実行
voting_ensemble_dir = os.path.join(dir_path, 'voting_ensemble')
os.makedirs(voting_ensemble_dir, exist_ok=True)
voting_ensemble_predictions = self.ensemble_methods.voting_ensemble_predict(predictions,
models,
test_features,
test_labels,
voting_ensemble_dir)
# ボルティングアンサンブルの評価
r2 = self.evaluator.evaluate_model_performance(test_labels,
voting_ensemble_predictions,
voting_ensemble_dir)
self.evaluator.plot_actual_predicted(test_labels,
voting_ensemble_predictions,
voting_ensemble_dir,
r2)
# 予測結果の保存
utils.save_predictions_to_csv(test_features,
test_labels,
voting_ensemble_predictions,
voting_ensemble_dir)
# 予測結果から的中率を計算
utils.analyze_and_save_win_rates(voting_ensemble_dir, self.train_test_split_date)
# ブレンディングアンサンブル予測の実行
blending_ensemble_dir = os.path.join(dir_path, 'blending_ensemble')
os.makedirs(blending_ensemble_dir, exist_ok=True)
blending_ensemble_predictions = self.ensemble_methods.blending_ensemble_predict(encoded_train_features,
encoded_test_features,
all_train_labels,
test_labels, models,
blending_ensemble_dir)
# ブレンディングアンサンブルモデルの評価
r2 = self.evaluator.evaluate_model_performance(test_labels,
blending_ensemble_predictions,
blending_ensemble_dir)
self.evaluator.plot_actual_predicted(test_labels,
blending_ensemble_predictions,
blending_ensemble_dir,
r2)
# 予測結果の保存
utils.save_predictions_to_csv(test_features,
test_labels,
blending_ensemble_predictions,
blending_ensemble_dir)
# 予測結果から的中率を計算
utils.analyze_and_save_win_rates(blending_ensemble_dir, self.train_test_split_date)
data_processor.py
# ==============================================================================
# レース結果データの前処理プログラム
# ==============================================================================
# プログラムの概要:
# 競馬レース予測のための機械学習に使用するデータの前処理を行うプログラム。
# 特徴量とラベルの準備、データの分割、カテゴリカル変数のエンコーディングを
# 実行する。前処理されたデータは各種モデルの学習に使用される。
#
# プログラムの主な機能:
# 1. データの準備と分割
# - 特徴量とラベル（レースタイム）の抽出
# - 日付による学習/テストデータの分割
# - 特徴量の前処理
# 2. カテゴリカル変数の処理
# - OrdinalEncoderによるエンコーディング
# - エンコーダーモデルの保存
# 3. データ形式の変換
# - DataFrame形式の維持
# - カラム情報の保持
#
# ==============================================================================
# 実行手順
# ==============================================================================
# 1. 必要なライブラリのインストール:
# pip install pandas numpy scikit-learn joblib
#
# 2. 設定値の確認:
# - 分割基準日の設定
# - エンコーダーの保存パス
# - 圧縮レベルの設定
#
# 3. 入力データの準備:
# - レース結果のDataFrame
# 必須カラム:
# - time: 予測対象のレースタイム
# - date: レース日付（YYYY-MM-DD形式）
# - その他の特徴量カラム
#
# 4. 実行方法:
# - model_creator.pyから呼び出される（直接実行は想定していない）
#
# 5. 処理の流れ:
# 1) レース結果データからtime列を分離して特徴量とラベルを作成
# 2) 指定された日付で学習データとテストデータに分割
# 3) カテゴリカル変数をOrdinalEncoderでエンコード
# 4) エンコーダーを保存
# 5) エンコードされたデータを返却
#
# ==============================================================================
# 出力データ
# ==============================================================================
# 1. エンコードされたデータ:
# - 学習用特徴量（DataFrame）
# - テスト用特徴量（DataFrame）
# - 学習用ラベル（ndarray）
# - テスト用ラベル（ndarray）
#
# 2. 保存ファイル:
# - oe_x.pkl: エンコーダーモデル（joblib形式）
#
# ==============================================================================
# プログラム間の関係
# ==============================================================================
# 1. 呼び出し元:
# - model_creator.py
# └─ データの前処理とエンコーディングを実行
#
# 2. データの受け渡し:
# - ModelTrainer: エンコードされたデータを提供
# - EnsembleMethods: 前処理済みデータを提供
#
# ==============================================================================
# 注意事項
# ==============================================================================
# 1. データ要件:
# - 入力データに欠損値が含まれていないこと
# - 日付データが正しいフォーマットであること
# - 必須カラムが全て存在すること
#
# 2. エンコーディング:
# - カテゴリカル変数は全てエンコードされる
# - エンコーダーは再利用のために保存される
# - カラム名は処理後も維持される
#
# 3. データ分割:
# - 分割基準日が適切な範囲にない場合はNoneを返す
# - 学習データが十分なサイズになるよう基準日を設定
#
# 4. メモリ使用:
# - 大規模データセット処理時はメモリ使用量に注意
# - エンコード処理は一時的にメモリ使用量が増加
#
# 5. 保存データ:
# - エンコーダーファイルは自動的に上書きされる
# - 圧縮レベル3でファイルサイズを最適化
#
import pandas as pd
import numpy as np
from sklearn.preprocessing import OrdinalEncoder
import joblib
class DataProcessor:
def __init__(self):
pass
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 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]
ensemble_methods.py
# ==============================================================================
# アンサンブル学習による予測モデル統合プログラム
# ==============================================================================
# プログラムの概要:
# 複数の機械学習モデル（XGBoost, LightGBM, CatBoost）の予測結果を
# 様々なアンサンブル手法で統合し、予測精度の向上を図るプログラム。
# 各アンサンブル手法の結果を評価・可視化する機能も提供する。
#
# プログラムの主な機能:
# 1. アンサンブル手法の実装
# - weighted_ensemble_predict(): 重み付けアンサンブル
# - average_ensemble_predict(): 平均アンサンブル
# - stacking_ensemble_predict(): スタッキングアンサンブル
# - voting_ensemble_predict(): 投票アンサンブル
# - blending_ensemble_predict(): ブレンディングアンサンブル
# 2. 特徴量重要度の分析
# - 各アンサンブル手法の特徴量重要度計算
# - 重要度の可視化とグラフ保存
# 3. モデル寄与度の分析
# - 各モデルの予測への寄与度計算
# - 寄与度の保存と可視化
#
# ==============================================================================
# 実行手順
# ==============================================================================
# 1. 必要なライブラリのインストール:
# pip install numpy xgboost lightgbm matplotlib joblib
#
# 2. 設定値の確認:
# - メタモデルのパラメータ設定
# - n_estimators: 100
# - learning_rate: 0.1
# - random_state: 42
# - 特徴量重要度の表示設定
# - 上位表示件数: 20件
#
# 3. 入力データの準備:
# - 各モデルの予測結果
# - 学習済みモデル
# - R2スコア
# - 特徴量データ
# - 正解ラベル
#
# 4. 実行方法:
# - model_creator.pyから呼び出される（直接実行は想定していない）
#
# 5. 処理の流れ:
# 1) 各アンサンブル手法での予測実行
# 2) 特徴量重要度の計算と可視化
# 3) モデル寄与度の計算
# 4) 結果の保存
#
# ==============================================================================
# 出力データ
# ==============================================================================
# 1. アンサンブル予測結果:
# - 各手法での予測値（numpy array）
#
# 2. モデル関連ファイル:
# - stacking_meta_model.pkl: スタッキング用メタモデル
# - blending_meta_model.pkl: ブレンディング用メタモデル
# - ensemble_weights.txt: 重み付けアンサンブルの重み
# - model_contributions.txt: 各モデルの寄与度
#
# 3. 可視化ファイル:
# - weighted_ensemble_feature_importance.svg
# - ensemble_feature_importance.svg
# - stacking_feature_importance.svg
# - voting_ensemble_feature_importance.svg
# - blending_feature_importance.svg
#
# ==============================================================================
# プログラム間の関係
# ==============================================================================
# 1. 呼び出し元:
# - model_creator.py
# └─ アンサンブル学習の実行を制御
#
# 2. データの受け取り:
# - ModelTrainer: 基本モデルの予測結果
# - DataProcessor: 前処理済みデータ
#
# 3. データの受け渡し:
# - Evaluator: アンサンブル予測結果の評価
#
# ==============================================================================
# 注意事項
# ==============================================================================
# 1. 実行環境:
# - GPU環境での実行を推奨
# - メモリ使用量に注意（特にスタッキング/ブレンディング時）
#
# 2. データ要件:
# - 全てのモデルの予測結果が必要
# - 予測値のシェイプが一致していること
# - 欠損値が含まれていないこと
#
# 3. アンサンブル手法の特徴:
# - 重み付け: モデルのR2スコアに基づく重み付け
# - スタッキング: メタモデルにXGBoostを使用
# - ブレンディング: 元の特徴量も利用
# - 投票: 予測値の平均化と寄与度計算
#
# 4. 保存データ:
# - SVGファイルは上書きされる
# - メタモデルは圧縮形式で保存
# - 重み設定は可読形式で保存
#
# 5. 性能に関する注意:
# - スタッキング/ブレンディングは計算コストが高い
# - 大規模データセットでは処理時間が長くなる可能性
# - メモリ使用量の監視が必要
#
import numpy as np
import xgboost as xgb
import lightgbm as lgb
import matplotlib.pyplot as plt
from sklearn.metrics import r2_score
import joblib
import os
class EnsembleMethods:
def __init__(self):
pass
def weighted_ensemble_predict(self, predictions, models, r2_scores, ensemble_save_dir_path):
"""
各モデルのR2スコアに基づいて重み付けを行い、アンサンブル予測を実行する。
重みの高いモデルの予測がより大きく影響を与える。
重み付けの結果とモデルの特徴量重要度も保存する。
:param dict predictions: 各モデルの予測結果。キーはモデル名、値は予測値のnp.ndarray。
:param dict models: 訓練済みモデル。キーはモデル名、値はモデルオブジェクト。
:param dict r2_scores: 各モデルのR2スコア。キーはモデル名、値はスコア。
:param str ensemble_save_dir_path: アンサンブル結果を保存するディレクトリパス。
:return: 重み付けアンサンブルによる予測結果。
:rtype: np.ndarray
"""
# R2スコアに基づいて重みを計算
total_r2 = sum(r2_scores.values())
weights = {model: score / total_r2 for model, score in r2_scores.items()}
# 重み付けアンサンブル予測
weighted_ensemble_predictions = np.zeros_like(list(predictions.values())[0])
for model, pred in predictions.items():
weighted_ensemble_predictions += weights[model] * pred
# アンサンブルモデルの特徴量重要度（重み付け平均）
self.plot_weighted_ensemble_feature_importance(models, weights, ensemble_save_dir_path)
# 重みの保存
with open(os.path.join(ensemble_save_dir_path, 'ensemble_weights.txt'), 'w') as f:
for model, weight in weights.items():
f.write(f"{model}: {weight:.4f}\n")
return weighted_ensemble_predictions
def average_ensemble_predict(self, predictions, models, test_features, test_labels, ensemble_save_dir_path):
"""
各モデルの予測結果の単純平均を計算してアンサンブル予測を行う。
すべてのモデルの予測に対して同じ重みを適用する。
アンサンブルモデルの特徴量重要度も計算して保存する。
:param dict predictions: 各モデルの予測結果。キーはモデル名、値は予測値のnp.ndarray。
:param dict models: 訓練済みモデル。キーはモデル名、値はモデルオブジェクト。
:param DataFrame test_features: テストデータの特徴量。
:param np.ndarray test_labels: テストデータの正解ラベル。
:param str ensemble_save_dir_path: アンサンブル結果を保存するディレクトリパス。
:return: 平均アンサンブルによる予測結果。
:rtype: np.ndarray
"""
# アンサンブル予測（単純平均）
average_ensemble_predictions = np.mean(list(predictions.values()), axis=0)
# アンサンブルモデルの特徴量重要度（個々のモデルの重要度の平均）
self.plot_average_ensemble_feature_importance(models, ensemble_save_dir_path)
return average_ensemble_predictions
def stacking_ensemble_predict(self, train_features, test_features, train_labels, test_labels, base_models,
ensemble_save_dir_path):
"""
スタッキングアンサンブルを実行する。
ベースモデルの予測を特徴量として使用し、メタモデル（XGBoost）で最終予測を行う。
メタモデルとその特徴量重要度を保存する。
:param DataFrame train_features: トレーニングデータの特徴量。
:param DataFrame test_features: テストデータの特徴量。
:param np.ndarray train_labels: トレーニングデータのラベル。
:param np.ndarray test_labels: テストデータのラベル。
:param dict base_models: ベースモデルの辞書。キーはモデル名、値はモデルオブジェクト。
:param str ensemble_save_dir_path: アンサンブル結果を保存するディレクトリパス。
:return: スタッキングアンサンブルによる予測結果。
:rtype: np.ndarray
"""
# ベースモデルの予測を生成
base_train_predictions = np.zeros((len(train_labels), len(base_models)))
base_test_predictions = np.zeros((len(test_labels), len(base_models)))
for i, (model_name, model) in enumerate(base_models.items()):
if model_name == 'xgboost':
dtrain = xgb.DMatrix(train_features)
dtest = xgb.DMatrix(test_features)
base_train_predictions[:, i] = model.predict(dtrain)
base_test_predictions[:, i] = model.predict(dtest)
elif model_name == 'lightgbm':
base_train_predictions[:, i] = model.predict(train_features)
base_test_predictions[:, i] = model.predict(test_features)
elif model_name == 'catboost':
base_train_predictions[:, i] = model.predict(train_features)
base_test_predictions[:, i] = model.predict(test_features)
# メタモデル（XGBoost）のトレーニング
meta_model = xgb.XGBRegressor(n_estimators=100, learning_rate=0.1, random_state=42)
meta_model.fit(base_train_predictions, train_labels)
# メタモデルによる最終予測
stacking_ensemble_predictions = meta_model.predict(base_test_predictions)
# スタッキングモデルの特徴量重要度
self.plot_stacking_feature_importance(meta_model, list(base_models.keys()), ensemble_save_dir_path)
# メタモデルの保存
joblib.dump(meta_model, os.path.join(ensemble_save_dir_path, 'stacking_meta_model.pkl'))
return stacking_ensemble_predictions
def voting_ensemble_predict(self, predictions, models, test_features, test_labels, ensemble_save_dir_path):
"""
ボルティングアンサンブルを実行する。
各モデルの予測を平均化し、各モデルの寄与度も計算する。
モデルの特徴量重要度と寄与度情報を保存する。
:param dict predictions: 各モデルの予測結果。キーはモデル名、値は予測値のnp.ndarray。
:param dict models: 訓練済みモデル。キーはモデル名、値はモデルオブジェクト。
:param DataFrame test_features: テストデータの特徴量。
:param np.ndarray test_labels: テストデータのラベル。
:param str ensemble_save_dir_path: アンサンブル結果を保存するディレクトリパス。
:return: ボルティングアンサンブルによる予測結果。
:rtype: np.ndarray
"""
# ボルティングアンサンブル予測（平均）
voting_ensemble_predictions = np.mean(list(predictions.values()), axis=0)
# ボルティングアンサンブルモデルの特徴量重要度（個々のモデルの重要度の平均）
self.plot_voting_ensemble_feature_importance(models, ensemble_save_dir_path)
# 各モデルの寄与度を保存
self.save_model_contributions(predictions, test_labels, ensemble_save_dir_path)
return voting_ensemble_predictions
def blending_ensemble_predict(self, train_features, test_features, train_labels, test_labels, base_models,
ensemble_save_dir_path):
"""
ブレンディングアンサンブルを実行する。
元の特徴量とベースモデルの予測を結合し、メタモデル（XGBoost）で最終予測を行う。
メタモデルと特徴量重要度を保存する。
:param DataFrame train_features: トレーニングデータの特徴量。
:param DataFrame test_features: テストデータの特徴量。
:param np.ndarray train_labels: トレーニングデータのラベル。
:param np.ndarray test_labels: テストデータのラベル。
:param dict base_models: ベースモデルの辞書。キーはモデル名、値はモデルオブジェクト。
:param str ensemble_save_dir_path: アンサンブル結果を保存するディレクトリパス。
:return: ブレンディングアンサンブルによる予測結果。
:rtype: np.ndarray
"""
# ベースモデルの予測を生成
base_train_predictions = np.zeros((len(train_labels), len(base_models)))
base_test_predictions = np.zeros((len(test_labels), len(base_models)))
for i, (model_name, model) in enumerate(base_models.items()):
if model_name == 'xgboost':
dtrain = xgb.DMatrix(train_features)
dtest = xgb.DMatrix(test_features)
base_train_predictions[:, i] = model.predict(dtrain)
base_test_predictions[:, i] = model.predict(dtest)
elif model_name == 'lightgbm':
base_train_predictions[:, i] = model.predict(train_features)
base_test_predictions[:, i] = model.predict(test_features)
elif model_name == 'catboost':
base_train_predictions[:, i] = model.predict(train_features)
base_test_predictions[:, i] = model.predict(test_features)
# 元の特徴量とベースモデルの予測を結合
blended_train_features = np.hstack((train_features, base_train_predictions))
blended_test_features = np.hstack((test_features, base_test_predictions))
# メタモデル（XGBoost）のトレーニング
meta_model = xgb.XGBRegressor(n_estimators=100, learning_rate=0.1, random_state=42)
meta_model.fit(blended_train_features, train_labels)
# メタモデルによる最終予測
blending_ensemble_predictions = meta_model.predict(blended_test_features)
# ブレンディングモデルの特徴量重要度
self.plot_blending_feature_importance(meta_model, train_features.columns, base_models.keys(),
ensemble_save_dir_path)
# メタモデルの保存
joblib.dump(meta_model, os.path.join(ensemble_save_dir_path, 'blending_meta_model.pkl'))
return blending_ensemble_predictions
def plot_weighted_ensemble_feature_importance(self, models, weights, ensemble_save_dir_path):
"""
重み付きアンサンブルモデルの特徴量重要度を計算し、可視化する。
各モデルの特徴量重要度に重みを適用して集計する。
:param dict models: 訓練済みモデル。キーはモデル名、値はモデルオブジェクト。
:param dict weights: 各モデルの重み。キーはモデル名、値は重み。
:param str ensemble_save_dir_path: 結果を保存するディレクトリパス。
:return: なし
"""
plt.figure(figsize=(10, 8))
feature_importance = {}
for model_type, model in models.items():
if model_type == 'xgboost':
importance = model.get_score(importance_type='gain')
elif model_type == 'lightgbm':
importance = dict(zip(model.feature_name(), model.feature_importance(importance_type='gain')))
elif model_type == 'catboost':
importance = dict(zip(model.feature_names_, model.get_feature_importance()))
for feature, value in importance.items():
if feature in feature_importance:
feature_importance[feature] += value * weights[model_type]
else:
feature_importance[feature] = value * weights[model_type]
# ソートして上位20個の特徴量を表示
sorted_importance = sorted(feature_importance.items(), key=lambda x: x[1], reverse=True)[:20]
features, values = zip(*sorted_importance)
plt.barh(range(len(features)), values)
plt.yticks(range(len(features)), features)
plt.title('Weighted Ensemble Feature Importance (Top 20)')
plt.xlabel('Weighted Average Feature Importance')
plt.ylabel('Features')
plt.tight_layout()
plt.savefig(os.path.join(ensemble_save_dir_path, 'weighted_ensemble_feature_importance.svg'))
plt.close()
def plot_average_ensemble_feature_importance(self, models, ensemble_save_dir_path):
"""
平均アンサンブルモデルの特徴量重要度を計算し、可視化する。
各モデルの特徴量重要度の平均を計算する。
:param dict models: 訓練済みモデル。キーはモデル名、値はモデルオブジェクト。
:param str ensemble_save_dir_path: 結果を保存するディレクトリパス。
:return: なし
"""
plt.figure(figsize=(10, 8))
feature_importance = {}
for model_type, model in models.items():
if model_type == 'xgboost':
importance = model.get_score(importance_type='gain')
elif model_type == 'lightgbm':
importance = dict(zip(model.feature_name(), model.feature_importance(importance_type='gain')))
elif model_type == 'catboost':
importance = dict(zip(model.feature_names_, model.get_feature_importance()))
for feature, value in importance.items():
if feature in feature_importance:
feature_importance[feature] += value
else:
feature_importance[feature] = value
# 平均化
for feature in feature_importance:
feature_importance[feature] /= len(models)
# ソートして上位20個の特徴量を表示
sorted_importance = sorted(feature_importance.items(), key=lambda x: x[1], reverse=True)[:20]
features, values = zip(*sorted_importance)
plt.barh(range(len(features)), values)
plt.yticks(range(len(features)), features)
plt.title('Ensemble Feature Importance (Top 20)')
plt.xlabel('Average Feature Importance')
plt.ylabel('Features')
plt.tight_layout()
plt.savefig(os.path.join(ensemble_save_dir_path, 'ensemble_feature_importance.svg'))
plt.close()
def plot_stacking_feature_importance(self, meta_model, base_model_names, ensemble_save_dir_path):
"""
スタッキングアンサンブルのメタモデルにおける各ベースモデルの重要度を可視化する。
:param XGBRegressor meta_model: 学習済みメタモデル。
:param list base_model_names: ベースモデルの名前リスト。
:param str ensemble_save_dir_path: 結果を保存するディレクトリパス。
:return: なし
"""
plt.figure(figsize=(10, 6))
importance = meta_model.feature_importances_
indices = np.argsort(importance)[::-1]
plt.title('Stacking Ensemble Feature Importance')
plt.bar(range(len(importance)), importance[indices])
plt.xticks(range(len(importance)), [base_model_names[i] for i in indices], rotation=45)
plt.xlabel('Base Models')
plt.ylabel('Feature Importance')
plt.tight_layout()
plt.savefig(os.path.join(ensemble_save_dir_path, 'stacking_feature_importance.svg'))
plt.close()
def plot_voting_ensemble_feature_importance(self, models, ensemble_save_dir_path):
"""
投票アンサンブルモデルの特徴量重要度を計算し、可視化する。
各モデルの特徴量重要度の平均を計算する。
:param dict models: 訓練済みモデル。キーはモデル名、値はモデルオブジェクト。
:param str ensemble_save_dir_path: 結果を保存するディレクトリパス。
:return: なし
"""
plt.figure(figsize=(10, 8))
feature_importance = {}
for model_type, model in models.items():
if model_type == 'xgboost':
importance = model.get_score(importance_type='gain')
elif model_type == 'lightgbm':
importance = dict(zip(model.feature_name(), model.feature_importance(importance_type='gain')))
elif model_type == 'catboost':
importance = dict(zip(model.feature_names_, model.get_feature_importance()))
for feature, value in importance.items():
if feature in feature_importance:
feature_importance[feature] += value
else:
feature_importance[feature] = value
# 平均化
for feature in feature_importance:
feature_importance[feature] /= len(models)
# ソートして上位20個の特徴量を表示
sorted_importance = sorted(feature_importance.items(), key=lambda x: x[1], reverse=True)[:20]
features, values = zip(*sorted_importance)
plt.barh(range(len(features)), values)
plt.yticks(range(len(features)), features)
plt.title('Voting Ensemble Feature Importance (Top 20)')
plt.xlabel('Average Feature Importance')
plt.ylabel('Features')
plt.tight_layout()
plt.savefig(os.path.join(ensemble_save_dir_path, 'voting_ensemble_feature_importance.svg'))
plt.close()
def plot_blending_feature_importance(self, meta_model, original_features, base_model_names, ensemble_save_dir_path):
"""
ブレンディングアンサンブルの特徴量重要度を可視化する。
元の特徴量とベースモデルの予測の両方に対する重要度を表示する。
:param XGBRegressor meta_model: 学習済みメタモデル。
:param list original_features: 元の特徴量の名前リスト。
:param list base_model_names: ベースモデルの名前リスト。
:param str ensemble_save_dir_path: 結果を保存するディレクトリパス。
:return: なし
"""
plt.figure(figsize=(12, 8))
importance = meta_model.feature_importances_
feature_names = list(original_features) + list(base_model_names)
indices = np.argsort(importance)[::-1]
plt.title('Blending Ensemble Feature Importance')
plt.bar(range(len(importance)), importance[indices])
plt.xticks(range(len(importance)), [feature_names[i] for i in indices], rotation=90)
plt.xlabel('Features')
plt.ylabel('Feature Importance')
plt.tight_layout()
plt.savefig(os.path.join(ensemble_save_dir_path, 'blending_feature_importance.svg'))
plt.close()
def save_model_contributions(self, predictions, test_labels, ensemble_save_dir_path):
"""
各モデルの予測精度（R2スコア）に基づく寄与度を計算し、保存する。
:param dict predictions: 各モデルの予測結果。キーはモデル名、値は予測値のnp.ndarray。
:param np.ndarray test_labels: テストデータの正解ラベル。
:param str ensemble_save_dir_path: 結果を保存するディレクトリパス。
:return: なし
"""
model_contributions = {}
for model_name, model_predictions in predictions.items():
model_contributions[model_name] = r2_score(test_labels, model_predictions)
total_contribution = sum(model_contributions.values())
normalized_contributions = {model: contrib / total_contribution for model, contrib in
model_contributions.items()}
with open(os.path.join(ensemble_save_dir_path, 'model_contributions.txt'), 'w') as f:
for model, contribution in normalized_contributions.items():
f.write(f"{model}: {contribution:.4f}\n")
evaluator.py
# ==============================================================================
# モデル評価・性能分析プログラム
# ==============================================================================
# プログラムの概要:
# 機械学習モデルの予測性能を評価し、結果を可視化・保存するプログラム。
# R2スコア、MAE、RMSEなどの評価指標の計算、予測結果の散布図作成、
# 特徴量重要度の分析を行う。各種評価結果はCSVおよびSVG形式で保存される。
#
# プログラムの主な機能:
# 1. 予測性能の評価
# - R2スコア（決定係数）の計算
# - MAE（平均絶対誤差）の計算
# - RMSE（平均二乗誤差の平方根）の計算
# 2. 結果の可視化
# - 実タイムと予測タイムの散布図作成
# - 特徴量重要度のグラフ作成
# 3. 評価結果の保存
# - 評価指標のCSVファイル保存
# - 可視化結果のSVGファイル保存
#
# ==============================================================================
# 実行手順
# ==============================================================================
# 1. 必要なライブラリのインストール:
# pip install numpy matplotlib scikit-learn japanize_matplotlib pandas
#
# 2. 設定値の確認:
# - グラフサイズ設定
# - 散布図: figsize=(6, 6)
# - 特徴量重要度: figsize=(10, 8)
# - 日本語フォント設定
# - 保存形式設定（SVG）
#
# 3. 入力データの準備:
# - テストデータの正解ラベル
# - モデルの予測結果
# - 評価対象のモデル
# - 結果保存用のディレクトリパス
#
# 4. 実行方法:
# - model_creator.pyから呼び出される（直接実行は想定していない）
#
# 5. 処理の流れ:
# 1) モデルの予測性能を評価（R2, MAE, RMSE）
# 2) 評価指標をCSVファイルに保存
# 3) 実タイムと予測タイムの散布図を作成
# 4) 特徴量重要度のグラフを作成
# 5) 可視化結果をSVGファイルとして保存
#
# ==============================================================================
# 出力データ
# ==============================================================================
# 1. 評価指標ファイル:
# - evaluation_results.csv
# - R2スコア
# - RMSE
# - MAE
#
# 2. 可視化ファイル:
# - 実タイムと予測タイムの散布図.svg
# - 予測タイムの重要度分析_[モデル名].svg
#
# ==============================================================================
# プログラム間の関係
# ==============================================================================
# 1. 呼び出し元:
# - model_creator.py
# └─ モデル評価の実行を制御
#
# 2. データの受け取り:
# - ModelTrainer: モデルと予測結果
# - EnsembleMethods: アンサンブル予測結果
#
# 3. データの利用:
# - ScoreCalculator: 評価指標を利用
#
# ==============================================================================
# 注意事項
# ==============================================================================
# 1. データ要件:
# - 予測値と正解値のシェイプが一致していること
# - 欠損値が含まれていないこと
# - 入力データが数値型であること
#
# 2. 可視化設定:
# - 日本語フォントが正しく設定されていること
# - グラフのサイズが適切であること
# - 軸ラベルが正しく表示されること
#
# 3. 保存データ:
# - ファイルは自動的に上書きされる
# - SVGファイルは高解像度で保存
# - 保存先ディレクトリが存在しない場合は自動作成
#
# 4. メモリ使用:
# - 大規模データセットでの可視化時はメモリ使用量に注意
# - グラフは適切なタイミングでクローズする
#
# 5. モデル固有の注意:
# - XGBoost: feature_importanceはgainタイプを使用
# - LightGBM: feature_importanceは独自の計算方法
# - CatBoost: feature_namesの指定が必要
#
import numpy as np
import matplotlib.pyplot as plt
from sklearn.metrics import r2_score, mean_absolute_error, mean_squared_error
import os
import xgboost as xgb
import japanize_matplotlib
import pandas as pd
class Evaluator:
def __init__(self):
pass
def evaluate_model_performance(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))
# 保存用のCSVパス
csv_path = os.path.join(model_save_dir_path, 'evaluation_results.csv')
# CSVに書き込むデータを準備
evaluation_data = {
'R2': [r2],
'RMSE': [rmse],
'MAE': [mae]
}
# データフレームを作成
df = pd.DataFrame(evaluation_data)
# CSVとして保存
df.to_csv(csv_path, index=False, encoding='utf-8-sig')
return r2
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, model, model_type, model_save_dir_path):
"""
モデルの特徴量重要度をプロットし、SVGファイルとして保存します。
:param model: 特徴量の重要度を表示するモデル（XGBoost、LightGBM、またはCatBoost）。
:param str model_type: モデルの種類（'xgboost'、'lightgbm'、または'catboost'）。
:param str model_save_dir_path: グラフを保存するディレクトリパス。
:return: なし
"""
plt.figure(figsize=(10, 8))
if model_type == 'xgboost':
xgb.plot_importance(model, ax=plt.gca(), importance_type='gain')
elif model_type == 'lightgbm':
importance = model.feature_importance(importance_type='gain')
feature_names = model.feature_name()
indices = np.argsort(importance)
plt.barh(range(len(indices)), importance[indices])
plt.yticks(range(len(indices)), [feature_names[i] for i in indices])
elif model_type == 'catboost':
feature_importance = model.get_feature_importance()
feature_names = model.feature_names_
sorted_idx = np.argsort(feature_importance)
pos = np.arange(sorted_idx.shape[0]) + .5
plt.barh(pos, feature_importance[sorted_idx], align='center')
plt.yticks(pos, np.array(feature_names)[sorted_idx])
else:
raise ValueError(f"Unsupported model type: {model_type}")
plt.title(f'予測タイムの重要度分析 ({model_type.capitalize()})')
plt.xlabel('Feature Importance')
plt.ylabel('Features')
plt.tight_layout()
plt.savefig(os.path.join(model_save_dir_path, f'予測タイムの重要度分析_{model_type}.svg'))
plt.close()generate_folder_list.py
# ==============================================================================
# モデル保存フォルダ管理プログラム
# ==============================================================================
# プログラムの概要:
# 競馬予測システムで生成された学習結果フォルダを管理するプログラム。
# YYYYMMdd_HHMMSS形式のフォルダを検索し、一覧をJSON形式で保存する。
# 生成されたJSONファイルは、Web UI（report2.html）でのフォルダ選択に
# 使用される。
#
# プログラムの主な機能:
# 1. フォルダ検索機能
# - 指定ディレクトリ内のフォルダ検索
# - YYYYMMdd_HHMMSS形式の検証
# - 日付順での降順ソート
# 2. JSON生成機能
# - フォルダ一覧のJSON形式変換
# - 自動的なディレクトリ作成
# - 既存JSONファイルの更新
#
# ==============================================================================
# 実行手順
# ==============================================================================
# 1. 必要なライブラリのインストール:
# - pythonの標準ライブラリのみを使用（追加のインストール不要）
#
# 2. 設定値の確認:
# target_dir: 検索対象のディレクトリパス
# output_path: 出力するJSONファイルのパス
#
# 3. 入力データの準備:
# - target_dirに検索対象のフォルダが存在することを確認
# - フォルダ名がYYYYMMdd_HHMMSS形式であることを確認
#
# 4. 実行方法:
# python generate_folder_list.py <対象ディレクトリ> <出力JSONパス>
# 例: python generate_folder_list.py . ./data/folder_list.json
#
# 5. 処理の流れ:
# 1) 指定ディレクトリ内のフォルダを検索
# 2) YYYYMMdd_HHMMSS形式のフォルダを抽出
# 3) 日付順に降順ソート
# 4) JSONファイルとして保存
#
# ==============================================================================
# 出力データ
# ==============================================================================
# 1. JSONファイル:
# - folder_list.json
# - フォルダ名の配列
# - 日付降順でソート済み
#
# 2. フォーマット例:
# [
# "20240119_235959",
# "20240119_225858",
# "20240119_215757"
# ]
#
# ==============================================================================
# プログラム間の関係
# ==============================================================================
# 1. 呼び出し元:
# - train_model.py
# └─ モデル作成完了後にフォルダリストを更新
#
# 2. 出力の利用:
# - report2.html
# └─ フォルダリストをドロップダウンメニューに表示
#
# ==============================================================================
# 注意事項
# ==============================================================================
# 1. フォルダ名の要件:
# - 厳密にYYYYMMdd_HHMMSS形式である必要がある
# - 15文字の固定長
# - 日付部分が8桁の数字
# - アンダースコアで区切られている
# - 時刻部分が6桁の数字
#
# 2. ファイル操作:
# - 出力ディレクトリが存在しない場合は自動的に作成
# - 既存のJSONファイルは上書きされる
# - ファイルパーミッションに注意
#
# 3. エラー処理:
# - 不適切なフォルダ名は自動的に除外
# - ディレクトリ権限エラーの可能性
# - JSON保存時のエラー処理
#
# 4. パフォーマンス:
# - 大量のフォルダが存在する場合の処理時間
# - メモリ使用量は通常小さい
#
# 5. 拡張性:
# - フォルダ名フォーマットの変更が可能
# - JSON以外の出力形式への対応が可能
# - ソート順の変更が可能
#
import os
import json
import sys
from pathlib import Path
def generate_folder_list(target_dir, output_path):
"""
指定されたディレクトリ内のフォルダ一覧を生成しJSONファイルとして保存する。
:param str target_dir: 検索対象のディレクトリパス。
:param str output_path: 出力するJSONファイルのパス。
:return: なし
:rtype: None
"""
try:
# 対象ディレクトリの存在確認
if not os.path.exists(target_dir):
raise FileNotFoundError(f"指定されたディレクトリが存在しません: {target_dir}")
# YYYYMMdd_HHMMSS形式のフォルダを取得
folders = [d for d in os.listdir(target_dir)
if os.path.isdir(os.path.join(target_dir, d)) and
len(d) == 15 and
d[:8].isdigit() and
d[8] == '_' and
d[9:].isdigit()]
# 日付順に降順ソート
folders.sort(reverse=True)
# 出力ディレクトリが存在しない場合は作成
os.makedirs(os.path.dirname(output_path), exist_ok=True)
# JSONファイルとして保存
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(folders, f, ensure_ascii=False, indent=2)
print(f"フォルダ一覧を生成しました: {output_path}")
print(f"フォルダ数: {len(folders)}")
except Exception as e:
print(f"エラーが発生しました: {str(e)}", file=sys.stderr)
sys.exit(1)
def main():
"""
コマンドライン引数を処理する。
使用例：
python generate_folder_list.py /path/to/target/dir /path/to/output/folder_list.json
"""
if len(sys.argv) != 3:
print("使用方法: python generate_folder_list.py <対象ディレクトリ> <出力JSONパス>")
print("例: python generate_folder_list.py . ./data/folder_list.json")
sys.exit(1)
target_dir = sys.argv[1]
output_path = sys.argv[2]
generate_folder_list(target_dir, output_path)
if __name__ == '__main__':
main()score_calculator.py
# ==============================================================================
# モデル性能スコア計算プログラム
# ==============================================================================
# プログラムの概要:
# 競馬予測モデルの評価スコアを計算し、的中率結果のCSVファイルに
# スコアを追加するプログラム。単勝的中率、R2スコア、RMSE、MAEに
# 重み付けを行い、総合的な性能評価スコアを算出する。
#
# プログラムの主な機能:
# 1. スコア計算機能
# - 単勝的中率への重み付け (0.4)
# - R2スコアへの重み付け (0.3)
# - RMSEへの重み付け (0.2)
# - MAEへの重み付け (0.1)
# 2. スコア正規化
# - RMSEの正規化（1/(1+RMSE)）
# - MAEの正規化（1/(1+MAE)）
# 3. 結果の更新
# - 的中率CSVへのスコア追加
# - 新規CSVファイルの生成
#
# ==============================================================================
# 実行手順
# ==============================================================================
# 1. 必要なライブラリのインストール:
# pip install pandas numpy
#
# 2. 設定値の確認:
# - スコア重み
# - 単勝的中率: 0.4
# - R2スコア: 0.3
# - RMSE: 0.2
# - MAE: 0.1
# - CSVエンコーディング設定
#
# 3. 入力データの準備:
# - 的中率結果.csv
# 必須カラム:
# - 単勝: 的中率（%表示）
# - R2: R2スコア
# - RMSE: 平均二乗誤差の平方根
# - MAE: 平均絶対誤差
#
# 4. 実行方法:
# - train_model.pyから呼び出される（直接実行は想定していない）
#
# 5. 処理の流れ:
# 1) 的中率CSVファイルの読み込み
# 2) 各指標の正規化処理
# 3) 重み付けスコアの計算
# 4) 結果を新しいCSVファイルとして保存
#
# ==============================================================================
# 出力データ
# ==============================================================================
# 1. スコア付き結果ファイル:
# - 的中率結果_with_score.csv
# - 元の的中率データ
# + スコア列（0-1の範囲）
#
# 2. スコアの計算式:
# score = (0.4 * win_rate +
# 0.3 * r2 +
# 0.2 * normalized_rmse +
# 0.1 * normalized_mae)
#
# ==============================================================================
# プログラム間の関係
# ==============================================================================
# 1. 呼び出し元:
# - train_model.py
# └─ スコア計算の実行を制御
#
# 2. データの受け取り:
# - Evaluator: 評価指標を提供
# - utils: 的中率データを提供
#
# ==============================================================================
# 注意事項
# ==============================================================================
# 1. データ要件:
# - 入力CSVファイルの必須カラム確認
# - パーセント表示の値の処理
# - 数値型への変換確認
#
# 2. スコア計算:
# - 重みの合計は1になるように設定
# - 各指標の正規化範囲の確認
# - ゼロ除算の防止
#
# 3. ファイル操作:
# - CSVファイルは'cp932'エンコーディングで保存
# - 既存ファイルは上書きされる
# - 出力パスの存在確認
#
# 4. エラー処理:
# - 欠損値の処理
# - 不正なデータ形式の処理
# - ファイル読み書きエラーの処理
#
# 5. 拡張性:
# - 新しい評価指標の追加が可能
# - 重み付けの調整が可能
# - 正規化方法の変更が可能
#
import pandas as pd
import numpy as np
class ScoreCalculator:
def calculate_model_score(win_rate, r2, rmse, mae):
"""
モデルのスコアを計算するメソッド。
:param float win_rate: 的中率 (0-1スケール)。
:param float r2: R²スコア。
:param float rmse: 平均二乗誤差平方根。
:param float mae: 平均絶対誤差。
:return: スコア。
"""
try:
# RMSEとMAEは小さいほど良いので逆数を計算
normalized_rmse = 1 / (1 + rmse)
normalized_mae = 1 / (1 + mae)
# 重み付けスコアを計算
score = (0.4 * win_rate +
0.3 * r2 +
0.2 * normalized_rmse +
0.1 * normalized_mae)
return score
except Exception as e:
print(f"スコア計算中にエラーが発生しました: {e}")
return 0.0
def add_scores_to_csv(file_path, output_path):
"""
的中率結果.csvにスコアを追加し、新しいCSVとして保存する。
:param str file_path: 入力CSVファイルのパス。
:param str output_path: 出力CSVファイルのパス。
"""
# ファイルを読み込む
df = pd.read_csv(file_path, encoding='cp932')
# スコア列を計算して追加
def calculate_row_score(row):
win_rate = float(row['単勝'].strip('%')) / 100
r2 = float(row['R2']) if 'R2' in row and not pd.isna(row['R2']) else 0
rmse = float(row['RMSE']) if 'RMSE' in row and not pd.isna(row['RMSE']) else 0
mae = float(row['MAE']) if 'MAE' in row and not pd.isna(row['MAE']) else 0
return ScoreCalculator.calculate_model_score(win_rate, r2, rmse, mae)
df['スコア'] = df.apply(calculate_row_score, axis=1)
# 保存
df.to_csv(output_path, index=False, encoding='cp932')
print(f"スコアを追加したCSVファイルを保存しました: {output_path}")
utils.py
# ==============================================================================
# 予測結果評価・保存用ユーティリティプログラム
# ==============================================================================
# プログラムの概要:
# 競馬予測システムにおけるレース予測結果の評価と保存に関する
# ユーティリティ関数を提供するプログラム。各種的中率の計算、
# 予測結果のCSV保存、評価結果の分析などの機能を提供する。
#
# プログラムの主な機能:
# 1. 的中率計算機能
# - 単勝の的中率計算
# - 複勝の的中率計算
# - 馬連の的中率計算
# - ワイドの的中率計算
# - 三連複の的中率計算
# 2. ランク付け機能
# - 実際のタイムによるランク付け
# - 予測タイムによるランク付け
# 3. データ保存機能
# - 予測結果のCSV保存
# - 的中率の分析と保存
#
# ==============================================================================
# 実行手順
# ==============================================================================
# 1. 必要なライブラリのインストール:
# pip install pandas numpy
#
# 2. 設定値の確認:
# - CSV保存時のエンコーディング設定
# - 出力ディレクトリパスの設定
# - 的中率計算条件の確認
#
# 3. 入力データの準備:
# - レース結果のDataFrame
# 必須カラム:
# - race_name: レース名
# - location: 開催場所
# - round: ラウンド
# - time: 実際のタイム
# - time_predict: 予測タイム
#
# 4. 実行方法:
# - model_creator.pyから呼び出される（直接実行は想定していない）
#
# 5. 処理の流れ:
# 1) レース結果データの読み込み
# 2) 実際の順位と予測順位の付与
# 3) 各種的中率の計算
# 4) 結果のCSV保存
#
# ==============================================================================
# 出力データ
# ==============================================================================
# 1. 予測結果ファイル:
# - 予測結果.csv
# - レース情報
# - 実際のタイム
# - 予測タイム
#
# 2. 的中率結果ファイル:
# - 的中率.csv
# - 単勝的中率
# - 複勝的中率
# - 馬連的中率
# - ワイド的中率
# - 三連複的中率
# - モデルパス
# - 分割日
#
# ==============================================================================
# プログラム間の関係
# ==============================================================================
# 1. 呼び出し元:
# - model_creator.py
# └─ 予測結果の評価と保存
#
# 2. データの利用:
# - ModelTrainer: 予測結果の提供
# - EnsembleMethods: アンサンブル予測結果の提供
# - ScoreCalculator: 的中率データの利用
#
# ==============================================================================
# 注意事項
# ==============================================================================
# 1. データ要件:
# - 必須カラムが全て存在すること
# - タイムデータが数値型であること
# - レース情報が正しく設定されていること
#
# 2. 的中率計算の条件:
# - 複勝の判定
# - 8頭以上: 3着以内
# - 8頭未満: 2着以内
# - 5頭未満のレースは判定対象外
#
# 3. ファイル保存:
# - CSVファイルは'cp932'エンコーディングで保存
# - エラー発生時は'ignore'オプションで処理継続
# - 的中率は追記モードで保存
#
# 4. 性能に関する注意:
# - 大量のレースデータ処理時はメモリ使用量に注意
# - DataFrame操作時の処理効率に注意
#
# 5. 拡張性:
# - 新しい的中率計算方法の追加が可能
# - 保存形式の変更が可能
# - 評価指標の追加が可能
#
import pandas as pd
import os
import csv
import datetime
def add_real_and_predicted_ranks(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 calculate_win_accuracy(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(ranked_race_results):
"""
複勝(英訳:place)の的中を確認する。
8頭以上の出走レースでは、3着以内に予測順位1～3の有無を確認する。
8頭未満が出走しているレースでは、2着以内に予測順位1～2の有無を確認する。
予想の当たり1/外れ0を返す
:param ranked_race_results:
:return:
"""
horses_num = len(ranked_race_results)
if horses_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(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(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(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 save_predictions_to_csv(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 analyze_and_save_win_rates(model_save_dir_path, train_test_split_date):
"""
予測結果に基づいて単勝、複勝、馬連、ワイド、三連複の的中率を計算し、CSVファイルに保存する。
:param model_save_dir_path: 予測結果を含むCSVファイルが保存されるディレクトリパス。
:param train_test_split_date: トレーニングデータとテストデータを分割する日付。
: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 = add_real_and_predicted_ranks(group_race_result)
num_hits_win += calculate_win_accuracy(race_result_assign_rank)
num_hits_place += calculate_place_accuracy(race_result_assign_rank)
num_hits_quinella += calculate_quinella_accuracy(race_result_assign_rank)
num_hits_quinella_place += calculate_quinella_place_accuracy(race_result_assign_rank)
num_hits_trio += calculate_trio_accuracy(race_result_assign_rank)
total_race += 1
current_date = datetime.datetime.now().strftime('%Y%m%d')
# evaluation_results.csvの読み込み
eval_csv_path = os.path.join(model_save_dir_path, 'evaluation_results.csv')
if os.path.exists(eval_csv_path):
eval_results = pd.read_csv(eval_csv_path)
r2 = eval_results['R2'].iloc[0]
rmse = eval_results['RMSE'].iloc[0]
mae = eval_results['MAE'].iloc[0]
else:
r2 = None
rmse = None
mae = None
# CSVファイルにヘッダーを書き込むかを判定する
# CSVファイルが存在しないあるいはファイルサイズが0の場合、ヘッダーを書き込むをTrueにする
csv_columns = ['作成日', '単勝', '複勝', '馬連', 'ワイド', '三連複', 'モデルパス', '分割日', 'R2', 'RMSE', 'MAE']
csv_file_path = f'{model_save_dir_path}/../../的中率結果.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,
train_test_split_date,
f'{r2:.5f}' if r2 is not None else 'N/A',
f'{rmse:.5f}' if rmse is not None else 'N/A',
f'{mae:.5f}' if mae is not None else 'N/A'
])
# 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, train_test_split_date])
以上です。
コメント