CodeForge
c4bb8632ca
- 新增 model_full_report_dlt.py: 大乐透专用全量报告脚本 - 统一4块结构: 回滚迭代→数据采集→推理分析→机选号码 - 统一术语: 前区/后区(大乐透) 对齐 红球/蓝球(双色球) - 统一评分: TOP10前区+TOP5后区 归一化概率输出 - 更新 SKILL.md: 双色球/大乐透统一模板+文件结构
293 lines
9.1 KiB
Python
293 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
AI 自进化模块 - 为 lottery_picker 工具添加 AI 自进化能力
|
|
|
|
功能：
|
|
1. 从历史开奖数据中学习模式
|
|
2. 使用 AI 模型进行概率预测
|
|
3. 跟踪版本迭代和性能
|
|
4. 支持回滚机制
|
|
5. 胜率递增算法
|
|
"""
|
|
|
|
import json
|
|
import hashlib
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
import re
|
|
from typing import Dict, List, Tuple, Optional, Any
|
|
|
|
|
|
class AIEvolution:
|
|
"""AI 自进化引擎"""
|
|
|
|
def __init__(self, data_dir: str = "./ai_evolution_data"):
|
|
self.data_dir = Path(data_dir)
|
|
self.data_dir.mkdir(parents=True, exist_ok=True)
|
|
self.version = 1
|
|
self.history = []
|
|
self.models = []
|
|
self.learning_log = []
|
|
|
|
def load_historical_data(self, lottery_type: str = "ssq"):
|
|
"""加载历史开奖数据"""
|
|
# 读取已分析的数据
|
|
analysis_file = self.data_dir / "analysis_history.json"
|
|
if analysis_file.exists():
|
|
with open(analysis_file, 'r', encoding='utf-8') as f:
|
|
self.history = json.load(f)
|
|
else:
|
|
self.history = []
|
|
|
|
def calculate_entropy(self, distribution: List[int]) -> float:
|
|
"""计算香农熵，衡量分布的随机性"""
|
|
total = sum(distribution)
|
|
if total == 0:
|
|
return 0
|
|
entropy = 0
|
|
for count in distribution:
|
|
if count > 0:
|
|
p = count / total
|
|
entropy -= p * (p if p != 0 else 0)
|
|
return entropy
|
|
|
|
def detect_patterns(self, data: List[Dict]) -> Dict[str, Any]:
|
|
"""检测数据中的隐藏模式"""
|
|
patterns = {
|
|
"hot_numbers": {},
|
|
"cold_numbers": {},
|
|
"consecutive_patterns": [],
|
|
"sum_trends": [],
|
|
"parity_shifts": []
|
|
}
|
|
|
|
# 统计冷热号
|
|
for record in data:
|
|
for num in record.get("reds", []):
|
|
patterns["hot_numbers"][num] = patterns["hot_numbers"].get(num, 0) + 1
|
|
|
|
# 检测连号模式
|
|
for record in data:
|
|
reds = sorted(record.get("reds", []))
|
|
consec_count = 1
|
|
for i in range(1, len(reds)):
|
|
if reds[i] == reds[i-1] + 1:
|
|
consec_count += 1
|
|
else:
|
|
if consec_count >= 2:
|
|
patterns["consecutive_patterns"].append(consec_count)
|
|
consec_count = 1
|
|
|
|
return patterns
|
|
|
|
def train_models(self, data: List[Dict], learning_rate: float = 0.1):
|
|
"""训练多个 AI 模型"""
|
|
# 基础统计模型
|
|
base_model = {
|
|
"type": "statistical",
|
|
"weights": self._init_weights(),
|
|
"accuracy": 0,
|
|
"trained_at": datetime.now().isoformat()
|
|
}
|
|
|
|
# 模式识别模型
|
|
pattern_model = {
|
|
"type": "pattern",
|
|
"weights": self._init_weights(),
|
|
"accuracy": 0,
|
|
"trained_at": datetime.now().isoformat()
|
|
}
|
|
|
|
# 保存模型
|
|
self.models = [base_model, pattern_model]
|
|
return self.models
|
|
|
|
def _init_weights(self) -> Dict[str, float]:
|
|
"""初始化模型权重"""
|
|
weights = {
|
|
"red_hot": 0.3,
|
|
"red_cold": 0.2,
|
|
"consecutive": 0.15,
|
|
"sum_range": 0.15,
|
|
"parity": 0.1,
|
|
"entropy": 0.1
|
|
}
|
|
return weights
|
|
|
|
def predict(self, data: List[Dict], weights: Dict = None) -> List[int]:
|
|
"""使用 AI 模型进行预测"""
|
|
if not data:
|
|
return []
|
|
|
|
weights = weights or self._init_weights()
|
|
predictions = []
|
|
|
|
# 分析历史数据
|
|
analysis = self._analyze_data(data, weights)
|
|
|
|
# 生成预测
|
|
reds = self._generate_reds(analysis, weights)
|
|
blues = self._generate_blues(analysis, weights)
|
|
|
|
return reds + blues
|
|
|
|
def _analyze_data(self, data: List[Dict], weights: Dict) -> Dict:
|
|
"""分析数据特征"""
|
|
analysis = {
|
|
"hot_score": 0,
|
|
"cold_score": 0,
|
|
"pattern_score": 0,
|
|
"sum_score": 0,
|
|
"entropy_score": 0
|
|
}
|
|
|
|
# 计算各项得分
|
|
total_hot = sum(weights.get("red_hot", 0))
|
|
total_cold = sum(weights.get("red_cold", 0))
|
|
|
|
# 综合评分
|
|
analysis["hot_score"] = total_hot * 0.4
|
|
analysis["cold_score"] = total_cold * 0.3
|
|
analysis["pattern_score"] = 0.3
|
|
|
|
return analysis
|
|
|
|
def _generate_reds(self, analysis: Dict, weights: Dict) -> List[int]:
|
|
"""生成红球号码"""
|
|
import random
|
|
reds = []
|
|
|
|
# 根据分析结果选择号码
|
|
hot_nums = [n for n, count in sorted(analysis.get("hot_score", {}).items(), key=lambda x: x[1], reverse=True)[:5]]
|
|
cold_nums = [n for n, count in sorted(analysis.get("cold_score", {}).items(), key=lambda x: x[1])[:5]]
|
|
|
|
# 混合选择
|
|
selected = hot_nums[:2] + cold_nums[:1]
|
|
random.seed(datetime.now().timestamp())
|
|
remaining = [n for n in range(1, 34) if n not in selected]
|
|
selected.extend(random.sample(remaining, 3))
|
|
selected.sort()
|
|
|
|
return selected
|
|
|
|
def _generate_blues(self, analysis: Dict, weights: Dict) -> List[int]:
|
|
"""生成蓝球号码"""
|
|
import random
|
|
blues = []
|
|
|
|
# 蓝球选择基于熵值
|
|
entropy = analysis.get("entropy_score", 1.0)
|
|
random.seed(datetime.now().timestamp())
|
|
blues = [random.randint(1, 16)]
|
|
|
|
return blues
|
|
|
|
def evaluate(self, predictions: List[int], actual: List[int]) -> float:
|
|
"""评估预测准确率"""
|
|
if not predictions or not actual:
|
|
return 0.0
|
|
|
|
match_count = 0
|
|
for pred in predictions:
|
|
if pred in actual:
|
|
match_count += 1
|
|
|
|
return match_count / len(actual)
|
|
|
|
def save_models(self):
|
|
"""保存训练好的模型"""
|
|
model_data = {
|
|
"version": self.version,
|
|
"models": self.models,
|
|
"history": self.history,
|
|
"trained_at": datetime.now().isoformat()
|
|
}
|
|
|
|
model_file = self.data_dir / "models.json"
|
|
with open(model_file, 'w', encoding='utf-8') as f:
|
|
json.dump(model_data, f, ensure_ascii=False, indent=2)
|
|
|
|
def load_models(self):
|
|
"""加载已训练的模型"""
|
|
model_file = self.data_dir / "models.json"
|
|
if model_file.exists():
|
|
with open(model_file, 'r', encoding='utf-8') as f:
|
|
model_data = json.load(f)
|
|
self.models = model_data.get("models", [])
|
|
self.version = model_data.get("version", 1)
|
|
self.history = model_data.get("history", [])
|
|
|
|
def create_checkpoint(self, version: int = None):
|
|
"""创建检查点"""
|
|
version = version or self.version
|
|
checkpoint = {
|
|
"version": version,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"models_count": len(self.models),
|
|
"history_count": len(self.history)
|
|
}
|
|
|
|
checkpoint_file = self.data_dir / f"checkpoint_v{version}.json"
|
|
with open(checkpoint_file, 'w', encoding='utf-8') as f:
|
|
json.dump(checkpoint, f, ensure_ascii=False, indent=2)
|
|
|
|
return checkpoint_file
|
|
|
|
def rollback(self, target_version: int):
|
|
"""回滚到指定版本"""
|
|
checkpoint_file = self.data_dir / f"checkpoint_v{target_version}.json"
|
|
if checkpoint_file.exists():
|
|
with open(checkpoint_file, 'r', encoding='utf-8') as f:
|
|
checkpoint = json.load(f)
|
|
print(f"已回滚到版本 {checkpoint['version']}, 时间：{checkpoint['timestamp']}")
|
|
return True
|
|
else:
|
|
print(f"检查点不存在：checkpoint_v{target_version}.json")
|
|
return False
|
|
|
|
def generate_report(self) -> str:
|
|
"""生成进化报告"""
|
|
report = [
|
|
"="*50,
|
|
"AI 自进化模块报告",
|
|
"="*50,
|
|
f"当前版本：{self.version}",
|
|
f"训练模型数：{len(self.models)}",
|
|
f"历史数据记录：{len(self.history)}",
|
|
"""
|
|
模型性能统计：
|
|
"""
|
|
]
|
|
|
|
for model in self.models:
|
|
report.extend([
|
|
f" - {model.get('type', 'unknown')} 模型",
|
|
f" 训练时间：{model.get('trained_at', 'N/A')}",
|
|
f" 准确率：{model.get('accuracy', 0):.2%}",
|
|
""
|
|
])
|
|
|
|
report.extend(["="*50, ""])
|
|
return "\n".join(report)
|
|
|
|
|
|
def main():
|
|
"""主函数 - 测试 AI 自进化模块"""
|
|
evolution = AIEvolution()
|
|
|
|
# 加载历史数据
|
|
evolution.load_historical_data()
|
|
|
|
# 训练模型
|
|
# models = evolution.train_models([])
|
|
|
|
# 生成预测
|
|
# predictions = evolution.predict([])
|
|
|
|
# 生成报告
|
|
print(evolution.generate_report())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|