get_historical_data
Retrieve historical lottery draw data for specified game types like Double Color Ball, 3D Lottery, Seven Happiness Lottery, and Happy 8, enabling analysis of past results and patterns.
Instructions
获取指定彩票类型的历史开奖数据
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| lottery_type | Yes | 彩票类型 | |
| periods | No | 获取期数 |
Implementation Reference
- src/swlc_mcp/server.py:1068-1090 (registration)MCP tool registration and input schema definition for get_historical_datatypes.Tool( name="get_historical_data", description="获取指定彩票类型的历史开奖数据", inputSchema={ "type": "object", "properties": { "lottery_type": { "type": "string", "enum": ["双色球", "福彩3D", "七乐彩", "快乐8"], "description": "彩票类型" }, "periods": { "type": "integer", "minimum": 1, "maximum": 1000, "default": 10, "description": "获取期数" } }, "required": ["lottery_type"] } ), types.Tool(
- src/swlc_mcp/server.py:1344-1363 (handler)MCP server tool call handler that invokes the service method and formats the textual responseelif name == "get_historical_data": lottery_type = arguments.get("lottery_type") periods = arguments.get("periods", 10) results = await lottery_service.get_historical_data(lottery_type, periods) if results: text_lines = [f"{lottery_type}历史开奖数据（最近{len(results)}期）：\n"] for result in results: if result.special_numbers: numbers_str = f"{' '.join(result.numbers)} + {' '.join(result.special_numbers)}" else: numbers_str = ' '.join(result.numbers) text_lines.append(f"期号：{result.period} 日期：{result.draw_date} 号码：{numbers_str}") return [types.TextContent(type="text", text="\n".join(text_lines))] else: return [types.TextContent(type="text", text="获取历史数据失败")]
- src/swlc_mcp/server.py:475-654 (handler)Core handler in SWLCService that fetches historical lottery data from database or network, checks data freshness and continuity, and handles syncing if necessaryasync def get_historical_data(self, lottery_type: str, periods: int = 10) -> List[LotteryResult]: """获取历史开奖数据""" try: # 首先尝试从数据库获取 db_results = self.db.get_historical_data(lottery_type, periods) # 检查是否需要从网络更新数据 should_update = False if not db_results: # 数据库没有数据，需要从网络获取 should_update = True elif len(db_results) < periods: # 数据库数据不足，需要从网络获取 should_update = True else: # 检查最新数据的新鲜度 latest_result = db_results[0] # 最新的数据 if not self._is_data_fresh(latest_result.get('draw_date', ''), lottery_type): should_update = True # 检查期号连续性 elif not self._check_period_continuity(db_results, lottery_type): logger.warning(f"{lottery_type}数据库中期号不连续，需要从网络更新") should_update = True if not should_update: logger.info(f"从本地数据库获取{lottery_type}历史数据") results = [] for item in db_results: if lottery_type == "双色球": result = LotteryResult( lottery_type="双色球", period=item['period'], draw_date=item['draw_date'], numbers=item['red_balls'], special_numbers=[item['blue_ball']], prize_pool=item['prize_pool'], sales_amount=item['sales_amount'] ) elif lottery_type == "福彩3D": result = LotteryResult( lottery_type="福彩3D", period=item['period'], draw_date=item['draw_date'], numbers=item['numbers'], sales_amount=item['sales_amount'] ) elif lottery_type == "七乐彩": result = LotteryResult( lottery_type="七乐彩", period=item['period'], draw_date=item['draw_date'], numbers=item['basic_numbers'], special_numbers=[item['special_number']], prize_pool=item['prize_pool'], sales_amount=item['sales_amount'] ) elif lottery_type == "快乐8": result = LotteryResult( lottery_type="快乐8", period=item['period'], draw_date=item['draw_date'], numbers=item['numbers'], prize_pool=item['prize_pool'], sales_amount=item['sales_amount'] ) else: continue results.append(result) return results # 从网络获取并保存数据 logger.info(f"从网络获取{lottery_type}历史数据") data = await self._fetch_lottery_data(lottery_type, periods) if not data or not data['result']: # 如果网络获取失败，尝试返回数据库中的可用数据 if db_results: logger.warning(f"网络获取{lottery_type}数据失败，返回数据库中的可用数据") return self._convert_db_results_to_lottery_results(db_results, lottery_type) return [] results = [] for item in data['result']: if lottery_type == "双色球": # 解析红球和蓝球 red_balls = item['red'].split(',') blue_ball = item['blue'] # 保存到数据库 self.db.save_ssq_result( period=item['code'], draw_date=item['date'], red_balls=red_balls, blue_ball=blue_ball ) result = LotteryResult( lottery_type="双色球", period=item['code'], draw_date=item['date'], numbers=red_balls, special_numbers=[blue_ball] ) elif lottery_type == "福彩3D": # 解析3D号码 numbers = item['red'].split(',') # 保存到数据库 self.db.save_3d_result( period=item['code'], draw_date=item['date'], numbers=numbers ) result = LotteryResult( lottery_type="福彩3D", period=item['code'], draw_date=item['date'], numbers=numbers ) elif lottery_type == "七乐彩": # 解析基本号码和特别号码 basic_numbers = item['red'].split(',') special_number = item['blue'] # 保存到数据库 self.db.save_qlc_result( period=item['code'], draw_date=item['date'], basic_numbers=basic_numbers, special_number=special_number ) result = LotteryResult( lottery_type="七乐彩", period=item['code'], draw_date=item['date'], numbers=basic_numbers, special_numbers=[special_number] ) elif lottery_type == "快乐8": # 解析快乐8号码 numbers = item['red'].split(',') # 保存到数据库 self.db.save_kl8_result( period=item['code'], draw_date=item['date'], numbers=numbers ) result = LotteryResult( lottery_type="快乐8", period=item['code'], draw_date=item['date'], numbers=numbers ) else: continue results.append(result) return results except Exception as e: logger.error(f"获取{lottery_type}历史数据失败: {e}") # 如果出错，尝试返回数据库中的可用数据 try: db_results = self.db.get_historical_data(lottery_type, periods) if db_results: logger.warning(f"返回数据库中的{lottery_type}数据作为备选") return self._convert_db_results_to_lottery_results(db_results, lottery_type) except: pass return []
- src/swlc_mcp/database.py:368-457 (helper)Database helper method that queries SQLite tables for historical data based on lottery type and limit, returning formatted dictsdef get_historical_data(self, lottery_type: str, limit: int = 10) -> List[Dict[str, Any]]: """获取历史开奖数据""" try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() if lottery_type == "双色球": cursor.execute(""" SELECT period, draw_date, red_balls, blue_ball, prize_pool, sales_amount FROM ssq_results ORDER BY period DESC LIMIT ? """, (limit,)) results = [] for row in cursor.fetchall(): results.append({ 'period': row[0], 'draw_date': row[1], 'red_balls': json.loads(row[2]), 'blue_ball': row[3], 'prize_pool': row[4], 'sales_amount': row[5] }) return results elif lottery_type == "福彩3D": cursor.execute(""" SELECT period, draw_date, numbers, sales_amount FROM fucai3d_results ORDER BY period DESC LIMIT ? """, (limit,)) results = [] for row in cursor.fetchall(): results.append({ 'period': row[0], 'draw_date': row[1], 'numbers': json.loads(row[2]), 'sales_amount': row[3] }) return results elif lottery_type == "七乐彩": cursor.execute(""" SELECT period, draw_date, basic_numbers, special_number, prize_pool, sales_amount FROM qilecai_results ORDER BY period DESC LIMIT ? """, (limit,)) results = [] for row in cursor.fetchall(): results.append({ 'period': row[0], 'draw_date': row[1], 'basic_numbers': json.loads(row[2]), 'special_number': row[3], 'prize_pool': row[4], 'sales_amount': row[5] }) return results elif lottery_type == "快乐8": cursor.execute(""" SELECT period, draw_date, numbers, prize_pool, sales_amount FROM kuaile8_results ORDER BY period DESC LIMIT ? """, (limit,)) results = [] for row in cursor.fetchall(): results.append({ 'period': row[0], 'draw_date': row[1], 'numbers': json.loads(row[2]), 'prize_pool': row[3], 'sales_amount': row[4] }) return results return [] except Exception as e: logger.error(f"获取历史数据失败: {e}") return [] def update_number_statistics(self, lottery_type: str, numbers: List[str]):