parameter_optimizer.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
参数迭代优化模块
自动遍历核心参数组合，每一组参数均做全样本历史回测，筛选最优参数
"""

import os
import logging
import pandas as pd
import json
from pathlib import Path
from datetime import datetime
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 配置日志
logger = logging.getLogger(__name__)

class ParameterOptimizer:
    """参数迭代优化类"""
    
    def __init__(self):
        """初始化参数优化对象"""
        self.optimization_dir = "results/optimization"
        Path(self.optimization_dir).mkdir(parents=True, exist_ok=True)
        
        # 初始化回测模块
        from src.backtester import Backtester
        self.backtester = Backtester()
        
        # 初始化大模型模块
        from src.model_api import ModelAPI
        self.model_api = ModelAPI()
        
        # 参数池
        self.param_pool = self._build_parameter_pool()
    
    def _build_parameter_pool(self):
        """构建参数池"""
        return {
            "learning_window": ["60日", "90日", "120日"],
            "profit_threshold": ["2%", "3%", "4%", "5%"],
            "stock_count": ["5只", "10只", "15只"],
            "volume_ratio_threshold": ["1.2", "1.5", "2.0"]
        }
    
    def _generate_parameter_combinations(self):
        """生成参数组合"""
        from itertools import product
        
        combinations = []
        for params in product(
            self.param_pool["learning_window"],
            self.param_pool["profit_threshold"],
            self.param_pool["stock_count"],
            self.param_pool["volume_ratio_threshold"]
        ):
            combinations.append({
                "learning_window": params[0],
                "profit_threshold": params[1],
                "stock_count": params[2],
                "volume_ratio_threshold": params[3]
            })
        
        logger.info(f"参数组合总数：{len(combinations)}")
        return combinations
    
    def run(self, param_combinations=None):
        """执行参数优化"""
        try:
            logger.info("开始参数优化...")
            
            # 使用传入的参数组合或生成所有组合
            if not param_combinations:
                param_combinations = self._generate_parameter_combinations()
            
            logger.info(f"待测试参数组合数：{len(param_combinations)}")
            
            # 初始化优化结果
            optimization_results = []
            
            # 对每个参数组合进行回测
            for idx, params in enumerate(param_combinations):
                logger.info(f"测试参数组合 {idx+1}/{len(param_combinations)}：{params}")
                
                try:
                    # 执行回测
                    report = self.backtester.run(params)
                    
                    # 记录优化结果
                    optimization_results.append({
                        "params": params,
                        "win_rate": report['win_rate'],
                        "avg_return": report['avg_return'],
                        "total_return": report['total_return'],
                        "max_drawdown": report['max_drawdown'],
                        "profit_loss_ratio": report['profit_loss_ratio'],
                        "avg_daily_return": report['avg_daily_return']
                    })
                    
                    logger.info(f"参数组合 {idx+1} 回测完成")
                    logger.info(f"胜率：{report['win_rate']:.2%}，收益率：{report['total_return']:.2%}，最大回撤：{report['max_drawdown']:.2%}")
                
                except Exception as e:
                    logger.error(f"参数组合 {idx+1} 回测失败：{e}")
                    continue
            
            if not optimization_results:
                logger.warning("无有效优化结果")
                return self._get_default_parameters()
            
            # 筛选最优参数组合
            best_params = self._select_best_parameters(optimization_results)
            
            # 保存优化结果
            self._save_optimization_results(optimization_results)
            
            logger.info("参数优化完成")
            logger.info(f"最优参数组合：{best_params}")
            
            return best_params
            
        except Exception as e:
            logger.error(f"参数优化失败：{e}")
            return self._get_default_parameters()
    
    def _select_best_parameters(self, optimization_results):
        """筛选最优参数组合"""
        logger.info("开始筛选最优参数组合...")
        
        # 转换为DataFrame以便分析
        df_results = pd.DataFrame([
            {
                "learning_window": res['params']['learning_window'],
                "profit_threshold": res['params']['profit_threshold'],
                "stock_count": res['params']['stock_count'],
                "volume_ratio_threshold": res['params']['volume_ratio_threshold'],
                "win_rate": res['win_rate'],
                "avg_return": res['avg_return'],
                "total_return": res['total_return'],
                "max_drawdown": res['max_drawdown'],
                "profit_loss_ratio": res['profit_loss_ratio'],
                "avg_daily_return": res['avg_daily_return']
            }
            for res in optimization_results
        ])
        
        # 保存优化结果到CSV
        csv_path = Path(self.optimization_dir) / "parameter_optimization_log.csv"
        df_results.to_csv(csv_path, index=False, encoding='utf-8-sig')
        
        logger.info(f"优化结果已保存到：{csv_path}")
        
        # 筛选最优参数（多目标优化）
        # 优先考虑胜率≥50%，盈亏比≥2
        valid_results = df_results[
            (df_results['win_rate'] >= 0.5) &
            (df_results['profit_loss_ratio'] >= 2)
        ]
        
        if valid_results.empty:
            logger.warning("无符合要求的参数组合，使用默认参数")
            return self._get_default_parameters()
        
        # 综合评分（权重：胜率50%，盈亏比30%，收益率20%）
        valid_results['score'] = (
            valid_results['win_rate'] * 0.5 +
            valid_results['profit_loss_ratio'] * 0.3 +
            valid_results['total_return'] * 0.2
        )
        
        # 选择评分最高的参数组合
        best_idx = valid_results['score'].idxmax()
        best_result = valid_results.loc[best_idx]
        
        # 转换为字典格式
        best_params = {
            "learning_window": best_result['learning_window'],
            "profit_threshold": best_result['profit_threshold'],
            "stock_count": best_result['stock_count'],
            "volume_ratio_threshold": best_result['volume_ratio_threshold']
        }
        
        logger.info(f"最优参数组合筛选完成：{best_params}")
        return best_params
    
    def _save_optimization_results(self, optimization_results):
        """保存优化结果"""
        try:
            # 保存为JSON文件
            json_path = Path(self.optimization_dir) / "parameter_optimization_results.json"
            with open(json_path, 'w', encoding='utf-8') as f:
                json.dump(optimization_results, f, ensure_ascii=False, indent=2)
            
            logger.info(f"优化结果已保存到：{json_path}")
            
            # 保存优化日志
            log_path = Path(self.optimization_dir) / "optimization_log.txt"
            with open(log_path, 'a', encoding='utf-8') as f:
                f.write(f"优化时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
                f.write(f"测试参数组合数：{len(optimization_results)}\n")
                f.write("-" * 80 + "\n")
            
            return True
            
        except Exception as e:
            logger.error(f"优化结果保存失败：{e}")
            return False
    
    def save_best_parameters(self, best_params):
        """保存最优参数到本地配置文件"""
        try:
            params_file = Path(self.optimization_dir) / "best_parameters.json"
            with open(params_file, 'w', encoding='utf-8') as f:
                json.dump(best_params, f, ensure_ascii=False, indent=2)
            
            logger.info(f"最优参数已保存到：{params_file}")
            return True
            
        except Exception as e:
            logger.error(f"最优参数保存失败：{e}")
            return False
    
    def load_best_parameters(self):
        """加载最优参数"""
        params_file = Path(self.optimization_dir) / "best_parameters.json"
        
        if params_file.exists():
            try:
                with open(params_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception as e:
                logger.error(f"最优参数加载失败：{e}")
        
        return self._get_default_parameters()
    
    def _get_default_parameters(self):
        """获取默认参数"""
        return {
            "learning_window": "90日",
            "profit_threshold": "3%",
            "stock_count": "10只",
            "volume_ratio_threshold": "1.5"
        }
    
    def get_optimization_summary(self):
        """获取优化结果汇总"""
        try:
            csv_path = Path(self.optimization_dir) / "parameter_optimization_log.csv"
            
            if csv_path.exists():
                df_summary = pd.read_csv(csv_path)
                return df_summary
            else:
                logger.warning("优化结果文件不存在")
                return pd.DataFrame()
                
        except Exception as e:
            logger.error(f"优化结果汇总获取失败：{e}")
            return pd.DataFrame()
    
    def test_optimization(self):
        """测试参数优化功能"""
        logger.info("开始测试参数优化功能...")
        
        # 测试参数池构建
        logger.info(f"参数池大小：{len(self.param_pool)}")
        
        # 测试参数组合生成（只生成3个组合用于测试）
        test_combinations = self._generate_parameter_combinations()[:3]
        logger.info(f"测试参数组合数：{len(test_combinations)}")
        
        # 测试执行优化（使用简化日期范围）
        logger.info("开始执行参数优化测试...")
        self.backtester.start_date = "20241201"
        self.backtester.end_date = "20241231"
        
        test_results = []
        for idx, params in enumerate(test_combinations):
            logger.info(f"测试参数组合 {idx+1}/{len(test_combinations)}：{params}")
            
            try:
                report = self.backtester.run(params, "20241201", "20241231")
                test_results.append({
                    "params": params,
                    "win_rate": report['win_rate'],
                    "avg_return": report['avg_return'],
                    "total_return": report['total_return'],
                    "max_drawdown": report['max_drawdown'],
                    "profit_loss_ratio": report['profit_loss_ratio'],
                    "avg_daily_return": report['avg_daily_return']
                })
                
                logger.info(f"参数组合 {idx+1} 回测完成")
                logger.info(f"胜率：{report['win_rate']:.2%}，收益率：{report['total_return']:.2%}，最大回撤：{report['max_drawdown']:.2%}")
                
            except Exception as e:
                logger.error(f"参数组合 {idx+1} 回测失败：{e}")
                continue
        
        # 测试结果分析
        if test_results:
            best_params = self._select_best_parameters(test_results)
            logger.info(f"最优参数组合：{best_params}")
        
        logger.info("参数优化功能测试完成")

if __name__ == "__main__":
    # 配置日志
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )
    
    # 测试参数优化模块
    optimizer = ParameterOptimizer()
    optimizer.test_optimization()
    # best_params = optimizer.run()
    # optimizer.save_best_parameters(best_params)
    # logger.info(f"最优参数：{best_params}")