daily_adjuster.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, timedelta
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

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

class DailyAdjuster:
    """每日自调整类"""
    
    def __init__(self):
        """初始化每日自调整对象"""
        self.adjustment_dir = "results/adjustments"
        Path(self.adjustment_dir).mkdir(parents=True, exist_ok=True)
        
        # 初始化数据管理模块
        from src.data_manager import DataManager
        self.data_manager = DataManager()
        
        # 初始化回测模块
        from src.backtester import Backtester
        self.backtester = Backtester()
        
        # 初始化参数优化模块
        from src.parameter_optimizer import ParameterOptimizer
        self.optimizer = ParameterOptimizer()
        
        # 初始化大模型模块
        from src.model_api import ModelAPI
        self.model_api = ModelAPI()
        
        # 模型有效性阈值
        self.accuracy_threshold = float(os.getenv("ACCURACY_THRESHOLD", 0.5))
        self.profit_loss_threshold = float(os.getenv("PROFIT_LOSS_THRESHOLD", 2.0))
    
    def run(self, trade_date=None):
        """执行每日自调整"""
        try:
            logger.info("开始每日自调整...")
            
            if not trade_date:
                trade_date = datetime.now().strftime('%Y%m%d')
            
            # 检查是否为交易日
            if not self._is_trading_day(trade_date):
                logger.warning(f"{trade_date} 不是交易日，每日自调整终止")
                return False
            
            # 完成当日数据下载
            logger.info("开始下载当日数据...")
            self.data_manager.update_history_data(trade_date, trade_date)
            logger.info("当日数据下载完成")
            
            # 完成当日数据回测
            logger.info("开始当日数据回测...")
            backtest_result = self._run_daily_backtest(trade_date)
            logger.info("当日数据回测完成")
            
            # 验证规律有效性
            logger.info("开始验证规律有效性...")
            is_valid = self._validate_trading_rules(backtest_result)
            
            if is_valid:
                logger.info("当前交易规律有效性良好，无需调整")
                return True
            else:
                logger.warning("当前交易规律有效性下降，需要重新优化")
                
                # 重新优化参数
                logger.info("开始重新优化参数...")
                best_params = self.optimizer.run()
                self.optimizer.save_best_parameters(best_params)
                
                # 重新挖掘规律
                logger.info("开始重新挖掘交易规律...")
                self._reextract_trading_rules(best_params)
                
                logger.info("每日自调整完成，已优化参数和规律")
                return True
                
        except Exception as e:
            logger.error(f"每日自调整失败：{e}")
            return False
    
    def _is_trading_day(self, date_str):
        """检查是否为交易日"""
        try:
            from chinese_calendar import is_workday
            date = datetime.strptime(date_str, '%Y%m%d').date()
            return is_workday(date)
        except Exception as e:
            logger.error(f"交易日检查失败：{e}")
            return False
    
    def _run_daily_backtest(self, trade_date):
        """执行当日数据回测"""
        try:
            logger.info(f"执行 {trade_date} 当日数据回测...")
            
            # 使用最优参数进行回测
            best_params = self.optimizer.load_best_parameters()
            
            # 回测日期范围：包含当日和前几日的数据
            start_date = (datetime.strptime(trade_date, '%Y%m%d') - timedelta(days=7)).strftime('%Y%m%d')
            end_date = trade_date
            
            report = self.backtester.run(best_params, start_date, end_date)
            
            # 保存当日回测报告
            self.backtester.save_report(report, f"{trade_date}_daily")
            
            logger.info(f"当日回测完成，胜率：{report['win_rate']:.2%}，盈亏比：{report['profit_loss_ratio']:.2f}")
            return report
            
        except Exception as e:
            logger.error(f"当日回测失败：{e}")
            return self.backtester._get_empty_report()
    
    def _validate_trading_rules(self, backtest_result):
        """验证交易规律有效性"""
        logger.info("验证交易规律有效性...")
        
        # 检查核心指标是否达标
        win_rate_valid = backtest_result['win_rate'] >= self.accuracy_threshold
        profit_loss_valid = backtest_result['profit_loss_ratio'] >= self.profit_loss_threshold
        
        logger.info(f"胜率验证：{win_rate_valid} ({backtest_result['win_rate']:.2%} ≥ {self.accuracy_threshold:.2%})")
        logger.info(f"盈亏比验证：{profit_loss_valid} ({backtest_result['profit_loss_ratio']:.2f} ≥ {self.profit_loss_threshold:.2f})")
        
        return win_rate_valid and profit_loss_valid
    
    def _reextract_trading_rules(self, best_params):
        """重新挖掘交易规律"""
        try:
            logger.info("开始重新挖掘交易规律...")
            
            # 获取历史数据
            history_data = self._get_history_data_for_rule_reextraction(best_params)
            
            # 获取当日实时数据和消息
            today = datetime.now().strftime('%Y%m%d')
            realtime_data = self.data_manager.get_real_time_data(today)
            news_data = self.data_manager.get_news_data(today)
            
            # 重新提取规律
            rules = self.model_api.extract_trading_rules(
                history_data,
                realtime_data,
                news_data,
                best_params
            )
            
            logger.info(f"重新提取交易规律完成，共 {len(rules.get('trading_rules', []))} 条")
            return rules
            
        except Exception as e:
            logger.error(f"交易规律重新挖掘失败：{e}")
            return None
    
    def _get_history_data_for_rule_reextraction(self, best_params):
        """获取规律重挖掘所需的历史数据"""
        try:
            # 根据学习窗口确定历史数据范围
            window_days = 90
            window_str = best_params.get('learning_window', '90日')
            if '60' in window_str:
                window_days = 60
            elif '120' in window_str:
                window_days = 120
            
            end_date = datetime.now()
            start_date = end_date - timedelta(days=window_days)
            
            logger.info(f"规律重挖掘历史数据范围：{start_date.strftime('%Y%m%d')} 到 {end_date.strftime('%Y%m%d')}")
            
            # 获取所有有效股票的历史双数据
            history_data = self.data_manager.get_all_stock_data_for_backtest(
                start_date.strftime('%Y%m%d'),
                end_date.strftime('%Y%m%d')
            )
            
            # 合并所有股票数据
            all_history = []
            for ts_code, df_stock in history_data.items():
                df_stock['ts_code'] = ts_code
                all_history.append(df_stock)
            
            if all_history:
                return pd.concat(all_history, ignore_index=True)
            else:
                logger.warning("未获取到规律重挖掘所需的历史数据")
                return pd.DataFrame()
                
        except Exception as e:
            logger.error(f"历史数据获取失败：{e}")
            return pd.DataFrame()
    
    def save_adjustment_report(self, adjustment_result, trade_date=None):
        """保存每日自调整报告"""
        try:
            if not trade_date:
                trade_date = datetime.now().strftime('%Y%m%d')
            
            report_path = Path(self.adjustment_dir) / f"{trade_date}_adjustment_report.json"
            
            with open(report_path, 'w', encoding='utf-8') as f:
                json.dump(adjustment_result, f, ensure_ascii=False, indent=2)
            
            logger.info(f"每日自调整报告已保存到：{report_path}")
            return True
            
        except Exception as e:
            logger.error(f"每日自调整报告保存失败：{e}")
            return False
    
    def load_adjustment_report(self, trade_date=None):
        """加载每日自调整报告"""
        try:
            if not trade_date:
                trade_date = datetime.now().strftime('%Y%m%d')
            
            report_path = Path(self.adjustment_dir) / f"{trade_date}_adjustment_report.json"
            
            if report_path.exists():
                with open(report_path, 'r', encoding='utf-8') as f:
                    return json.load(f)
            else:
                logger.warning(f"每日自调整报告不存在：{report_path}")
                return None
                
        except Exception as e:
            logger.error(f"每日自调整报告加载失败：{e}")
            return None
    
    def test_adjustment(self):
        """测试每日自调整功能"""
        logger.info("开始测试每日自调整功能...")
        
        # 测试交易日检查
        test_date = datetime.now().strftime('%Y%m%d')
        is_trading = self._is_trading_day(test_date)
        logger.info(f"{test_date} 是交易日：{is_trading}")
        
        # 测试当日数据下载
        logger.info("测试当日数据下载...")
        try:
            self.data_manager.update_history_data(test_date, test_date)
            logger.info("当日数据下载成功")
        except Exception as e:
            logger.warning(f"当日数据下载测试失败：{e}")
        
        # 测试当日回测
        logger.info("测试当日回测...")
        try:
            backtest_result = self._run_daily_backtest(test_date)
            logger.info(f"当日回测结果：胜率 {backtest_result['win_rate']:.2%}，盈亏比 {backtest_result['profit_loss_ratio']:.2f}")
        except Exception as e:
            logger.warning(f"当日回测测试失败：{e}")
        
        # 测试规律有效性验证
        logger.info("测试规律有效性验证...")
        try:
            is_valid = self._validate_trading_rules({
                'win_rate': 0.65,
                'profit_loss_ratio': 2.3
            })
            logger.info(f"规律有效性验证结果：{is_valid}")
        except Exception as e:
            logger.warning(f"规律有效性验证测试失败：{e}")
        
        logger.info("每日自调整功能测试完成")

if __name__ == "__main__":
    # 配置日志
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )
    
    # 测试每日自调整模块
    adjuster = DailyAdjuster()
    adjuster.test_adjustment()
    # adjuster.run()