get_historical_data
Retrieve historical lottery draw data for specified types, such as Double Color Ball or 3D Lottery, enabling analysis or tracking trends over a set number of periods.
Instructions
获取指定彩票类型的历史开奖数据
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| lottery_type | Yes | 彩票类型 | |
| periods | No | 获取期数 |
Implementation Reference
- src/swlc_mcp/server.py:939-959 (registration)Registration of the 'get_historical_data' tool in the MCP server's list_tools method, including its schema.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"] } ),
- src/swlc_mcp/server.py:1122-1141 (handler)The MCP tool handler implementation that executes get_historical_data by calling the service method and formatting the output.elif 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:412-587 (helper)Core helper method in SWLCService that fetches historical data from DB or network, handles caching logic, and returns LotteryResult list.async 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 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-456 (helper)Database layer helper that queries SQLite for historical data based on lottery_type and limit, returning list of dicts.def 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 []