機械学習予測モデル
LightGBMを使って競馬の着順予測モデルを構築します。
必要なライブラリ
pip install lightgbm scikit-learn pandas numpy
Step 1: データの準備
過去3ヶ月分のデータを取得し、学習用データを準備します。
import pandas as pd
import numpy as np
import requests
from datetime import datetime, timedelta
# ダッシュボードで取得した値をそれぞれ貼り付け
API_KEY = "API_KEY" # APIキー（アプリ共通）
AUTH_KEY = "YOUR_AUTH_KEY" # 認証キー (x-user-key)
def fetch(endpoint, start_date, end_date):
"""過去データを期間指定で取得して DataFrame にする"""
url = f"https://dev.api.bigtime.world/v1/{endpoint}?api_key={API_KEY}"
headers = {
"Content-Type": "application/json",
"x-user-key": AUTH_KEY
}
payload = {
"start_racedate": start_date,
"end_racedate": end_date
}
response = requests.post(url, json=payload, headers=headers)
data = response.json()
# 成功時は result.status == 200。エラー時はトップレベルに error/message が返る
if data.get("error") or data["result"]["status"] != 200:
raise Exception(data.get("message", "Unknown error"))
return pd.DataFrame(data["result"]["data"])
def fetch_training_data(start_date, end_date):
"""学習用データを組み立てる
/dky（出馬表・前日確定の指数）に、
/dra（レース条件）と /dse（着順＝正解ラベル）を結合する。
"""
ky = fetch("dky", start_date, end_date) # 特徴量（前日時点で分かる情報）
ra = fetch("dra", start_date, end_date) # 距離・トラック種別など
se = fetch("dse", start_date, end_date) # 着順・馬体重（結果）
df = ky.merge(
ra[['racekey', 'distance', 'track_type', 'racecourse_cd', 'class_cd']],
on='racekey', how='left'
)
df = df.merge(
se[['racekey', 'horse_number', 'order_of_finish',
'horse_weight', 'weight_cycling']],
on=['racekey', 'horse_number'], how='inner'
)
return df
# 過去3ヶ月のデータを取得
end_date = datetime.now().strftime("%Y%m%d")
start_date = (datetime.now() - timedelta(days=90)).strftime("%Y%m%d")
df = fetch_training_data(start_date, end_date)
print(f"総データ数: {len(df)}件")Step 2: 特徴量の選択と前処理
予測に使用する特徴量を選択し、欠損値処理を行います。
# 使用する特徴量（すべて前日時点で確定している値）
feature_columns = [
'adjust_idm', # IDM（総合能力指数）/dky
'training_index', # 調教指数 /dky
'stable_index', # 厩舎指数 /dky
'criteria_win_odds', # 基準単勝オッズ /dky
'criteria_win_popularity', # 基準単勝人気 /dky
'jockey_expect_winrate', # 騎手期待勝率 /dky
'foottype_cd', # 脚質 /dky
'weight_to_carry', # 斤量 /dky
'horse_age', # 馬齢 /dky
'rotation', # ローテーション（中n週）/dky
'distance', # 距離 /dra
'track_type', # トラック種別（1:芝, 2:ダート, 3:障害）/dra
]
# 馬体重（horse_weight）と増減（weight_cycling）は当日情報。
# 当日運用まで含めるなら /tpd から取得して同じ列名で足す。
# ターゲット（1着かどうか）
df['target'] = (df['order_of_finish'] == 1).astype(int)
# 特徴量の前処理
def preprocess_features(df, feature_columns):
X = df[feature_columns].copy()
# 数値変換
for col in X.columns:
X[col] = pd.to_numeric(X[col], errors='coerce')
# 欠損値を中央値で補完
X = X.fillna(X.median())
return X
X = preprocess_features(df, feature_columns)
y = df['target']
print(f"特徴量の形状: {X.shape}")
print(f"1着データ数: {y.sum()} ({y.mean()*100:.1f}%)")Step 3: モデルの学習
LightGBMで二値分類モデルを学習します。
import lightgbm as lgb
from sklearn.model_selection import train_test_split
# 訓練データとテストデータに分割
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42
)
# LightGBMのパラメータ
params = {
'objective': 'binary',
'metric': 'auc',
'boosting_type': 'gbdt',
'num_leaves': 31,
'learning_rate': 0.05,
'feature_fraction': 0.8,
'bagging_fraction': 0.8,
'bagging_freq': 5,
'verbose': -1,
'seed': 42
}
# データセットの作成
train_data = lgb.Dataset(X_train, label=y_train)
valid_data = lgb.Dataset(X_test, label=y_test, 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)
]
)
print(f"\n最適なイテレーション数: {model.best_iteration}")Step 4: モデルの評価
学習したモデルの性能を評価します。
from sklearn.metrics import roc_auc_score, classification_report
import matplotlib.pyplot as plt
# 予測
y_pred_proba = model.predict(X_test, num_iteration=model.best_iteration)
# AUCスコア
auc_score = roc_auc_score(y_test, y_pred_proba)
print(f"AUCスコア: {auc_score:.4f}")
# 閾値を設定して分類
threshold = 0.5
y_pred = (y_pred_proba >= threshold).astype(int)
print("\n分類レポート:")
print(classification_report(y_test, y_pred, target_names=['2着以下', '1着']))
# 特徴量重要度の可視化
importance = pd.DataFrame({
'feature': feature_columns,
'importance': model.feature_importance(importance_type='gain')
}).sort_values('importance', ascending=False)
plt.figure(figsize=(10, 6))
plt.barh(importance['feature'], importance['importance'])
plt.xlabel('重要度')
plt.title('特徴量重要度')
plt.gca().invert_yaxis()
plt.tight_layout()
plt.savefig('feature_importance.png', dpi=150)
plt.show()
print("\n特徴量重要度:")
print(importance)Step 5: 実際の予測
今週のレースデータを取得して予測を行います。
def fetch_today(endpoint, racedate):
"""今週データを日付指定で取得する"""
url = f"https://dev.api.bigtime.world/v1/{endpoint}?api_key={API_KEY}"
headers = {
"Content-Type": "application/json",
"x-user-key": AUTH_KEY
}
response = requests.post(url, json={"racedate": racedate}, headers=headers)
data = response.json()
if data.get("error") or data["result"]["status"] != 200:
raise Exception(data.get("message", "Unknown error"))
return pd.DataFrame(data["result"]["data"])
def predict_race(racedate):
"""指定日のレースを予測"""
# 学習時と同じ構成（出馬表 + レース条件）で組み立てる
df_today = fetch_today("tky", racedate)
ra = fetch_today("tra", racedate)
df_today = df_today.merge(
ra[['racekey', 'distance', 'track_type', 'racecourse_cd', 'class_cd']],
on='racekey', how='left'
)
# 特徴量の前処理
X_today = preprocess_features(df_today, feature_columns)
# 予測
predictions = model.predict(X_today, num_iteration=model.best_iteration)
# 結果をDataFrameに追加
df_today['win_probability'] = predictions
# レース毎に上位3頭を表示
for racekey in df_today['racekey'].unique():
race_df = df_today[df_today['racekey'] == racekey]
race_df = race_df.sort_values('win_probability', ascending=False)
print(f"\n=== レース: {racekey} ===")
for _, row in race_df.head(3).iterrows():
print(f" {row['horse_number']}番 {row['horse_name']}: "
f"{row['win_probability']*100:.1f}%")
return df_today
# 予測実行
result = predict_race("20250105")注意事項
- • このモデルは教育目的のサンプルです
- • 実際の馬券購入には、より高度な特徴量設計が必要です
- • 競馬予測には不確実性があり、損失のリスクがあります
モデルの保存と読み込み
# モデルを保存
model.save_model('horse_race_model.lgb')
# モデルを読み込み
loaded_model = lgb.Booster(model_file='horse_race_model.lgb')