diff --git a/lottery_picker/data/SKILL.md b/lottery_picker/data/SKILL.md deleted file mode 100644 index cea3218..0000000 --- a/lottery_picker/data/SKILL.md +++ /dev/null @@ -1,379 +0,0 @@ ---- -name: lottery_picker -description: Use this skill when the user mentions lottery-related keywords such as "双色球", "大乐透", "彩票", "今日彩票", "机选", "买彩票", or asks for lottery number recommendations, analysis, or results. Triggers on any lottery-related query. The skill provides data collection, analysis, recommendation, and rollback iteration for SSQ (双色球) and DLT (大乐透). ---- - -# 彩票智能分析技能 (lottery_picker) - -## 概述 - -双色球和大乐透的智能号码分析与机选推荐工具。基于**6阶段数学建模分析系统**(泊松频次→负二项/伽马拟合→贝叶斯后验→K-Means聚类→ILP约束求解),提供数据采集、推理分析、机选推荐和回滚迭代完整流程。 - -## 触发关键词 - -双色球、大乐透、彩票、今日彩票、机选、买彩票 - -## 工作流程 - -当用户输入彩票关键词时,按以下步骤执行: - -### Step 1: 调用 full_report - -**双色球**(6阶段数学建模): -```bash -python model_full_report.py -``` - -**大乐透**(基础统计): -```bash -python lottery_tool.py --action full_report --lottery dlt --count 5 -``` - -### Step 2: 解析4块结构化输出 - -工具输出4块内容,按以下框架整理为 Markdown 格式返回给用户。 - -**输出原则(订阅产品标准)**: -- **数据真实** — 直接来自工具输出,不编造 -- **过程严谨** — 4块结构完整,逻辑清晰 -- **呈现简明** — 关键信息突出,去冗余 -- **界面美观** — Markdown表格+分隔线,视觉层次清晰 - ---- - -#### [1/4] 回滚迭代 - -| 指标 | 数值 | -|:---|:---| -| 当前版本号 | **v{N}** | -| 版本胜率 | X%(N期推荐 / M次盈利)| -| 累计中奖收益 | **RMB X.XX** | - ---- - -#### [2/4] 数据采集 - -| 项目 | 内容 | -|:---|:---| -| 采集范围 | 第X期 ~ 第Y期(近100期)| -| 上一期号 | **第X期** | -| 开奖日期 | YYYY-MM-DD | -| 开奖号码 | 红球/前区 **XX XX XX XX XX** · 蓝球/后区 **XX** | - ---- - -#### [3/4] 推理分析 - -| 分析版本 | 数据源 | -|:---|:---| -| **v{N} (6阶段数学建模)** | 500.com | - -**步骤1 · 频率分析(TOP10红球)** - -| 排名 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | -|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:| -| 号码 | **XX** | **XX** | **XX** | **XX** | **XX** | **XX** | **XX** | **XX** | **XX** | **XX** | -| 评分 | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | - -**步骤2 · 蓝球分析(TOP5)** - -| 排名 | 1 | 2 | 3 | 4 | 5 | -|:---:|:---:|:---:|:---:|:---:|:---:| -| 号码 | **XX** | **XX** | **XX** | **XX** | **XX** | -| 评分 | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | - -**步骤3 · 模式识别** - -| 奇偶比 | 大小比 | 和值区间 | -|:---:|:---:|:---:| -| X:X | X:X | [XX.X, XX.X] | - -**步骤4 · 权重体系(6阶段数学模型)** - -``` -Score = 0.5×贝叶斯后验 + 0.3×泊松频次 + 0.2×伽马遗漏 -方法链路: 泊松频次→负二项/伽马拟合→贝叶斯后验→K-Means→ILP约束 -``` - ---- - -#### [4/4] 机选号码 - -> 本次期号: **第X期** | 即将开奖: **YYYY-MM-DD (周X)** - -| # | 红球/前区 | 蓝球/后区 | 评分 | 奇偶 | 大小 | 和值 | -|:---:|:---|:---:|:---:|:---:|:---:|:---:| -| **①** | **XX XX XX XX XX** | **XX** | **X.XXXX** ⭐ | X:X | X:X | XXX | -| ② | XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | -| ③ | XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | -| ④ | XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | -| ⑤ | XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | - -### Step 3: 回滚迭代(开奖后) - -当用户告知开奖结果时: - -```bash -# 自动回滚(带收益记录) -python lottery_tool.py --action check --lottery {ssq|dlt} --earnings {金额} -``` - -- `--earnings` 参数:用户中奖后反馈的实际收益金额(>0表示盈利) -- 回滚后版本号自增,但**不影响当天已输出的推荐结果** - -## 核心设计原则 - -### 确定性输出 -- 相同输入保证相同输出 -- 种子公式: `MD5(固定种子 + 彩种 + 日期 + 分析版本号)` -- 版本号仅在数据源变化时递增(新期号到来),回滚迭代不改变版本号 - -### 结果变化因素(仅以下三者) -1. **数据源变了** — 开奖后采集到新数据 -2. **逻辑推理因素变了** — 分析算法/权重体系调整 -3. **分析步骤变了** — 代码执行流程修改 - -**版本迭代/回滚优化本身不应改变当天已输出的推荐结果。** - -### 输出标准(订阅产品级) -- **数据真实** — 直接来自工具输出,不编造 -- **过程严谨** — 4块结构完整,逻辑清晰 -- **呈现简明** — 关键信息突出,去冗余 -- **界面美观** — Markdown表格+分隔线,视觉层次清晰 - -### 数据源 -| 彩种 | 主数据源 | 备用数据源 | -|------|---------|-----------| -| 双色球 | 500.com HTML | 17500.cn API | -| 大乐透 | 体彩官方API | 无 | - -### 开奖周期 -- 双色球: 周二、四、日 21:15 -- 大乐透: 周一、三、六 20:25 - -## 输出示例 - -### 双色球输出模板 - -```markdown -# 🎱 双色球 · 第XXXXX期 - -> 今晚 **21:15** 开奖 · YYYY-MM-DD 周X - ---- - -## [1/4] 回滚迭代 - -| 指标 | 数值 | -|:---|:---| -| 当前版本号 | **v{N}** | -| 版本胜率 | X%(N期推荐 / M次盈利)| -| 累计中奖收益 | **RMB X.XX** | - ---- - -## [2/4] 数据采集 - -| 项目 | 内容 | -|:---|:---| -| 采集范围 | 第XXXXX期 ~ 第XXXXX期(近100期)| -| 上一期号 | **第XXXXX期** | -| 开奖日期 | YYYY-MM-DD | -| 开奖号码 | 红球 **XX XX XX XX XX XX** · 蓝球 **XX** | - ---- - -## [3/4] 推理分析 - -| 分析版本 | 数据源 | -|:---|:---| -| **v{N}** | 500.com | - -**步骤1 · 频率分析(TOP10红球)** - -| 排名 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | -|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:| -| 号码 | **XX** | **XX** | **XX** | **XX** | **XX** | **XX** | **XX** | **XX** | **XX** | **XX** | -| 评分 | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | - -**步骤2 · 蓝球分析(TOP5)** - -| 排名 | 1 | 2 | 3 | 4 | 5 | -|:---:|:---:|:---:|:---:|:---:|:---:| -| 号码 | **XX** | **XX** | **XX** | **XX** | **XX** | -| 评分 | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | - -**步骤3 · 模式识别** - -| 奇偶比 | 大小比 | 和值区间 | -|:---:|:---:|:---:| -| X:X | X:X | [XX.X, XX.X] | - -**步骤4 · 权重体系(6阶段数学模型)** - -``` -Score = 0.5×贝叶斯后验 + 0.3×泊松频次 + 0.2×伽马遗漏 -方法链路: 泊松频次→负二项/伽马拟合→贝叶斯后验→K-Means→ILP约束 -``` - ---- - -## [4/4] 机选号码 - -| # | 红球 | 蓝球 | 评分 | 奇偶 | 大小 | 和值 | -|:---:|:---|:---:|:---:|:---:|:---:|:---:| -| **①** | **XX XX XX XX XX XX** | **XX** | **X.XXXX** ⭐ | X:X | X:X | XXX | -| ② | XX XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | -| ③ | XX XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | -| ④ | XX XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | -| ⑤ | XX XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | - ---- - -> 📌 **数据真实** · 过程严谨 · 呈现简明 · 界面美观 -> -> 开奖后把实际号码发我,帮你回滚迭代校正精度 🍀 -``` - -### 双色球输出模板 - -```markdown -# 🎱 双色球 · 第XXXXX期 - -> 今晚 **21:15** 开奖 · YYYY-MM-DD 周X - ---- - -## [1/4] 回滚迭代 - -| 指标 | 数值 | -|:---|:---| -| 当前版本号 | **v{N}** | -| 版本胜率 | X%(N期推荐 / M次盈利)| -| 累计中奖收益 | **RMB X.XX** | - ---- - -## [2/4] 数据采集 - -| 项目 | 内容 | -|:---|:---| -| 采集范围 | 第XXXXX期 ~ 第XXXXX期(近100期)| -| 上一期号 | **第XXXXX期** | -| 开奖日期 | YYYY-MM-DD | -| 开奖号码 | 红球 **XX XX XX XX XX XX** · 蓝球 **XX** | - ---- - -## [3/4] 推理分析 - -| 分析版本 | 数据源 | -|:---|:---| -| **v{N}** | 500.com | - -**步骤1 · 频率分析(TOP10红球)** - -| 排名 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | -|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:| -| 号码 | **XX** | **XX** | ... | | | | | | | | -| 评分 | X.XXX | X.XXX | ... | | | | | | | | - -**步骤2 · 蓝球分析(TOP5)** - -| 排名 | 1 | 2 | 3 | 4 | 5 | -|:---:|:---:|:---:|:---:|:---:|:---:| -| 号码 | **XX** | **XX** | **XX** | **XX** | **XX** | -| 评分 | X.XXX | X.XXX | X.XXX | X.XXX | X.XXX | - -**步骤3 · 模式识别** - -| 奇偶比 | 大小比 | 和值区间 | -|:---:|:---:|:---:| -| X:X | X:X | [XX.X, XX.X] | - -**步骤4 · 权重体系(6阶段数学模型)** - -``` -Score = 0.5×贝叶斯后验 + 0.3×泊松频次 + 0.2×伽马遗漏 -方法链路: 泊松频次→负二项/伽马拟合→贝叶斯后验→K-Means→ILP约束 -``` - ---- - -## [4/4] 机选号码 - -| # | 红球 | 蓝球 | 评分 | 奇偶 | 大小 | 和值 | -|:---:|:---|:---:|:---:|:---:|:---:|:---:| -| **①** | **XX XX XX XX XX XX** | **XX** | **X.XXXX** ⭐ | X:X | X:X | XXX | -| ② | XX XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | -| ③ | XX XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | -| ④ | XX XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | -| ⑤ | XX XX XX XX XX XX | XX | X.XXXX | X:X | X:X | XXX | - ---- - -> 📌 **数据真实** · 过程严谨 · 呈现简明 · 界面美观 -> -> 开奖后把实际号码发我,帮你回滚迭代校正精度 🍀 -``` - -### 确定性输出 -- 相同输入保证相同输出 -- 种子公式: `MD5(固定种子 + 彩种 + 日期 + 分析版本号)` -- 版本号仅在数据源变化时递增(新期号到来),回滚迭代不改变版本号 - -### 结果变化因素(仅以下三者) -1. **数据源变了** — 开奖后采集到新数据 -2. **逻辑推理因素变了** — 分析算法/权重体系调整 -3. **分析步骤变了** — 代码执行流程修改 - -**版本迭代/回滚优化本身不应改变当天已输出的推荐结果。** - -### 数据源 -| 彩种 | 主数据源 | 备用数据源 | -|------|---------|-----------| -| 双色球 | 500.com HTML | 17500.cn API | -| 大乐透 | 体彩官方API | 无 | - -### 开奖周期 -- 双色球: 周二、四、日 21:15 -- 大乐透: 周一、三、六 20:25 - -## 收益跟踪 - -`feedback.json` 中记录: -- `profitable_wins`: 产生实际盈利的中奖次数 -- `total_earnings`: 累计中奖收益金额(RMB) -- 版本胜率 = profitable_wins / total_recs * 100% - -用户中奖后告知金额,通过 `--earnings` 参数记录。 - -## 文件结构 - -``` -skills/lottery_skill/ -├── lottery_tool.py # 主工具 -├── lottery_mcp_server.py # MCP Server 封装 -├── SKILL.md # 本文件 -├── data/ # 数据持久化目录 -│ ├── config.json # 配置(种子、迭代计数) -│ ├── ssq_history.json # 双色球历史数据 -│ ├── dlt_history.json # 大乐透历史数据 -│ ├── analysis_ssq.json # 双色球分析结果 -│ ├── analysis_dlt.json # 大乐透分析结果 -│ └── feedback.json # 回滚反馈(含收益) -└── templates/ # 模板文件 -``` - -## CLI 参数 - -| 参数 | 值 | 说明 | -|------|-----|------| -| `--action` | full_report | 4块结构化输出(推荐) | -| `--action` | recommend | 仅输出推荐号码 | -| `--action` | analyze | 仅输出分析报告 | -| `--action` | fetch | 仅采集数据 | -| `--action` | check | 回滚迭代 | -| `--lottery` | ssq / dlt / all | 彩种 | -| `--count` | 5 | 推荐组数 | -| `--earnings` | 0.0 | 中奖收益(check专用) | diff --git a/lottery_picker/data/__init__.py b/lottery_picker/data/__init__.py deleted file mode 100644 index ce26f28..0000000 --- a/lottery_picker/data/__init__.py +++ /dev/null @@ -1,2 +0,0 @@ -"""彩票智能分析技能包""" -__version__ = "1.0.0" diff --git a/lottery_picker/data/blueball_optimizer.py b/lottery_picker/data/blueball_optimizer.py deleted file mode 100644 index c1ba49d..0000000 --- a/lottery_picker/data/blueball_optimizer.py +++ /dev/null @@ -1,444 +0,0 @@ -""" -蓝球算法专项优化 -目标:将蓝球命中率从6%提升至15%+ -技术:深度学习 + 强化学习 + 组合优化 -""" - -import numpy as np -import random -from collections import Counter, defaultdict -from typing import List, Dict, Tuple -import math - - -class BlueBallOptimizer: - """蓝球算法优化器""" - - def __init__(self, lottery_type: str, persistence): - self.lt = lottery_type - self.p = persistence - self.blue_max = 16 if lottery_type == "ssq" else 12 - - # 历史数据 - self.training_history = [] - - # 模型参数 - self.markov_chain = defaultdict(lambda: defaultdict(float)) - self.pattern_memory = [] # 记忆最近模式 - self.reward_history = [] # 奖励历史 - self.learning_rate = 0.1 - - print(f"[蓝球优化] 初始化 {lottery_type} (范围: 1-{self.blue_max})") - - def load_blue_data(self) -> List[List[int]]: - """加载蓝球历史数据""" - data = self.p.history(self.lt) - if not data: - return [] - - blue_sequences = [] - for record in data[:100]: # 最近100期 - blues = record["blue"] if isinstance(record["blue"], list) else [int(record["blue"])] - blue_sequences.append(blues) - - return blue_sequences - - def analyze_patterns(self, sequences: List[List[int]]) -> Dict: - """分析蓝球模式""" - if len(sequences) < 20: - return {"status": "insufficient_data"} - - # 1. 基础频率分析 - all_blues = [num for seq in sequences for num in seq] - freq_counter = Counter(all_blues) - total = len(all_blues) - - # 2. 奇偶模式 - odd_even_patterns = [] - for seq in sequences: - # 统计奇偶组合 - odd_count = sum(1 for n in seq if n % 2 == 1) - pattern = f"O{odd_count}_E{len(seq)-odd_count}" - odd_even_patterns.append(pattern) - - # 3. 大小分区 - size_patterns = [] - mid_point = self.blue_max // 2 - for seq in sequences: - big_count = sum(1 for n in seq if n > mid_point) - pattern = f"B{big_count}_S{len(seq)-big_count}" - size_patterns.append(pattern) - - # 4. 热号冷号分析 - hot_numbers = [n for n, count in freq_counter.most_common(5)] - - # 冷号(最近20期未出) - recent_numbers = [num for seq in sequences[-20:] for num in seq] - cold_numbers = [n for n in range(1, self.blue_max+1) if n not in recent_numbers] - - # 5. 间隔分析 - interval_analysis = self._analyze_intervals(sequences) - - return { - "status": "analyzed", - "total_samples": len(sequences), - "frequency_dist": dict(freq_counter.most_common()), - "hot_numbers": hot_numbers, - "cold_numbers": cold_numbers[:5], - "odd_even_patterns": Counter(odd_even_patterns), - "size_patterns": Counter(size_patterns), - "interval_stats": interval_analysis, - "most_likely_pattern": Counter(odd_even_patterns).most_common(1)[0][0] if odd_even_patterns else "" - } - - def _analyze_intervals(self, sequences: List[List[int]]) -> Dict: - """分析间隔模式""" - last_seen = {n: -1 for n in range(1, self.blue_max+1)} - intervals = defaultdict(list) - - for i, seq in enumerate(sequences): - for num in seq: - if last_seen[num] != -1: - interval = i - last_seen[num] - intervals[num].append(interval) - last_seen[num] = i - - # 统计间隔特征 - interval_stats = {} - for num in range(1, self.blue_max+1): - if num in intervals and intervals[num]: - ivals = intervals[num] - interval_stats[num] = { - "avg_interval": np.mean(ivals), - "min_interval": min(ivals), - "max_interval": max(ivals), - "std_interval": np.std(ivals) if len(ivals) > 1 else 0 - } - else: - interval_stats[num] = {"avg_interval": 999, "min_interval": 999, "max_interval": 999} - - return interval_stats - - def build_markov_chain(self, sequences: List[List[int]]): - """构建高阶马尔科夫链(考虑更多历史状态)""" - if len(sequences) < 30: - return - - # 二阶马尔科夫链 - for i in range(2, len(sequences)): - prev1 = tuple(sorted(sequences[i-2])) - prev2 = tuple(sorted(sequences[i-1])) - curr = tuple(sorted(sequences[i])) - - # 状态编码 - state = (prev1, prev2) - self.markov_chain[state][curr] += 1 - - # 转换为概率 - for state, transitions in self.markov_chain.items(): - total = sum(transitions.values()) - for key in transitions: - transitions[key] /= total - - print(f"[蓝球优化] 构建马尔科夫链: {len(self.markov_chain)}个状态") - - def train_deep_model(self, sequences: List[List[int]]): - """训练深度学习模型""" - if len(sequences) < 40: - print("[蓝球优化] 数据不足,跳过深度学习训练") - return - - # 特征提取 - X = [] # 特征 - y = [] # 目标(下一期蓝球) - - for i in range(1, len(sequences)): - # 当前期特征 - current = sequences[i-1] - - # 历史特征 - historical = sequences[max(0, i-5):i-1] - - # 构建特征向量 - features = self._extract_features(current, historical) - - # 目标:下一期蓝球 - target = sequences[i] - - X.append(features) - y.append(target) - - self.training_history = list(zip(X, y)) - print(f"[蓝球优化] 深度学习训练: {len(X)}个样本") - - def _extract_features(self, current: List[int], historical: List[List[int]]) -> List[float]: - """提取特征向量""" - features = [] - - # 1. 当前期特征 - features.extend(current) - features.extend([0] * (2 - len(current))) # 填充 - - # 2. 奇偶比例 - odd_ratio = sum(1 for n in current if n % 2 == 1) / len(current) - features.append(odd_ratio) - - # 3. 大小比例 - mid = self.blue_max // 2 - big_ratio = sum(1 for n in current if n > mid) / len(current) - features.append(big_ratio) - - # 4. 历史平均值 - if historical: - all_nums = [num for seq in historical for num in seq] - features.append(np.mean(all_nums) / self.blue_max if all_nums else 0.5) - else: - features.append(0.5) - - # 5. 间隔特征 - if len(current) > 0: - # 数字间距 - if len(current) > 1: - spacing = max(current) - min(current) - features.append(spacing / self.blue_max) - else: - features.append(0) - else: - features.append(0) - - return features - - def predict_with_deep_learning(self, historical: List[List[int]], current: List[int], top_k=5) -> List[int]: - """深度学习预测""" - if not self.training_history: - return self._fallback_prediction(historical, current, top_k) - - # 提取特征 - features = self._extract_features(current, historical) - - # 简单最近邻预测 - predictions = [] - feature_arr = np.array(features) - - for train_X, train_y in self.training_history[-50:]: - # 计算相似度 - sim = np.dot(feature_arr, np.array(train_X)) / ( - np.linalg.norm(feature_arr) * np.linalg.norm(train_X) + 1e-8) - - if sim > 0.7: # 相似度阈值 - predictions.extend(train_y) - - if predictions: - # 统计出现频率 - freq = Counter(predictions) - top_numbers = [num for num, _ in freq.most_common(top_k)] - return sorted(top_numbers) - - return self._fallback_prediction(historical, current, top_k) - - def _fallback_prediction(self, historical: List[List[int]], current: List[int], top_k=5) -> List[int]: - """备用预测算法""" - # 合并历史数据 - all_historical = [num for seq in historical for num in seq] - - if not all_historical: - # 随机选择 - return random.sample(range(1, self.blue_max+1), min(top_k, self.blue_max)) - - # 频率分析 - freq_counter = Counter(all_historical) - - # 热号 + 冷号平衡策略 - hot_numbers = [num for num, _ in freq_counter.most_common(10)] - - # 寻找冷号(长期未出) - recent_history = [num for seq in historical[-20:] for num in seq] if len(historical) >= 20 else [] - cold_numbers = [n for n in range(1, self.blue_max+1) if n not in recent_history] - - # 组合策略:70%热号 + 30%冷号 - hot_count = int(top_k * 0.7) - cold_count = top_k - hot_count - - selected = hot_numbers[:hot_count] - if cold_numbers and cold_count > 0: - selected.extend(random.sample(cold_numbers, min(cold_count, len(cold_numbers)))) - - # 如果不够,补随机 - while len(selected) < top_k: - candidate = random.randint(1, self.blue_max) - if candidate not in selected: - selected.append(candidate) - - return sorted(selected[:top_k]) - - def predict_with_markov(self, historical: List[List[int]]) -> List[int]: - """马尔科夫链预测""" - if len(historical) < 3 or not self.markov_chain: - return [] - - # 获取最近状态 - state = ( - tuple(sorted(historical[-2])) if len(historical) >= 2 else (), - tuple(sorted(historical[-1])) - ) - - if state in self.markov_chain: - # 获取最可能的下一状态 - transitions = self.markov_chain[state] - if transitions: - # 按概率降序排序 - sorted_transitions = sorted(transitions.items(), key=lambda x: x[1], reverse=True) - if sorted_transitions: - # 返回最可能的状态 - return list(sorted_transitions[0][0]) - - return [] - - def hybrid_prediction(self, historical: List[List[int]], current: List[int], - method_weights=[0.3, 0.3, 0.4]) -> List[int]: - """混合预测(加权组合)""" - predictions = [] - - # 方法1:深度学习预测 - dl_pred = self.predict_with_deep_learning(historical, current, top_k=3) - if dl_pred: - predictions.extend([(n, method_weights[0]) for n in dl_pred]) - - # 方法2:马尔科夫预测 - mk_pred = self.predict_with_markov(historical) - if mk_pred: - predictions.extend([(n, method_weights[1]) for n in mk_pred]) - - # 方法3:频率分析 - freq_pred = self._fallback_prediction(historical, current, top_k=3) - predictions.extend([(n, method_weights[2]) for n in freq_pred]) - - if not predictions: - return [] - - # 加权选择 - weighted_scores = {} - for num, weight in predictions: - if num in weighted_scores: - weighted_scores[num] += weight - else: - weighted_scores[num] = weight - - # 选择得分最高的 - sorted_numbers = sorted(weighted_scores.items(), key=lambda x: x[1], reverse=True) - - # 确定蓝球数量 - target_count = 1 if self.lt == "ssq" else 2 - return [num for num, _ in sorted_numbers[:target_count]] - - def learn_from_feedback(self, predicted: List[int], actual: List[int], reward: float): - """从反馈中学习""" - # 记录奖励 - self.reward_history.append({ - "predicted": predicted, - "actual": actual, - "reward": reward, - "timestamp": len(self.reward_history) - }) - - # 调整学习率 (优化速率) - if reward > 0: - self.learning_rate *= 1.05 # 成功则加快学习 - else: - self.learning_rate *= 0.95 # 失败则减慢 - - self.learning_rate = max(0.01, min(0.5, self.learning_rate)) - - # 更新模式记忆 - hit_numbers = set(predicted) & set(actual) - miss_numbers = set(predicted) - set(actual) - - pattern = { - "hit_numbers": list(hit_numbers), - "miss_numbers": list(miss_numbers), - "reward": reward, - "total_hits": len(hit_numbers) - } - - self.pattern_memory.append(pattern) - if len(self.pattern_memory) > 50: - self.pattern_memory = self.pattern_memory[-50:] - - return { - "learning_rate": self.learning_rate, - "reward_history_len": len(self.reward_history), - "avg_recent_reward": np.mean([r["reward"] for r in self.reward_history[-10:]]) if self.reward_history else 0 - } - - -def test_blue_ball_optimization(): - """测试蓝球优化""" - print("="*60) - print("蓝球算法优化测试") - print("="*60) - - from lottery_tool import Persistence - - p = Persistence() - - for lt in ["ssq", "dlt"]: - print(f"\n--- {'双色球' if lt == 'ssq' else '大乐透'} ---") - - optimizer = BlueBallOptimizer(lt, p) - sequences = optimizer.load_blue_data() - - if len(sequences) < 30: - print(f" 数据不足: {len(sequences)}期 (<30期)") - continue - - print(f" 数据量: {len(sequences)}期") - - # 分析模式 - analysis = optimizer.analyze_patterns(sequences) - if analysis["status"] == "analyzed": - print(f" 热号: {analysis['hot_numbers']}") - print(f" 冷号: {analysis['cold_numbers']}") - print(f" 最常见模式: {analysis['most_likely_pattern']}") - - # 训练模型 - optimizer.build_markov_chain(sequences) - optimizer.train_deep_model(sequences) - - # 预测测试(最后10期) - test_size = min(10, len(sequences) - 5) - hits = 0 - total = 0 - - for i in range(len(sequences)-test_size, len(sequences)-1): - historical = sequences[:i] - current = sequences[i] - - # 混合预测 - prediction = optimizer.hybrid_prediction(historical, current) - - # 实际下一期 - next_actual = sequences[i+1] - - # 计算命中 - hit_count = len(set(prediction) & set(next_actual)) - hits += hit_count - total += len(prediction) - - if i >= len(sequences) - 5: # 显示最后几期 - pred_str = " ".join(f"{n:02d}" for n in prediction) - act_str = " ".join(f"{n:02d}" for n in next_actual) - hit_marker = "✓" if hit_count > 0 else "✗" - print(f" 期{i+1}: 预测 [{pred_str}] | 实际 [{act_str}] | 命中{hit_count} {hit_marker}") - - accuracy = hits / total * 100 if total > 0 else 0 - print(f" 测试准确率: {accuracy:.2f}% ({hits}/{total})") - - if accuracy > 0: - # 与随机概率对比 - random_chance = (1 / (16 if lt == "ssq" else 12)) * 100 if lt == "ssq" else (2 / 12) * 100 - improvement = accuracy - random_chance - print(f" 相比随机提升: {improvement:+.2f}% (随机概率: {random_chance:.2f}%)") - - -if __name__ == "__main__": - test_blue_ball_optimization() \ No newline at end of file diff --git a/lottery_picker/data/enhanced_lottery.py b/lottery_picker/data/enhanced_lottery.py deleted file mode 100644 index 0bf8034..0000000 --- a/lottery_picker/data/enhanced_lottery.py +++ /dev/null @@ -1,758 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -增强版彩票智能分析系统 v2.0 -=============================== -核心升级: - 1. 采集近200期数据(原100期) - 2. AI逐期回滚推演:从第200期开始->模拟分析机选->比对实开->回滚优化 - 3. 方法论自优化:每轮回滚后总结方法论并迭代升级 - 4. 目标:胜出率(概率占比) >= 51% - -用法: - python enhanced_lottery.py --action full_report --lottery ssq --count 5 - python enhanced_lottery.py --action rollback --lottery ssq - python enhanced_lottery.py --action self_evolve --lottery ssq --iterations 10 -""" - -import json, os, sys, math, random, copy, datetime -from collections import Counter - -BASE_DIR = os.path.dirname(os.path.abspath(__file__)) -DATA_DIR = os.path.join(BASE_DIR, "data") -os.makedirs(DATA_DIR, exist_ok=True) - -LOTTERY_RULES = { - "ssq": { - "name": "双色球", - "red_count": 6, "red_max": 33, "blue_count": 1, "blue_max": 16, - "draw_days": [2, 4, 7], "draw_time": "21:15", - }, - "dlt": { - "name": "大乐透", - "red_count": 5, "red_max": 35, "blue_count": 2, "blue_max": 12, - "draw_days": [1, 3, 6], "draw_time": "20:25", - } -} - -DEFAULT_WEIGHTS = { - "frequency": 0.25, - "hot_number": 0.20, - "missing": 0.15, - "tail": 0.10, - "zone": 0.08, - "odd_even": 0.06, - "big_small": 0.05, - "sum_value": 0.04, - "probability": 0.04, - "consecutive": 0.03, -} - - -# ============================================================ -# 增强版分析引擎(200期 + 10维度深度分析) -# ============================================================ - -def _normalize_rec(rec): - """标准化记录,兼容新旧格式(blue vs blues)""" - if "blues" not in rec or not rec["blues"]: - if "blue" in rec: - rec["blues"] = rec["blue"] if isinstance(rec["blue"], list) else [rec["blue"]] - return rec - - -def _get_blues(rec): - """获取blue值,兼容新旧格式""" - if "blues" in rec and rec["blues"]: - raw = rec["blues"] - elif "blue" in rec: - raw = rec["blue"] - else: - raw = [] - return raw if isinstance(raw, list) else [raw] - - -class EnhancedAnalyzer: - """使用近200期数据,10维度权重可调,输出结构化分析结果""" - - def __init__(self, data, lt, weights=None): - self.lt = lt - self.rule = LOTTERY_RULES[lt] - # 标准化并取近200期 - normed = [_normalize_rec(d) for d in data] - self.data = sorted(normed, key=lambda x: int(x["issue"]), reverse=True)[:200] - self.weights = weights if weights else copy.deepcopy(DEFAULT_WEIGHTS) - self.rmax = self.rule["red_max"] - self.bmax = self.rule["blue_max"] - self.rcnt = self.rule["red_count"] - self.bcnt = self.rule["blue_count"] - self.mid = self.rmax / 2 - - def analyze(self): - """完整分析,返回结构化分析结果""" - if not self.data: - return None - - # 1. 频率分析 - all_r, all_b = [], [] - for rec in self.data: - all_r.extend(int(x) for x in rec["reds"]) - bl = _get_blues(rec) - all_b.extend(int(x) for x in bl) - - rf = Counter(all_r) - bf = Counter(all_b) - max_rf = max(rf.values()) or 1 - max_bf = max(bf.values()) or 1 - rf_score = {i: rf.get(i, 0) / max_rf for i in range(1, self.rmax + 1)} - bf_score = {i: bf.get(i, 0) / max_bf for i in range(1, self.bmax + 1)} - - # 2. 热号分析(最近20期) - hot = self.data[:20] - hot_r = [int(x) for r in hot for x in r["reds"]] - hot_rf = Counter(hot_r) - max_hr = max(hot_rf.values()) or 1 - hot_score = {i: hot_rf.get(i, 0) / max_hr for i in range(1, self.rmax + 1)} - - # 3. 遗漏值分析 - r_miss = {} - for n in range(1, self.rmax + 1): - cnt = 0 - for r in self.data: - if n in [int(x) for x in r["reds"]]: - break - cnt += 1 - r_miss[n] = cnt - max_rm = max(r_miss.values()) or 1 - rm_score = {n: math.sin(m / max_rm * math.pi) * 0.8 + 0.2 for n, m in r_miss.items()} - - b_miss = {} - for n in range(1, self.bmax + 1): - cnt = 0 - for r in self.data: - ns = [int(x) for x in _get_blues(r)] - if n in ns: - break - cnt += 1 - b_miss[n] = cnt - max_bm = max(b_miss.values()) or 1 - bm_score = {n: math.sin(m / max_bm * math.pi) * 0.8 + 0.2 for n, m in b_miss.items()} - - # 4. 奇偶模式 - oe_counter = Counter() - for rec in self.data: - ns = [int(x) for x in rec["reds"]] - odd_count = sum(1 for n in ns if n % 2 == 1) - oe_counter[f"{odd_count}:{self.rcnt - odd_count}"] += 1 - most_oe = oe_counter.most_common(1)[0][0] if oe_counter else "3:3" - - # 5. 大小模式 - bs_counter = Counter() - for rec in self.data: - ns = [int(x) for x in rec["reds"]] - big_count = sum(1 for n in ns if n > self.mid) - bs_counter[f"{big_count}:{self.rcnt - big_count}"] += 1 - most_bs = bs_counter.most_common(1)[0][0] if bs_counter else "3:3" - - # 6. 和值分析 - sums = [sum(int(x) for x in r["reds"]) for r in self.data] - avg_sum = sum(sums) / len(sums) if sums else 0 - std_sum = math.sqrt(sum((s - avg_sum) ** 2 for s in sums) / len(sums)) if sums else 0 - - # 7. 尾数分布 - tail_counter = Counter() - for rec in self.data: - for ns in rec["reds"]: - tail_counter[int(ns) % 10] += 1 - max_tf = max(tail_counter.values()) or 1 - tail_score = {t: tail_counter.get(t, 0) / max_tf for t in range(10)} - - # 8. 区间分布 - zones = [(1, 11), (12, 22), (23, self.rmax)] if self.lt == "ssq" else [(1, 12), (13, 24), (25, self.rmax)] - zone_balance = [] - for zs, ze in zones: - count = sum(1 for r in self.data for n in [int(x) for x in r["reds"]] if zs <= n <= ze) - expected = len(self.data) * self.rcnt / len(zones) - zone_balance.append(count / expected if expected else 1.0) - - # 9. 连号分析 - consec_counter = Counter() - for rec in self.data: - ns = sorted(int(x) for x in rec["reds"]) - consec = sum(1 for i in range(len(ns) - 1) if ns[i + 1] - ns[i] == 1) - consec_counter[consec] += 1 - most_consec = consec_counter.most_common(1)[0][0] if consec_counter else 0 - - # 10. 综合评分 - r_comp = {} - for n in range(1, self.rmax + 1): - s = rf_score.get(n, 0) * self.weights["frequency"] - s += hot_score.get(n, 0) * self.weights["hot_number"] - s += rm_score.get(n, 0.5) * self.weights["missing"] - s += tail_score.get(n % 10, 0.5) * self.weights["tail"] - for i, (zs, ze) in enumerate(zones): - if zs <= n <= ze and zone_balance[i] < 0.85: - s += 0.10 - break - r_comp[n] = round(s, 4) - - # 蓝球复合评分 - recent_blue = [] - for r in self.data[:20]: - recent_blue.extend(int(x) for x in (r["blues"] if isinstance(r["blues"], list) else [r["blues"]])) - rbf = Counter(recent_blue) - max_rbf = max(rbf.values()) or 1 - b_comp = {} - for n in range(1, self.bmax + 1): - s = bf_score.get(n, 0) * 0.35 + bm_score.get(n, 0.5) * 0.40 + (rbf.get(n, 0) / max_rbf) * 0.25 - b_comp[n] = round(s, 4) - - result = { - "data_range": { - "total": len(self.data), - "start": self.data[-1]["issue"], - "end": self.data[0]["issue"] - }, - "red_composite": r_comp, - "blue_composite": b_comp, - "patterns": { - "most_oe": most_oe, - "most_bs": most_bs, - "avg_sum": round(avg_sum, 1), - "std_sum": round(std_sum, 1), - "sum_range": [round(avg_sum - std_sum, 1), round(avg_sum + std_sum, 1)], - "zone_balance": {f"Z{i+1}": round(zb, 3) for i, zb in enumerate(zone_balance)}, - "most_consec": most_consec, - }, - "weights_used": copy.deepcopy(self.weights), - } - return result - - def recommend(self, analysis, count=5): - """基于分析结果生成推荐号码""" - if not analysis: - return [] - - r_comp = analysis["red_composite"] - b_comp = analysis["blue_composite"] - pat = analysis["patterns"] - - sorted_r = sorted(r_comp.keys(), key=lambda n: r_comp[n], reverse=True) - sorted_b = sorted(b_comp.keys(), key=lambda n: b_comp[n], reverse=True) - - target_oe = [int(x) for x in pat["most_oe"].split(":")] - target_bs = [int(x) for x in pat["most_bs"].split(":")] - sum_min, sum_max = pat["sum_range"] - - seed = int(datetime.datetime.now().timestamp() * 1000) % 1000000 - rng = random.Random(seed) - - results = [] - for _ in range(count * 3): - top_r = sorted_r[:20] - weights_r = [max(r_comp[n], 0) for n in top_r] - total_w = sum(weights_r) - if total_w <= 0: - weights_r = [1] * len(top_r) - total_w = len(top_r) - probs_r = [w / total_w for w in weights_r] - - reds = set() - attempts = 0 - while len(reds) < self.rcnt and attempts < 100: - n = rng.choices(top_r, weights=probs_r, k=1)[0] - reds.add(n) - attempts += 1 - if len(reds) < self.rcnt: - remaining = [n for n in range(1, self.rmax + 1) if n not in reds] - rng.shuffle(remaining) - reds.update(remaining[:self.rcnt - len(reds)]) - - reds = sorted(reds) - - top_b = sorted_b[:8] - weights_b = [max(b_comp[n], 0) for n in top_b] - total_bw = sum(weights_b) or 1 - probs_b = [w / total_bw for w in weights_b] - if self.bcnt > 1: - blues = list(rng.choices(top_b, weights=probs_b, k=self.bcnt)) - # 确保后区号码不重复 - while len(set(blues)) < len(blues): - for j in range(len(blues)): - if blues.count(blues[j]) > 1: - remaining = [n for n in range(1, self.bmax + 1) if n not in blues[:j] + blues[j+1:]] - blues[j] = rng.choice(remaining) - else: - blues = list(rng.choices(top_b, weights=probs_b, k=self.bcnt)) - - score = sum(r_comp.get(n, 0) for n in reds) / self.rcnt - score += sum(b_comp.get(n, 0) for n in blues) / self.bcnt - score /= 2 - - oe = sum(1 for n in reds if n % 2 == 1) - bs = sum(1 for n in reds if n > self.mid) - total_sum = sum(reds) - - bonus = 0 - if oe == target_oe[0]: - bonus += 0.05 - if bs == target_bs[0]: - bonus += 0.05 - if sum_min <= total_sum <= sum_max: - bonus += 0.05 - - results.append({ - "reds": reds, "blues": blues, - "score": round(score + bonus, 4), - "oe": f"{oe}:{self.rcnt - oe}", - "bs": f"{bs}:{self.rcnt - bs}", - "sum": total_sum, - }) - - seen = set() - unique = [] - for r in sorted(results, key=lambda x: x["score"], reverse=True): - key = tuple(r["reds"]) + tuple(r["blues"]) - if key not in seen: - seen.add(key) - unique.append(r) - - return unique[:count] - - -# ============================================================ -# AI 逐期回滚推演引擎 -# ============================================================ - -class AIRollbackEngine: - """ - AI逐期回滚推演: - 从第200期开始 -> 使用当时可用的历史数据进行分析 -> 生成推荐 - -> 与实开号码比对 -> 记录命中率 -> 自动调整权重 - -> 进入下一期... 直至最新一期 - """ - - def __init__(self, lt, data, initial_weights=None): - self.lt = lt - self.rule = LOTTERY_RULES[lt] - # 标准化所有记录(兼容 blue/blues 混用) - normed = [_normalize_rec(d) for d in data] - self.all_data = sorted(normed, key=lambda x: int(x["issue"])) # 从旧到新 - self.weights = initial_weights if initial_weights else copy.deepcopy(DEFAULT_WEIGHTS) - self.rollback_history = [] - self.methodology_log = [] - - def run(self): - """执行完整回滚推演:从第200期开始逐期推演至最新一期""" - total_periods = len(self.all_data) - if total_periods < 200: - return {"error": f"数据不足200期,当前仅{total_periods}期", "available": total_periods} - - start_idx = 199 # 第200期(0-based索引) - total_simulations = 0 - total_red_hits = 0 - total_blue_hits = 0 - possible_red = 0 - possible_blue = 0 - methodology_snapshots = [] - - print(f"{'='*60}") - print(f" AI逐期回滚推演引擎 - {self.rule['name']}") - print(f" 数据总量: {total_periods}期 | 推演范围: 第200期 ~ 第{total_periods}期") - print(f" 推演总轮次: {total_periods - start_idx} 期") - print(f"{'='*60}") - - for i in range(start_idx, total_periods): - current_issue = self.all_data[i]["issue"] - historical = self.all_data[:i] - - actual = self.all_data[i] - actual_reds = [int(x) for x in actual["reds"]] - actual_blues = [int(x) for x in _get_blues(actual)] - - analyzer = EnhancedAnalyzer(historical, self.lt, self.weights) - analysis = analyzer.analyze() - if not analysis: - continue - - recommendations = analyzer.recommend(analysis, count=5) - if not recommendations: - continue - - best_pick = recommendations[0] - pred_reds = best_pick["reds"] - pred_blues = best_pick["blues"] - - red_hit = len(set(pred_reds) & set(actual_reds)) - blue_hit = len(set(pred_blues) & set(actual_blues)) - - total_red_hits += red_hit - total_blue_hits += blue_hit - possible_red += self.rule["red_count"] - possible_blue += self.rule["blue_count"] - total_simulations += 1 - - self.rollback_history.append({ - "issue": current_issue, - "iteration": total_simulations, - "pred_reds": pred_reds, - "pred_blues": pred_blues, - "actual_reds": actual_reds, - "actual_blues": actual_blues, - "red_hit": red_hit, - "blue_hit": blue_hit, - "score": best_pick["score"], - "weights_at_time": copy.deepcopy(self.weights), - }) - - self._optimize_weights(red_hit, blue_hit, pred_reds, actual_reds) - - if total_simulations % 5 == 0: - red_acc = total_red_hits / possible_red * 100 if possible_red else 0 - blue_acc = total_blue_hits / possible_blue * 100 if possible_blue else 0 - methodology_snapshots.append({ - "iteration": total_simulations, - "issue": current_issue, - "weights": copy.deepcopy(self.weights), - "red_accuracy": round(red_acc, 2), - "blue_accuracy": round(blue_acc, 2), - }) - - if total_simulations % 10 == 0 or total_simulations == 1: - red_acc = total_red_hits / possible_red * 100 if possible_red else 0 - blue_acc = total_blue_hits / possible_blue * 100 if possible_blue else 0 - overall_acc = (total_red_hits + total_blue_hits) / (possible_red + possible_blue) * 100 - print(f" [推演 {total_simulations:3d}] 期{current_issue} " - f"红{red_hit}/{self.rule['red_count']} 蓝{blue_hit}/{self.rule['blue_count']} " - f"| 累计 红{red_acc:.1f}% 蓝{blue_acc:.1f}% 综合{overall_acc:.1f}%") - - # 最终统计 - avg_red = total_red_hits / possible_red * 100 if possible_red else 0 - avg_blue = total_blue_hits / possible_blue * 100 if possible_blue else 0 - overall_hits = total_red_hits + total_blue_hits - overall_possible = possible_red + possible_blue - overall_acc = overall_hits / overall_possible * 100 if overall_possible else 0 - win_probability = overall_acc - - hit_dist = {0: 0, 1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0} - for entry in self.rollback_history: - total_hit = entry["red_hit"] + entry["blue_hit"] - if total_hit in hit_dist: - hit_dist[total_hit] += 1 - - result = { - "lottery": self.lt, - "total_simulations": total_simulations, - "start_issue": self.all_data[start_idx]["issue"], - "end_issue": self.all_data[-1]["issue"], - "final_weights": self.weights, - "accuracy": { - "red_accuracy": round(avg_red, 2), - "blue_accuracy": round(avg_blue, 2), - "overall_accuracy": round(overall_acc, 2), - "win_probability": round(win_probability, 2), - }, - "hits": { - "total_red_hits": total_red_hits, - "total_blue_hits": total_blue_hits, - "possible_red": possible_red, - "possible_blue": possible_blue, - }, - "hit_distribution": hit_dist, - "methodology_snapshots": methodology_snapshots, - "weight_evolution": self._get_weight_evolution(), - } - return result - - def _optimize_weights(self, red_hit, blue_hit, pred_reds, actual_reds): - """根据推演结果动态调整10维度权重""" - iteration = len(self.rollback_history) + 1 - lr = max(0.003, 0.01 - iteration * 0.0001) - - hit_ratio = (red_hit + blue_hit) / (self.rule["red_count"] + self.rule["blue_count"]) - - if hit_ratio >= 0.5: - self.weights["frequency"] = min(0.35, self.weights["frequency"] * (1 + lr)) - self.weights["hot_number"] = min(0.30, self.weights["hot_number"] * (1 + lr * 0.8)) - else: - self.weights["missing"] = min(0.30, self.weights["missing"] * (1 + lr)) - self.weights["probability"] = min(0.15, self.weights["probability"] * (1 + lr * 1.2)) - self.weights["frequency"] = max(0.15, self.weights["frequency"] * (1 - lr * 0.5)) - - pred_oe = sum(1 for n in pred_reds if n % 2 == 1) - actual_oe = sum(1 for n in actual_reds if n % 2 == 1) - if pred_oe == actual_oe: - self.weights["odd_even"] = min(0.12, self.weights["odd_even"] * (1 + lr)) - - pred_sum = sum(pred_reds) - actual_sum = sum(actual_reds) - if abs(pred_sum - actual_sum) <= 15: - self.weights["sum_value"] = min(0.10, self.weights["sum_value"] * (1 + lr)) - - pred_consec = sum(1 for i in range(len(pred_reds) - 1) if pred_reds[i + 1] - pred_reds[i] == 1) - actual_consec = sum(1 for i in range(len(actual_reds) - 1) if actual_reds[i + 1] - actual_reds[i] == 1) - if pred_consec == actual_consec: - self.weights["consecutive"] = min(0.08, self.weights["consecutive"] * (1 + lr)) - - total = sum(self.weights.values()) - if total > 0: - for k in self.weights: - self.weights[k] /= total - - def _get_weight_evolution(self): - if not self.rollback_history: - return [] - interval = max(1, len(self.rollback_history) // 20) - evolution = [] - for i, entry in enumerate(self.rollback_history): - if i % interval == 0 or i == len(self.rollback_history) - 1: - evolution.append({ - "iteration": entry["iteration"], - "issue": entry["issue"], - "weights": entry["weights_at_time"], - }) - return evolution - - -# ============================================================ -# 自进化系统(多轮迭代 + 方法论总结升级) -# ============================================================ - -class SelfEvolvingSystem: - """多轮迭代运行,每轮完成后总结方法论并升级,目标胜出率 >= 51%""" - - def __init__(self, lt, data): - self.lt = lt - self.data = data - self.rule = LOTTERY_RULES[lt] - self.best_weights = copy.deepcopy(DEFAULT_WEIGHTS) - self.best_accuracy = 0 - self.iteration_results = [] - self.methodology_log = [] - - def run_iterations(self, max_iterations=20): - print(f"\n{'='*60}") - print(f" 🔬 自进化系统启动 - {self.rule['name']}") - print(f" 目标:胜出率(概率占比) >= 51%") - print(f" 最大迭代轮次: {max_iterations}") - print(f"{'='*60}") - - for iteration in range(1, max_iterations + 1): - print(f"\n{'─'*60}") - print(f" 第 {iteration}/{max_iterations} 轮迭代") - print(f"{'─'*60}") - - if iteration == 1: - current_weights = copy.deepcopy(DEFAULT_WEIGHTS) - else: - current_weights = self._mutate_weights(self.best_weights, iteration) - - engine = AIRollbackEngine(self.lt, self.data, current_weights) - result = engine.run() - - if "error" in result: - print(f" ❌ 错误: {result['error']}") - break - - accuracy = result["accuracy"]["win_probability"] - self.iteration_results.append({ - "iteration": iteration, - "accuracy": accuracy, - "weights": result["final_weights"], - "simulations": result["total_simulations"], - }) - - self._summarize_methodology(iteration, result) - - if accuracy > self.best_accuracy: - self.best_accuracy = accuracy - self.best_weights = result["final_weights"] - print(f" 🏆 新最佳胜出率: {accuracy:.2f}%") - else: - print(f" 本轮胜出率: {accuracy:.2f}% (最佳: {self.best_accuracy:.2f}%)") - - if self.best_accuracy >= 51.0: - print(f"\n{'='*60}") - print(f" 🎯 目标达成!胜出率 {self.best_accuracy:.2f}% >= 51%") - print(f"{'='*60}") - break - - return self._final_report() - - def _mutate_weights(self, base_weights, iteration): - mutated = copy.deepcopy(base_weights) - mutation_rate = max(0.005, 0.03 - iteration * 0.001) - for k in mutated: - delta = random.uniform(-mutation_rate, mutation_rate) - mutated[k] = max(0.01, mutated[k] + delta) - total = sum(mutated.values()) - for k in mutated: - mutated[k] /= total - return mutated - - def _summarize_methodology(self, iteration, result): - acc = result["accuracy"] - weights = result["final_weights"] - dist = result.get("hit_distribution", {}) - - sorted_w = sorted(weights.items(), key=lambda x: x[1], reverse=True) - top_dims = sorted_w[:3] - weak_dims = sorted_w[-2:] - - summary_lines = [ - f"\n === 方法论总结 - 第{iteration}轮 ===", - f" 胜出率: {acc['win_probability']:.2f}% (红球{acc['red_accuracy']:.1f}% 蓝球{acc['blue_accuracy']:.1f}%)", - f" 最优三维度:", - ] - for dim, w in top_dims: - summary_lines.append(f" - {dim}: {w:.4f}") - summary_lines.append(f" 最弱二维度:") - for dim, w in weak_dims: - summary_lines.append(f" - {dim}: {w:.4f}") - summary_lines.append(f" 命中分布:") - for level in sorted(dist.keys()): - if dist.get(level, 0) > 0: - pct = dist[level] / result["total_simulations"] * 100 - summary_lines.append(f" - {level}个命中: {dist[level]}次 ({pct:.1f}%)") - - summary = "\n".join(summary_lines) - print(summary) - self.methodology_log.append(summary) - - mpath = os.path.join(DATA_DIR, f"methodology_{self.lt}_iter{iteration}.json") - with open(mpath, "w", encoding="utf-8") as f: - json.dump({ - "iteration": iteration, - "accuracy": acc, - "weights": weights, - "hit_distribution": dist, - "snapshots": result.get("methodology_snapshots", []), - }, f, ensure_ascii=False, indent=2) - print(f" [方法论已保存] {mpath}") - - def _final_report(self): - report = { - "lottery": self.lt, - "total_iterations": len(self.iteration_results), - "best_accuracy": self.best_accuracy, - "best_weights": self.best_weights, - "iteration_summary": self.iteration_results, - } - rpath = os.path.join(DATA_DIR, f"self_evolve_{self.lt}_final.json") - with open(rpath, "w", encoding="utf-8") as f: - json.dump(report, f, ensure_ascii=False, indent=2) - return report - - -# ============================================================ -# CLI 主入口 -# ============================================================ - -def load_history(lt): - path = os.path.join(DATA_DIR, f"{lt}_history.json") - if os.path.exists(path): - try: - with open(path, "r", encoding="utf-8") as f: - return json.load(f) - except: - pass - return [] - - -def print_header(): - print("\n" + "*" * 60) - print(" 双色球 & 大乐透 * 增强智能分析系统 v2.0") - print(" AI回滚推演 | 方法论自优化 | 200期深度分析") - print(" " + datetime.datetime.now().strftime("%Y-%m-%d %H:%M")) - print("*" * 60) - - -def show_full_report(lt, count=5): - rule = LOTTERY_RULES[lt] - data = load_history(lt) - if not data: - print(f"\n ⚠️ {rule['name']}暂无数据,请先采集") - print(f" python3 lottery_tool.py --action fetch --lottery {lt}") - return - - analyzer = EnhancedAnalyzer(data, lt) - analysis = analyzer.analyze() - if not analysis: - print(f"\n ⚠️ {rule['name']}分析失败") - return - - print(f"\n{'─'*60}") - print(f" {rule['name']} - 增强分析报告 (200期)") - print(f"{'─'*60}") - dr = analysis["data_range"] - print(f" 数据范围: {dr['total']}期 ({dr['start']} ~ {dr['end']})") - w = analyzer.weights - print(f" 权重体系: freq={w['frequency']:.3f} hot={w['hot_number']:.3f} miss={w['missing']:.3f}") - - pat = analysis["patterns"] - print(f"\n 模式分析:") - print(f" 奇偶比: {pat['most_oe']} | 大小比: {pat['most_bs']}") - print(f" 和值范围: {pat['sum_range'][0]} ~ {pat['sum_range'][1]} (均{pat['avg_sum']})") - print(f" 区间分布: {pat['zone_balance']}") - - recs = analyzer.recommend(analysis, count=count) - print(f"\n 🎯 AI推荐号码 ({count}组):") - for i, rec in enumerate(recs, 1): - rs = " ".join(f"{n:02d}" for n in rec["reds"]) - bs = " ".join(f"{n:02d}" for n in rec["blues"]) - print(f" [{i}] 红球: {rs} 蓝球: {bs} (评分:{rec['score']:.3f} OE:{rec['oe']} BS:{rec['bs']})") - - -def main(): - import argparse - parser = argparse.ArgumentParser(description="增强版彩票智能分析系统 v2.0") - parser.add_argument("--action", - choices=["report", "rollback", "self_evolve"], - default="report", - help="report: 200期分析推荐 | rollback: 逐期回滚推演 | self_evolve: 自进化") - parser.add_argument("--lottery", choices=["ssq", "dlt", "all"], default="ssq") - parser.add_argument("--count", type=int, default=5, help="推荐组数") - parser.add_argument("--iterations", type=int, default=10, help="自进化迭代轮次") - args = parser.parse_args() - - print_header() - lotteries = ["ssq", "dlt"] if args.lottery == "all" else [args.lottery] - - if args.action == "report": - for lt in lotteries: - show_full_report(lt, args.count) - return - - if args.action == "rollback": - for lt in lotteries: - data = load_history(lt) - if len(data) < 200: - print(f"\n ⚠️ {LOTTERY_RULES[lt]['name']}数据不足200期 (当前{len(data)}期)") - print(f" 请先采集: python3 lottery_tool.py --action fetch --lottery {lt}") - continue - engine = AIRollbackEngine(lt, data) - result = engine.run() - if "error" not in result: - acc = result["accuracy"] - print(f"\n === 回滚推演完成 ===") - print(f" 综合胜出率: {acc['win_probability']:.2f}%") - print(f" 红球准确率: {acc['red_accuracy']:.1f}% | 蓝球: {acc['blue_accuracy']:.1f}%") - - if args.action == "self_evolve": - for lt in lotteries: - data = load_history(lt) - if len(data) < 200: - print(f"\n ⚠️ {LOTTERY_RULES[lt]['name']}数据不足200期 (当前{len(data)}期)") - continue - system = SelfEvolvingSystem(lt, data) - final = system.run_iterations(max_iterations=args.iterations) - print(f"\n === 自进化最终结果 ===") - print(f" 最佳胜出率: {final['best_accuracy']:.2f}%") - print(f" 迭代总轮次: {final['total_iterations']}") - print(f" 最优权重:") - for k, v in sorted(final['best_weights'].items(), key=lambda x: x[1], reverse=True): - print(f" {k}: {v:.4f}") - - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/lottery_picker/data/ensemble_learning.py b/lottery_picker/data/ensemble_learning.py deleted file mode 100644 index 626a219..0000000 --- a/lottery_picker/data/ensemble_learning.py +++ /dev/null @@ -1,725 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -大乐透/双色球 集成学习模块 -方案3:多模型集成 + 权重优化训练 -================================== -包含5种分析模型: -1. 频率加权模型 (Frequency Model) -2. 贝叶斯平滑模型 (Bayesian Smoothing Model) -3. 马尔可夫链模型 (Markov Chain Model) -4. K-Means聚类模型 (K-Means Clustering Model) -5. 残差回归模型 (Residual Regression Model) - -通过历史数据回测学习各模型的最优权重 -""" - -import json, math, random, hashlib -from collections import Counter, defaultdict - -# ==================== 基础工具函数 ==================== - -def sigmoid(x): - """Sigmoid激活函数""" - return 1 / (1 + math.exp(-max(min(x, 500), -500))) - -def normalize(scores): - """归一化分数,使总和为1""" - total = sum(scores.values()) - if total == 0: - return {k: 1.0/len(scores) for k in scores} - return {k: v/total for k, v in scores.items()} - -def weighted_choice(weights, count=1): - """从权重分布中无放回选择指定数量的元素""" - items = list(weights.keys()) - if len(items) == 0: - return [] - if count >= len(items): - return items[:] - result = [] - remaining = items[:] - remaining_weights = weights.copy() - for _ in range(count): - total = sum(remaining_weights.values()) - if total <= 0: - choice = random.choice(remaining) - else: - r = random.random() * total - cum = 0 - for item in remaining: - cum += remaining_weights[item] - if r <= cum: - choice = item - break - result.append(choice) - remaining.remove(choice) - del remaining_weights[choice] - return result - -# ==================== 模型1:频率加权模型 ==================== - -class FrequencyModel: - """基于历史频率的加权模型""" - def __init__(self, lt, data, rule): - self.lt = lt - self.data = data - self.rule = rule - self.rcnt = rule["red_count"] - self.rmax = rule["red_max"] - self.bcnt = rule["blue_count"] - self.bmax = rule["blue_max"] - self.red_score = {} - self.blue_score = {} - self.analyze() - - def analyze(self): - rule, rmax, bmax, rcnt, bcnt = self.rule, self.rule["red_max"], self.rule["blue_max"], self.rule["red_count"], self.rule["blue_count"] - all_r, all_b = [], [] - for rec in self.data: - all_r.extend(int(x) for x in rec["reds"]) - bl = rec["blue"] - # 修复:蓝球处理 - 支持整数和字符串两种格式 - if isinstance(bl, str): - all_b.append(int(bl)) - elif isinstance(bl, list): - all_b.extend(bl) - else: - all_b.append(bl) - - # 频率统计 - rf = Counter(all_r) - bf = Counter(all_b) - - max_rf = max(rf.values()) if rf else 1 - max_bf = max(bf.values()) if bf else 1 - - # 归一化 - self.red_score = {i: rf.get(i, 0) / max_rf for i in range(1, rmax + 1)} - self.blue_score = {i: bf.get(i, 0) / max_bf for i in range(1, bmax + 1)} - - # 平滑:避免0分 - for i in range(1, rmax + 1): - if self.red_score[i] == 0: - self.red_score[i] = 0.01 - for i in range(1, bmax + 1): - if self.blue_score[i] == 0: - self.blue_score[i] = 0.01 - - # 归一化到[0,1] - max_r = max(self.red_score.values()) - max_b = max(self.blue_score.values()) - if max_r > 0: - for i in self.red_score: - self.red_score[i] /= max_r - if max_b > 0: - for i in self.blue_score: - self.blue_score[i] /= max_b - - -# ==================== 模型2:贝叶斯平滑模型 ==================== - -class BayesianModel: - """贝叶斯先验 + 似然平滑模型""" - def __init__(self, lt, data, rule): - self.lt = lt - self.data = data - self.rule = rule - self.rcnt = rule["red_count"] - self.rmax = rule["red_max"] - self.bcnt = rule["blue_count"] - self.bmax = rule["blue_max"] - self.prior = {i: 1.0 / rule["red_max"] for i in range(1, rule["red_max"] + 1)} - self.red_score = {} - self.blue_score = {} - self.analyze() - - def analyze(self): - rule, rmax, bmax, rcnt, bcnt = self.rule, self.rule["red_max"], self.rule["blue_max"], self.rule["red_count"], self.rule["blue_count"] - all_r, all_b = [], [] - for rec in self.data: - all_r.extend(int(x) for x in rec["reds"]) - bl = rec["blue"] - # 修复:蓝球处理 - 支持整数和字符串两种格式 - if isinstance(bl, str): - all_b.append(int(bl)) - elif isinstance(bl, list): - all_b.extend(bl) - else: - all_b.append(bl) - - # 似然 - rf = Counter(all_r) - bf = Counter(all_b) - - # 贝叶斯后验:后验 ∝ 先验 × 似然 - # 使用伪计数:先验+1,似然+1,避免0 - alpha = 1 # 先验伪计数 - beta = 1 # 似然伪计数 - total_alpha = rmax * alpha - total_beta = len(all_r) + beta * rmax - - self.red_score = {} - for i in range(1, rmax + 1): - # 后验均值 = (α + count) / (α + β * rmax) - posterior = (alpha + rf.get(i, 0)) / (total_alpha + beta) - self.red_score[i] = posterior - - total_blue = len(all_b) + beta * bmax - self.blue_score = {} - for i in range(1, bmax + 1): - posterior = (alpha + bf.get(i, 0)) / (total_alpha + beta) - self.blue_score[i] = posterior - - # 归一化 - max_r = max(self.red_score.values()) - max_b = max(self.blue_score.values()) - if max_r > 0: - for i in self.red_score: - self.red_score[i] /= max_r - if max_b > 0: - for i in self.blue_score: - self.blue_score[i] /= max_b - - -# ==================== 模型3:马尔可夫链模型 ==================== - -class MarkovModel: - """基于马尔可夫链的状态转移模型""" - def __init__(self, lt, data, rule): - self.lt = lt - self.data = data - self.rule = rule - self.rcnt = rule["red_count"] - self.rmax = rule["red_max"] - self.bcnt = rule["blue_count"] - self.bmax = rule["blue_max"] - self.red_score = {} - self.blue_score = {} - self.analyze() - - def analyze(self): - rule, rmax, bmax, rcnt, bcnt = self.rule, self.rule["red_max"], self.rule["blue_max"], self.rule["red_count"], self.rule["blue_count"] - - # 构建状态转移矩阵(考虑最近5期作为状态) - # 这里简化为:计算每个号码在历史数据中的"惯性" - # 即:一个号码出现后,后续几期再次出现的概率 - - # 使用frozenset作为不可变集合作为字典键 - transition = defaultdict(lambda: defaultdict(int)) - for i in range(1, len(self.data)): - prev = frozenset(int(x) for x in self.data[i-1]["reds"]) - curr = set(int(x) for x in self.data[i]["reds"]) - for num in curr: - transition[prev][num] += 1 - - # 计算每个号码的"热度"(结合近期出现频率和转移概率) - recent_data = self.data[:20] if len(self.data) >= 20 else self.data - recent_reds = [] - for rec in recent_data: - recent_reds.extend(int(x) for x in rec["reds"]) - - rf = Counter(recent_reds) - total_recent = len(recent_reds) - - # 综合评分:近期频率 + 转移惯性 - self.red_score = {} - for i in range(1, rmax + 1): - freq_score = rf.get(i, 0) / total_recent if total_recent > 0 else 0 - # 惯性得分:如果上一期出现过,给予额外权重 - last_reds = set(int(x) for x in self.data[0]["reds"]) if self.data else set() - inertia = 1.5 if i in last_reds else 1.0 - self.red_score[i] = (freq_score * inertia + 0.01) # 加0.01平滑 - - # 蓝球简化:直接用近期频率 - recent_blues = [] - for rec in recent_data: - bl = rec["blue"] - recent_blues.append(int(bl) if isinstance(bl, str) else (bl[0] if isinstance(bl, list) else bl)) - - bf = Counter(recent_blues) - total_blue = len(recent_blues) - self.blue_score = {} - for i in range(1, bmax + 1): - self.blue_score[i] = (bf.get(i, 0) / total_blue * 0.8 + 1.0 / bmax * 0.2) if total_blue > 0 else (1.0 / bmax) - - # 归一化 - max_r = max(self.red_score.values()) if self.red_score else 1 - max_b = max(self.blue_score.values()) if self.blue_score else 1 - for i in self.red_score: - self.red_score[i] /= max_r - for i in self.blue_score: - self.blue_score[i] /= max_b - - -# ==================== 模型4:K-Means聚类模型 ==================== - -class KMeansModel: - """基于历史聚类中心的推荐模型""" - def __init__(self, lt, data, rule): - self.lt = lt - self.data = data - self.rule = rule - self.rcnt = rule["red_count"] - self.rmax = rule["red_max"] - self.bcnt = rule["blue_count"] - self.bmax = rule["blue_max"] - self.red_score = {} - self.blue_score = {} - self.analyze() - - def analyze(self): - rule, rmax, bmax, rcnt, bcnt = self.rule, self.rule["red_max"], self.rule["blue_max"], self.rule["red_count"], self.rule["blue_max"] - - # 提取所有历史红球组合 - clusters = [] - for rec in self.data: - nums = sorted(int(x) for x in rec["reds"]) - if len(nums) == rcnt: - clusters.append(nums) - - if len(clusters) < 3: - # 数据不足,退化为频率模型 - rf = Counter() - for nums in clusters: - for n in nums: - rf[n] += 1 - max_f = max(rf.values()) if rf else 1 - self.red_score = {i: rf.get(i, 0) / max_f for i in range(1, rmax + 1)} - return - - # 简化版K-Means:使用质心法 - # 计算每个位置的平均值(排序后) - sorted_clusters = [sorted(c) for c in clusters] - n_clusters = min(5, len(sorted_clusters)) - - # 计算每个位置的聚类中心 - position_means = [[] for _ in range(rcnt)] - for cluster in sorted_clusters: - for pos in range(rcnt): - position_means[pos].append(cluster[pos]) - - # 计算每个号码的"接近中心"得分 - position_avg = [sum(p) / len(p) for p in position_means if p] - - self.red_score = {} - for i in range(1, rmax + 1): - # 计算该号码在所有位置上偏离平均值的程度 - deviations = [] - for pos, avg in enumerate(position_avg): - if len(sorted_clusters) > pos: - cluster_at_pos = [c[pos] for c in sorted_clusters if len(c) > pos] - if cluster_at_pos: - deviations.append(abs(i - avg)) - avg_dev = sum(deviations) / len(deviations) if deviations else rmax / 2 - # 得分:偏离越小得分越高(越接近典型组合) - score = max(0, 1 - avg_dev / (rmax * 0.5)) - self.red_score[i] = max(score, 0.01) # 最小0.01 - - # 蓝球:直接用频率 - all_blues = [] - for rec in self.data: - bl = rec["blue"] - all_blues.append(int(bl) if isinstance(bl, str) else (bl[0] if isinstance(bl, list) else bl)) - bf = Counter(all_blues) - max_bf = max(bf.values()) if bf else 1 - self.blue_score = {i: bf.get(i, 0) / max_bf for i in range(1, bmax + 1)} - for i in range(1, bmax + 1): - if self.blue_score[i] == 0: - self.blue_score[i] = 0.01 - - # 归一化 - max_r = max(self.red_score.values()) if self.red_score else 1 - max_b = max(self.blue_score.values()) if self.blue_score else 1 - for i in self.red_score: - self.red_score[i] /= max_r - for i in self.blue_score: - self.blue_score[i] /= max_b - - -# ==================== 模型5:残差回归模型 ==================== - -class ResidualModel: - """残差分析 + 回归预测模型""" - def __init__(self, lt, data, rule): - self.lt = lt - self.data = data - self.rule = rule - self.rcnt = rule["red_count"] - self.rmax = rule["red_max"] - self.bcnt = rule["blue_count"] - self.bmax = rule["blue_max"] - self.red_score = {} - self.blue_score = {} - self.analyze() - - def analyze(self): - rule, rmax, bmax, rcnt, bcnt = self.rule, self.rule["red_max"], self.rule["blue_max"], self.rule["red_count"], self.rule["blue_count"] - - # 计算每期的和值、奇偶比、大小比作为特征 - sums = [] - for rec in self.data: - s = sum(int(x) for x in rec["reds"]) - sums.append(s) - - if len(sums) < 10: - # 数据不足,退化为均匀分布 - self.red_score = {i: 1.0 / rmax for i in range(1, rmax + 1)} - self.blue_score = {i: 1.0 / bmax for i in range(1, bmax + 1)} - return - - # 计算和值的移动平均(最近10期) - window = min(10, len(sums)) - recent_sums = sums[-window:] - avg_sum = sum(recent_sums) / len(recent_sums) - std_sum = math.sqrt(sum((s - avg_sum) ** 2 for s in recent_sums) / len(recent_sums)) - - # 计算每个号码在"和值附近"组合中的出现频率 - # 思路:哪些号码更常出现在和值接近均值的组合中 - score_by_num = {i: 0.0 for i in range(1, rmax + 1)} - count_by_num = {i: 0 for i in range(1, rmax + 1)} - - for rec in self.data: - nums = [int(x) for x in rec["reds"]] - s = sum(nums) - # 计算该组合和值与均值的距离 - dist = abs(s - avg_sum) - if dist <= std_sum * 2: # 落在2个标准差内 - for n in nums: - score_by_num[n] += 1 / (dist + 1) # 距离越近得分越高 - count_by_num[n] += 1 - - # 归一化 - max_score = max(score_by_num.values()) if max(score_by_num.values()) > 0 else 1 - self.red_score = {} - for i in range(1, rmax + 1): - if count_by_num[i] > 0: - self.red_score[i] = (score_by_num[i] / count_by_num[i]) / max_score - else: - self.red_score[i] = 1.0 / rmax # 均匀分布 - - # 蓝球:和值相关性分析 - blues_by_sum = defaultdict(list) - for rec in self.data: - bl = rec["blue"] - bl_val = int(bl) if isinstance(bl, str) else (bl[0] if isinstance(bl, list) else bl) - s = sum(int(x) for x in rec["reds"]) - blues_by_sum[s].append(bl_val) - - self.blue_score = {} - for i in range(1, bmax + 1): - # 计算该蓝球在"和值附近"出现的频率 - total = 0 - for s, blues in blues_by_sum.items(): - dist = abs(s - avg_sum) - if dist <= std_sum * 2: - total += len([b for b in blues if b == i]) / (dist + 1) - self.blue_score[i] = max(total / 100, 1.0 / bmax) - - # 归一化 - max_r = max(self.red_score.values()) if self.red_score else 1 - max_b = max(self.blue_score.values()) if self.blue_score else 1 - for i in self.red_score: - self.red_score[i] /= max_r - for i in self.blue_score: - self.blue_score[i] /= max_b - - -# ==================== 集成学习引擎 ==================== - -class EnsembleLearner: - """集成学习引擎:多模型加权 + 回测优化权重""" - - def __init__(self, lt, data, rule): - self.lt = lt - self.data = data - self.rule = rule - self.rcnt = rule["red_count"] - self.rmax = rule["red_max"] - self.bcnt = rule["blue_count"] - self.bmax = rule["blue_max"] - - # 初始化5个模型 - self.models = { - 'freq': FrequencyModel(lt, data, rule), - 'bayes': BayesianModel(lt, data, rule), - 'markov': MarkovModel(lt, data, rule), - 'kmeans': KMeansModel(lt, data, rule), - 'residual': ResidualModel(lt, data, rule) - } - - # 初始权重(等权) - self.weights = {m: 1.0 / len(self.models) for m in self.models} - self.initial_weights = self.weights.copy() - - # 回测结果 - self.backtest_results = None - - def get_combined_score(self): - """获取集成后的综合评分""" - red_combined = {} - blue_combined = {} - - for model_name, model in self.models.items(): - w = self.weights[model_name] - for i in range(1, self.rmax + 1): - red_combined[i] = red_combined.get(i, 0) + w * model.red_score[i] - for i in range(1, self.bmax + 1): - blue_combined[i] = blue_combined.get(i, 0) + w * model.blue_score[i] - - # 归一化 - max_r = max(red_combined.values()) if red_combined else 1 - max_b = max(blue_combined.values()) if blue_combined else 1 - for i in red_combined: - red_combined[i] /= max_r - for i in blue_combined: - blue_combined[i] /= max_b - - return red_combined, blue_combined - - def _get_combined_for_weights(self, weights): - """根据给定权重获取组合评分""" - red_c = {} - blue_c = {} - for model_name, model in self.models.items(): - ww = weights[model_name] - for i in range(1, self.rmax + 1): - red_c[i] = red_c.get(i, 0) + ww * model.red_score[i] - for i in range(1, self.bmax + 1): - blue_c[i] = blue_c.get(i, 0) + ww * model.blue_score[i] - return red_c, blue_c - - def _predict_from_scores(self, red_c, blue_c, rmax, bmax, rcnt, bcnt): - """根据评分预测号码""" - sorted_r = sorted(red_c.items(), key=lambda x: x[1], reverse=True) - top_r = [n for n, s in sorted_r[:rmax]] - sorted_b = sorted(blue_c.items(), key=lambda x: x[1], reverse=True) - top_b = [n for n, s in sorted_b[:bmax]] - - sel = [] - cand = top_r[:] - weights_r = [red_c.get(n, 0.5) for n in cand] - while len(sel) < rcnt and cand: - tw = sum(weights_r) - if tw <= 0: pk = random.choice(cand) - else: - rv = random.random() * tw - cum = 0 - for i, ww in enumerate(weights_r): - cum += ww - if rv <= cum: - pk = cand[i] - break - if pk not in sel: sel.append(pk) - idx = cand.index(pk); cand.pop(idx); weights_r.pop(idx) - sel.sort() - - bls = [] - cand_b = top_b[:] - weights_b = [blue_c.get(n, 0.5) for n in cand_b] - while len(bls) < bcnt and cand_b: - tw = sum(weights_b) - if tw <= 0: pk = random.choice(cand_b) - else: - rv = random.random() * tw - cum = 0 - for i, ww in enumerate(weights_b): - cum += ww - if rv <= cum: - pk = cand_b[i] - break - if pk not in bls: bls.append(pk) - idx = cand_b.index(pk); cand_b.pop(idx); weights_b.pop(idx) - bls.sort() - - return sel, bls - - def backtest_optimize(self, test_periods=30): - """ - 通过回测优化各模型权重 - 使用简单网格搜索 + 梯度上升混合方法 - """ - rule = self.rule - rmax, bmax = self.rule["red_max"], self.rule["blue_max"] - - print(f"\n{'='*60}") - print(f"🧠 集成学习权重优化 | 彩种: {rule['name']} | 数据期数: {len(self.data)}") - print(f"{'='*60}") - - # 准备回测数据(按时间顺序) - sorted_data = sorted(self.data, key=lambda x: int(x["issue"])) - if len(sorted_data) < test_periods + 10: - test_periods = max(10, len(sorted_data) - 10) - - # 评分函数:计算推荐与实际匹配的分数 - def evaluate(weights, data_slice): - red_comp, blue_comp = self._get_combined_for_weights(weights) - pred_r, pred_b = self._predict_from_scores(red_comp, blue_comp, rmax, bmax, rule["red_count"], rule["blue_count"]) - - actual = data_slice - act_r = [int(x) for x in actual["reds"]] - act_b = actual["blue"] if isinstance(actual["blue"], list) else [int(actual["blue"])] - - # 评分:命中红球数*2 + 命中蓝球数 - hit_r = sum(1 for n in pred_r if n in act_r) - hit_b = sum(1 for n in pred_b if n in act_b if (isinstance(act_b, list) and n in act_b) or (isinstance(act_b, int) and n == act_b)) - return hit_r * 2 + hit_b - - # 简单网格搜索优化权重 - best_weights = self.weights.copy() - best_score = float('-inf') - - # 初始评估 - current_score = 0 - for i in range(test_periods): - if i + 10 < len(sorted_data): - slice_data = sorted_data[i + 10:i + 11] # 下一期作为测试 - current_score += evaluate(self.weights, slice_data[0]) - - print(f" 初始等权总分: {current_score:.1f}") - - # 简单梯度上升(每个模型权重微调) - improved = True - iterations = 0 - max_iterations = 20 - - while improved and iterations < max_iterations: - improved = False - iterations += 1 - - for model_name in self.weights: - for delta in [-0.05, 0.05]: # 微调权重 - test_w = self.weights.copy() - test_w[model_name] = max(0.01, min(0.9, test_w[model_name] + delta)) - - # 归一化 - total = sum(test_w.values()) - test_w = {k: v/total for k, v in test_w.items()} - - score = 0 - for i in range(test_periods): - if i + 10 < len(sorted_data): - slice_data = sorted_data[i + 10:i + 11] - score += evaluate(test_w, slice_data[0]) - - if score > best_score: - best_score = score - best_weights = test_w.copy() - improved = True - - self.weights = best_weights - self.backtest_results = { - 'iterations': iterations, - 'best_score': best_score, - 'test_periods': test_periods, - 'model_weights': best_weights - } - - # 输出各模型权重 - print(f" 优化后权重:") - for m, w in sorted(self.weights.items(), key=lambda x: -x[1]): - model_names = { - 'freq': '频率加权', - 'bayes': '贝叶斯平滑', - 'markov': '马尔可夫链', - 'kmeans': 'K-Means聚类', - 'residual': '残差回归' - } - print(f" {model_names.get(m, m):<12s}: {w:.4f}") - - print(f" 最终集成总分: {best_score:.1f} (测试{test_periods}期)") - - # 返回综合评分 - return self.get_combined_score() - - def get_recommendations(self, count=5): - """基于集成评分生成推荐""" - red_comp, blue_comp = self.get_combined_score() - rmax, bmax = self.rule["red_max"], self.rule["blue_max"] - rcnt, bcnt = self.rule["red_count"], self.rule["blue_count"] - - sorted_r = sorted(red_comp.items(), key=lambda x: x[1], reverse=True) - top_r = [n for n, s in sorted_r[:rmax]] - sorted_b = sorted(blue_comp.items(), key=lambda x: x[1], reverse=True) - top_b = [n for n, s in sorted_b[:bmax]] - - seed = hashlib.md5(f"{self.lt}_ensemble".encode()).hexdigest() - res = [] - for ci in range(count): - cs = hashlib.md5(f"ensemble_c{ci}_{seed}".encode()).hexdigest() - cr = random.Random(cs) - - sel = [] - cand = top_r[:] - w = [red_comp.get(n, 0.5) for n in cand] - while len(sel) < rcnt and cand: - tw = sum(w) - if tw <= 0: pk = cr.choice(cand) - else: - rv = cr.random() * tw - cum = 0 - for i, ww in enumerate(w): - cum += ww - if rv <= cum: - pk = cand[i] - break - if pk not in sel: sel.append(pk) - idx = cand.index(pk); cand.pop(idx); w.pop(idx) - sel.sort() - - bls = [] - cand_b = top_b[:] - wb = [blue_comp.get(n, 0.5) for n in cand_b] - while len(bls) < bcnt and cand_b: - tw = sum(wb) - if tw <= 0: pk = cr.choice(cand_b) - else: - rv = cr.random() * tw - cum = 0 - for i, ww in enumerate(wb): - cum += ww - if rv <= cum: - pk = cand_b[i] - break - if pk not in bls: bls.append(pk) - idx = cand_b.index(pk); cand_b.pop(idx); wb.pop(idx) - bls.sort() - - score = round(sum(red_comp.get(n, 0) for n in sel) + sum(blue_comp.get(n, 0) for n in bls), 4) - mid = rmax / 2 - res.append({ - "reds": sel, - "blues": bls, - "score": score, - "odd_even": f"{sum(1 for n in sel if n % 2 == 1)}:{rcnt - sum(1 for n in sel if n % 2 == 1)}", - "big_small": f"{sum(1 for n in sel if n > mid)}:{rcnt - sum(1 for n in sel if n > mid)}", - "sum": sum(sel) - }) - - res.sort(key=lambda x: x["score"], reverse=True) - return res - - -# ==================== 主函数 ==================== - -def ensemble_recommendation(lt, data, rule, count=5): - """集成学习推荐主接口""" - learner = EnsembleLearner(lt, data, rule) - red_comp, blue_comp = learner.backtest_optimize(test_periods=30) - recommendations = learner.get_recommendations(count=count) - - return { - 'weights': learner.weights, - 'backtest': learner.backtest_results, - 'recommendations': recommendations, - 'combined_scores': { - 'red': red_comp, - 'blue': blue_comp - } - } - - -if __name__ == "__main__": - # 简单测试 - print("集成学习模块加载成功!") - print("可用模型:频率加权、贝叶斯平滑、马尔可夫链、K-Means聚类、残差回归") \ No newline at end of file diff --git a/lottery_picker/data/evolution_tool.py b/lottery_picker/data/evolution_tool.py deleted file mode 100644 index dc4aaa8..0000000 --- a/lottery_picker/data/evolution_tool.py +++ /dev/null @@ -1,285 +0,0 @@ -#!/usr/bin/env python3 -""" -回滚迭代跟踪工具 - 管理 AI 进化版本和回滚机制 -""" - -import json -import os -from datetime import datetime -from typing import Dict, List, Optional, Any -from pathlib import Path - - -class EvolutionTracker: - """回滚迭代跟踪器""" - - def __init__(self, data_dir: str = "./evolution_tracking"): - self.data_dir = Path(data_dir) - self.data_dir.mkdir(parents=True, exist_ok=True) - self.version_file = self.data_dir / "versions.json" - self.performance_log = self.data_dir / "performance.json" - self.rollback_log = self.data_dir / "rollbacks.json" - - def initialize(self): - """初始化跟踪器""" - # 初始化版本记录 - if not self.version_file.exists(): - initial_version = { - "version": 0, - "created_at": datetime.now().isoformat(), - "performance": { - "win_rate": 0.0, - "total_bets": 0, - "total_winnings": 0.0 - } - } - self.save_version(initial_version) - - # 初始化性能日志 - if not self.performance_log.exists(): - self.save_performance({ - "history": [], - "current": { - "total_bets": 0, - "win_rate": 0.0, - "earnings": 0.0, - "total_winnings": 0.0 - } - }) - - # 初始化回滚日志 - if not self.rollback_log.exists(): - self.save_rollbacks({ - "history": [], - "current_version": 0 - }) - - def save_version(self, version_data: Dict): - """保存版本信息""" - with open(self.version_file, 'w', encoding='utf-8') as f: - json.dump(version_data, f, ensure_ascii=False, indent=2) - - def load_version(self) -> Dict: - """加载当前版本""" - if self.version_file.exists(): - with open(self.version_file, 'r', encoding='utf-8') as f: - return json.load(f) - return {} - - def record_performance(self, version: int, win_rate: float, earnings: float): - """记录性能数据""" - performance = self.load_performance() - - entry = { - "version": version, - "recorded_at": datetime.now().isoformat(), - "win_rate": win_rate, - "earnings": earnings, - "total_bets": performance["current"].get("total_bets", 0), - "total_winnings": performance["current"].get("total_winnings", 0.0) - } - - performance["history"].append(entry) - - # 更新当前性能 - current = performance["current"] - current["total_bets"] += 1 - current["win_rate"] = win_rate - current["earnings"] = earnings - current["total_winnings"] = current.get("total_winnings", 0.0) + earnings - - self.save_performance(performance) - - def load_performance(self) -> Dict: - """加载性能日志""" - if self.performance_log.exists(): - with open(self.performance_log, 'r', encoding='utf-8') as f: - return json.load(f) - return {"history": [], "current": {}} - - def save_performance(self, performance: Dict): - """保存性能日志""" - with open(self.performance_log, 'w', encoding='utf-8') as f: - json.dump(performance, f, ensure_ascii=False, indent=2) - - def record_rollback(self, from_version: int, to_version: int, reason: str): - """记录回滚操作""" - rollbacks = self.load_rollbacks() - - entry = { - "from_version": from_version, - "to_version": to_version, - "reason": reason, - "timestamp": datetime.now().isoformat() - } - - rollbacks["history"].append(entry) - rollbacks["current_version"] = to_version - - self.save_rollbacks(rollbacks) - - def load_rollbacks(self) -> Dict: - """加载回滚日志""" - if self.rollback_log.exists(): - with open(self.rollback_log, 'r', encoding='utf-8') as f: - return json.load(f) - return {"history": [], "current_version": 0} - - def save_rollbacks(self, rollbacks: Dict): - """保存回滚日志""" - with open(self.rollback_log, 'w', encoding='utf-8') as f: - json.dump(rollbacks, f, ensure_ascii=False, indent=2) - - def get_version_history(self) -> List[Dict]: - """获取版本历史""" - performance = self.load_performance() - return performance.get("history", []) - - def get_best_version(self) -> Optional[Dict]: - """获取性能最好的版本""" - history = self.get_version_history() - if not history: - return None - - best = max(history, key=lambda x: x.get("win_rate", 0)) - return best - - def compare_versions(self, version1: int, version2: int) -> Dict: - """比较两个版本的性能""" - history = self.get_version_history() - - v1_data = next((h for h in history if h["version"] == version1), None) - v2_data = next((h for h in history if h["version"] == version2), None) - - if not v1_data or not v2_data: - return {"error": "版本不存在"} - - comparison = { - "version1": version1, - "version2": version2, - "v1_win_rate": v1_data.get("win_rate", 0), - "v2_win_rate": v2_data.get("win_rate", 0), - "v1_earnings": v1_data.get("earnings", 0.0), - "v2_earnings": v2_data.get("earnings", 0.0), - "better_version": version1 if v1_data.get("win_rate", 0) > v2_data.get("win_rate", 0) else version2 - } - - return comparison - - def generate_report(self) -> str: - """生成进化报告""" - performance = self.load_performance() - rollbacks = self.load_rollbacks() - - report = [ - "="*60, - "AI 进化跟踪报告", - "="*60, - f"当前版本:{rollbacks.get('current_version', 0)}", - f"回滚历史数:{len(rollbacks.get('history', []))}", - "\n当前性能数据:", - f" 胜率:{performance['current'].get('win_rate', 0):.2%}", - f" 累计投注:{performance['current'].get('total_bets', 0)}", - f" 累计收益:¥{performance['current'].get('total_winnings', 0.0):.2f}", - "\n版本历史:", - ] - - history = performance.get("history", []) - if history: - for i, entry in enumerate(history[-10:], 1): # 显示最后 10 个 - report.extend([ - f" [v{entry['version']}] 胜率:{entry['win_rate']:.2%}, " - f"收益:¥{entry['earnings']:.2f}, " - f"记录时间:{entry['recorded_at']}", - "" - ]) - else: - report.append(" 暂无历史数据") - - report.extend(["\n回滚历史:", ""]) - - if rollbacks.get("history"): - for i, entry in enumerate(rollbacks["history"]): - report.extend([ - f" 回滚 #{i+1}: v{entry['from_version']} → v{entry['to_version']}, " - f"原因:{entry['reason']}, " - f"时间:{entry['timestamp']}", - "" - ]) - else: - report.append(" 暂无回滚记录") - - report.extend(["="*60, ""]) - return "\n".join(report) - - def optimize_weights(self, learning_rate: float = 0.1, epochs: int = 100) -> Dict: - """优化模型权重 - 模拟进化过程""" - import random - - # 初始化权重 - weights = { - "red_hot": 0.3, - "red_cold": 0.2, - "consecutive": 0.15, - "sum_range": 0.15, - "parity": 0.1, - "entropy": 0.1 - } - - # 模拟进化过程 - for epoch in range(epochs): - # 模拟数据输入 - sample_data = [random.randint(1, 33) for _ in range(10)] - - # 计算损失 - loss = sum((w - random.random())**2 for w in weights.values()) - - # 梯度下降 - for key in weights: - gradient = sum(random.random() for _ in range(10)) / len(sample_data) - weights[key] -= learning_rate * gradient - - # 归一化权重 - total = sum(weights.values()) - weights = {k: round(v/total, 4) for k, v in weights.items()} - - return { - "optimized_weights": weights, - "epochs_completed": epochs, - "final_loss": round(loss, 4) - } - - -def main(): - """主函数 - 测试回滚跟踪工具""" - tracker = EvolutionTracker() - tracker.initialize() - - # 模拟一些版本记录 - versions = [ - (1, 0.05, 100.0), - (2, 0.12, 150.0), - (3, 0.08, 120.0), - (4, 0.15, 200.0), - (5, 0.18, 250.0) - ] - - for version, win_rate, earnings in versions: - tracker.record_performance(version, win_rate, earnings) - - # 模拟回滚 - tracker.record_rollback(5, 4, "性能下降") - - # 生成报告 - print(tracker.generate_report()) - - # 优化权重 - optimized = tracker.optimize_weights() - print("\n优化结果:") - print(f" 完成轮数:{optimized['epochs_completed']}") - print(f" 最终损失:{optimized['final_loss']}") - print(f" 优化权重:{optimized['optimized_weights']}") - - -if __name__ == "__main__": - main() diff --git a/lottery_picker/data/lottery_tool.py b/lottery_picker/data/lottery_tool.py deleted file mode 100644 index 7a16ba1..0000000 --- a/lottery_picker/data/lottery_tool.py +++ /dev/null @@ -1,1314 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -双色球 & 大乐透 号码分析与机选工具 -================================== -功能: 数据采集/分析/推荐/回滚/持久化 -数据源: 双色球(500.com) 大乐透(体彩API) -""" - -import json, re, os, hashlib, datetime, subprocess, random, math, sys -from collections import Counter - -# 集成学习模块 -try: - from ensemble_learning import ensemble_recommendation -except ImportError: - ensemble_recommendation = None - -BASE_DIR = os.path.dirname(os.path.abspath(__file__)) -DATA_DIR = os.path.join(BASE_DIR, "data") -os.makedirs(DATA_DIR, exist_ok=True) - -LOTTERY_RULES = { - "ssq": { - "name": "双色球", "source": "500.com", - "red_count": 6, "red_max": 33, "blue_count": 1, "blue_max": 16, - "draw_days": [2, 4, 7], "draw_time": "21:15", - "api_url": "https://datachart.500.com/ssq/history/newinc/history.php?start={start}&end={end}&limit=100", - "backup_url": "https://www.17500.cn/api/kaijiang/getlist", - "backup_type": "post" - }, - "dlt": { - "name": "大乐透", "source": "中国体育彩票", - "red_count": 5, "red_max": 35, "blue_count": 2, "blue_max": 12, - "draw_days": [1, 3, 6], "draw_time": "20:25", - "api_url": "https://webapi.sporttery.cn/gateway/lottery/getHistoryPageListV1.qry?gameNo=85&provinceId=0&pageSize=100&is498=0&pageNo=1" - } -} -WEEKDAY = {1:"周一",2:"周二",3:"周三",4:"周四",5:"周五",6:"周六",7:"周日"} - - -class Persistence: - def __init__(self): - self.config = self.load("config.json", { - "version": "1.0.0", "seed": "LOTTERY_SKILL_V1.0.0", "iteration_count": 0, - "ssq": {"latest_issue":"","latest_fetch_start":"","latest_fetch_end":""}, - "dlt": {"latest_issue":"","latest_fetch_start":"","latest_fetch_end":""} - }) - def path(self, n): return os.path.join(DATA_DIR, n) - def load(self, n, d=None): - p=self.path(n) - if os.path.exists(p): - try: return json.load(open(p,'r',encoding='utf-8')) - except: pass - return d if d is not None else {} - def save(self, n, d): - json.dump(d, open(self.path(n),'w',encoding='utf-8'), ensure_ascii=False, indent=2) - def history(self, lt): return self.load(f"{lt}_history.json", []) - def save_history(self,lt,r): self.save(f"{lt}_history.json", r) - def analysis(self,lt): return self.load(f"analysis_{lt}.json", {}) - def save_analysis(self,lt,d): self.save(f"analysis_{lt}.json", d) - def feedback(self): - return self.load("feedback.json", { - "iterations":[],"accuracy_history":[], - "ssq":{"total":0,"correct_reds":0,"correct_blues":0,"best":0, - "profitable_wins":0,"total_earnings":0.0}, - "dlt":{"total":0,"correct_reds":0,"correct_blues":0,"best":0, - "profitable_wins":0,"total_earnings":0.0} - }) - def save_feedback(self,d): self.save("feedback.json", d) - def save_config(self): self.save("config.json", self.config) - def get_seed(self,lt,date=None): - if not date: date = datetime.datetime.now().strftime("%Y-%m-%d") - # 种子基于:固定种子 + 彩种 + 日期 + 分析版本号 - # version 从 analysis_{lt}.json 读取(由 show_analysis 写入), - # FeedbackLoop.check() 回滚时不修改该字段,保证同天内结果一致。 - # 仅当数据源变化→重新分析→版本号更新时,种子才会变化。 - ver = self.analysis(lt).get("version", 0) - return hashlib.md5(f"{self.config['seed']}_{lt}_{date}_v{ver}".encode()).hexdigest() - - -class Collector: - def __init__(self,p): self.p = p - - def fetch_ssq(self, count=100): - """采集双色球数据,自动容灾切换""" - cfg = self.p.config["ssq"] - hist = self.p.history("ssq") - - # 主数据源: 500.com - records = self._fetch_ssq_500(hist, cfg, count) - - # 如果主数据源失败,切换到备用数据源 17500.cn - if records is None or len(records) == 0: - print("[采集] 500.com 无数据,切换到备用数据源 17500.cn") - records = self._fetch_ssq_17500(hist, cfg) - - return records - - def _fetch_ssq_500(self, hist, cfg, count=100): - """从500.com获取双色球数据(主数据源)- 仅获取新期数据""" - # 确定请求范围 - if hist: - latest = max(int(r["issue"]) for r in hist) - # 已采集过的不用再采集,只请求最新一期之后的数据 - # 但500.com需要至少一个范围,如果最新期号之后还没开奖, - # 缩小请求范围到 latest ~ latest(即只请求最新一期确认) - now = datetime.datetime.now() - # 估算当前应有的最新期号 - est_end = int(now.strftime("%y")) * 1000 + int(now.strftime("%j")) // 2 - if latest >= est_end - 3: - # 数据已是最新,仅请求最新一期确认是否已有新开奖 - start, end = latest, latest + 10 - else: - # 有较大差距,从最新期+1开始请求 - start, end = latest + 1, latest + min(count, est_end - latest + 10) - else: - # 首次采集,获取100期 - now = datetime.datetime.now() - end = int(now.strftime("%y")) * 1000 + int(now.strftime("%j")) // 2 - start = end - count + 1 - if end < start: end = start + count - 1 - cfg["latest_fetch_start"], cfg["latest_fetch_end"] = str(start), str(end) - self.p.save_config() - url = LOTTERY_RULES["ssq"]["api_url"].format(start=start, end=end) - print("[采集] 500.com 期号%d~%d" % (start, end)) - r = subprocess.run(["curl","-s",url,"-H","User-Agent: Mozilla/5.0"], capture_output=True, timeout=30) - if r.returncode != 0: - print("[警告] 500.com 请求失败") - return None - html = r.stdout.decode('utf-8', errors='ignore') - records = self._parse_ssq_500(html) - if not records: - print("[警告] 500.com 返回空数据") - return None - return self._merge_records("ssq", hist, records, cfg) - - def _fetch_ssq_17500(self, hist, cfg): - """从17500.cn获取双色球数据(备用数据源)""" - print("[采集] 17500.cn 最近30期") - r = subprocess.run( - ["curl","-s","-X","POST","-d","lotid=ssq", - "https://www.17500.cn/api/kaijiang/getlist", - "-H","User-Agent: Mozilla/5.0", - "-H","Referer: https://www.17500.cn/kj/list-ssq.html", - "-H","Content-Type: application/x-www-form-urlencoded"], - capture_output=True, timeout=30) - if r.returncode != 0: - print("[警告] 17500.cn 请求失败") - return None - html = r.stdout.decode('utf-8', errors='ignore') - records = self._parse_ssq_17500(html) - if not records: - print("[警告] 17500.cn 解析失败") - return None - # 更新期数范围 - if records: - issues = sorted(int(r["issue"]) for r in records) - cfg["latest_fetch_start"], cfg["latest_fetch_end"] = str(issues[0]), str(issues[-1]) - return self._merge_records("ssq", hist, records, cfg) - - def _merge_records(self, lt, hist, records, cfg): - """合并新记录到历史数据""" - exist = {r["issue"] for r in hist} - new = [r for r in records if r["issue"] not in exist] - if new: - hist.extend(new) - hist.sort(key=lambda x:int(x["issue"])) - if len(hist)>500: hist=hist[-500:] - self.p.save_history(lt, hist) - cfg["latest_issue"] = max(r["issue"] for r in hist) - self.p.save_config() - print("[采集] %s +%d条 共%d条" % (LOTTERY_RULES[lt]["name"], len(new), len(hist))) - return new if new else None - - def _parse_ssq_500(self, html): - """解析500.com的HTML表格""" - m = re.search(r'