競馬AI予測システム設計：データ収集から機械学習モデルまで完全ガイド

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

-- 馬テーブル
CREATE TABLE horses (
    horse_id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    birth_date DATE,
    gender CHAR(1) CHECK (gender IN ('牡', '牝', 'セ')),
    trainer_id INTEGER REFERENCES trainers(trainer_id),
    bloodline_id INTEGER REFERENCES bloodlines(bloodline_id),
    color VARCHAR(20),
    origin VARCHAR(50),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- レース結果テーブル（時系列データ）
CREATE TABLE race_results (
    result_id SERIAL,
    race_id INTEGER NOT NULL REFERENCES races(race_id),
    horse_id INTEGER NOT NULL REFERENCES horses(horse_id),
    jockey_id INTEGER NOT NULL REFERENCES jockeys(jockey_id),
    post_position SMALLINT,
    finish_position SMALLINT,
    finish_time NUMERIC(6, 1),
    last_3f_time NUMERIC(4, 1),
    weight NUMERIC(4, 1),
    weight_diff NUMERIC(3, 1),
    odds NUMERIC(6, 1),
    popularity SMALLINT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (result_id, race_id)
);

-- TimescaleDBでハイパーテーブル化
SELECT create_hypertable('race_results', 'created_at');

-- 血統テーブル
CREATE TABLE bloodlines (
    bloodline_id SERIAL PRIMARY KEY,
    sire VARCHAR(100),           -- 父
    dam VARCHAR(100),            -- 母
    broodmare_sire VARCHAR(100), -- 母父
    sire_line VARCHAR(50),       -- 父系統
    INDEX idx_sire (sire),
    INDEX idx_broodmare_sire (broodmare_sire)
);

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304

import pandas as pd
import numpy as np
from typing import List, Dict, Tuple

class FeatureEngineer:
    """競馬予測のための特徴量エンジニアリング"""

    def __init__(self, n_past_races: int = 5):
        self.n_past_races = n_past_races

    def create_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """全特徴量を生成"""
        features = pd.DataFrame(index=df.index)

        # 馬の基本特徴量
        features = pd.concat([
            features,
            self._create_horse_basic_features(df),
            self._create_past_performance_features(df),
            self._create_bloodline_features(df),
            self._create_jockey_trainer_features(df),
            self._create_race_condition_features(df),
            self._create_odds_features(df),
            self._create_training_features(df)
        ], axis=1)

        return features

    def _create_horse_basic_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """馬の基本特徴量"""
        return pd.DataFrame({
            # 年齢（レース日 - 誕生日）
            'age': (df['race_date'] - df['birth_date']).dt.days / 365.25,

            # 性別（ワンホットエンコーディング）
            'is_male': (df['gender'] == '牡').astype(int),
            'is_female': (df['gender'] == '牝').astype(int),
            'is_gelding': (df['gender'] == 'セ').astype(int),

            # 馬体重
            'weight': df['weight'],
            'weight_diff': df['weight_diff'],

            # 体重変動の絶対値（大きな変動は不安要素）
            'weight_diff_abs': df['weight_diff'].abs(),

            # 適正体重からの乖離（馬ごとの平均体重との差）
            'weight_deviation': df.groupby('horse_id')['weight'].transform(
                lambda x: x - x.rolling(10, min_periods=1).mean()
            )
        }, index=df.index)

    def _create_past_performance_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """過去成績の特徴量"""
        features = {}

        # 馬ごとにソートして過去成績を計算
        df_sorted = df.sort_values(['horse_id', 'race_date'])

        for i in range(1, self.n_past_races + 1):
            # 過去i走前の着順
            features[f'finish_pos_{i}'] = df_sorted.groupby('horse_id')['finish_position'].shift(i)

            # 過去i走前の上がり3F
            features[f'last_3f_{i}'] = df_sorted.groupby('horse_id')['last_3f_time'].shift(i)

        # 過去N走の統計
        past_positions = df_sorted.groupby('horse_id')['finish_position'].apply(
            lambda x: x.shift(1).rolling(self.n_past_races, min_periods=1)
        )

        features['avg_finish_pos'] = past_positions.mean()
        features['min_finish_pos'] = past_positions.min()
        features['std_finish_pos'] = past_positions.std()

        # 勝率・連対率・複勝率
        features['win_rate'] = df_sorted.groupby('horse_id').apply(
            lambda x: (x['finish_position'].shift(1) == 1).rolling(20, min_periods=1).mean()
        ).reset_index(level=0, drop=True)

        features['place_rate'] = df_sorted.groupby('horse_id').apply(
            lambda x: (x['finish_position'].shift(1) <= 2).rolling(20, min_periods=1).mean()
        ).reset_index(level=0, drop=True)

        features['show_rate'] = df_sorted.groupby('horse_id').apply(
            lambda x: (x['finish_position'].shift(1) <= 3).rolling(20, min_periods=1).mean()
        ).reset_index(level=0, drop=True)

        # 距離別成績
        features['distance_win_rate'] = self._calculate_condition_rate(
            df_sorted, 'distance_category', 'win'
        )

        # 馬場別成績
        features['surface_win_rate'] = self._calculate_condition_rate(
            df_sorted, 'surface', 'win'
        )

        # コース別成績
        features['course_win_rate'] = self._calculate_condition_rate(
            df_sorted, 'racecourse_id', 'win'
        )

        return pd.DataFrame(features, index=df.index)

    def _calculate_condition_rate(
        self, df: pd.DataFrame, condition_col: str, rate_type: str
    ) -> pd.Series:
        """条件別の成績を計算"""
        if rate_type == 'win':
            threshold = 1
        elif rate_type == 'place':
            threshold = 2
        else:
            threshold = 3

        return df.groupby(['horse_id', condition_col]).apply(
            lambda x: (x['finish_position'].shift(1) <= threshold).rolling(10, min_periods=1).mean()
        ).reset_index(level=[0, 1], drop=True)

    def _create_bloodline_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """血統特徴量"""
        # 父系統別の距離適性スコア（事前に集計したマスタを使用）
        sire_distance_score = self._load_sire_distance_scores()
        broodmare_sire_score = self._load_broodmare_sire_scores()

        return pd.DataFrame({
            # 父系統の距離適性
            'sire_sprint_score': df['sire'].map(sire_distance_score.get('sprint', {})),
            'sire_mile_score': df['sire'].map(sire_distance_score.get('mile', {})),
            'sire_middle_score': df['sire'].map(sire_distance_score.get('middle', {})),
            'sire_long_score': df['sire'].map(sire_distance_score.get('long', {})),

            # 母父系統の馬場適性
            'broodmare_turf_score': df['broodmare_sire'].map(broodmare_sire_score.get('turf', {})),
            'broodmare_dirt_score': df['broodmare_sire'].map(broodmare_sire_score.get('dirt', {})),

            # 父系統の勝率
            'sire_win_rate': df['sire'].map(self._calculate_sire_win_rate(df)),

            # インブリード係数（近親交配度）
            'inbreeding_coefficient': df.apply(self._calculate_inbreeding, axis=1)
        }, index=df.index)

    def _create_jockey_trainer_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """騎手・調教師特徴量"""
        return pd.DataFrame({
            # 騎手の直近成績
            'jockey_win_rate': df.groupby('jockey_id').apply(
                lambda x: (x['finish_position'].shift(1) == 1).rolling(50, min_periods=1).mean()
            ).reset_index(level=0, drop=True),

            'jockey_place_rate': df.groupby('jockey_id').apply(
                lambda x: (x['finish_position'].shift(1) <= 3).rolling(50, min_periods=1).mean()
            ).reset_index(level=0, drop=True),

            # 調教師の直近成績
            'trainer_win_rate': df.groupby('trainer_id').apply(
                lambda x: (x['finish_position'].shift(1) == 1).rolling(50, min_periods=1).mean()
            ).reset_index(level=0, drop=True),

            # 騎手×馬の相性（過去の組み合わせ成績）
            'jockey_horse_chemistry': self._calculate_jockey_horse_chemistry(df),

            # 騎手×コースの成績
            'jockey_course_win_rate': df.groupby(['jockey_id', 'racecourse_id']).apply(
                lambda x: (x['finish_position'].shift(1) == 1).rolling(20, min_periods=1).mean()
            ).reset_index(level=[0, 1], drop=True),

            # 騎手の乗り替わりフラグ
            'jockey_change': (df.groupby('horse_id')['jockey_id'].shift(1) != df['jockey_id']).astype(int)
        }, index=df.index)

    def _create_race_condition_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """レース条件特徴量"""
        return pd.DataFrame({
            # 距離カテゴリ
            'distance': df['distance'],
            'is_sprint': (df['distance'] <= 1400).astype(int),
            'is_mile': ((df['distance'] > 1400) & (df['distance'] <= 1800)).astype(int),
            'is_middle': ((df['distance'] > 1800) & (df['distance'] <= 2200)).astype(int),
            'is_long': (df['distance'] > 2200).astype(int),

            # 馬場
            'is_turf': (df['surface'] == '芝').astype(int),
            'is_dirt': (df['surface'] == 'ダ').astype(int),

            # 馬場状態
            'condition_good': (df['condition'] == '良').astype(int),
            'condition_slightly_heavy': (df['condition'] == '稍重').astype(int),
            'condition_heavy': (df['condition'] == '重').astype(int),
            'condition_bad': (df['condition'] == '不良').astype(int),

            # コース形態
            'is_right_handed': (df['course_direction'] == '右').astype(int),
            'is_left_handed': (df['course_direction'] == '左').astype(int),

            # グレード
            'is_g1': (df['grade'] == 'G1').astype(int),
            'is_g2': (df['grade'] == 'G2').astype(int),
            'is_g3': (df['grade'] == 'G3').astype(int),
            'is_op': (df['grade'] == 'OP').astype(int),

            # 出走頭数
            'num_runners': df['num_runners'],

            # 枠順
            'post_position': df['post_position'],

            # 枠順の有利不利（コース×距離で異なる）
            'post_position_advantage': self._calculate_post_advantage(df)
        }, index=df.index)

    def _create_odds_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """オッズ特徴量"""
        return pd.DataFrame({
            # 単勝オッズ
            'odds': df['odds'],
            'odds_log': np.log1p(df['odds']),

            # 人気順
            'popularity': df['popularity'],

            # オッズと人気の乖離（穴馬指標）
            'odds_popularity_gap': df['odds'].rank() - df['popularity'],

            # オッズの変動（前日比）
            'odds_change': df['odds'] - df['odds_previous_day'],
            'odds_change_rate': (df['odds'] - df['odds_previous_day']) / df['odds_previous_day'],

            # 支持率（オッズから逆算）
            'support_rate': 0.8 / df['odds'],  # 控除率20%を仮定

            # オッズの偏差（そのレース内での相対的な人気）
            'odds_zscore': df.groupby('race_id')['odds'].transform(
                lambda x: (x - x.mean()) / x.std()
            )
        }, index=df.index)

    def _create_training_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """調教特徴量"""
        return pd.DataFrame({
            # 最終調教タイム
            'training_time': df['training_time'],

            # 調教タイムの偏差
            'training_time_deviation': df.groupby('horse_id')['training_time'].transform(
                lambda x: x - x.rolling(5, min_periods=1).mean()
            ),

            # 調教評価（A, B, C → 数値化）
            'training_score': df['training_evaluation'].map({'A': 3, 'B': 2, 'C': 1, 'D': 0}),

            # 調教コース（坂路、ウッド、ポリ等）
            'training_course_hill': (df['training_course'] == '坂路').astype(int),
            'training_course_wood': (df['training_course'] == 'ウッド').astype(int),
            'training_course_poly': (df['training_course'] == 'ポリ').astype(int),

            # 調教本数
            'training_count': df['training_count'],

            # 併せ馬の有無
            'has_companion_training': df['has_companion_training'].astype(int)
        }, index=df.index)

    def _load_sire_distance_scores(self) -> Dict[str, Dict[str, float]]:
        """父系統の距離適性スコアを読み込む（事前に集計）"""
        # 実際はDBまたはファイルから読み込む
        return {
            'sprint': {},
            'mile': {},
            'middle': {},
            'long': {}
        }

    def _load_broodmare_sire_scores(self) -> Dict[str, Dict[str, float]]:
        """母父系統の馬場適性スコアを読み込む"""
        return {
            'turf': {},
            'dirt': {}
        }

    def _calculate_sire_win_rate(self, df: pd.DataFrame) -> Dict[str, float]:
        """父系統別の勝率を計算"""
        return df.groupby('sire').apply(
            lambda x: (x['finish_position'] == 1).mean()
        ).to_dict()

    def _calculate_inbreeding(self, row: pd.Series) -> float:
        """インブリード係数を計算（簡易版）"""
        # 実際は5代血統表から計算
        return 0.0

    def _calculate_jockey_horse_chemistry(self, df: pd.DataFrame) -> pd.Series:
        """騎手と馬の相性スコアを計算"""
        return df.groupby(['jockey_id', 'horse_id']).apply(
            lambda x: (x['finish_position'].shift(1) <= 3).rolling(5, min_periods=1).mean()
        ).reset_index(level=[0, 1], drop=True)

    def _calculate_post_advantage(self, df: pd.DataFrame) -> pd.Series:
        """枠順の有利不利スコア"""
        # コース×距離別の枠順成績を事前に集計
        # 内枠有利、外枠有利などをスコア化
        return pd.Series(0, index=df.index)  # 簡易実装

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

import lightgbm as lgb
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import log_loss, roc_auc_score
import numpy as np
import pandas as pd
from typing import Tuple, Dict, List

class HorseRacingModel:
    """競馬予測モデル"""

    def __init__(self, model_type: str = 'lightgbm'):
        self.model_type = model_type
        self.models = {}
        self.feature_importance = None

    def train(
        self,
        X: pd.DataFrame,
        y: pd.Series,
        target_type: str = 'win',  # 'win', 'place', 'show'
        n_splits: int = 5
    ) -> Dict[str, float]:
        """時系列を考慮したクロスバリデーションで学習"""

        # 時系列分割
        tscv = TimeSeriesSplit(n_splits=n_splits)

        scores = {
            'train_logloss': [],
            'valid_logloss': [],
            'train_auc': [],
            'valid_auc': []
        }

        # ターゲット変数の作成
        if target_type == 'win':
            y_binary = (y == 1).astype(int)
        elif target_type == 'place':
            y_binary = (y <= 2).astype(int)
        else:  # show
            y_binary = (y <= 3).astype(int)

        fold_models = []

        for fold, (train_idx, valid_idx) in enumerate(tscv.split(X)):
            X_train, X_valid = X.iloc[train_idx], X.iloc[valid_idx]
            y_train, y_valid = y_binary.iloc[train_idx], y_binary.iloc[valid_idx]

            # LightGBMパラメータ
            params = {
                'objective': 'binary',
                'metric': 'binary_logloss',
                'boosting_type': 'gbdt',
                'num_leaves': 31,
                'learning_rate': 0.05,
                'feature_fraction': 0.8,
                'bagging_fraction': 0.8,
                'bagging_freq': 5,
                'verbose': -1,
                'n_jobs': -1,
                'seed': 42
            }

            # データセット作成
            train_data = lgb.Dataset(X_train, label=y_train)
            valid_data = lgb.Dataset(X_valid, label=y_valid, reference=train_data)

            # 学習
            model = lgb.train(
                params,
                train_data,
                num_boost_round=1000,
                valid_sets=[train_data, valid_data],
                valid_names=['train', 'valid'],
                callbacks=[
                    lgb.early_stopping(stopping_rounds=50),
                    lgb.log_evaluation(period=100)
                ]
            )

            fold_models.append(model)

            # 予測
            train_pred = model.predict(X_train)
            valid_pred = model.predict(X_valid)

            # スコア計算
            scores['train_logloss'].append(log_loss(y_train, train_pred))
            scores['valid_logloss'].append(log_loss(y_valid, valid_pred))
            scores['train_auc'].append(roc_auc_score(y_train, train_pred))
            scores['valid_auc'].append(roc_auc_score(y_valid, valid_pred))

            print(f"Fold {fold + 1}: "
                  f"Valid LogLoss = {scores['valid_logloss'][-1]:.4f}, "
                  f"Valid AUC = {scores['valid_auc'][-1]:.4f}")

        # モデル保存
        self.models[target_type] = fold_models

        # 特徴量重要度の集計
        self._calculate_feature_importance(fold_models, X.columns)

        # 平均スコアを返す
        return {
            'mean_train_logloss': np.mean(scores['train_logloss']),
            'mean_valid_logloss': np.mean(scores['valid_logloss']),
            'mean_train_auc': np.mean(scores['train_auc']),
            'mean_valid_auc': np.mean(scores['valid_auc'])
        }

    def predict(
        self,
        X: pd.DataFrame,
        target_type: str = 'win'
    ) -> np.ndarray:
        """予測"""
        if target_type not in self.models:
            raise ValueError(f"Model for {target_type} not trained")

        # 全foldモデルの予測を平均
        predictions = np.zeros(len(X))
        for model in self.models[target_type]:
            predictions += model.predict(X)
        predictions /= len(self.models[target_type])

        return predictions

    def predict_proba(self, X: pd.DataFrame) -> pd.DataFrame:
        """全ての確率を予測"""
        return pd.DataFrame({
            'win_prob': self.predict(X, 'win'),
            'place_prob': self.predict(X, 'place'),
            'show_prob': self.predict(X, 'show')
        }, index=X.index)

    def _calculate_feature_importance(
        self,
        models: List[lgb.Booster],
        feature_names: pd.Index
    ) -> None:
        """特徴量重要度を計算"""
        importance = np.zeros(len(feature_names))

        for model in models:
            importance += model.feature_importance(importance_type='gain')

        importance /= len(models)

        self.feature_importance = pd.DataFrame({
            'feature': feature_names,
            'importance': importance
        }).sort_values('importance', ascending=False)

    def get_feature_importance(self, top_n: int = 20) -> pd.DataFrame:
        """特徴量重要度を取得"""
        if self.feature_importance is None:
            raise ValueError("Model not trained yet")
        return self.feature_importance.head(top_n)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

from sklearn.linear_model import LogisticRegression
import xgboost as xgb
import catboost as cb

class EnsembleModel:
    """アンサンブル予測モデル"""

    def __init__(self):
        self.base_models = {
            'lightgbm': HorseRacingModel('lightgbm'),
            'xgboost': self._create_xgboost_model(),
            'catboost': self._create_catboost_model()
        }
        self.meta_model = LogisticRegression()

    def _create_xgboost_model(self):
        """XGBoostモデル"""
        return xgb.XGBClassifier(
            n_estimators=1000,
            max_depth=6,
            learning_rate=0.05,
            subsample=0.8,
            colsample_bytree=0.8,
            early_stopping_rounds=50,
            eval_metric='logloss',
            random_state=42
        )

    def _create_catboost_model(self):
        """CatBoostモデル"""
        return cb.CatBoostClassifier(
            iterations=1000,
            depth=6,
            learning_rate=0.05,
            loss_function='Logloss',
            early_stopping_rounds=50,
            verbose=100,
            random_state=42
        )

    def train_stacking(
        self,
        X: pd.DataFrame,
        y: pd.Series,
        n_splits: int = 5
    ) -> None:
        """スタッキングで学習"""
        tscv = TimeSeriesSplit(n_splits=n_splits)

        # ベースモデルの予測を格納
        base_predictions = np.zeros((len(X), len(self.base_models)))

        for i, (name, model) in enumerate(self.base_models.items()):
            print(f"Training {name}...")

            fold_preds = np.zeros(len(X))
            fold_count = np.zeros(len(X))

            for train_idx, valid_idx in tscv.split(X):
                X_train, X_valid = X.iloc[train_idx], X.iloc[valid_idx]
                y_train, y_valid = y.iloc[train_idx], y.iloc[valid_idx]

                if name == 'lightgbm':
                    model.train(X_train, y_train)
                    fold_preds[valid_idx] = model.predict(X_valid, 'win')
                else:
                    model.fit(X_train, y_train)
                    fold_preds[valid_idx] = model.predict_proba(X_valid)[:, 1]

                fold_count[valid_idx] += 1

            base_predictions[:, i] = fold_preds / np.maximum(fold_count, 1)

        # メタモデルの学習
        print("Training meta model...")
        self.meta_model.fit(base_predictions, y)

    def predict(self, X: pd.DataFrame) -> np.ndarray:
        """アンサンブル予測"""
        base_predictions = np.zeros((len(X), len(self.base_models)))

        for i, (name, model) in enumerate(self.base_models.items()):
            if name == 'lightgbm':
                base_predictions[:, i] = model.predict(X, 'win')
            else:
                base_predictions[:, i] = model.predict_proba(X)[:, 1]

        return self.meta_model.predict_proba(base_predictions)[:, 1]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

def evaluate_model(
    y_true: np.ndarray,
    y_pred: np.ndarray,
    odds: np.ndarray,
    race_ids: np.ndarray
) -> Dict[str, float]:
    """モデルの評価"""

    results = {}

    # 基本指標
    results['auc'] = roc_auc_score(y_true, y_pred)
    results['logloss'] = log_loss(y_true, y_pred)

    # レースごとの的中率
    race_results = []
    for race_id in np.unique(race_ids):
        mask = race_ids == race_id
        race_pred = y_pred[mask]
        race_true = y_true[mask]
        race_odds = odds[mask]

        # 予測1位の的中
        pred_winner_idx = np.argmax(race_pred)
        actual_winner = race_true[pred_winner_idx]

        race_results.append({
            'race_id': race_id,
            'hit': actual_winner,
            'odds': race_odds[pred_winner_idx] if actual_winner else 0
        })

    race_df = pd.DataFrame(race_results)

    # 的中率
    results['hit_rate'] = race_df['hit'].mean()

    # 回収率
    total_bet = len(race_df) * 100  # 1レース100円
    total_return = (race_df['hit'] * race_df['odds'] * 100).sum()
    results['return_rate'] = total_return / total_bet

    # 期待値がプラスの馬券数
    # （予測確率 × オッズ > 1 の馬券）
    expected_values = y_pred * odds
    results['positive_ev_rate'] = (expected_values > 1).mean()

    return results

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Optional
import redis
import json

app = FastAPI(title="競馬AI予測API")

# Redis接続
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# モデルロード
model = EnsembleModel.load('models/ensemble_v1.pkl')
feature_engineer = FeatureEngineer()

class PredictionRequest(BaseModel):
    race_id: str
    include_analysis: bool = False

class HorsePrediction(BaseModel):
    horse_id: str
    horse_name: str
    win_prob: float
    place_prob: float
    show_prob: float
    expected_value: float
    recommendation: str

class RacePrediction(BaseModel):
    race_id: str
    race_name: str
    predictions: List[HorsePrediction]
    analysis: Optional[str] = None
    updated_at: str

@app.get("/api/v1/predictions/{race_id}", response_model=RacePrediction)
async def get_race_prediction(race_id: str, include_analysis: bool = False):
    """レースの予測を取得"""

    # キャッシュ確認
    cache_key = f"prediction:{race_id}"
    cached = redis_client.get(cache_key)

    if cached:
        return json.loads(cached)

    # DBからレースデータ取得
    race_data = await get_race_data(race_id)
    if not race_data:
        raise HTTPException(status_code=404, detail="Race not found")

    # 特徴量生成
    features = feature_engineer.create_features(race_data)

    # 予測
    predictions = model.predict_proba(features)

    # 期待値計算
    predictions['expected_value'] = predictions['win_prob'] * race_data['odds']

    # レコメンデーション生成
    predictions['recommendation'] = predictions.apply(
        lambda x: generate_recommendation(x), axis=1
    )

    # レスポンス作成
    response = RacePrediction(
        race_id=race_id,
        race_name=race_data['race_name'].iloc[0],
        predictions=[
            HorsePrediction(
                horse_id=row['horse_id'],
                horse_name=row['horse_name'],
                win_prob=row['win_prob'],
                place_prob=row['place_prob'],
                show_prob=row['show_prob'],
                expected_value=row['expected_value'],
                recommendation=row['recommendation']
            )
            for _, row in predictions.iterrows()
        ],
        analysis=generate_analysis(predictions) if include_analysis else None,
        updated_at=datetime.now().isoformat()
    )

    # キャッシュ保存（5分間）
    redis_client.setex(cache_key, 300, json.dumps(response.dict()))

    return response

def generate_recommendation(row: pd.Series) -> str:
    """馬券レコメンデーションを生成"""
    if row['expected_value'] > 1.5 and row['win_prob'] > 0.15:
        return "強く推奨（単勝・馬単軸）"
    elif row['expected_value'] > 1.2 and row['show_prob'] > 0.5:
        return "推奨（複勝・ワイド）"
    elif row['show_prob'] > 0.4:
        return "検討（三連複ヒモ）"
    else:
        return "見送り"

def generate_analysis(predictions: pd.DataFrame) -> str:
    """レース分析を生成"""
    top_horse = predictions.iloc[predictions['win_prob'].argmax()]
    second_horse = predictions.iloc[predictions['win_prob'].argsort()[-2]]

    analysis = f"""
    【レース展望】
    本命: {top_horse['horse_name']}（勝率: {top_horse['win_prob']*100:.1f}%）
    対抗: {second_horse['horse_name']}（勝率: {second_horse['win_prob']*100:.1f}%）

    【期待値分析】
    期待値1.0以上: {(predictions['expected_value'] > 1).sum()}頭
    最高期待値: {predictions['expected_value'].max():.2f}（{predictions.iloc[predictions['expected_value'].argmax()]['horse_name']}）

    【推奨馬券】
    - 単勝: {top_horse['horse_name']}
    - 馬連: {top_horse['horse_name']} - {second_horse['horse_name']}
    """

    return analysis

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

import mlflow
from datetime import datetime, timedelta

class ModelMonitor:
    """モデルの精度をモニタリング"""

    def __init__(self, model_name: str, alert_threshold: float = 0.05):
        self.model_name = model_name
        self.alert_threshold = alert_threshold
        self.baseline_metrics = self._load_baseline_metrics()

    def log_prediction(
        self,
        race_id: str,
        predictions: pd.DataFrame,
        actual_results: pd.DataFrame
    ) -> None:
        """予測結果をログ"""
        # 的中判定
        predicted_winner = predictions.iloc[predictions['win_prob'].argmax()]['horse_id']
        actual_winner = actual_results[actual_results['finish_position'] == 1]['horse_id'].iloc[0]
        is_hit = predicted_winner == actual_winner

        # MLflowにログ
        with mlflow.start_run(run_name=f"prediction_{race_id}"):
            mlflow.log_param("race_id", race_id)
            mlflow.log_param("num_runners", len(predictions))
            mlflow.log_metric("hit", int(is_hit))

            if is_hit:
                odds = actual_results[actual_results['horse_id'] == predicted_winner]['odds'].iloc[0]
                mlflow.log_metric("return", odds * 100)
            else:
                mlflow.log_metric("return", 0)

    def calculate_daily_metrics(self, date: datetime) -> Dict[str, float]:
        """日次メトリクスを計算"""
        # MLflowから当日の予測結果を取得
        runs = mlflow.search_runs(
            filter_string=f"params.date = '{date.strftime('%Y-%m-%d')}'"
        )

        if len(runs) == 0:
            return {}

        metrics = {
            'hit_rate': runs['metrics.hit'].mean(),
            'return_rate': runs['metrics.return'].sum() / (len(runs) * 100),
            'num_predictions': len(runs)
        }

        return metrics

    def check_degradation(self, current_metrics: Dict[str, float]) -> List[str]:
        """精度劣化をチェック"""
        alerts = []

        for metric, value in current_metrics.items():
            if metric in self.baseline_metrics:
                baseline = self.baseline_metrics[metric]
                degradation = (baseline - value) / baseline

                if degradation > self.alert_threshold:
                    alerts.append(
                        f"{metric}: {degradation*100:.1f}%劣化 "
                        f"(baseline: {baseline:.3f}, current: {value:.3f})"
                    )

        return alerts

    def detect_feature_drift(
        self,
        current_features: pd.DataFrame,
        reference_features: pd.DataFrame
    ) -> Dict[str, float]:
        """特徴量ドリフトを検出"""
        from scipy.stats import ks_2samp

        drift_scores = {}

        for col in current_features.columns:
            if current_features[col].dtype in ['float64', 'int64']:
                statistic, p_value = ks_2samp(
                    reference_features[col].dropna(),
                    current_features[col].dropna()
                )
                drift_scores[col] = statistic

        return drift_scores

    def _load_baseline_metrics(self) -> Dict[str, float]:
        """ベースラインメトリクスを読み込む"""
        # 過去30日間の平均をベースラインとする
        return {
            'hit_rate': 0.25,
            'return_rate': 0.82
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'ml-team',
    'depends_on_past': False,
    'start_date': datetime(2026, 1, 1),
    'email_on_failure': True,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'horse_racing_model_retrain',
    default_args=default_args,
    description='週次モデル再学習パイプライン',
    schedule_interval='0 3 * * 1',  # 毎週月曜3時
    catchup=False
)

def extract_training_data(**context):
    """学習データを抽出"""
    # 過去2年分のデータを抽出
    end_date = datetime.now()
    start_date = end_date - timedelta(days=730)

    query = f"""
    SELECT * FROM race_results
    WHERE race_date BETWEEN '{start_date}' AND '{end_date}'
    """
    # データ抽出処理...

def create_features(**context):
    """特徴量を生成"""
    # 特徴量エンジニアリング処理...

def train_model(**context):
    """モデルを学習"""
    model = EnsembleModel()
    model.train_stacking(X, y)
    model.save(f"models/ensemble_{datetime.now().strftime('%Y%m%d')}.pkl")

def evaluate_model(**context):
    """モデルを評価"""
    # 直近1ヶ月のデータで評価
    metrics = evaluate_model(y_test, y_pred, odds_test, race_ids_test)

    if metrics['return_rate'] < 0.75:
        raise ValueError(f"回収率が基準以下: {metrics['return_rate']}")

def deploy_model(**context):
    """モデルをデプロイ"""
    # Blue-Greenデプロイ
    # 新モデルをステージング環境にデプロイ
    # A/Bテスト開始

# タスク定義
extract_task = PythonOperator(
    task_id='extract_training_data',
    python_callable=extract_training_data,
    dag=dag
)

feature_task = PythonOperator(
    task_id='create_features',
    python_callable=create_features,
    dag=dag
)

train_task = PythonOperator(
    task_id='train_model',
    python_callable=train_model,
    dag=dag
)

evaluate_task = PythonOperator(
    task_id='evaluate_model',
    python_callable=evaluate_model,
    dag=dag
)

deploy_task = PythonOperator(
    task_id='deploy_model',
    python_callable=deploy_model,
    dag=dag
)

# 依存関係
extract_task >> feature_task >> train_task >> evaluate_task >> deploy_task