JVLink MCP Server

Instructions

Implementation Reference

• src/jvlink_mcp_server/server.py:543-559 (handler)

  The MCP tool handler for "sire_stats" in the server.py file, which invokes the underlying high-level API function.

  @mcp.tool(name="sire_stats")
  def analyze_sire_stats(
      sire_name: str,
      venue: Optional[str] = None,
      distance: Optional[int] = None,
      year_from: Optional[str] = None
  ) -> dict:
      """種牡馬（父馬）の産駒成績を分析
  
      種牡馬名を指定して、産駒の勝率・複勝率を調べられます。
      距離や競馬場でフィルタリングすると、血統の適性傾向が見えます。
      """
      with DatabaseConnection() as db:
          return _get_sire_stats(
              db, sire_name=sire_name, venue=venue,
              distance=distance, year_from=year_from
          )

• src/jvlink_mcp_server/database/high_level_api.py:534-676 (handler)

  The actual implementation of the "sire_stats" logic in the database high-level API.

  def get_sire_stats(
      db_connection,
      sire_name: str,
      venue: Optional[str] = None,
      distance: Optional[int] = None,
      year_from: Optional[str] = None
  ) -> Dict[str, Any]:
      """種牡馬（父馬）成績を取得
  
      Args:
          db_connection: DatabaseConnectionインスタンス
          sire_name: 種牡馬名（部分一致検索）
          venue: 競馬場名（日本語、例: '東京', '中山'）
          distance: 距離（メートル、例: 1600）
          year_from: 集計開始年（例: '2023'）
  
      Returns:
          dict: {
              'sire_name': 種牡馬名,
              'total_runs': 総出走数,
              'wins': 1着回数,
              'places_2': 2着以内回数,
              'places_3': 3着以内回数,
              'win_rate': 勝率（%）,
              'place_rate_2': 連対率（%）,
              'place_rate_3': 複勝率（%）,
              'conditions': 適用した条件の説明
          }
  
      Example:
          >>> result = get_sire_stats(db_conn, 'ディープインパクト', venue='東京', distance=1600)
          >>> print(f"{result['sire_name']}: 勝率 {result['win_rate']:.1f}%")
      """
      conditions = []
      query_params: List = []
      condition_desc = [f"種牡馬: {sire_name}（部分一致）"]
  
      # 種牡馬名（部分一致）- NL_UMのKetto3InfoBamei1が父馬名
      conditions.append("u.Ketto3InfoBamei1 LIKE ?")
      query_params.append('%' + sire_name + '%')
  
      # 確定着順がNULLでない（INTEGER型）
      conditions.append("s.KakuteiJyuni IS NOT NULL")
      conditions.append("s.KakuteiJyuni > 0")
  
      # 競馬場
      if venue:
          venue_code = VENUE_CODES.get(venue)
          if not venue_code:
              raise ValueError(f"不明な競馬場名: {venue}. 有効な値: {list(VENUE_CODES.keys())}")
          conditions.append("s.JyoCD = ?")
          query_params.append(venue_code)
          condition_desc.append(f"{venue}競馬場")
  
      # 年（INTEGER型）
      if year_from:
          year_val = _validate_year(year_from)
          conditions.append("s.Year >= ?")
          query_params.append(year_val)
          condition_desc.append(f"{year_val}年以降")
  
      # 距離（INTEGER型、NL_RAと結合が必要）
      if distance:
          conditions.append("r.Kyori = ?")
          query_params.append(distance)
          condition_desc.append(f"{distance}m")
  
      where_clause = " AND ".join(conditions)
  
      # NL_UMとJOINして父馬名(Ketto3InfoBamei1)を取得
      if distance:
          query = f"""
          SELECT
              u.Ketto3InfoBamei1 as sire_name,
              COUNT(*) as total_runs,
              SUM(CASE WHEN s.KakuteiJyuni = 1 THEN 1 ELSE 0 END) as wins,
              SUM(CASE WHEN s.KakuteiJyuni IN (1, 2) THEN 1 ELSE 0 END) as places_2,
              SUM(CASE WHEN s.KakuteiJyuni IN (1, 2, 3) THEN 1 ELSE 0 END) as places_3
          FROM NL_SE s
          JOIN NL_UM u ON s.KettoNum = u.KettoNum
          JOIN NL_RA r
              ON s.Year = r.Year
              AND s.MonthDay = r.MonthDay
              AND s.JyoCD = r.JyoCD
              AND s.Kaiji = r.Kaiji
              AND s.Nichiji = r.Nichiji
              AND s.RaceNum = r.RaceNum
          WHERE {where_clause}
          GROUP BY u.Ketto3InfoBamei1
          """
      else:
          query = f"""
          SELECT
              u.Ketto3InfoBamei1 as sire_name,
              COUNT(*) as total_runs,
              SUM(CASE WHEN s.KakuteiJyuni = 1 THEN 1 ELSE 0 END) as wins,
              SUM(CASE WHEN s.KakuteiJyuni IN (1, 2) THEN 1 ELSE 0 END) as places_2,
              SUM(CASE WHEN s.KakuteiJyuni IN (1, 2, 3) THEN 1 ELSE 0 END) as places_3
          FROM NL_SE s
          JOIN NL_UM u ON s.KettoNum = u.KettoNum
          WHERE {where_clause}
          GROUP BY u.Ketto3InfoBamei1
          """
  
      # クエリ実行
      df = db_connection.execute_safe_query(query, params=tuple(query_params))
  
      if df.empty:
          return {
              'sire_name': sire_name,
              'total_runs': 0,
              'wins': 0,
              'places_2': 0,
              'places_3': 0,
              'win_rate': 0.0,
              'place_rate_2': 0.0,
              'place_rate_3': 0.0,
              'conditions': ', '.join(condition_desc),
              'query': query
          }
  
      # 複数の種牡馬がマッチする可能性があるため、合計を計算
      total_runs = int(df['total_runs'].sum())
      wins = int(df['wins'].sum())
      places_2 = int(df['places_2'].sum())
      places_3 = int(df['places_3'].sum())
  
      # マッチした種牡馬名を取得（最も出走数が多い種牡馬）
      matched_sire = df.loc[df['total_runs'].idxmax(), 'sire_name'] if not df.empty else sire_name
  
      return {
          'sire_name': matched_sire,
          'total_runs': total_runs,
          'wins': wins,
          'places_2': places_2,
          'places_3': places_3,
          'win_rate': (wins / total_runs * 100) if total_runs > 0 else 0.0,
          'place_rate_2': (places_2 / total_runs * 100) if total_runs > 0 else 0.0,
          'place_rate_3': (places_3 / total_runs * 100) if total_runs > 0 else 0.0,
          'conditions': ', '.join(condition_desc),
          'matched_sires': df['sire_name'].tolist(),
          'query': query
      }