#!/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()