#!/usr/bin/env python3 """ Googleスプレッドシート（予想見解&実績）を取得し、同一レースをAIで予想して sheet_forecast_compare に保存する。 既定の公開CSVエクスポートURL（共有設定が「リンクを知っている全員」など）を使用。 """ from __future__ import annotations import argparse import csv import io import json import os import re import sys import urllib.request from datetime import datetime sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from ai.stage1_screening import run_stage1 from collectors.db_util import connect DEFAULT_SHEET_CSV_URL = ( "https://docs.google.com/spreadsheets/d/" "1WoyUi9Vuh0jbb4N8K7ofCOCY3qf8e7fy/export?format=csv&gid=702668700" ) def _norm_name(s: str) -> str: s = (s or "").replace("　", "").replace(" ", "").strip() return s def _parse_yen(s: str) -> int | None: if s is None or str(s).strip() == "": return None t = re.sub(r"[,，円￥¥\s]", "", str(s).strip()) if t == "" or t == "-": return None try: return int(float(t)) except ValueError: return None def _parse_race_number(cell: str) -> int | None: m = re.search(r"(\d+)", str(cell or "")) return int(m.group(1)) if m else None def _parse_honmei_name(marks: str) -> str | None: m = re.search(r"◎\s*([^/\n／]+)", marks or "") if not m: return None return m.group(1).strip() def _load_track_name_to_code() -> dict[str, str]: conn = connect() try: with conn.cursor() as cur: cur.execute("SELECT code, name FROM tracks WHERE circuit = 'JRA'") return {str(r["name"]): str(r["code"]) for r in cur.fetchall()} finally: conn.close() def _resolve_race_id(cur, race_date: str, track_code: str, race_number: int) -> int | None: cur.execute( """ SELECT r.id FROM races r 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' ORDER BY r.id DESC LIMIT 1 """, (race_date, track_code, race_number), ) row = cur.fetchone() return int(row["id"]) if row else None def _match_umaban(cur, race_id: int, horse_name: str) -> int | None: target = _norm_name(horse_name) if not target: return None cur.execute( """ SELECT re.horse_number, h.name FROM race_entries re JOIN horses h ON h.id = re.horse_id WHERE re.race_id = %s AND re.is_scratched = 0 """, (race_id,), ) rows = cur.fetchall() for row in rows: if _norm_name(row["name"]) == target: return int(row["horse_number"]) for row in rows: nm = _norm_name(row["name"]) if len(target) >= 2 and (target in nm or nm in target): return int(row["horse_number"]) 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 return int(row["horse_number"]), str(row["name"]) def _top3_results(cur, race_id: int) -> list[dict]: cur.execute( """ SELECT rr.finish_position, re.horse_number, h.name AS horse_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 BETWEEN 1 AND 3 ORDER BY rr.finish_position ASC """, (race_id,), ) out = [] for r in cur.fetchall(): out.append( { "finish_position": int(r["finish_position"]), "horse_number": int(r["horse_number"]), "horse_name": str(r["horse_name"] or ""), } ) return out def _reason_to_text(reason: str) -> str: m = { "lap_improved": "直近でラップ改善が見られる", "long_layoff_rebound": "休養明けの反発が見込める", "finish_sharp_recovery": "着順の急回復が確認できる", "closing_gain": "差し脚の伸びが安定している", "surface_fit_plus": "同条件の馬場適性が高い", "surface_fit_minus": "馬場適性に不安がある", "distance_fit_plus": "距離適性が噛み合っている", "distance_fit_minus": "距離適性に課題がある", "pace_match_front": "展開的に前が残りやすい想定", "pace_match_closer": "展開的に差しが届く想定", "pace_risk_front": "前半で脚を使わされる展開リスクがある", "pace_risk_closer": "スロー寄りで差し遅れリスクがある", "rotation_mid_plus": "間隔が程よく状態維持しやすい", "rotation_tight_minus": "間隔が詰まり反動リスクがある", "layoff_long_minus": "長期休養明けで割引が必要", "jockey_form_plus": "騎手の通算3着内率が高水準", "jockey_form_plus_mid": "騎手の通算3着内率がやや高め", "jockey_form_plus_light": "騎手の通算3着内率がやや有利", "jockey_form_minus": "騎手の通算3着内率が低水準", "jockey_form_minus_mid": "騎手の通算3着内率がやや低め", "jockey_form_minus_light": "騎手の通算3着内率がやや不利", "jockey_win_plus": "騎手の通算勝率が高め", "jockey_win_plus_light": "騎手の通算勝率がやや高め", "jockey_win_minus": "騎手の通算勝率が低め", "jockey_win_minus_light": "騎手の通算勝率がやや低め", } return m.get(reason, reason) def _tansho_marks(rows: list[dict], limit: int = 4) -> list[dict]: """単勝の印（◎〇▲△）を上位から付与して返す。""" marks = ["◎本命", "〇対抗", "▲単穴", "△連下"] out = [] for i, h in enumerate(rows[:limit]): out.append( { "mark": marks[i] if i < len(marks) else f"{i+1}", "horse_number": int(h.get("horse_number") or 0), "horse_name": str(h.get("horse_name") or ""), "score_accuracy": h.get("score_accuracy"), "stage3_prob": h.get("stage3_prob"), "win_odds": h.get("win_odds"), "expected_roi": h.get("expected_roi"), "expected_roi_aggressive": h.get("expected_roi_aggressive"), "stage1_5_reasons": _safe_list(h.get("stage1_5_reasons")), "stage2_reasons": _safe_list(h.get("stage2_reasons")), "running_style": h.get("running_style"), "pace_scenario": h.get("pace_scenario"), } ) return out def _format_mark_lines(picks: list[dict]) -> str: lines = [] for p in picks: no = p.get("horse_number") nm = p.get("horse_name") or "" lines.append(f"{p.get('mark')} #{no} {nm}".strip()) return "\n".join(lines) if lines else "（選出なし）" def _risk_note(reasons: list[str]) -> str: neg = [r for r in reasons if "minus" in r or "risk" in r or "layoff" in r] if neg: return "不安: " + "、".join([_reason_to_text(x) for x in neg[:2]]) return "不安: 大きな減点材料は少なめ" def _make_comments_for_axis(picks: list[dict], axis_name: str) -> tuple[str, str]: """ 予想コメント（3行）と予想詳細（10行）を返す。 """ if not picks: return f"予想コメント：{axis_name}の選出ができませんでした。", f"予想詳細：{axis_name}の選出ができませんでした。" top = picks[0] reasons = [*(top.get("stage1_5_reasons") or []), *(top.get("stage2_reasons") or [])] reasons_txt = "、".join([_reason_to_text(x) for x in reasons[:2]]) or "基礎評価が上位" prob_pct = round(float(top.get("stage3_prob") or 0.0) * 100.0, 1) odds = top.get("win_odds") # 3行コメント short = "\n".join( [ f"予想コメント：{axis_name}は{top.get('mark')} #{top.get('horse_number')} {top.get('horse_name')}", f"根拠: {reasons_txt}", f"{_risk_note(reasons)}", ] ) # 10行詳細（最大10行） lines = [] lines.append(f"予想詳細：{axis_name}（単勝4頭）") for p in picks[:4]: pb = round(float(p.get("stage3_prob") or 0.0) * 100.0, 1) od = p.get("win_odds") lines.append(f"{p.get('mark')} #{p.get('horse_number')} {p.get('horse_name')} / 勝率{pb}% / 単勝{od if od is not None else '—'}") lines.append(f"◎の根拠: {reasons_txt}") lines.append(f"◎の走法/展開: 脚質 {top.get('running_style') or '—'} / 展開 {top.get('pace_scenario') or '—'}") lines.append(f"◎のオッズ: 単勝 {odds if odds is not None else '—'} / モデル勝率 {prob_pct}%") lines.append(_risk_note(reasons)) lines.append("買い方: 単勝は◎〜△の中から当日の気配とオッズで最終決定") lines.append("注意: 印はAIスコア順（堅実/妙味/高回収で並びが変わります）") detail = "\n".join(lines[:10]) return short, detail def _make_ai_comment( best_acc: dict | None, best_val: dict | None, best_hi: dict | None = None ) -> tuple[str, str, str]: if best_acc: acc_reasons = [*_safe_list(best_acc.get("stage1_5_reasons")), *_safe_list(best_acc.get("stage2_reasons"))] acc_txt = "、".join([_reason_to_text(x) for x in acc_reasons[:2]]) if acc_txt == "": acc_txt = "基礎点が上位で、堅実に走れる想定" comment_acc = ( f"堅実軸は#{best_acc.get('horse_number')} {best_acc.get('horse_name')}。" f"{acc_txt}を踏まえ、軸として信頼度を高く見ています。" ) else: comment_acc = "堅実軸コメントを生成できませんでした。" if best_val: odds = best_val.get("win_odds") roi = best_val.get("expected_roi") val_reasons = [*_safe_list(best_val.get("stage2_reasons"))] val_txt = "、".join([_reason_to_text(x) for x in val_reasons[:1]]) if val_txt == "": val_txt = "オッズと能力バランス" comment_val = ( f"妙味軸は#{best_val.get('horse_number')} {best_val.get('horse_name')}。" f"{val_txt}の条件で、オッズ妙味を狙う判断です。" f"期待値はROI {roi if roi is not None else '—'}" f"{' / 単勝オッズ ' + str(odds) if odds is not None else ''}。" ) else: comment_val = "妙味軸コメントを生成できませんでした。" if best_hi: odds = best_hi.get("win_odds") roi_a = best_hi.get("expected_roi_aggressive") prob_pct = round(float(best_hi.get("stage3_prob") or 0.0) * 100.0, 1) comment_hi = ( f"高回収狙いは#{best_hi.get('horse_number')} {best_hi.get('horse_name')}。" f"単勝オッズ{odds if odds is not None else '—'}・モデル勝率約{prob_pct}%前提で、" f"的中時の払戻を重視した期待値（指標{roi_a if roi_a is not None else '—'}）を最優先しています。" f"（500%回収は運用・オッズ次第で相対的に狙う軸です）" ) else: comment_hi = "高回収狙いコメントを生成できませんでした。" return comment_acc, comment_val, comment_hi def _safe_list(v) -> list: if isinstance(v, list): return v return [] def download_csv(url: str, timeout: int = 60) -> str: req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 (sheet import)"}) with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.read().decode("utf-8") def upsert_row(conn, rec: dict) -> None: cols = [ "race_date", "track_code", "race_number", "race_id", "sheet_race_name", "sheet_marks", "sheet_honmei_name", "sheet_honmei_umaban", "sheet_hit", "sheet_payout_yen", "sheet_pl_yen", "ai_rank1_umaban", "ai_rank1_name", "ai_rank1_score", "ai_rank2_umaban", "ai_rank2_name", "ai_rank2_score", "ai_value1_umaban", "ai_value1_name", "ai_value1_roi", "ai_hi_ev1_umaban", "ai_hi_ev1_name", "ai_hi_ev1_roi", "winner_umaban", "winner_name", "match_honmei_winner", "match_ai_rank1_winner", "match_ai_rank2_winner", "match_ai_value1_winner", "match_ai_hi_ev1_winner", "match_ai_rank1_honmei", "match_ai_rank2_honmei", "match_ai_value1_honmei", "match_ai_hi_ev1_honmei", "status", "error_message", "ai_snapshot_json", "sheet_source_url", ] placeholders = ", ".join(["%s"] * len(cols)) updates = ", ".join([f"{c} = VALUES({c})" for c in cols if c not in ("race_date", "track_code", "race_number")]) sql = f""" INSERT INTO sheet_forecast_compare ({", ".join(cols)}) VALUES ({placeholders}) ON DUPLICATE KEY UPDATE {updates} """ with conn.cursor() as cur: cur.execute(sql, [rec[c] for c in cols]) conn.commit() def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--url", default=DEFAULT_SHEET_CSV_URL, help="スプレッドシート CSV エクスポートURL") ap.add_argument("--csv-file", default=None, help="ローカルCSV（オフライン時）") ap.add_argument("--limit", type=int, default=0, help="0=全件") ap.add_argument("--dry-run", action="store_true") ap.add_argument("--lookback-runs", type=int, default=10) ap.add_argument("--pass-rate", type=float, default=0.5) ap.add_argument("--stage3-min-prob", type=float, default=0.02) ap.add_argument("--stage3-odds-cap", type=float, default=80.0) ap.add_argument("--stage3-aggressive-min-prob", type=float, default=0.0) ap.add_argument("--stage3-aggressive-odds-cap", type=float, default=300.0) args = ap.parse_args() if args.csv_file: with open(args.csv_file, "r", encoding="utf-8") as f: text = f.read() else: text = download_csv(args.url) track_map = _load_track_name_to_code() reader = csv.DictReader(io.StringIO(text)) rows = list(reader) if args.limit and args.limit > 0: rows = rows[: args.limit] conn = connect() ok = 0 for i, row in enumerate(rows): race_date = (row.get("日付") or "").strip() place = (row.get("場所") or "").strip() marks = row.get("印と馬名") or "" try: datetime.strptime(race_date, "%Y-%m-%d") except ValueError: continue track_code = track_map.get(place) race_num = _parse_race_number(row.get("レース番号") or "") if not track_code or race_num is None: continue rec: dict = { "race_date": race_date, "track_code": track_code, "race_number": race_num, "race_id": None, "sheet_race_name": (row.get("レース名") or "").strip() or None, "sheet_marks": marks.strip() or None, "sheet_honmei_name": None, "sheet_honmei_umaban": None, "sheet_hit": (row.get("的中_不的中") or "").strip() or None, "sheet_payout_yen": _parse_yen(row.get("払戻金")), "sheet_pl_yen": _parse_yen(row.get("収支")), "ai_rank1_umaban": None, "ai_rank1_name": None, "ai_rank1_score": None, "ai_rank2_umaban": None, "ai_rank2_name": None, "ai_rank2_score": None, "ai_value1_umaban": None, "ai_value1_name": None, "ai_value1_roi": None, "ai_hi_ev1_umaban": None, "ai_hi_ev1_name": None, "ai_hi_ev1_roi": None, "winner_umaban": None, "winner_name": None, "match_honmei_winner": None, "match_ai_rank1_winner": None, "match_ai_rank2_winner": None, "match_ai_value1_winner": None, "match_ai_hi_ev1_winner": None, "match_ai_rank1_honmei": None, "match_ai_rank2_honmei": None, "match_ai_value1_honmei": None, "match_ai_hi_ev1_honmei": None, "status": "ok", "error_message": None, "ai_snapshot_json": None, "sheet_source_url": args.url[:512] if args.url else None, } try: with conn.cursor() as cur: rid = _resolve_race_id(cur, race_date, track_code, race_num) if not rid: rec["status"] = "race_not_found" rec["error_message"] = "DBに該当レースなし" if not args.dry_run: upsert_row(conn, rec) continue rec["race_id"] = rid honmei = _parse_honmei_name(marks) rec["sheet_honmei_name"] = honmei with conn.cursor() as cur: if honmei: rec["sheet_honmei_umaban"] = _match_umaban(cur, rid, honmei) if rec["sheet_honmei_umaban"] is None: rec["status"] = "honmei_not_matched" rec["error_message"] = f"馬名不一致: {honmei}" w_um, w_name = _winner(cur, rid) rec["winner_umaban"] = w_um rec["winner_name"] = w_name race_result_top3 = _top3_results(cur, rid) payload = run_stage1( race_id=rid, lookback_runs=args.lookback_runs, pass_rate=args.pass_rate, enable_stage1_5=True, enable_stage2=True, enable_stage3=True, 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, ) horses = payload.get("horses") or [] by_acc = sorted(horses, key=lambda x: int(x.get("rank_accuracy") or 999)) by_val = sorted(horses, key=lambda x: int(x.get("rank_value") or 999)) by_hi = sorted(horses, key=lambda x: int(x.get("rank_value_aggressive") or 999)) picks_acc = _tansho_marks(by_acc, limit=4) picks_val = _tansho_marks(by_val, limit=4) picks_hi = _tansho_marks(by_hi, limit=4) if by_acc: h1 = by_acc[0] rec["ai_rank1_umaban"] = int(h1["horse_number"]) rec["ai_rank1_name"] = str(h1.get("horse_name") or "") rec["ai_rank1_score"] = float(h1.get("score_accuracy") or h1.get("score") or 0) if len(by_acc) >= 2: h2 = by_acc[1] rec["ai_rank2_umaban"] = int(h2["horse_number"]) rec["ai_rank2_name"] = str(h2.get("horse_name") or "") rec["ai_rank2_score"] = float(h2.get("score_accuracy") or h2.get("score") or 0) if by_val: v1 = by_val[0] rec["ai_value1_umaban"] = int(v1["horse_number"]) rec["ai_value1_name"] = str(v1.get("horse_name") or "") roi = v1.get("expected_roi") rec["ai_value1_roi"] = float(roi) if roi is not None else None if by_hi: h1 = by_hi[0] rec["ai_hi_ev1_umaban"] = int(h1["horse_number"]) rec["ai_hi_ev1_name"] = str(h1.get("horse_name") or "") roi_h = h1.get("expected_roi_aggressive") rec["ai_hi_ev1_roi"] = float(roi_h) if roi_h is not None else None comment_acc, comment_val, comment_hi = _make_ai_comment( by_acc[0] if by_acc else None, by_val[0] if by_val else None, by_hi[0] if by_hi else None, ) c3_acc, c10_acc = _make_comments_for_axis(picks_acc, "堅実") c3_val, c10_val = _make_comments_for_axis(picks_val, "妙味") c3_hi, c10_hi = _make_comments_for_axis(picks_hi, "高回収") sh = rec["sheet_honmei_umaban"] w = rec["winner_umaban"] r1 = rec["ai_rank1_umaban"] r2 = rec["ai_rank2_umaban"] v1u = rec["ai_value1_umaban"] h1u = rec["ai_hi_ev1_umaban"] if w is not None: rec["match_honmei_winner"] = 1 if (sh is not None and sh == w) else 0 rec["match_ai_rank1_winner"] = 1 if (r1 is not None and r1 == w) else 0 rec["match_ai_rank2_winner"] = 1 if (r2 is not None and r2 == w) else 0 rec["match_ai_value1_winner"] = 1 if (v1u is not None and v1u == w) else 0 rec["match_ai_hi_ev1_winner"] = 1 if (h1u is not None and h1u == w) else 0 if sh is not None: rec["match_ai_rank1_honmei"] = 1 if (r1 is not None and r1 == sh) else 0 rec["match_ai_rank2_honmei"] = 1 if (r2 is not None and r2 == sh) else 0 rec["match_ai_value1_honmei"] = 1 if (v1u is not None and v1u == sh) else 0 rec["match_ai_hi_ev1_honmei"] = 1 if (h1u is not None and h1u == sh) else 0 rec["ai_snapshot_json"] = json.dumps( { "race": payload.get("race"), "config": payload.get("config"), "summary": payload.get("summary"), "horses_top5_accuracy": by_acc[:5], "horses_top5_value": by_val[:5], "horses_top5_hi_ev": by_hi[:5], "tansho_picks_accuracy": picks_acc, "tansho_picks_value": picks_val, "tansho_picks_hi_ev": picks_hi, "tansho_marks_text_accuracy": _format_mark_lines(picks_acc), "tansho_marks_text_value": _format_mark_lines(picks_val), "tansho_marks_text_hi_ev": _format_mark_lines(picks_hi), "race_result_top3": race_result_top3, "ai_comment_accuracy": comment_acc, "ai_comment_value": comment_val, "ai_comment_hi_ev": comment_hi, "ai_comment_accuracy_3lines": c3_acc, "ai_comment_accuracy_10lines": c10_acc, "ai_comment_value_3lines": c3_val, "ai_comment_value_10lines": c10_val, "ai_comment_hi_ev_3lines": c3_hi, "ai_comment_hi_ev_10lines": c10_hi, }, ensure_ascii=False, default=str, ) if not args.dry_run: upsert_row(conn, rec) ok += 1 print(f"[{i+1}/{len(rows)}] OK race_id={rid} {race_date} {place} {race_num}R", flush=True) except Exception as e: rec["status"] = "error" rec["error_message"] = str(e)[:500] if not args.dry_run: upsert_row(conn, rec) print(f"[{i+1}/{len(rows)}] ERR {race_date} {place} {race_num}R: {e}", flush=True) conn.close() print(f"done imported={ok} total_rows={len(rows)}") if __name__ == "__main__": main()