JVLink MCP Server

Instructions

Implementation Reference

• src/jvlink_mcp_server/database/high_level_api.py:173-203 (handler)

  The public API function that calls the implementation of favorite_performance for JRA data.

  def get_favorite_performance(
      db_connection,
      venue: Optional[str] = None,
      ninki: int = 1,
      grade: Optional[str] = None,
      year_from: Optional[str] = None,
      distance: Optional[int] = None
  ) -> Dict[str, Any]:
      """人気別成績を取得
  
      Args:
          db_connection: DatabaseConnectionインスタンス
          venue: 競馬場名（日本語、例: '東京', '中山'）
          ninki: 人気順位（1-18）
          grade: グレードコード（'G1', 'G2', 'G3' または 'A', 'B', 'C'）
          year_from: 集計開始年（例: '2023'）
          distance: 距離（メートル、例: 1600）
  
      Returns:
          dict: 勝率・連対率・複勝率等
  
      Example:
          >>> result = get_favorite_performance(
          ...     db_conn, venue='東京', ninki=1, grade='G1', year_from='2020'
          ... )
          >>> print(f"勝率: {result['win_rate']:.1f}%")
      """
      return _favorite_performance_impl(
          db_connection, venue=venue, ninki=ninki, grade=grade,
          year_from=year_from, distance=distance, source='jra'
      )

• src/jvlink_mcp_server/database/high_level_api.py:85-170 (handler)

  The actual handler implementation for favorite_performance, supporting both JRA and NAR via the 'source' parameter.

  def _favorite_performance_impl(
      db_connection,
      venue: Optional[str] = None,
      ninki: int = 1,
      grade: Optional[str] = None,
      year_from: Optional[str] = None,
      distance: Optional[int] = None,
      source: str = 'jra'
  ) -> Dict[str, Any]:
      """人気別成績の共通実装（JRA/NAR兼用）"""
      tables = _SOURCE_TABLES[source]
      conditions = []
      query_params: List = []
      condition_desc = []
  
      conditions.append("Ninki = ?")
      query_params.append(ninki)
      condition_desc.append(f"{ninki}番人気")
      if source == 'nar':
          condition_desc.append("NAR地方競馬")
  
      conditions.append("KakuteiJyuni IS NOT NULL")
      conditions.append("KakuteiJyuni > 0")
  
      if venue:
          venue_code = _resolve_venue(venue, source)
          conditions.append("s.JyoCD = ?")
          query_params.append(venue_code)
          condition_desc.append(f"{venue}競馬場")
  
      if grade:
          grade_code = GRADE_CODES.get(grade.upper())
          if not grade_code:
              raise ValueError(f"不明なグレード: {grade}. 有効な値: {list(GRADE_CODES.keys())}")
          conditions.append("r.GradeCD = ?")
          query_params.append(grade_code)
          condition_desc.append(f"グレード{grade}")
  
      if year_from:
          year_val = _validate_year(year_from)
          conditions.append("s.Year >= ?")
          query_params.append(year_val)
          condition_desc.append(f"{year_val}年以降")
  
      if distance:
          conditions.append("r.Kyori = ?")
          query_params.append(distance)
          condition_desc.append(f"{distance}m")
  
      where_clause = " AND ".join(conditions)
      need_join = grade or distance
  
      if need_join:
          query = f"""
          SELECT COUNT(*) as total,
              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 {tables['se']} s
          JOIN {tables['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}
          """
      else:
          query = f"""
          SELECT COUNT(*) as total,
              SUM(CASE WHEN KakuteiJyuni = 1 THEN 1 ELSE 0 END) as wins,
              SUM(CASE WHEN KakuteiJyuni IN (1, 2) THEN 1 ELSE 0 END) as places_2,
              SUM(CASE WHEN KakuteiJyuni IN (1, 2, 3) THEN 1 ELSE 0 END) as places_3
          FROM {tables['se']} s
          WHERE {where_clause}
          """
  
      df = db_connection.execute_safe_query(query, params=tuple(query_params))
  
      if df.empty or df.iloc[0]['total'] == 0:
          return {**_compute_rates(0, 0, 0, 0),
                  'conditions': ', '.join(condition_desc), 'query': query}
  
      row = df.iloc[0]
      result = _compute_rates(int(row['total']), int(row['wins']),
                             int(row['places_2']), int(row['places_3']))
      result['conditions'] = ', '.join(condition_desc)
      result['query'] = query
      return result