JUKUAN/Lib/research/future_ma5_deviation_analysis.py

"""
期货合约5日移动平均线偏离分析
研究期货合约在大幅偏离其5天移动平均线（K5）后是否出现重大反转走势

本程序实现完整的分析流程：
1. 数据结构设置 - 获取主力期货合约数据
2. 数据收集 - 扩展时间范围获取OHLC数据
3. 距离计算 - 计算价格相对K5的上下偏离距离
4. 阈值确定 - 使用百分位数作为极端偏差阈值
5. 信号检测与分析 - 识别极端偏差并分析后续走势
6. 结果输出 - 生成Range_1偏差统计和信号分析结果CSV

注：程序使用动态获取的主力合约进行分析，确保分析结果基于真实的交易活动

作者: jukuan研究团队
日期: 2025-09
适用平台: 聚宽在线研究平台
"""

import pandas as pd
import numpy as np
from jqdata import *
import datetime
import warnings
warnings.filterwarnings('ignore')

# =====================================================================================
# 分析配置参数 - 集中配置部分
# =====================================================================================

class AnalysisConfig:
    """期货5日移动平均线偏离分析配置参数"""
    
    # ==================== 时间范围设置 ====================
    ANALYSIS_START_DATE = datetime.datetime(2025, 8, 1)  # 分析开始日期
    ANALYSIS_END_DATE = datetime.datetime(2025, 8, 20)   # 分析结束日期
    
    # ==================== 时间扩展参数 ====================
    HISTORICAL_DAYS = 365      # 历史分析期：优势期前天数
    FORWARD_DAYS = 20         # 正向分析期：优势期后天数
    FORWARD_MONITOR_DAYS = 10  # 前向监控窗口：交易日数
    
    # ==================== 偏差阈值设置 ====================
    DEVIATION_PERCENTILE = 90  # 偏差阈值百分位数（范围：0-100）
    
    # ==================== 分析规模控制 ====================
    MAX_ANALYSIS_CONTRACTS = 10  # 最大分析合约数量限制
    
    
    # ==================== 移动平均线参数 ====================
    MA_PERIOD = 5  # K线移动平均线周期（天数）
    
    # ==================== 输出设置 ====================
    OUTPUT_ENCODING = 'utf-8-sig'  # 输出文件编码格式
    VERBOSE_LOGGING = True          # 是否打印详细日志
    
    @classmethod
    def print_config(cls):
        """打印当前配置信息"""
        print("=== 期货5日移动平均线偏离分析配置 ===")
        print(f"分析时间范围: {cls.ANALYSIS_START_DATE.strftime('%Y-%m-%d')} 至 {cls.ANALYSIS_END_DATE.strftime('%Y-%m-%d')}")
        print(f"历史分析期: {cls.HISTORICAL_DAYS}天")
        print(f"前向分析期: {cls.FORWARD_DAYS}天")
        print(f"前向监控窗口: {cls.FORWARD_MONITOR_DAYS}个交易日")
        print(f"偏差阈值百分位数: {cls.DEVIATION_PERCENTILE}%")
        print(f"最大分析合约数: {cls.MAX_ANALYSIS_CONTRACTS}")
        print(f"合约类型: 主力合约")
        print(f"移动平均线周期: {cls.MA_PERIOD}日")
        print(f"详细日志: {'开启' if cls.VERBOSE_LOGGING else '关闭'}")
        print("=" * 50)

