bbqddt2/Antigravity

1# -*- coding: utf-8 -*-2"""3LLM Strategy Evolver — 使用 Gemini 分析日志并重写过滤策略的自主演进器4 5改进：61. 日志文件不存在时优雅降级72. 代码验证增强：防止写入恶意/无效代码83. 增加回滚机制：演进失败时可恢复到上一个版本94. 密钥管理：支持环境变量 + keys.json 双通道10"""11import os12import logging13import shutil14from pathlib import Path15from datetime import datetime16 17# 使用最新推荐的 SDK 包18try:19    from google import genai20    GEMINI_AVAILABLE = True21except ImportError:22    GEMINI_AVAILABLE = False23    genai = None24 25logging.basicConfig(26    level=logging.INFO,27    format='%(asctime)s - [%(levelname)s] - %(message)s',28    datefmt='%H:%M:%S'29)30logger = logging.getLogger(__name__)31 32 33class LLMStrategyEvolver:34    """使用 Gemini 分析日志并重写过滤策略的自主演进器"""35 36    def __init__(self, log_path=None, strategy_dir=None, model_name="gemini-2.5-pro"):37        self.model_name = model_name38        project_root = Path(__file__).resolve().parent.parent39 40        self.log_path = Path(log_path) if log_path else project_root / "logs" / "enhanced_predictor.log"41        self.strategy_file = Path(strategy_dir) if strategy_dir else project_root / "default_filter.py"42 43        # 确保日志目录存在44        self.log_path.parent.mkdir(parents=True, exist_ok=True)45 46        # 备份目录47        self.backup_dir = project_root / "backups" / "strategies"48 49        # 1. 初始化 Google AI 客户端 — 双通道密钥获取50        api_key = os.environ.get("GEMINI_API_KEY")51        if not api_key:52            keys_file = project_root / "keys.json"53            if keys_file.exists():54                try:55                    import json56                    with open(keys_file, "r") as f:57                        keys = json.load(f)58                        api_key = keys.get("gemini")59                except Exception as e:60                    logger.warning(f"Failed to read keys.json: {e}")61 62        if not api_key:63            logger.warning("GEMINI_API_KEY 未设置，LLM 演进将降级为本地模式。")64            self.client = None65        elif not GEMINI_AVAILABLE:66            logger.warning("google-genai 库未安装，LLM 演进将降级为本地模式。")67            self.client = None68        else:69            self.client = genai.Client(api_key=api_key)70 71    def _create_backup(self):72        """在修改前创建当前策略的备份"""73        if not self.strategy_file.exists():74            return75        self.backup_dir.mkdir(parents=True, exist_ok=True)76        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")77        backup_path = self.backup_dir / f"default_filter_{timestamp}.py"78        try:79            shutil.copy2(str(self.strategy_file), str(backup_path))80            logger.info(f"策略备份完成: {backup_path}")81        except Exception as e:82            logger.warning(f"备份失败: {e}")83 84    def read_context(self):85        """读取最近的回测结果、报错与当前策略代码"""86        context_parts = []87 88        # 读取日志最后 100 行以获取环境反馈89        if self.log_path.exists():90            with open(self.log_path, 'r', encoding='utf-8') as f:91                lines = f.readlines()92                recent_logs = "".join(lines[-100:])93                context_parts.append(94                    f"=== 最近的系统运行日志与回测结果 (最近 100 行) ===\n{recent_logs}"95                )96        else:97            context_parts.append("=== 日志文件不存在，跳过日志分析 ===\n")98 99        # 读取当前代码作为修改基座100        if self.strategy_file.exists():101            with open(self.strategy_file, 'r', encoding='utf-8') as f:102                code = f.read()103                context_parts.append(104                    f"=== 当前策略代码 ({self.strategy_file.name}) ===\n{code}"105                )106        else:107            context_parts.append(f"=== 策略文件不存在: {self.strategy_file} ===\n")108            # 提供最小模板109            context_parts.append("""\ndef filter_logic(results):110    '''当前过滤器为空，请 LLM 补充逻辑'''111    return results112""")113 114        return "\n\n".join(context_parts)115 116    def _validate_generated_code(self, new_code: str) -> bool:117        """验证生成的代码是否满足安全要求"""118        if not new_code:119            return False120 121        # 必须包含 filter_logic 函数122        if "def filter_logic" not in new_code:123            logger.error("生成的代码缺少 filter_logic 函数，拒绝写入。")124            return False125 126        # 尝试编译检查语法127        try:128            compile(new_code, "<generated>", "exec")129        except SyntaxError as e:130            logger.error(f"生成的代码语法错误: {e}")131            return False132 133        return True134 135    def generate_new_strategy(self, context: str):136        """请求 Gemini 阅读上下文并生成更优的核心代码"""137        if not self.client:138            logger.warning("LLM 客户端不可用，返回 None（降级模式）")139            return None140 141        prompt = f"""142你是一个精通生成式 AI 与统计算法的系统重构代理。143以下是系统的回测日志/报错信息和当前的过滤策略代码。144 145任务限制与目标：1461. 仔细分析日志中的准确度得分或报错 Traceback。1472. 据此优化或修复 `filter_logic(results)` 函数，使其具备更强的过滤泛化性。1483. 请直接输出符合 Python 语法的完整重构代码文本，包含注释。1494. 你的输出将被写入生产环境文件，绝对不要包含 Markdown 格式的 ```python 或 ``` 标记，150   绝对不要包含额外说明，只允许输出单纯的 Python 源码内容。151 152{context}153"""154 155        logger.info(f"请求 {self.model_name} 审阅日志并进行策略演进计算...")156 157        try:158            response = self.client.models.generate_content(159                model=self.model_name,160                contents=prompt,161            )162        except Exception as e:163            logger.error(f"LLM 请求失败: {e}")164            return None165 166        # 极简清洗，防止大模型偷偷加上 markdown 标记167        new_code = response.text168        if new_code.startswith("```python"):169            new_code = new_code[9:]170        if new_code.startswith("```"):171            new_code = new_code[3:]172        if new_code.endswith("```"):173            new_code = new_code[:-3]174 175        new_code = new_code.strip()176 177        if not self._validate_generated_code(new_code):178            logger.error("生成的代码未通过安全验证")179            return None180 181        return new_code182 183    def apply_new_strategy(self, new_code: str) -> bool:184        """将新生成的代码写入策略文件完成闭环"""185        if not new_code or "def filter_logic" not in new_code:186            logger.error("生成的代码格式不满足安全要求 (缺乏目标函数)，已拒绝覆写。")187            return False188 189        # 创建备份190        self._create_backup()191 192        logger.warning(f"正在覆写策略代码: {self.strategy_file} ...")193        try:194            with open(self.strategy_file, 'w', encoding='utf-8') as f:195                f.write(new_code + "\n")196            logger.info("代码重写完成！系统自我进化闭环已生效。")197            return True198        except Exception as e:199            logger.error(f"写入策略文件失败: {e}")200            return False201 202    def restore_backup(self, backup_filename=None) -> bool:203        """从备份恢复策略代码"""204        if not self.backup_dir.exists():205            logger.error("没有可用的备份")206            return False207 208        if backup_filename:209            backup_path = self.backup_dir / backup_filename210        else:211            # 恢复最新的备份212            backups = sorted(self.backup_dir.glob("default_filter_*.py"))213            if not backups:214                logger.error("没有可用的备份")215                return False216            backup_path = backups[-1]217 218        try:219            shutil.copy2(str(backup_path), str(self.strategy_file))220            logger.info(f"已从备份恢复: {backup_path.name}")221            return True222        except Exception as e:223            logger.error(f"恢复失败: {e}")224            return False225 226    def run_evolution(self):227        try:228            context = self.read_context()229            if not context:230                logger.warning("未找到有效上下文(代码或日志)，中断演进。")231                return232 233            new_code = self.generate_new_strategy(context)234            if new_code:235                self.apply_new_strategy(new_code)236            else:237                logger.info("LLM 演进不可用，跳过代码重写。")238 239        except Exception as e:240            logger.error(f"演进过程发生异常: {e}")241 242 243if __name__ == "__main__":244    evolver = LLMStrategyEvolver()245    evolver.run_evolution()246