生成AI予測モデル

pip install openai anthropic pandas requests

# 環境変数の設定
# OPENAI_API_KEY=sk-...
# ANTHROPIC_API_KEY=sk-ant-...
# JRDB_API_KEY=your_jrdb_api_key
# JRDB_AUTH_KEY=your_jrdb_auth_key

def fetch(endpoint, racedate):
    """今週データを取得して DataFrame にする"""
    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()

    # 成功時は 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_race_data(racedate, racekey):
    """出馬表・レース条件・直前情報をまとめて1レース分取り出す"""
    ky = fetch("tky", racedate)   # 指数・騎手・血統など
    ra = fetch("tra", racedate)   # レース名・距離・トラック種別
    pd_ = fetch("tpd", racedate)  # 馬体重・当日オッズ

    df = ky.merge(
        ra[['racekey', 'race_name', 'distance', 'track_type',
            'racecourse_cd', 'turf_condition_cd']],
        on='racekey', how='left'
    ).merge(
        pd_[['racekey', 'horse_number', 'horse_weight',
             'weight_cycling', 'win_odds']],
        on=['racekey', 'horse_number'], how='left'
    )

    return df[df['racekey'] == racekey]

# コード → 表示名（コード表ページを参照）
SEX = {'1': '牡', '2': '牝', '3': 'セ'}
FOOTTYPE = {'1': '逃げ', '2': '先行', '3': '差し', '4': '追込',
            '5': '好位差し', '6': '自在'}
TRACK = {'1': '芝', '2': 'ダート', '3': '障害'}
CONDITION = {'1': '良', '2': '稍重', '3': '重', '4': '不良'}
RACECOURSE = {'01': '札幌', '02': '函館', '03': '福島', '04': '新潟', '05': '東京',
              '06': '中山', '07': '中京', '08': '京都', '09': '阪神', '10': '小倉'}

def format_horse_info(row):
    """馬情報をテキスト形式に変換"""
    info = f"""
【{row['horse_number']}番 {row['horse_name']}】
・性齢: {SEX.get(str(row['sex_cd']), '?')}{row['horse_age']}歳
・騎手: {row['jockey_name']}
・調教師: {row['trainer_name']}
・馬体重: {row['horse_weight']}kg（{row['weight_cycling']}kg）
・オッズ: {row['win_odds']}倍（基準{row['criteria_win_popularity']}番人気）
・IDM: {row['adjust_idm']}
・前走: {row['lastrace_racekey1']}
・脚質: {FOOTTYPE.get(str(row['foottype_cd']), '?')}
"""
    return info.strip()

def format_race_info(df):
    """レース情報をテキスト形式に変換"""
    # レース基本情報（1行目から取得）
    race = df.iloc[0]

    race_info = f"""
【レース情報】
・レース名: {race.get('race_name', 'Unknown')}
・競馬場: {RACECOURSE.get(str(race.get('racecourse_cd')), 'Unknown')}
・距離: {race.get('distance', 'Unknown')}m {TRACK.get(str(race.get('track_type')), '')}
・馬場状態: {CONDITION.get(str(race.get('turf_condition_cd')), 'Unknown')}
・頭数: {len(df)}頭
"""

    horses_info = "\n【出走馬一覧】\n"
    for _, row in df.iterrows():
        horses_info += format_horse_info(row) + "\n"

    return race_info + horses_info

def predict_with_llm(race_text, model="gpt-4o"):
    """LLMでレース予測を行う"""

    system_prompt = """あなたは競馬予想の専門家です。
与えられたレース情報を分析し、上位3頭を予測してください。

予測の際は以下の観点を考慮してください：
- 各馬の能力指数（IDM）
- 距離適性と脚質
- 馬体重の増減
- 騎手・調教師の成績
- 前走からの上積み

回答は以下のJSON形式で返してください：
{
  "predictions": [
    {"rank": 1, "horse_number": 1, "horse_name": "馬名", "reason": "選出理由"},
    {"rank": 2, "horse_number": 2, "horse_name": "馬名", "reason": "選出理由"},
    {"rank": 3, "horse_number": 3, "horse_name": "馬名", "reason": "選出理由"}
  ],
  "race_analysis": "レース展開の予想",
  "risk_factors": ["リスク要因1", "リスク要因2"]
}"""

    response = openai_client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": race_text}
        ],
        response_format={"type": "json_object"},
        temperature=0.3  # 安定した出力のため低めに設定
    )

    result = json.loads(response.choices[0].message.content)
    return result

