CodeForge
c4bb8632ca
- 新增 model_full_report_dlt.py: 大乐透专用全量报告脚本 - 统一4块结构: 回滚迭代→数据采集→推理分析→机选号码 - 统一术语: 前区/后区(大乐透) 对齐 红球/蓝球(双色球) - 统一评分: TOP10前区+TOP5后区 归一化概率输出 - 更新 SKILL.md: 双色球/大乐透统一模板+文件结构
286 lines
9.6 KiB
Python
286 lines
9.6 KiB
Python
#!/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()
|