JVLink MCP Server

Instructions

Implementation Reference

• src/jvlink_mcp_server/database/high_level_api.py:326-441 (handler)

  The implementation of the get_frame_stats logic which queries the database for frame-based performance.

  def get_frame_stats(
      db_connection,
      venue: Optional[str] = None,
      distance: Optional[int] = None,
      year_from: Optional[str] = None
  ) -> pd.DataFrame:
      """枠番別成績を取得
  
      Args:
          db_connection: DatabaseConnectionインスタンス
          venue: 競馬場名（日本語、例: '東京', '中山'）
          distance: 距離（メートル、例: 1600）
          year_from: 集計開始年（例: '2023'）
  
      Returns:
          DataFrame: 枠番別の成績
              - wakuban: 枠番（1-8）
              - total: 総出走数
              - wins: 1着回数
              - places_2: 2着以内回数
              - places_3: 3着以内回数
              - win_rate: 勝率（%）
              - place_rate_2: 連対率（%）
              - place_rate_3: 複勝率（%）
  
      Example:
          >>> df = get_frame_stats(db_conn, venue='東京', distance=1600)
          >>> print(df.to_string())
      """
      # WHERE条件を構築
      conditions = []
      query_params: List = []
      condition_desc = []
  
      # 確定着順がNULLでない（INTEGER型）
      conditions.append("s.KakuteiJyuni IS NOT NULL")
      conditions.append("s.KakuteiJyuni > 0")
      conditions.append("s.Wakuban IS NOT NULL")
      conditions.append("s.Wakuban > 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_RAと結合
      if distance:
          query = f"""
          SELECT
              s.Wakuban as wakuban,
              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 NL_SE s
          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 s.Wakuban
          ORDER BY s.Wakuban
          """
      else:
          query = f"""
          SELECT
              Wakuban as wakuban,
              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 NL_SE s
          WHERE {where_clause}
          GROUP BY Wakuban
          ORDER BY Wakuban
          """
  
      # クエリ実行
      df = db_connection.execute_safe_query(query, params=tuple(query_params))
  
      if df.empty:
          return pd.DataFrame(columns=['wakuban', 'total', 'wins', 'places_2', 'places_3',
                                      'win_rate', 'place_rate_2', 'place_rate_3'])
  
      # 勝率・連対率・複勝率を計算
      df['win_rate'] = (df['wins'] / df['total'] * 100).round(2)
      df['place_rate_2'] = (df['places_2'] / df['total'] * 100).round(2)
      df['place_rate_3'] = (df['places_3'] / df['total'] * 100).round(2)
  
      # メタデータを属性として追加
      df.attrs['conditions'] = ', '.join(condition_desc) if condition_desc else '全条件'
      df.attrs['query'] = query
  
      return df

• src/jvlink_mcp_server/server.py:504-520 (handler)

  MCP tool handler 'analyze_frame_stats' registered as "frame_stats".

  @mcp.tool(name="frame_stats")
  def analyze_frame_stats(
      venue: Optional[str] = None,
      distance: Optional[int] = None,
      year_from: Optional[str] = None
  ) -> dict:
      """枠番（1〜8枠）別の成績を分析
  
      内枠・外枠の有利不利を調べられます。
      競馬場や距離でフィルタリングすると、コース特性が見えます。
      """
      with DatabaseConnection() as db:
          df = _get_frame_stats(db, venue=venue, distance=distance, year_from=year_from)
          return {
              "data": df.to_dict(orient="records"),
              "conditions": df.attrs.get("conditions", ""),
              "columns": df.columns.tolist()