#!/usr/bin/env python3 """ 結果確定済みの全レース（既定: 予想結果 p6_latest と同じ母集団）について、 勝率(New)×単勝オッズの期待値(New)で ◎〜△ を付け、all_race_ai_profile_compare（ev_new）へ保存する。 predictions が無いレースは stage1（p6_latest 相当）をその場で実行して材料を作る。 例: python3 scripts/import_all_race_ev_new_compare.py python3 scripts/import_all_race_ev_new_compare.py --only-missing python3 scripts/import_all_race_ev_new_compare.py --from-date 2024-01-01 --write-predictions """ from __future__ import annotations import argparse import json import os import re import sys from typing import Any sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from ai.prediction_profiles import get_profiles from ai.stage1_screening import run_stage1, write_predictions from ai.win_prob_ev_new import attach_win_prob_and_ev, marks_by_ev_new from collectors.db_util import connect PROFILE_KEY = "ev_new" PROFILE_LABEL = "予想結果（New・期待値）" STAGE1_PROFILE = "p6_latest" def _parse_horse_number(combination: str) -> int | None: c = (combination or "").strip() if re.fullmatch(r"\d{1,2}", c): return int(c) if c.startswith("hid:"): return None return None def _winner(cur, race_id: int) -> tuple[int | None, str | None]: cur.execute( """ SELECT re.horse_number, h.name FROM race_results rr JOIN race_entries re ON re.race_id = rr.race_id AND re.horse_id = rr.horse_id JOIN horses h ON h.id = re.horse_id WHERE rr.race_id = %s AND rr.finish_position = 1 LIMIT 1 """, (race_id,), ) row = cur.fetchone() if not row: return None, None hn = row.get("horse_number") return (int(hn) if hn is not None else None), str(row.get("name") or "") def _tansho_odds_map(cur, race_id: int) -> dict[int, float]: cur.execute( """ SELECT combination, min_odds, max_odds FROM odds_lines WHERE race_id = %s AND odds_type = 'tansho' """, (race_id,), ) out: dict[int, float] = {} for row in cur.fetchall(): comb = str(row.get("combination") or "").strip() if not comb.isdigit(): continue no = int(comb) mx = row.get("max_odds") mn = row.get("min_odds") v = mx if mx is not None else mn if v is None: continue fv = float(v) if fv > 0: out[no] = fv return out def _horses_from_stage1_payload(payload: dict[str, Any], odds_map: dict[int, float]) -> list[dict]: horses: list[dict] = [] for row in payload.get("horses") or []: hn = row.get("horse_number") if hn is not None: try: hn = int(hn) except (TypeError, ValueError): hn = None ms = row.get("score_accuracy") if ms is None: ms = row.get("score") model_score = float(ms) if ms is not None else None odds = None if hn is not None and hn in odds_map: odds = odds_map[hn] if odds is None and row.get("win_odds") is not None: wo = float(row["win_odds"]) if wo > 0: odds = wo horses.append( { "horse_number": hn, "horse_name": str(row.get("horse_name") or ""), "model_score": model_score, "odds": odds, } ) return horses def _select_targets( cur, *, from_date: str | None, to_date: str | None, limit: int | None, use_p6_universe: bool, only_missing: bool, ) -> list[dict]: params: list = [] if use_p6_universe: where = ["p.profile_key = %s", "p.status = 'ok'"] params.append(STAGE1_PROFILE) if from_date: where.append("p.race_date >= %s") params.append(from_date) if to_date: where.append("p.race_date <= %s") params.append(to_date) if only_missing: where.append( """NOT EXISTS ( SELECT 1 FROM all_race_ai_profile_compare e WHERE e.profile_key = %s AND e.race_id = p.race_id AND e.status = 'ok' )""" ) params.append(PROFILE_KEY) sql = f""" SELECT p.race_id, p.race_date, p.track_code, p.race_number, p.race_name FROM all_race_ai_profile_compare p WHERE {" AND ".join(where)} ORDER BY p.race_date DESC, p.track_code ASC, p.race_number ASC """ else: where = ["r.circuit = 'JRA'"] if from_date: where.append("r.race_date >= %s") params.append(from_date) if to_date: where.append("r.race_date <= %s") params.append(to_date) if only_missing: where.append( """NOT EXISTS ( SELECT 1 FROM all_race_ai_profile_compare e WHERE e.profile_key = %s AND e.race_id = r.id AND e.status = 'ok' )""" ) params.append(PROFILE_KEY) sql = f""" SELECT r.id AS race_id, r.race_date, t.code AS track_code, r.race_number, r.race_name FROM races r JOIN tracks t ON t.id = r.track_id WHERE {" AND ".join(where)} AND EXISTS ( SELECT 1 FROM race_results rr WHERE rr.race_id = r.id AND rr.finish_position = 1 ) ORDER BY r.race_date DESC, t.code ASC, r.race_number ASC """ if limit and limit > 0: sql += f" LIMIT {int(limit)}" cur.execute(sql, params) return list(cur.fetchall()) def _load_horses_from_predictions(cur, race_id: int) -> list[dict]: cur.execute( """ SELECT combination, win_probability, expected_odds, note FROM predictions WHERE race_id = %s AND prediction_type = 'stage1_screening_accuracy' AND bet_type = 'horse' ORDER BY rank_in_pred ASC """, (race_id,), ) pred_rows = list(cur.fetchall()) if not pred_rows: return [] odds_map = _tansho_odds_map(cur, race_id) horses: list[dict] = [] for row in pred_rows: comb = str(row.get("combination") or "") hn = _parse_horse_number(comb) note = {} raw_note = row.get("note") if raw_note: try: note = json.loads(raw_note) if isinstance(raw_note, str) else dict(raw_note) except json.JSONDecodeError: note = {} name = str(note.get("horse_name") or "") model_score = row.get("win_probability") ms = float(model_score) if model_score is not None else None odds = None if hn is not None and hn in odds_map: odds = odds_map[hn] if odds is None and row.get("expected_odds") is not None: eo = float(row["expected_odds"]) if eo > 0: odds = eo if odds is None and note.get("win_odds") is not None: wo = float(note["win_odds"]) if wo > 0: odds = wo horses.append( { "horse_number": hn, "horse_name": name, "model_score": ms, "odds": odds, } ) return horses def _load_horses_for_race( cur, race_id: int, *, profile, model_path: str | None, model_meta_path: str | None, write_predictions_flag: bool, ) -> list[dict]: horses = _load_horses_from_predictions(cur, race_id) if horses: return horses payload = run_stage1( race_id=race_id, lookback_runs=profile.lookback_runs, pass_rate=profile.pass_rate, enable_stage1_5=profile.enable_stage1_5, enable_stage2=profile.enable_stage2, enable_stage3=profile.enable_stage3, stage3_min_prob=profile.stage3_min_prob, stage3_odds_cap=profile.stage3_odds_cap, stage3_aggressive_min_prob=profile.stage3_aggressive_min_prob, stage3_aggressive_odds_cap=profile.stage3_aggressive_odds_cap, model_path=model_path, model_meta_path=model_meta_path, ) if write_predictions_flag: write_predictions(payload) odds_map = _tansho_odds_map(cur, race_id) return _horses_from_stage1_payload(payload, odds_map) def _upsert(conn, rec: dict) -> None: cols = [ "profile_key", "profile_label", "race_id", "race_date", "track_code", "race_number", "race_name", "ai_honmei_umaban", "ai_honmei_name", "ai_honmei_score", "ai_taikou_umaban", "ai_taikou_name", "ai_taikou_score", "ai_tanana_umaban", "ai_tanana_name", "ai_tanana_score", "ai_renka_umaban", "ai_renka_name", "ai_renka_score", "winner_umaban", "winner_name", "match_honmei_winner", "match_taikou_winner", "match_honmei_or_taikou_winner", "status", "error_message", "ai_snapshot_json", ] placeholders = ", ".join(["%s"] * len(cols)) updates = ", ".join([f"{c}=VALUES({c})" for c in cols if c not in ("profile_key", "race_id")]) sql = f""" INSERT INTO all_race_ai_profile_compare ({", ".join(cols)}) VALUES ({placeholders}) ON DUPLICATE KEY UPDATE {updates} """ with conn.cursor() as cur: cur.execute(sql, [rec.get(c) for c in cols]) conn.commit() def _process_race( conn, t: dict, *, profile, model_path: str | None, model_meta_path: str | None, write_predictions_flag: bool, dry_run: bool, ) -> tuple[str, str | None]: """Returns ('ok'|'err'|'skip', message).""" race_id = int(t["race_id"]) rec = { "profile_key": PROFILE_KEY, "profile_label": PROFILE_LABEL, "race_id": race_id, "race_date": t["race_date"], "track_code": t["track_code"], "race_number": int(t["race_number"]), "race_name": t.get("race_name"), "ai_honmei_umaban": None, "ai_honmei_name": None, "ai_honmei_score": None, "ai_taikou_umaban": None, "ai_taikou_name": None, "ai_taikou_score": None, "ai_tanana_umaban": None, "ai_tanana_name": None, "ai_tanana_score": None, "ai_renka_umaban": None, "ai_renka_name": None, "ai_renka_score": None, "winner_umaban": None, "winner_name": None, "match_honmei_winner": None, "match_taikou_winner": None, "match_honmei_or_taikou_winner": None, "status": "ok", "error_message": None, "ai_snapshot_json": None, } with conn.cursor() as cur: horses = _load_horses_for_race( cur, race_id, profile=profile, model_path=model_path, model_meta_path=model_meta_path, write_predictions_flag=write_predictions_flag, ) if not horses: return "skip", "no horses" attach_win_prob_and_ev(horses) with_ev = [h for h in horses if h.get("ev_new") is not None] if not with_ev: rec["status"] = "error" rec["error_message"] = "期待値(New)を算出できる馬がいません（オッズ不足）" if not dry_run: _upsert(conn, rec) return "err", "no odds" marks = marks_by_ev_new(with_ev) if len(marks) >= 1: rec["ai_honmei_umaban"] = marks[0].get("horse_number") rec["ai_honmei_name"] = marks[0].get("horse_name") rec["ai_honmei_score"] = marks[0].get("ev_new") if len(marks) >= 2: rec["ai_taikou_umaban"] = marks[1].get("horse_number") rec["ai_taikou_name"] = marks[1].get("horse_name") rec["ai_taikou_score"] = marks[1].get("ev_new") if len(marks) >= 3: rec["ai_tanana_umaban"] = marks[2].get("horse_number") rec["ai_tanana_name"] = marks[2].get("horse_name") rec["ai_tanana_score"] = marks[2].get("ev_new") if len(marks) >= 4: rec["ai_renka_umaban"] = marks[3].get("horse_number") rec["ai_renka_name"] = marks[3].get("horse_name") rec["ai_renka_score"] = marks[3].get("ev_new") with conn.cursor() as cur: w_umaban, w_name = _winner(cur, race_id) rec["winner_umaban"] = w_umaban rec["winner_name"] = w_name h = rec["ai_honmei_umaban"] o = rec["ai_taikou_umaban"] w = rec["winner_umaban"] if w is not None: rec["match_honmei_winner"] = 1 if (h is not None and h == w) else 0 rec["match_taikou_winner"] = 1 if (o is not None and o == w) else 0 rec["match_honmei_or_taikou_winner"] = ( 1 if ((h is not None and h == w) or (o is not None and o == w)) else 0 ) hon_ev = marks[0].get("ev_new") if marks else None rec["ai_snapshot_json"] = json.dumps( { "method": "ev_new", "stage1_profile": STAGE1_PROFILE, "marks": marks, "horses_ev_ranked": sorted( with_ev, key=lambda x: float(x.get("ev_new") or 0), reverse=True, )[:12], "honmei_confidence": { "score_0_100": min(100, int(round(float(hon_ev or 0) * 50))), "label": "期待値(New)◎", "ev_new_honmei": hon_ev, }, }, ensure_ascii=False, default=str, ) if not dry_run: _upsert(conn, rec) return "ok", f"hon_ev={hon_ev}" def main() -> None: ap = argparse.ArgumentParser(description="期待値(New)ベースの ◎〜△ を全結果レースへ取込") ap.add_argument("--from-date", default=None, help="YYYY-MM-DD") ap.add_argument("--to-date", default=None, help="YYYY-MM-DD") ap.add_argument("--limit", type=int, default=0, help="0=全件") ap.add_argument( "--all-finished-races", action="store_true", help="p6_latest 母集団ではなく、1着確定の全JRAレースを対象", ) ap.add_argument( "--only-missing", action="store_true", help="ev_new が未登録または status!=ok のレースだけ処理", ) ap.add_argument( "--write-predictions", action="store_true", help="stage1 実行時に predictions テーブルへも保存（重いがレース詳細と共有）", ) ap.add_argument("--model-path", default=None) ap.add_argument("--model-meta-path", default=None) ap.add_argument("--dry-run", action="store_true") args = ap.parse_args() profile = get_profiles(STAGE1_PROFILE)[0] conn = connect() ok = err = skip = 0 try: with conn.cursor() as cur: targets = _select_targets( cur, from_date=args.from_date, to_date=args.to_date, limit=(args.limit if args.limit > 0 else None), use_p6_universe=not args.all_finished_races, only_missing=args.only_missing, ) total = len(targets) print( f"targets={total} universe={'p6_latest' if not args.all_finished_races else 'all_finished'} " f"only_missing={args.only_missing} write_predictions={args.write_predictions} dry_run={args.dry_run}", flush=True, ) for i, t in enumerate(targets, start=1): race_id = int(t["race_id"]) try: status, msg = _process_race( conn, t, profile=profile, model_path=args.model_path, model_meta_path=args.model_meta_path, write_predictions_flag=args.write_predictions, dry_run=args.dry_run, ) if status == "ok": ok += 1 print(f"[{i}/{total}] OK race_id={race_id} {msg}", flush=True) elif status == "skip": skip += 1 print(f"[{i}/{total}] SKIP race_id={race_id} {msg}", flush=True) else: err += 1 print(f"[{i}/{total}] ERR race_id={race_id} {msg}", flush=True) except Exception as e: err += 1 if not args.dry_run: rec = { "profile_key": PROFILE_KEY, "profile_label": PROFILE_LABEL, "race_id": race_id, "race_date": t["race_date"], "track_code": t["track_code"], "race_number": int(t["race_number"]), "race_name": t.get("race_name"), "status": "error", "error_message": str(e)[:500], } _upsert(conn, rec) print(f"[{i}/{total}] ERR race_id={race_id}: {e}", flush=True) finally: conn.close() print(f"done ok={ok} err={err} skip={skip} total={ok + err + skip}", flush=True) if err: raise SystemExit(1) if __name__ == "__main__": main()