class FutureMA5DeviationAnalyzer:
    """期货5日移动平均线偏离分析器"""
    
    def __init__(self, config=None):
        """初始化分析器"""
        # 使用配置类参数
        if config is None:
            config = AnalysisConfig
        
        self.config = config
        self.historical_days = config.HISTORICAL_DAYS
        self.forward_days = config.FORWARD_DAYS  
        self.deviation_percentile = config.DEVIATION_PERCENTILE
        self.forward_monitor_days = config.FORWARD_MONITOR_DAYS
        self.analysis_start_date = config.ANALYSIS_START_DATE
        self.analysis_end_date = config.ANALYSIS_END_DATE
        self.max_analysis_contracts = config.MAX_ANALYSIS_CONTRACTS
        self.ma_period = config.MA_PERIOD
        self.output_encoding = config.OUTPUT_ENCODING
        self.verbose_logging = config.VERBOSE_LOGGING
        
        # 期货品种与交易所的映射关系
        self.exchange_map = {
            # 大连商品交易所 (XDCE)
            'A': 'XDCE', 'B': 'XDCE', 'C': 'XDCE', 'CS': 'XDCE', 'FB': 'XDCE', 
            'I': 'XDCE', 'J': 'XDCE', 'JD': 'XDCE', 'JM': 'XDCE', 'L': 'XDCE',
            'M': 'XDCE', 'P': 'XDCE', 'PP': 'XDCE', 'PG': 'XDCE', 'RR': 'XDCE',
            'V': 'XDCE', 'Y': 'XDCE', 'EB': 'XDCE', 'EG': 'XDCE', 'LH': 'XDCE',
            
            # 郑州商品交易所 (XZCE)
            'AP': 'XZCE', 'CF': 'XZCE', 'CY': 'XZCE', 'FG': 'XZCE', 'JR': 'XZCE',
            'LR': 'XZCE', 'MA': 'XZCE', 'OI': 'XZCE', 'PM': 'XZCE', 'RI': 'XZCE',
            'RM': 'XZCE', 'RS': 'XZCE', 'SF': 'XZCE', 'SM': 'XZCE', 'SR': 'XZCE',
            'TA': 'XZCE', 'WH': 'XZCE', 'ZC': 'XZCE', 'CJ': 'XZCE', 
            'ME': 'XZCE', 'PF': 'XZCE', 'PK': 'XZCE', 'RO': 'XZCE', 'SA': 'XZCE',
            'TC': 'XZCE', 'UR': 'XZCE', 'WS': 'XZCE', 'WT': 'XZCE',
            
            # 上海期货交易所 (XSGE)
            'AG': 'XSGE', 'AL': 'XSGE', 'AU': 'XSGE', 'BU': 'XSGE', 'CU': 'XSGE',
            'FU': 'XSGE', 'HC': 'XSGE', 'NI': 'XSGE', 'PB': 'XSGE', 'RB': 'XSGE',
            'RU': 'XSGE', 'SN': 'XSGE', 'SP': 'XSGE', 'SS': 'XSGE', 'WR': 'XSGE',
            'ZN': 'XSGE',
            
            # 上海国际能源交易中心 (XINE)
            'BC': 'XINE', 'LU': 'XINE', 'NR': 'XINE', 'SC': 'XINE',
            
            # 中金所 (CCFX)
            'IC': 'CCFX', 'IF': 'CCFX', 'IH': 'CCFX', 'T': 'CCFX', 'TF': 'CCFX', 'TS': 'CCFX',
            
            # 广期所 (GFEX)
            'SI': 'GFEX', 'LC': 'GFEX', 'PS': 'GFEX'
        }
        
        # 存储结果的字典
        self.contract_data = {}  # 主合约数据结构
        self.signal_results = {}  # 信号检测结果
        
        if self.verbose_logging:
            print("初始化期货5日移动平均线偏离分析器")
            print(f"配置参数 - 历史分析期: {self.historical_days}天, 前向分析期: {self.forward_days}天")
            print(f"偏差阈值: {self.deviation_percentile}分位数, 监控窗口: {self.forward_monitor_days}个交易日")
            print(f"最大分析合约数: {self.max_analysis_contracts}, 移动平均线周期: {self.ma_period}日")
            print(f"合约类型: 主力合约")
    
    def _is_trading_day(self, check_date):
        """
        检查是否为交易日
        
        参数:
            check_date (date): 要检查的日期
        
        返回:
            bool: 是否为交易日
        """
        try:
            # 使用聚宽API检查交易日
            trade_days = get_trade_days(end_date=check_date, count=1)
            if len(trade_days) > 0:
                return trade_days[0].date() == check_date
            return False
        except:
            # 如果API调用失败，简单判断是否为工作日
            return check_date.weekday() < 5
    
    def _get_symbol_dominant_history(self, symbol, start_date, end_date):
        """
        获取单个标的的主力合约历史变化
        
        参数:
            symbol (str): 标的编码
            start_date (date): 开始日期
            end_date (date): 结束日期
        
        返回:
            list: 主力合约记录列表
        """
        results = []
        current_date = start_date
        current_dominant = None
        period_start = None
        
        if self.verbose_logging:
            print(f"  获取 {symbol} 主力合约历史...")
        
        # 按日遍历时间范围
        while current_date <= end_date:
            
            # 跳过非交易日
            if not self._is_trading_day(current_date):
                current_date += datetime.timedelta(days=1)
                continue
            
            try:
                # 获取当日主力合约
                dominant_contract = get_dominant_future(symbol, current_date)
                
                if dominant_contract is None:
                    current_date += datetime.timedelta(days=1)
                    continue
                    
                # 如果是新的主力合约
                if dominant_contract != current_dominant:
                    
                    # 如果不是第一个合约，结束上一个合约的记录
                    if current_dominant is not None and period_start is not None:
                        results.append({
                            'symbol': symbol,
                            'dominant_contract': current_dominant,
                            'start_date': period_start,
                            'end_date': current_date - datetime.timedelta(days=1)
                        })
                    
                    # 开始新合约的记录
                    current_dominant = dominant_contract
                    period_start = current_date
                    
                    if self.verbose_logging:
                        print(f"    {current_date}: 主力合约变更为 {dominant_contract}")
                
            except Exception as e:
                if self.verbose_logging:
                    print(f"    获取 {current_date} 的主力合约时出错: {str(e)}")
            
            current_date += datetime.timedelta(days=1)
        
        # 处理最后一个合约
        if current_dominant is not None and period_start is not None:
            results.append({
                'symbol': symbol,
                'dominant_contract': current_dominant,
                'start_date': period_start,
                'end_date': end_date
            })
        
        return results
    
    def get_dominant_contracts_history(self, symbol_list, start_date, end_date):
        """
        获取指定时间范围内多个标的的主力合约历史变化
        
        参数:
            symbol_list (list): 标的编码列表
            start_date (date): 开始日期
            end_date (date): 结束日期
        
        返回:
            pandas.DataFrame: 包含主力合约变化历史的数据框
        """
        if self.verbose_logging:
            print(f"获取主力合约数据...")
            print(f"标的列表: {symbol_list}")
            print(f"时间范围: {start_date} 至 {end_date}")
        
        # 存储所有结果的列表
        all_results = []
        
        # 逐个处理每个标的
        for symbol in symbol_list:
            if self.verbose_logging:
                print(f"\n处理标的: {symbol}")
            
            try:
                # 获取该标的在指定时间范围内的主力合约历史
                symbol_results = self._get_symbol_dominant_history(symbol, start_date, end_date)
                all_results.extend(symbol_results)
                
            except Exception as e:
                if self.verbose_logging:
                    print(f"处理标的 {symbol} 时发生错误: {str(e)}")
                continue
        
        # 将结果转换为DataFrame
        if all_results:
            result_df = pd.DataFrame(all_results)
            result_df = result_df.sort_values(['symbol', 'start_date']).reset_index(drop=True)
            
            if self.verbose_logging:
                print(f"\n成功获取 {len(result_df)} 条主力合约记录")
            return result_df
        else:
            if self.verbose_logging:
                print("\n未获取到任何主力合约数据")
            return pd.DataFrame(columns=['symbol', 'dominant_contract', 'start_date', 'end_date'])

    def build_contract_structure(self):
        """
        构建期货主力合约数据结构
        返回结构：Key=完整合约代码, Value=[start_date, end_date]
        """
        if self.verbose_logging:
            print("\n=== 步骤1: 构建期货主力合约数据结构 ===")
        
        # 分析时间范围
        analysis_start = self.analysis_start_date
        analysis_end = self.analysis_end_date
        
        if self.verbose_logging:
            print(f"分析时间范围: {analysis_start.strftime('%Y-%m-%d')} 至 {analysis_end.strftime('%Y-%m-%d')}")
            print(f"合约类型: 主力合约")
        
        # 构建主力合约数据结构
        contract_dict = self._build_dominant_contract_structure(analysis_start, analysis_end)
        
        self.contract_data = contract_dict
        if self.verbose_logging:
            print(f"构建完成，共{len(contract_dict)}个期货合约")
        return contract_dict
    
    def _build_dominant_contract_structure(self, analysis_start, analysis_end):
        """构建基于主力合约的数据结构"""
        if self.verbose_logging:
            print("使用主力合约获取逻辑...")
        
        # 获取要分析的期货品种列表
        symbol_list = list(self.exchange_map.keys())[:self.max_analysis_contracts]
        
        # 获取主力合约历史数据
        dominant_df = self.get_dominant_contracts_history(
            symbol_list, 
            analysis_start.date(), 
            analysis_end.date()
        )
        
        contract_dict = {}
        
        if not dominant_df.empty:
            # 将DataFrame转换为字典结构
            for _, row in dominant_df.iterrows():
                contract_code = row['dominant_contract']
                start_date = row['start_date'] if isinstance(row['start_date'], datetime.date) else row['start_date'].date()
                end_date = row['end_date'] if isinstance(row['end_date'], datetime.date) else row['end_date'].date()
                
                contract_dict[contract_code] = [start_date, end_date]
                
                if self.verbose_logging:
                    print(f"  添加主力合约: {contract_code} ({start_date} 至 {end_date})")
        
        return contract_dict
    
    def collect_extended_data(self):
        """
        数据收集：为每个合约收集扩展时间范围的数据
        扩展期限：从(start_date-365天)到(end_date+100天)
        """
        if self.verbose_logging:
            print("\n=== 步骤2: 收集扩展时间范围的数据 ===")
        
        extended_data = {}
        
        for contract_code, period in self.contract_data.items():
            start_date_1, end_date_1 = period
            
            # 计算扩展期限
            start_date_2 = start_date_1 - datetime.timedelta(days=self.historical_days)
            end_date_2 = end_date_1 + datetime.timedelta(days=self.forward_days)
            
            if self.verbose_logging:
                print(f"获取 {contract_code} 数据")
                print(f"  原始期间: {start_date_1} 至 {end_date_1}")
                print(f"  扩展期间: {start_date_2} 至 {end_date_2}")
            
            try:
                # 获取日线数据
                data = get_price(
                    contract_code,
                    start_date=start_date_2,
                    end_date=end_date_2,
                    frequency='daily',
                    fields=['open', 'close', 'high', 'low', 'volume'],
                    skip_paused=False,
                    panel=False
                )
                
                if data is not None and len(data) > 0:
                    # 计算移动平均线
                    data['K5'] = data['close'].rolling(window=self.ma_period).mean()
                    
                    extended_data[contract_code] = {
                        'data': data,
                        'original_start': start_date_1,
                        'original_end': end_date_1,
                        'extended_start': start_date_2,
                        'extended_end': end_date_2
                    }
                    
                    if self.verbose_logging:
                        print(f"  成功获取{len(data)}条数据记录")
                else:
                    if self.verbose_logging:
                        print(f"  未获取到数据")
                    
            except Exception as e:
                if self.verbose_logging:
                    print(f"  获取数据失败: {str(e)}")
                continue
        
        if self.verbose_logging:
            print(f"数据收集完成，共{len(extended_data)}个合约有效数据")
        return extended_data
    
    def calculate_k5_distances(self, extended_data):
        """
        计算距离：每个交易日价格相对K5的上下偏离距离
        """
        if self.verbose_logging:
            print("\n=== 步骤3: 计算K5偏离距离 ===")
        
        distance_data = {}
        
        for contract_code, contract_info in extended_data.items():
            data = contract_info['data']
            if self.verbose_logging:
                print(f"计算 {contract_code} 的K5距离")
            
            # 初始化距离列
            data['upper_distance'] = np.nan
            data['lower_distance'] = np.nan
            
            # 计算每日的上下偏离距离
            for idx, row in data.iterrows():
                if pd.isna(row['K5']):  # K5不可用则跳过
                    continue
                
                K5 = row['K5']
                high = row['high']
                low = row['low']
                
                # 计算K5与高低价的差值
                k5_minus_high = K5 - high
                k5_minus_low = K5 - low
                
                if k5_minus_high > 0 and k5_minus_low > 0:
                    # 高低都低于K5
                    data.loc[idx, 'upper_distance'] = np.nan
                    data.loc[idx, 'lower_distance'] = abs(K5 - low)
                    
                elif k5_minus_high < 0 and k5_minus_low < 0:
                    # 高低都在K5以上
                    data.loc[idx, 'upper_distance'] = abs(K5 - high)
                    data.loc[idx, 'lower_distance'] = np.nan
                    
                elif k5_minus_high > 0 and k5_minus_low < 0:
                    # 高低于K5，低高于K5（K5介于高与低之间）
                    data.loc[idx, 'upper_distance'] = abs(K5 - high)
                    data.loc[idx, 'lower_distance'] = abs(K5 - low)
                
                # 处理零情况
                if abs(k5_minus_high) < 1e-6:  # high == K5
                    data.loc[idx, 'upper_distance'] = 0.0
                if abs(k5_minus_low) < 1e-6:   # low == K5
                    data.loc[idx, 'lower_distance'] = 0.0
            
            distance_data[contract_code] = contract_info
            
            # 统计有效距离数据
            valid_upper = data['upper_distance'].notna().sum()
            valid_lower = data['lower_distance'].notna().sum()
            if self.verbose_logging:
                print(f"  上偏离距离有效数据: {valid_upper}条, 下偏离距离有效数据: {valid_lower}条")
        
        if self.verbose_logging:
            print("K5距离计算完成")
        return distance_data
    
    def determine_thresholds(self, distance_data):
        """
        阈值确定：使用Range_1数据计算百分位数阈值
        Range_1: start_date_2 <= date < start_date_1
        """
        if self.verbose_logging:
            print(f"\n=== 步骤4: 确定{self.deviation_percentile}分位数阈值 ===")
        
        threshold_data = {}
        
        for contract_code, contract_info in distance_data.items():
            data = contract_info['data']
            start_date_1 = contract_info['original_start']
            start_date_2 = contract_info['extended_start']
            
            if self.verbose_logging:
                print(f"计算 {contract_code} 的阈值")
                print(f"  Range_1期间: {start_date_2} 至 {start_date_1}")
            
            # 筛选Range_1数据
            range_1_mask = (data.index.date >= start_date_2) & (data.index.date < start_date_1)
            range_1_data = data[range_1_mask]
            
            if len(range_1_data) == 0:
                if self.verbose_logging:
                    print(f"  Range_1无有效数据，跳过")
                continue
            
            # 计算上下偏离距离的分布和阈值
            upper_distances = range_1_data['upper_distance'].dropna()
            lower_distances = range_1_data['lower_distance'].dropna()
            
            # 百分位数阈值
            percentile_value = self.deviation_percentile / 100.0
            upper_threshold = upper_distances.quantile(percentile_value) if len(upper_distances) > 0 else np.nan
            lower_threshold = lower_distances.quantile(percentile_value) if len(lower_distances) > 0 else np.nan
            
            threshold_data[contract_code] = {
                'contract_info': contract_info,
                'upper_threshold': upper_threshold,
                'lower_threshold': lower_threshold,
                'range_1_stats': {
                    'total_days': len(range_1_data),
                    'upper_valid_days': len(upper_distances),
                    'lower_valid_days': len(lower_distances),
                    'upward_deviation_days': len(lower_distances[lower_distances >= lower_threshold]) if not pd.isna(lower_threshold) else 0,
                    'downward_deviation_days': len(upper_distances[upper_distances >= upper_threshold]) if not pd.isna(upper_threshold) else 0
                }
            }
            
            if self.verbose_logging:
                print(f"  上偏离{self.deviation_percentile}分位数阈值: {upper_threshold:.4f}" if not pd.isna(upper_threshold) else f"  上偏离阈值: 无效")
                print(f"  下偏离{self.deviation_percentile}分位数阈值: {lower_threshold:.4f}" if not pd.isna(lower_threshold) else f"  下偏离阈值: 无效")
            
        if self.verbose_logging:
            print("阈值确定完成")
        return threshold_data
    
    def detect_signals_and_analyze(self, threshold_data):
        """
        信号检测与分析：在Range_2期间扫描极端偏差并分析后续走势
        Range_2: start_date_1 <= date <= end_date_2
        """
        if self.verbose_logging:
            print("\n=== 步骤5: 信号检测与前瞻性分析 ===")
        
        signal_dict = {}
        
        for contract_code, threshold_info in threshold_data.items():
            contract_info = threshold_info['contract_info']
            data = contract_info['data']
            start_date_1 = contract_info['original_start']
            end_date_2 = contract_info['extended_end']
            
            upper_threshold = threshold_info['upper_threshold']
            lower_threshold = threshold_info['lower_threshold']
            
            if self.verbose_logging:
                print(f"\n分析 {contract_code} 的信号")
                print(f"  Range_2期间: {start_date_1} 至 {end_date_2}")
            
            # 筛选Range_2数据
            range_2_mask = (data.index.date >= start_date_1) & (data.index.date <= end_date_2)
            range_2_data = data[range_2_mask]
            
            contract_signals = {}
            
            # 扫描极端偏差天数
            for idx, (signal_date, row) in enumerate(range_2_data.iterrows()):
                upper_distance = row['upper_distance']
                lower_distance = row['lower_distance']
                
                # 检查是否触发阈值
                trigger_upper = not pd.isna(upper_distance) and not pd.isna(upper_threshold) and upper_distance >= upper_threshold
                trigger_lower = not pd.isna(lower_distance) and not pd.isna(lower_threshold) and lower_distance >= lower_threshold
                
                if trigger_upper or trigger_lower:
                    # 确定偏离方向和参考价格
                    if trigger_lower:  # lower_distance触发表示向上偏离（价格在K5之上）
                        deviation_direction = "up"
                        deviation_distance = lower_distance
                        reference_category = "low"
                        reference_price = row['low']
                    else:  # upper_distance触发表示向下偏离（价格在K5之下）
                        deviation_direction = "down"
                        deviation_distance = upper_distance
                        reference_category = "high"
                        reference_price = row['high']
                    
                    if self.verbose_logging:
                        print(f"  发现信号: {signal_date.strftime('%Y-%m-%d')} {deviation_direction}偏离 距离{deviation_distance:.4f}")
                    
                    # 前瞻性分析
                    forward_analysis = self._analyze_forward_performance(
                        range_2_data, idx, deviation_direction, reference_price, self.forward_monitor_days
                    )
                    
                    # 计算最大利润和风险
                    max_profit_info = self._find_max_profit(forward_analysis)
                    max_risk_info = self._find_max_risk(forward_analysis)
                    
                    contract_signals[signal_date] = {
                        "deviation_direction": deviation_direction,
                        "deviation_distance": float(deviation_distance),
                        "reference_category": reference_category,
                        "reference_price": float(reference_price),
                        "forward_analysis": forward_analysis,
                        "max_profit": max_profit_info,
                        "max_risk": max_risk_info
                    }
            
            if contract_signals:
                signal_dict[contract_code] = contract_signals
                if self.verbose_logging:
                    print(f"  检测到{len(contract_signals)}个有效信号")
            else:
                if self.verbose_logging:
                    print(f"  未检测到有效信号")
        
        if self.verbose_logging:
            print(f"\n信号检测完成，共{sum(len(signals) for signals in signal_dict.values())}个信号")
        self.signal_results = signal_dict
        return signal_dict
    
    def _analyze_forward_performance(self, data, signal_idx, deviation_direction, reference_price, monitor_days):
        """分析信号后续N个交易日的表现"""
        forward_analysis = {}
        
        for i in range(1, monitor_days + 1):
            future_idx = signal_idx + i
            if future_idx < len(data):
                future_row = data.iloc[future_idx]
                future_date = data.index[future_idx]
                
                # 根据偏离方向选择对应价格
                if deviation_direction == "up":
                    corresponding_price = future_row['low']
                    corresponding_return = (reference_price - corresponding_price) / reference_price
                    reverse_return = (future_row['high'] - reference_price) / reference_price
                else:  # down
                    corresponding_price = future_row['high']
                    corresponding_return = (corresponding_price - reference_price) / reference_price
                    reverse_return = (reference_price - future_row['low']) / reference_price
                
                forward_analysis[future_date] = {
                    "date": future_date.strftime('%Y-%m-%d'),
                    "corresponding_price": float(corresponding_price),
                    "corresponding_return": float(corresponding_return),
                    "reverse_category": "high" if deviation_direction == "up" else "low",
                    "reverse_return": float(reverse_return)
                }
        
        return forward_analysis
    
    def _find_max_profit(self, forward_analysis):
        """找到最大利润日期和收益率"""
        if not forward_analysis:
            return {"date": None, "return": 0.0}
        
        max_return = max(day_data["corresponding_return"] for day_data in forward_analysis.values())
        max_date = None
        
        for date, day_data in forward_analysis.items():
            if day_data["corresponding_return"] == max_return:
                max_date = date.strftime('%Y-%m-%d')
                break
        
        return {"date": max_date, "return": float(max_return)}
    
    def _find_max_risk(self, forward_analysis):
        """找到最大风险日期和收益率"""
        if not forward_analysis:
            return {"date": None, "return": 0.0}
        
        max_risk = max(day_data["reverse_return"] for day_data in forward_analysis.values())
        max_risk_date = None
        
        for date, day_data in forward_analysis.items():
            if day_data["reverse_return"] == max_risk:
                max_risk_date = date.strftime('%Y-%m-%d')
                break
        
        return {"date": max_risk_date, "return": float(max_risk)}
    
    def generate_csv_outputs(self, threshold_data, signal_dict):
        """生成两个CSV文件输出"""
        if self.verbose_logging:
            print("\n=== 步骤6: 生成CSV输出文件 ===")
        
        timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
        
        # CSV 1: Range_1偏差统计
        self._generate_range1_stats_csv(threshold_data, timestamp)
        
        # CSV 2: 信号分析结果
        self._generate_signal_analysis_csv(signal_dict, timestamp)
        
        if self.verbose_logging:
            print("CSV文件生成完成")
    
    def _generate_range1_stats_csv(self, threshold_data, timestamp):
        """生成Range_1偏差统计CSV"""
        stats_data = []
        
        for contract_code, threshold_info in threshold_data.items():
            stats = threshold_info['range_1_stats']
            total_days = stats['total_days']
            
            upward_ratio = stats['upward_deviation_days'] / total_days if total_days > 0 else 0
            downward_ratio = stats['downward_deviation_days'] / total_days if total_days > 0 else 0
            
            stats_data.append({
                'contract_code': contract_code,
                'upward_deviation_ratio': round(upward_ratio, 4),
                'downward_deviation_ratio': round(downward_ratio, 4)
            })
        
        if stats_data:
            stats_df = pd.DataFrame(stats_data)
            filename = f'range1_deviation_stats_{timestamp}.csv'
            stats_df.to_csv(filename, index=False, encoding=self.output_encoding)
            if self.verbose_logging:
                print(f"Range_1偏差统计保存至: {filename}")
    
    def _generate_signal_analysis_csv(self, signal_dict, timestamp):
        """生成信号分析结果CSV"""
        signal_data = []
        
        for contract_code, contract_signals in signal_dict.items():
            for signal_date, signal_info in contract_signals.items():
                signal_data.append({
                    'contract_code': contract_code,
                    'signal_date': signal_date.strftime('%Y-%m-%d'),
                    'deviation_direction': signal_info['deviation_direction'],
                    'max_profit_date': signal_info['max_profit']['date'],
                    'max_profit_return': round(signal_info['max_profit']['return'], 4),
                    'max_risk_date': signal_info['max_risk']['date'],
                    'max_risk_return': round(signal_info['max_risk']['return'], 4)
                })
        
        if signal_data:
            signal_df = pd.DataFrame(signal_data)
            filename = f'signal_analysis_results_{timestamp}.csv'
            signal_df.to_csv(filename, index=False, encoding=self.output_encoding)
            if self.verbose_logging:
                print(f"信号分析结果保存至: {filename}")
    
    def run_complete_analysis(self):
        """执行完整的分析流程"""
        if self.verbose_logging:
            print("开始执行期货5日移动平均线偏离分析")
            print("=" * 60)
        
        try:
            # 步骤1: 构建合约数据结构
            contract_structure = self.build_contract_structure()
            
            # 步骤2: 收集扩展数据
            extended_data = self.collect_extended_data()
            if not extended_data:
                if self.verbose_logging:
                    print("未获取到有效数据，分析终止")
                return
            
            # 步骤3: 计算K5距离
            distance_data = self.calculate_k5_distances(extended_data)
            
            # 步骤4: 确定阈值
            threshold_data = self.determine_thresholds(distance_data)
            if not threshold_data:
                if self.verbose_logging:
                    print("未能确定有效阈值，分析终止")
                return
            
            # 步骤5: 信号检测与分析
            signal_dict = self.detect_signals_and_analyze(threshold_data)
            
            # 步骤6: 生成输出文件
            self.generate_csv_outputs(threshold_data, signal_dict)
            
            # 分析汇总
            total_contracts = len(contract_structure)
            valid_contracts = len(threshold_data)
            total_signals = sum(len(signals) for signals in signal_dict.values())
            
            if self.verbose_logging:
                print("\n" + "=" * 60)
                print("分析完成汇总:")
                print(f"总合约数: {total_contracts}")
                print(f"有效合约数: {valid_contracts}")
                print(f"检测信号数: {total_signals}")
            
            return {
                'contract_structure': contract_structure,
                'threshold_data': threshold_data,
                'signal_results': signal_dict,
                'summary': {
                    'total_contracts': total_contracts,
                    'valid_contracts': valid_contracts,
                    'total_signals': total_signals
                }
            }
            
        except Exception as e:
            if self.verbose_logging:
                print(f"分析过程中出现错误: {str(e)}")
                import traceback
                traceback.print_exc()
            return None


# =====================================================================================
# 主程序入口
# =====================================================================================

def run_ma5_deviation_analysis(config=None):
    """运行期货5日移动平均线偏离分析"""
    if config is None:
        config = AnalysisConfig
    
    # 打印配置信息
    config.print_config()
    
    # 创建分析器并运行
    analyzer = FutureMA5DeviationAnalyzer(config)
    results = analyzer.run_complete_analysis()
    return results

# 执行分析
if __name__ == "__main__":
    print("期货5日移动平均线偏离分析工具")
    print("研究期货合约在大幅偏离K5后的反转走势规律")
    print("使用动态获取的主力合约进行精准分析")
    print("适用于聚宽在线研究平台")
    
    results = run_ma5_deviation_analysis()
    
    if results:
        print("\n✅ 分析执行成功！")
        summary = results['summary']
        print(f"📊 结果摘要:")
        print(f"   - 分析合约数: {summary['total_contracts']}")
        print(f"   - 有效数据合约: {summary['valid_contracts']}")
        print(f"   - 检测信号总数: {summary['total_signals']}")
    else:
        print("\n❌ 分析执行失败，请检查错误信息")