# 実行例
df = fetch_race_data("20250105", "2025010506010101")
race_text = format_race_info(df)
prediction = predict_with_llm(race_text)

print("=== AI予測結果 ===")
for p in prediction["predictions"]:
    print(f"{p['rank']}位: {p['horse_number']}番 {p['horse_name']}")
    print(f"   理由: {p['reason']}")
print(f"\n展開予想: {prediction['race_analysis']}")

def predict_with_cot(race_text, model="gpt-4o"):
    """Chain of Thoughtを使った詳細な予測"""

    system_prompt = """あなたは競馬予想の専門家です。
以下のステップで分析を行い、最終的な予測を導き出してください。

## 分析ステップ

### Step 1: 能力評価
各馬のIDM、過去成績から基礎能力を評価してください。

### Step 2: 適性評価
今回のコース（距離、馬場状態）への適性を評価してください。

### Step 3: 状態評価
馬体重の増減、休み明け、ローテーションから現在の状態を評価してください。

### Step 4: 展開予想
逃げ・先行馬の数、ペースの予想を行ってください。

### Step 5: 総合判断
上記を踏まえ、各馬の勝利確率を推定し、上位3頭を選出してください。

## 出力形式
各ステップの分析結果を示した後、最終的な予測をJSON形式で出力してください。

最終出力JSON:
{
  "step1_ability": {"top3": ["馬名1", "馬名2", "馬名3"], "analysis": "分析"},
  "step2_aptitude": {"favored": ["馬名"], "concerns": ["馬名"], "analysis": "分析"},
  "step3_condition": {"good": ["馬名"], "bad": ["馬名"], "analysis": "分析"},
  "step4_pace": {"prediction": "ハイペース/ミドルペース/スローペース", "key_horses": ["馬名"], "analysis": "分析"},
  "final_predictions": [
    {"rank": 1, "horse_number": 1, "horse_name": "馬名", "win_probability": 0.25, "reason": "理由"},
    {"rank": 2, "horse_number": 2, "horse_name": "馬名", "win_probability": 0.18, "reason": "理由"},
    {"rank": 3, "horse_number": 3, "horse_name": "馬名", "win_probability": 0.12, "reason": "理由"}
  ],
  "confidence": "high/medium/low",
  "confidence_reason": "自信度の理由"
}"""

    response = openai_client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": race_text}
        ],
        response_format={"type": "json_object"},
        temperature=0.2
    )

    result = json.loads(response.choices[0].message.content)
    return result

import lightgbm as lgb

def hybrid_prediction(df, model_path='horse_race_model.lgb'):
    """MLモデルとLLMを組み合わせた予測"""

    # 1. 機械学習モデルで予測
    model = lgb.Booster(model_file=model_path)
    feature_columns = [...]  # 学習時と同じ特徴量
    X = preprocess_features(df, feature_columns)
    ml_predictions = model.predict(X, num_iteration=model.best_iteration)

    df['ml_probability'] = ml_predictions
    df['ml_rank'] = df['ml_probability'].rank(ascending=False)

    # 2. ML予測結果を含むテキストを作成
    race_text = format_race_info(df)
    ml_summary = "\n【機械学習モデルの予測】\n"
    ml_top5 = df.nsmallest(5, 'ml_rank')
    for _, row in ml_top5.iterrows():
        ml_summary += f"・{row['horse_number']}番 {row['horse_name']}: "
        ml_summary += f"勝率 {row['ml_probability']*100:.1f}%\n"

    combined_text = race_text + ml_summary

    # 3. LLMで最終判断
    system_prompt = """あなたは競馬予想の専門家です。
レース情報と機械学習モデルの予測結果を参考に、最終的な予測を行ってください。

機械学習モデルの予測は過去データに基づく客観的な評価ですが、
以下の点は考慮されていない可能性があります：
- 当日の馬場状態の変化
- 騎手の戦略的な騎乗
- 展開の有利不利
- 調子の上昇/下降トレンド

これらを考慮して、機械学習の予測を補正してください。

出力形式:
{
  "ml_agreement": ["MLの予測に同意する馬番"],
  "ml_disagreement": [
    {"horse_number": 1, "ml_rank": 3, "my_rank": 1, "reason": "理由"}
  ],
  "final_predictions": [
    {"rank": 1, "horse_number": 1, "horse_name": "馬名", "confidence": 0.8},
    {"rank": 2, "horse_number": 2, "horse_name": "馬名", "confidence": 0.6},
    {"rank": 3, "horse_number": 3, "horse_name": "馬名", "confidence": 0.5}
  ],
  "betting_advice": "馬券のアドバイス"
}"""

    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": combined_text}
        ],
        response_format={"type": "json_object"},
        temperature=0.3
    )

    return json.loads(response.choices[0].message.content)

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

