#!/usr/bin/env python3 """第1段階（基礎能力足切り）スクリーニング。""" from __future__ import annotations import argparse import json import os import statistics import sys from datetime import date from typing import Any sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from ai.stage1_5_adjustments import compute_stage1_5_adjustment # noqa: E402 from ai.stage2_factors import build_race_context, compute_stage2_adjustment, infer_running_style # noqa: E402 from ai.stage3_value import apply_ev_forecast, apply_stage3_value, attach_win_odds # noqa: E402 from ai.win_rate import assign_rank_accuracy, assign_win_rates # noqa: E402 from ai.feature_engineering import compute_stage1_features, get_feature_columns # noqa: E402 from ai.empirical_hit_rates import fetch_ev_forecast_empirical_calibration_for_track # noqa: E402 from ai.race_confidence import compute_honmei_confidence # noqa: E402 from collectors.db_util import connect # noqa: E402 # (model_path, meta_path) ごとに Booster をキャッシュ（全レース一括バッチの I/O 削減） _STAGE1_MODEL_CACHE: dict[tuple[str, str], tuple[list[tuple[Any, float]], list[str]]] = {} def clear_stage1_model_cache() -> None: """メモリ解放やモデル差し替え検証時に呼ぶ（通常の単発実行では不要）。""" _STAGE1_MODEL_CACHE.clear() def _stage1_model_cache_key(model_path: str, model_meta_path: str | None) -> tuple[str, str]: return (os.path.abspath(model_path), os.path.abspath(model_meta_path) if model_meta_path else "") def _load_model_cached(model_path: str, model_meta_path: str | None = None) -> tuple[list[tuple[Any, float]], list[str]]: key = _stage1_model_cache_key(model_path, model_meta_path) if key not in _STAGE1_MODEL_CACHE: _STAGE1_MODEL_CACHE[key] = _load_model(model_path, model_meta_path) return _STAGE1_MODEL_CACHE[key] def _apply_within_race_score_spread(rows: list[dict], *, z_scale: float = 2.8) -> None: """同一レース内でスコアの上下差をやや強める（相対評価）。第2段階の直後・第3段階の前に適用。""" if len(rows) < 3: return scores = [float(r["score"]) for r in rows] mu = statistics.fmean(scores) try: sigma = statistics.pstdev(scores) except statistics.StatisticsError: return if sigma < 0.35: return for r in rows: z = (float(r["score"]) - mu) / sigma adj = max(-4.5, min(4.5, z * z_scale)) r["score"] = round(min(100.0, max(0.0, float(r["score"]) + adj)), 2) def _stage1_score(metrics: dict) -> float: runs = int(metrics["recent_runs"]) if runs == 0: # 戦績ゼロは判断情報不足。最低限の評価から開始。 return 25.0 win_rate = float(metrics["recent_win_rate"]) top3_rate = float(metrics["recent_top3_rate"]) avg_finish = float(metrics["recent_avg_finish"]) graded_top5_rate = float(metrics["recent_graded_top5_rate"]) exp_score = min(runs, 10) / 10.0 consistency = max(0.0, 1.0 - ((avg_finish - 1.0) / 10.0)) score = ( top3_rate * 45.0 + win_rate * 25.0 + consistency * 20.0 + exp_score * 10.0 + graded_top5_rate * 15.0 ) # コンテキスト適性: 近走勢い・芝ダ/距離/競馬場での過去複勝率・上がり・斤量 mom = float(metrics.get("form_momentum") or 0.0) ss = float(metrics.get("same_surface_top3_rate") or top3_rate) db = float(metrics.get("distance_band_top3_rate") or top3_rate) st = float(metrics.get("same_track_top3_rate") or top3_rate) neutral = max(0.05, min(0.85, top3_rate)) ctx = ( mom * 12.0 + (ss - neutral) * 18.0 + (db - neutral) * 15.0 + (st - neutral) * 16.0 ) lf = float(metrics.get("recent_avg_last_3f") or 0.0) if lf > 0: # 上がりが速い（秒が小さい）ほど加点（目安 30〜42 秒） ctx += max(-5.0, min(5.0, (40.0 - lf) * 0.65)) cw = float(metrics.get("carry_weight_kg") or 0.0) if cw > 0: # 斤量が軽いほど若干加点（54〜58kg 想定） ctx += max(-4.0, min(4.0, (57.0 - cw) * 0.45)) ctx = max(-22.0, min(22.0, ctx)) score = score + ctx return round(min(100.0, score), 2) def _load_model(model_path: str, meta_path: str | None = None) -> tuple[list[tuple[Any, float]], list[str]]: """meta.json に ensemble.models がある場合は複数 Booster を読み、(booster, weight) のリストを返す。""" try: import lightgbm as lgb except ImportError as e: raise RuntimeError("lightgbm が未インストールです。`python3.11 -m pip install lightgbm` を実行してください。") from e feature_columns = get_feature_columns() models_weights: list[tuple[Any, float]] = [] if meta_path and os.path.isfile(meta_path): with open(meta_path, "r", encoding="utf-8") as fr: meta = json.load(fr) feature_columns = meta.get("feature_columns") or feature_columns entries = (meta.get("ensemble") or {}).get("models") or [] if entries: meta_dir = os.path.dirname(os.path.abspath(meta_path)) for ent in entries: rel = (ent.get("path") or "").strip() w = float(ent.get("weight") or 1.0) fp = rel if os.path.isabs(rel) else os.path.join(meta_dir, rel) if not os.path.isfile(fp): raise FileNotFoundError(f"ensemble model file not found: {fp}") models_weights.append((lgb.Booster(model_file=fp), w)) s = sum(w for _, w in models_weights) if s > 0 and abs(s - 1.0) > 1e-5: models_weights = [(b, w / s) for b, w in models_weights] return models_weights, feature_columns if not os.path.isfile(model_path): raise FileNotFoundError(f"model file not found: {model_path}") booster = lgb.Booster(model_file=model_path) return [(booster, 1.0)], feature_columns def _model_win_prob( models_weights: list[tuple[Any, float]], feature_columns: list[str], metrics: dict, distance_m: float, surface_turf: float, ) -> float: row = metrics.copy() row["distance_m"] = distance_m row["surface_turf"] = surface_turf values = [[float(row.get(col, 0.0) or 0.0) for col in feature_columns]] prob = 0.0 for booster, w in models_weights: prob += w * float(booster.predict(values)[0]) return max(0.0, min(1.0, float(prob))) def _model_score( models_weights: list[tuple[Any, float]], feature_columns: list[str], metrics: dict, distance_m: float, surface_turf: float, ) -> float: return round(_model_win_prob(models_weights, feature_columns, metrics, distance_m, surface_turf) * 100.0, 2) def _resolve_race_id(cur, race_id: int | None, race_date: str | None, track_code: str | None, race_number: int | None): if race_id: return int(race_id) if not (race_date and track_code and race_number): raise ValueError("--race-id または --race-date/--track-code/--race-number の指定が必要です") sql = """ SELECT r.id FROM races r INNER JOIN tracks t ON t.id = r.track_id WHERE r.race_date = %s AND t.code = %s AND r.race_number = %s AND r.circuit = 'JRA' LIMIT 1 """ cur.execute(sql, (race_date, track_code, race_number)) row = cur.fetchone() if not row: raise ValueError("指定条件のレースが見つかりません") return int(row["id"]) def run_stage1( race_id: int, lookback_runs: int = 10, pass_rate: float = 0.5, enable_stage1_5: bool = True, enable_stage2: bool = True, enable_stage3: bool = True, stage3_min_prob: float = 0.02, stage3_odds_cap: float = 80.0, stage3_aggressive_min_prob: float = 0.0, stage3_aggressive_odds_cap: float = 300.0, model_path: str | None = None, model_meta_path: str | None = None, *, forecast_pipeline: str = "legacy", rank_mode: str = "accuracy", ) -> dict: scoring_mode = "rule" models_weights: list[tuple[Any, float]] = [] model_features: list[str] = [] if model_path: models_weights, model_features = _load_model_cached(model_path, model_meta_path) scoring_mode = "lightgbm" conn = connect() try: with conn.cursor() as cur: cur.execute( """ SELECT r.id, r.race_date, r.race_name, r.race_number, r.distance_m, r.surface, t.code AS track_code, t.name AS track_name FROM races r INNER JOIN tracks t ON t.id = r.track_id WHERE r.id = %s LIMIT 1 """, (race_id,), ) race = cur.fetchone() if not race: raise ValueError("race_id が存在しません") cur.execute( """ SELECT re.race_id, re.horse_id, re.horse_number, re.bracket_number, re.carry_weight, re.odds_win, re.jockey_id, h.name AS horse_name FROM race_entries re INNER JOIN horses h ON h.id = re.horse_id WHERE re.race_id = %s AND re.is_scratched = 0 ORDER BY re.horse_number ASC """, (race_id,), ) entries = cur.fetchall() if not entries: raise ValueError("出走馬データがありません") history_sql = """ SELECT rr.finish_position, rr.race_time_seconds, rr.last_3f_time, rr.passing_order, r.grade, r.race_date, r.distance_m, r.surface, t.code AS track_code FROM race_results rr INNER JOIN races r ON r.id = rr.race_id INNER JOIN tracks t ON t.id = r.track_id WHERE rr.horse_id = %s AND r.race_date < %s AND r.circuit = 'JRA' AND rr.finish_position > 0 ORDER BY r.race_date DESC, rr.race_id DESC LIMIT %s """ jockey_ids = [int(e["jockey_id"]) for e in entries if e.get("jockey_id") is not None] jockey_stats: dict[int, dict] = {} if jockey_ids: placeholders = ",".join(["%s"] * len(jockey_ids)) jockey_sql = f""" SELECT re.jockey_id, COUNT(*) AS runs, SUM(CASE WHEN rr.finish_position BETWEEN 1 AND 3 THEN 1 ELSE 0 END) AS top3, SUM(CASE WHEN rr.finish_position = 1 THEN 1 ELSE 0 END) AS wins FROM race_entries re INNER JOIN race_results rr ON rr.race_id = re.race_id AND rr.horse_id = re.horse_id INNER JOIN races r ON r.id = re.race_id WHERE re.jockey_id IN ({placeholders}) AND r.circuit = 'JRA' AND r.race_date < %s AND rr.finish_position > 0 GROUP BY re.jockey_id """ cur.execute(jockey_sql, (*jockey_ids, race["race_date"])) for row in cur.fetchall(): runs = int(row["runs"] or 0) top3 = int(row["top3"] or 0) wins = int(row["wins"] or 0) jockey_stats[int(row["jockey_id"])] = { "runs": runs, "top3_rate": (float(top3) / runs) if runs > 0 else 0.0, "win_rate": (float(wins) / runs) if runs > 0 else 0.0, } results: list[dict] = [] for entry in entries: cur.execute(history_sql, (entry["horse_id"], race["race_date"], int(lookback_runs))) history = cur.fetchall() metrics = compute_stage1_features( history, target_surface=str(race.get("surface") or ""), target_distance_m=float(race.get("distance_m") or 0) or None, target_track_code=str(race.get("track_code") or ""), carry_weight_kg=float(entry.get("carry_weight") or 0) or None, ) metrics["recent_runs"] = int(metrics["recent_runs"]) model_win_prob: float | None = None if scoring_mode == "lightgbm" and models_weights: model_win_prob = _model_win_prob( models_weights, model_features, metrics, float(race.get("distance_m") or 0.0), 1.0 if (race.get("surface") or "") == "芝" else 0.0, ) score = round(model_win_prob * 100.0, 2) else: score = _stage1_score(metrics) model_win_prob = round(max(0.0, min(1.0, score / 100.0)), 6) stage1_5_bonus = 0.0 stage1_5_reasons: list[str] = [] if enable_stage1_5: adjustment = compute_stage1_5_adjustment(history) stage1_5_bonus = float(adjustment["bonus"]) stage1_5_reasons = adjustment["reasons"] score = round(min(100.0, score + stage1_5_bonus), 2) running_style = infer_running_style(history) horse_number = entry.get("horse_number") horse_number_int = int(horse_number) if horse_number is not None else None results.append( { "race_id": int(entry["race_id"]), "horse_id": int(entry["horse_id"]), # JV の馬番未確定は race_entries.horse_number が NULL（元は "00"） "horse_number": horse_number_int, "bracket_number": int(entry["bracket_number"]) if entry.get("bracket_number") is not None else None, "jockey_id": int(entry["jockey_id"]) if entry.get("jockey_id") is not None else None, "horse_name": entry["horse_name"], "entry_odds_win": float(entry["odds_win"]) if entry.get("odds_win") is not None else None, "model_win_prob": model_win_prob, "score": score, "stage1_5_bonus": stage1_5_bonus, "stage1_5_reasons": stage1_5_reasons, "running_style": running_style, "_history": history, **metrics, } ) if enable_stage2: race_context = build_race_context(results) for row in results: jockey_info = jockey_stats.get(row["jockey_id"] or -1, {}) stage2 = compute_stage2_adjustment( history_rows=row["_history"], race_date=race["race_date"] if isinstance(race["race_date"], date) else None, race_surface=race.get("surface"), race_distance=int(race.get("distance_m") or 0) if race.get("distance_m") else None, running_style=row["running_style"], race_context=race_context, jockey_top3_rate=jockey_info.get("top3_rate"), jockey_win_rate=jockey_info.get("win_rate"), jockey_samples=int(jockey_info.get("runs") or 0), track_code=race.get("track_code"), ) row["stage2_bonus"] = float(stage2["bonus"]) row["stage2_reasons"] = stage2["reasons"] row["pace_scenario"] = stage2["pace_scenario"] row["score"] = round(min(100.0, max(0.0, row["score"] + row["stage2_bonus"])), 2) else: for row in results: row["stage2_bonus"] = 0.0 row["stage2_reasons"] = [] row["pace_scenario"] = "off" _apply_within_race_score_spread(results) # 第3段階: 単勝オッズを取り込み、期待値軸を付与 odds_map: dict[int, float] = {} cur.execute( """ SELECT combination, min_odds, max_odds FROM odds_lines WHERE race_id = %s AND odds_type = 'tansho' """, (race_id,), ) for o in cur.fetchall(): combo = (o.get("combination") or "").strip() if combo.isdigit(): horse_number = int(combo) odd = o.get("min_odds") if odd is None: odd = o.get("max_odds") if odd is not None: odds_map[horse_number] = float(odd) # odds_lines にない場合は race_entries.odds_win で補完 for row in results: if row["horse_number"] not in odds_map and row.get("entry_odds_win") is not None: odds_map[row["horse_number"]] = float(row["entry_odds_win"]) if forecast_pipeline in ("base", "ev"): keep_base = sorted(results, key=lambda x: float(x.get("score") or 0.0), reverse=True) keep_count = max(1, int(round(len(keep_base) * pass_rate))) threshold = keep_base[keep_count - 1]["score"] for row in results: row["score_accuracy"] = float(row["score"]) row["passed"] = bool( row["score_accuracy"] >= threshold or float(row["race_level_rescue"]) >= 1.0 ) assign_rank_accuracy(results) assign_win_rates(results) odds_attached = 0 if forecast_pipeline == "ev" and enable_stage3: odds_attached = attach_win_odds(results, odds_map) apply_ev_forecast( results, enabled=True, min_prob_for_value=stage3_min_prob, odds_cap=stage3_odds_cap, ) else: odds_attached = attach_win_odds(results, odds_map) for row in results: row["score_accuracy"] = float(row["score"]) apply_stage3_value( results, enabled=enable_stage3, min_prob_for_value=stage3_min_prob, odds_cap=stage3_odds_cap, aggressive_min_prob=stage3_aggressive_min_prob, aggressive_odds_cap=stage3_aggressive_odds_cap, use_legacy_normalized_marks=True, apply_prob_odds_marks=(rank_mode == "prob_odds"), ) keep_base = sorted(results, key=lambda x: float(x["score_accuracy"]), reverse=True) keep_count = max(1, int(round(len(keep_base) * pass_rate))) threshold = keep_base[keep_count - 1]["score_accuracy"] for row in results: row.pop("_history", None) if "passed" not in row: row["passed"] = bool( row["score_accuracy"] >= threshold or float(row["race_level_rescue"]) >= 1.0 ) passed_count = sum(1 for row in results if row["passed"]) payload = { "race": { "race_id": int(race["id"]), "race_date": race["race_date"].isoformat() if isinstance(race["race_date"], date) else str(race["race_date"]), "track_code": race["track_code"], "track_name": race["track_name"], "race_number": int(race["race_number"]), "race_name": race["race_name"], }, "config": { "lookback_runs": int(lookback_runs), "pass_rate": pass_rate, "scoring_mode": scoring_mode, "lightgbm_ensemble_n": len(models_weights) if scoring_mode == "lightgbm" else 0, "stage1_5_enabled": enable_stage1_5, "stage2_enabled": enable_stage2, "stage3_enabled": enable_stage3, "stage3_min_prob": stage3_min_prob, "stage3_odds_cap": stage3_odds_cap, "stage3_aggressive_min_prob": stage3_aggressive_min_prob, "stage3_aggressive_odds_cap": stage3_aggressive_odds_cap, "forecast_pipeline": forecast_pipeline, "pass_score_threshold": threshold, }, "summary": { "entry_count": len(results), "passed_count": passed_count, "failed_count": len(results) - passed_count, "odds_attached_count": odds_attached, }, "horses": results, } try: emp = fetch_ev_forecast_empirical_calibration_for_track(cur, str(race.get("track_code") or "")) if emp is not None: emp["track_name"] = str(race.get("track_name") or "") payload["empirical_hit_rates_ev"] = emp except Exception: pass try: hc = compute_honmei_confidence(payload["horses"], payload.get("empirical_hit_rates_ev")) if hc is not None: payload["honmei_confidence"] = hc except Exception: pass return payload finally: conn.close() def write_predictions(payload: dict) -> int: race_id = int(payload["race"]["race_id"]) conn = connect() try: with conn.cursor() as cur: cur.execute( """ DELETE FROM predictions WHERE race_id = %s AND prediction_type IN ( 'stage1_screening_accuracy', 'stage1_screening_value', 'stage1_screening_hi_ev' ) AND bet_type = 'horse' """, (race_id,), ) inserted = 0 for row in payload["horses"]: comb = ( str(int(row["horse_number"])) if row.get("horse_number") is not None else f"hid:{int(row['horse_id'])}" ) note_obj: dict = { "stage": "stage1", "horse_name": row["horse_name"], "horse_number": row.get("horse_number"), "bracket_number": row.get("bracket_number"), "passed": row["passed"], "score": row["score"], "score_accuracy": row.get("score_accuracy"), "stage3_prob": row.get("stage3_prob"), "win_odds": row.get("win_odds"), "expected_roi": row.get("expected_roi"), "win_probability": row.get("win_probability"), "win_rate": row.get("win_rate"), "model_win_prob": row.get("model_win_prob"), "cut": row.get("cut"), "cut_reason": row.get("cut_reason"), "rank_accuracy": row.get("rank_accuracy"), "rank_value": row.get("rank_value"), "rank_value_aggressive": row.get("rank_value_aggressive"), "expected_roi_aggressive": row.get("expected_roi_aggressive"), "metrics": { "recent_runs": row["recent_runs"], "recent_win_rate": row["recent_win_rate"], "recent_top3_rate": row["recent_top3_rate"], "recent_avg_finish": row["recent_avg_finish"], "recent_graded_top5_rate": row["recent_graded_top5_rate"], "race_level_rescue": row["race_level_rescue"], "stage1_5_bonus": row.get("stage1_5_bonus", 0.0), "stage1_5_reasons": row.get("stage1_5_reasons", []), "stage2_bonus": row.get("stage2_bonus", 0.0), "stage2_reasons": row.get("stage2_reasons", []), "running_style": row.get("running_style"), "pace_scenario": row.get("pace_scenario"), }, } if int(row.get("rank_prob_odds") or row.get("rank_accuracy") or 999) == 1 and payload.get( "honmei_confidence" ): note_obj["honmei_confidence"] = payload["honmei_confidence"] note = json.dumps(note_obj, ensure_ascii=False) cur.execute( """ INSERT INTO predictions ( race_id, prediction_type, bet_type, combination, win_probability, expected_odds, expected_roi, rank_in_pred, note ) VALUES (%s, 'stage1_screening_accuracy', 'horse', %s, %s, %s, %s, %s, %s) """, ( race_id, comb, float(row.get("stage3_prob", 0.0) * 100.0), float(row["win_odds"]) if row.get("win_odds") is not None else None, float(row.get("prob_times_odds")) if row.get("prob_times_odds") is not None else (float(row["expected_roi"]) if row.get("expected_roi") is not None else None), int(row.get("rank_prob_odds") or row.get("rank_accuracy") or row["rank"]), note, ), ) inserted += 1 cur.execute( """ INSERT INTO predictions ( race_id, prediction_type, bet_type, combination, win_probability, expected_odds, expected_roi, rank_in_pred, note ) VALUES (%s, 'stage1_screening_value', 'horse', %s, %s, %s, %s, %s, %s) """, ( race_id, comb, float(row.get("stage3_prob", 0.0) * 100.0), float(row["win_odds"]) if row.get("win_odds") is not None else None, float(row["expected_roi"]) if row.get("expected_roi") is not None else None, int(row.get("rank_value") or row["rank"]), note, ), ) inserted += 1 cur.execute( """ INSERT INTO predictions ( race_id, prediction_type, bet_type, combination, win_probability, expected_odds, expected_roi, rank_in_pred, note ) VALUES (%s, 'stage1_screening_hi_ev', 'horse', %s, %s, %s, %s, %s, %s) """, ( race_id, comb, float(row.get("stage3_prob", 0.0) * 100.0), float(row["win_odds"]) if row.get("win_odds") is not None else None, float(row["expected_roi_aggressive"]) if row.get("expected_roi_aggressive") is not None else None, int(row.get("rank_value_aggressive") or row["rank"]), note, ), ) inserted += 1 conn.commit() return inserted finally: conn.close() def main() -> None: parser = argparse.ArgumentParser(description="第1段階（基礎能力足切り）スクリーニング") parser.add_argument("--race-id", type=int, default=None) parser.add_argument("--race-date", type=str, default=None, help="YYYY-MM-DD") parser.add_argument("--track-code", type=str, default=None) parser.add_argument("--race-number", type=int, default=None) parser.add_argument("--lookback-runs", type=int, default=10) parser.add_argument("--pass-rate", type=float, default=0.5, help="通過率(0.0-1.0)") parser.add_argument("--disable-stage1-5", action="store_true", help="第1.5段階補正を無効化") parser.add_argument("--disable-stage2", action="store_true", help="第2段階補正を無効化") parser.add_argument("--disable-stage3", action="store_true", help="第3段階(オッズ妙味統合)を無効化") parser.add_argument("--stage3-min-prob", type=float, default=0.02, help="妙味計算の最低信頼確率") parser.add_argument("--stage3-odds-cap", type=float, default=80.0, help="妙味計算時のオッズ上限") parser.add_argument( "--stage3-aggressive-min-prob", type=float, default=0.0, help="高回収狙い(大穴寄り期待値)の最低信頼確率減算（既定0=妙味より緩い）", ) parser.add_argument( "--stage3-aggressive-odds-cap", type=float, default=300.0, help="高回収狙いのオッズ上限（妙味より広く取る既定）", ) parser.add_argument("--model-path", type=str, default=None, help="LightGBMモデルファイル(.txt)") parser.add_argument("--model-meta-path", type=str, default=None, help="モデルメタ情報JSON") parser.add_argument("--write-predictions", action="store_true", help="predictionsへ保存") parser.add_argument("--json", action="store_true", help="JSON出力") args = parser.parse_args() if not (0.0 < args.pass_rate <= 1.0): raise SystemExit("--pass-rate は 0.0 より大きく 1.0 以下で指定してください") if args.lookback_runs <= 0: raise SystemExit("--lookback-runs は 1 以上で指定してください") if args.stage3_min_prob < 0.0 or args.stage3_min_prob >= 1.0: raise SystemExit("--stage3-min-prob は 0.0 以上 1.0 未満で指定してください") if args.stage3_odds_cap <= 0: raise SystemExit("--stage3-odds-cap は 0 より大きく指定してください") if args.stage3_aggressive_min_prob < 0.0 or args.stage3_aggressive_min_prob >= 1.0: raise SystemExit("--stage3-aggressive-min-prob は 0.0 以上 1.0 未満で指定してください") if args.stage3_aggressive_odds_cap <= 0: raise SystemExit("--stage3-aggressive-odds-cap は 0 より大きく指定してください") conn = connect() try: with conn.cursor() as cur: race_id = _resolve_race_id(cur, args.race_id, args.race_date, args.track_code, args.race_number) finally: conn.close() payload = run_stage1( race_id=race_id, lookback_runs=args.lookback_runs, pass_rate=args.pass_rate, enable_stage1_5=not args.disable_stage1_5, enable_stage2=not args.disable_stage2, enable_stage3=not args.disable_stage3, stage3_min_prob=args.stage3_min_prob, stage3_odds_cap=args.stage3_odds_cap, stage3_aggressive_min_prob=args.stage3_aggressive_min_prob, stage3_aggressive_odds_cap=args.stage3_aggressive_odds_cap, model_path=args.model_path, model_meta_path=args.model_meta_path, ) inserted = 0 if args.write_predictions: inserted = write_predictions(payload) if args.json: print(json.dumps(payload, ensure_ascii=False, indent=2)) if args.write_predictions: print(json.dumps({"predictions_inserted": inserted}, ensure_ascii=False)) return race = payload["race"] summary = payload["summary"] print( f"[stage1] {race['race_date']} {race['track_name']} {race['race_number']}R " f"{race['race_name'] or ''}".strip() ) emp = payload.get("empirical_hit_rates_ev") if emp: ts = emp.get("tansho_marks") or {} ubx = emp.get("umaren_box_top4") or emp.get("umaren_box_top3") or {} ub = ubx.get("box_six_tickets") or ubx.get("box_three_tickets") or {} print( "[期待値予想·参考的中率] 同一競馬場の過去集計 — " f"◎的中率={ts.get('honmei_hit_rate')} ◎or〇={ts.get('top2_hit_rate')} " f"馬連BOX(6点)的中率={ub.get('hit_rate')} " f"(単勝サンプル={ts.get('sample_races_with_winner')} / BOX分母={ubx.get('denominator_races')})" ) hc = payload.get("honmei_confidence") if hc: hm = hc.get("honmei") or {} print( f"[本命·自信度] {hc.get('score_0_100')}/100（{hc.get('label')}） " f"#{hm.get('horse_number')} {hm.get('horse_name')}" ) print( f"entries={summary['entry_count']} passed={summary['passed_count']} " f"failed={summary['failed_count']} threshold={payload['config']['pass_score_threshold']} " f"mode={payload['config']['scoring_mode']} stage1_5={payload['config']['stage1_5_enabled']} " f"stage2={payload['config']['stage2_enabled']} stage3={payload['config']['stage3_enabled']} " f"odds_attached={summary.get('odds_attached_count', 0)} " f"min_prob={payload['config']['stage3_min_prob']} odds_cap={payload['config']['stage3_odds_cap']} " f"agg_min={payload['config'].get('stage3_aggressive_min_prob')} " f"agg_cap={payload['config'].get('stage3_aggressive_odds_cap')}" ) for row in payload["horses"]: mark = "PASS" if row["passed"] else "DROP" rescue = " rescue" if row["race_level_rescue"] else "" roi_text = "-" if row.get("expected_roi") is None else f"{row['expected_roi']:.2f}" hn = row.get("horse_number") hn_text = "?" if hn is None else f"{int(hn):>2}" print( f"{row['rank']:>2}. [{mark}] #{hn_text} {row['horse_name']} " f"score={row['score']:>5} runs={row['recent_runs']} " f"b1.5={row.get('stage1_5_bonus', 0.0)} b2={row.get('stage2_bonus', 0.0)} " f"rA={row.get('rank_accuracy')} rV={row.get('rank_value')} " f"rHi={row.get('rank_value_aggressive')} roi={roi_text}{rescue}" ) if args.write_predictions: print(f"predictions inserted: {inserted}") if __name__ == "__main__": main()