SWLC MCP Server

Instructions

Implementation Reference

• src/swlc_mcp/server.py:1112-1140 (registration)

  Registration of the analyze_seq_numbers tool in list_tools(), including input schema definition.
  types.Tool( name="analyze_seq_numbers", description="分析号码连续出现概率（滑窗），返回理论值与实测值", inputSchema={ "type": "object", "properties": { "lottery_type": { "type": "string", "enum": ["双色球", "福彩3D", "七乐彩", "快乐8"], "description": "彩票类型" }, "periods": { "type": "integer", "minimum": 5, "maximum": 1000, "default": 100, "description": "分析期数" }, "sequence_length": { "type": "integer", "minimum": 1, "maximum": 10, "default": 2, "description": "连续期数" } }, "required": ["lottery_type"] } ),
• src/swlc_mcp/server.py:1393-1413 (handler)

  MCP tool handler in call_tool() that invokes the analysis service and formats the response as text.
  elif name == "analyze_seq_numbers": lottery_type = arguments.get("lottery_type") periods = arguments.get("periods", 100) sequence_length = arguments.get("sequence_length", 2) result = await lottery_service.analyze_seq_numbers( lottery_type=lottery_type, periods=periods, sequence_length=sequence_length, ) detail = result.get("counts", {}) text = ( f"{lottery_type} 连续出现分析（最近{result['periods_used']}期，连续{sequence_length}期）：\n\n" f"- 理论概率: {result['theoretical_prob']:.8f}\n" f"- 实测概率: {result['empirical_prob']:.8f}\n" f"- 计数: {detail.get('hits', 0)} / {detail.get('windows', 0)}\n" f"- 号码池: {result['pool_size']}，每期开出: {result['numbers_per_draw']}\n" f"- 最长连出分布: {result.get('max_run_distribution', {})}" ) return [types.TextContent(type="text", text=text)]
• src/swlc_mcp/server.py:890-983 (handler)

  Core implementation in SWLCService class: computes theoretical and empirical probabilities of numbers appearing consecutively over specified periods using sliding window analysis on historical data.
  async def analyze_seq_numbers( self, lottery_type: str, periods: int, sequence_length: int, ) -> Dict[str, Any]: """ 分析号码连续出现概率（滑窗） Args: lottery_type: 彩票类型（需在 lottery_codes 中） periods: 使用的期数窗口 sequence_length: 连续出现期数 Returns: dict: 理论值、实测值、计数等 """ if sequence_length < 1: raise ValueError("sequence_length must be >= 1") # 获取历史数据 results = await self.get_historical_data(lottery_type, periods) if not results: raise ValueError("未获取到历史数据") num_draws = len(results) if sequence_length > num_draws: raise ValueError("sequence_length 大于可用期数") # 抽取号码（仅主号码） rows: List[List[int]] = [] for r in results: try: rows.append([int(n) for n in r.numbers]) except Exception: # 如果存在非数字，直接跳过该期 continue if not rows: raise ValueError("历史数据格式异常，缺少号码") numbers_per_draw = len(rows[0]) # 不同彩票的号码池大小 pool_sizes = { "双色球": 33, "福彩3D": 10, "七乐彩": 30, "快乐8": 80, } pool_size = pool_sizes.get(lottery_type) if not pool_size: raise ValueError(f"不支持的彩票类型: {lottery_type}") if numbers_per_draw > pool_size: raise ValueError("每期号码数量大于号码池大小，数据异常") # 理论：单号在一期开出的概率 p_single = numbers_per_draw / pool_size theoretical = p_single ** sequence_length # 实测：滑窗计数 from collections import Counter balls = range(1, pool_size + 1) total_windows = (num_draws - sequence_length + 1) * pool_size hit_count = 0 max_run = {} for b in balls: # 滑窗连续 for i in range(num_draws - sequence_length + 1): if all(b in rows[i + j] for j in range(sequence_length)): hit_count += 1 # 最长连出 cur = longest = 0 for reds in rows: if b in reds: cur += 1 longest = max(longest, cur) else: cur = 0 max_run[b] = longest empirical = hit_count / total_windows if total_windows else 0 max_run_dist = Counter(max_run.values()) return { "lottery_type": lottery_type, "periods_used": num_draws, "sequence_length": sequence_length, "pool_size": pool_size, "numbers_per_draw": numbers_per_draw, "theoretical_prob": theoretical, "empirical_prob": empirical, "counts": { "hits": hit_count, "windows": total_windows, }, "max_run_distribution": dict(sorted(max_run_dist.items())), }