async def predict_race_async(race_text, racekey):
    """非同期でレース予測を行う"""
    try:
        response = await async_client.chat.completions.create(
            model="gpt-4o-mini",  # バッチ処理にはコスト効率の良いモデルを使用
            messages=[
                {"role": "system", "content": PREDICTION_PROMPT},
                {"role": "user", "content": race_text}
            ],
            response_format={"type": "json_object"},
            temperature=0.3
        )
        result = json.loads(response.choices[0].message.content)
        result['racekey'] = racekey
        return result
    except Exception as e:
        return {"racekey": racekey, "error": str(e)}

async def predict_all_races(racedate):
    """1日分の全レースを予測"""

    # 全レースデータを取得
    url = f"https://dev.api.bigtime.world/v1/tky?api_key={API_KEY}"
    headers = {"Content-Type": "application/json", "x-user-key": AUTH_KEY}
    payload = {"racedate": racedate}

    response = requests.post(url, json=payload, headers=headers)
    df = pd.DataFrame(response.json()["result"]["data"])

    # レース毎にグループ化
    races = df.groupby('racekey')

    # 非同期タスクを作成
    tasks = []
    for racekey, race_df in races:
        race_text = format_race_info(race_df)
        tasks.append(predict_race_async(race_text, racekey))

    # 並列実行（レート制限に注意）
    results = await asyncio.gather(*tasks)

    return results

# 実行
# results = asyncio.run(predict_all_races("20250105"))

def evaluate_predictions(predictions, actual_results):
    """予測精度を評価"""

    metrics = {
        'total_races': 0,
        'top1_hit': 0,      # 1着的中
        'top3_hit': 0,      # 3着以内的中
        'trifecta_hit': 0,  # 3連単的中
    }

    for pred, actual in zip(predictions, actual_results):
        metrics['total_races'] += 1

        pred_top3 = [p['horse_number'] for p in pred['final_predictions'][:3]]
        actual_top3 = actual['top3_horses']

        # 1着的中
        if pred_top3[0] == actual_top3[0]:
            metrics['top1_hit'] += 1

        # 3着以内的中（予測1位が3着以内に入っているか）
        if pred_top3[0] in actual_top3:
            metrics['top3_hit'] += 1

        # 3連単的中
        if pred_top3 == actual_top3:
            metrics['trifecta_hit'] += 1

    # 的中率を計算
    n = metrics['total_races']
    print(f"=== 予測精度評価 ===")
    print(f"対象レース数: {n}")
    print(f"1着的中率: {metrics['top1_hit']/n*100:.1f}%")
    print(f"3着内率: {metrics['top3_hit']/n*100:.1f}%")
    print(f"3連単的中率: {metrics['trifecta_hit']/n*100:.1f}%")

    return metrics

def improve_prompt_with_feedback(original_prompt, failed_cases):
    """失敗ケースを分析してプロンプトを改善"""

    analysis_prompt = f"""以下の競馬予測で失敗したケースを分析してください。

【元のプロンプト】
{original_prompt}

【失敗ケース】
{json.dumps(failed_cases, ensure_ascii=False, indent=2)}

以下の観点で分析し、プロンプトの改善案を提案してください：
1. 共通する失敗パターンはあるか
2. 見落としている評価観点はあるか
3. 具体的なプロンプト修正案

出力形式:
{{
  "failure_patterns": ["パターン1", "パターン2"],
  "missing_factors": ["要因1", "要因2"],
  "improved_prompt": "改善されたプロンプト全文"
}}"""

    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": analysis_prompt}],
        response_format={"type": "json_object"}
    )

    return json.loads(response.choices[0].message.content)