JUKUAN/Lib/research/future_candlestick_hatch_analysis.py

"""
期货烛台影线形态分析
研究期货合约中带有明显影线的K线对未来价格走势的预测能力

本程序实现完整的分析流程：
1. 数据结构设置 - 获取主力期货合约数据
2. 数据收集 - 扩展时间范围获取OHLC数据
3. 实体长度计算 - 计算历史平均实体长度作为阈值
4. 影线形态识别 - 识别满足条件的K线形态
5. 未来表现分析 - 分析影线方向与价格走势的关系
6. 结果输出 - 生成影线形态分析结果CSV

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

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

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

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

class CandlestickHatchConfig:
    """期货烛台影线形态分析配置参数"""
    
    # ==================== 时间范围设置 ====================
    ANALYSIS_START_DATE = datetime.datetime(2025, 8, 1)  # 分析开始日期
    ANALYSIS_END_DATE = datetime.datetime(2025, 8, 20)   # 分析结束日期
    
    # ==================== 时间扩展参数 ====================
    HISTORICAL_DAYS = 365      # 历史数据回溯期：优势期前天数
    FORWARD_DAYS = 15          # 正向分析期：优势期后天数
    FUTURE_MONITOR_DAYS = 15   # 未来表现跟踪期：交易日数
    
    # ==================== 影线形态参数 ====================
    HATCH_TO_BODY_RATIO = 1.2  # 影线与实体长度比率阈值
    OPPOSITE_HATCH_RATIO = 0.5  # 相反方向影线与实体长度比率阈值
    
    # ==================== 形态验证阈值 ====================
    MAX_REVENUE_THRESHOLD = 0.02   # 最大盈利阈值
    MAX_LOSS_THRESHOLD = 0.01     # 最大亏损阈值
    
    # ==================== 分析规模控制 ====================
    MAX_ANALYSIS_CONTRACTS = -1  # 最大分析合约数量限制
    
    # ==================== 输出设置 ====================
    OUTPUT_ENCODING = 'utf-8-sig'  # 输出文件编码格式
    VERBOSE_LOGGING = True          # 是否打印详细日志
    
    # ==================== 交易仓位管理 ====================
    INITIAL_TRADE_AMOUNT = 20000    # 单笔交易初始金额
    LOSS_RATE = 0.01               # 亏损交易的亏损率（可配置）
    PROFIT_DRAWDOWN_RATE = 0.02    # 盈利交易的回撤率（可配置）
    
    @classmethod
    def print_config(cls):
        """打印当前配置信息"""
        print("=== 期货烛台影线形态分析配置 ===")
        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.FUTURE_MONITOR_DAYS}个交易日")
        print(f"影线与实体长度比率阈值: {cls.HATCH_TO_BODY_RATIO}")
        print(f"相反方向影线比率阈值: {cls.OPPOSITE_HATCH_RATIO}")
        print(f"最大盈利阈值: {cls.MAX_REVENUE_THRESHOLD}")
        print(f"最大亏损阈值: {cls.MAX_LOSS_THRESHOLD}")
        print(f"最大分析合约数: {cls.MAX_ANALYSIS_CONTRACTS}")
        print(f"合约类型: 主力合约")
        print(f"详细日志: {'开启' if cls.VERBOSE_LOGGING else '关闭'}")
        print(f"初始交易金额: {cls.INITIAL_TRADE_AMOUNT}")
        print(f"亏损交易亏损率: {cls.LOSS_RATE}")
        print(f"盈利交易回撤率: {cls.PROFIT_DRAWDOWN_RATE}")
        print("=" * 50)

class FutureCandlestickHatchAnalyzer:
    """期货烛台影线形态分析器"""
    
    def __init__(self, config=None):
        """初始化分析器"""
        # 使用配置类参数
        if config is None:
            config = CandlestickHatchConfig
        
        self.config = config
        self.historical_days = config.HISTORICAL_DAYS
        self.forward_days = config.FORWARD_DAYS  
        self.future_monitor_days = config.FUTURE_MONITOR_DAYS
        self.hatch_to_body_ratio = config.HATCH_TO_BODY_RATIO
        self.opposite_hatch_ratio = config.OPPOSITE_HATCH_RATIO
        self.max_revenue_threshold = config.MAX_REVENUE_THRESHOLD
        self.max_loss_threshold = config.MAX_LOSS_THRESHOLD
        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.output_encoding = config.OUTPUT_ENCODING
        self.verbose_logging = config.VERBOSE_LOGGING
        
        # 交易仓位管理参数
        self.initial_trade_amount = config.INITIAL_TRADE_AMOUNT
        self.loss_rate = config.LOSS_RATE
        self.profit_drawdown_rate = config.PROFIT_DRAWDOWN_RATE
        
        # 期货品种与交易所的映射关系
        self.exchange_map = {
            # 大连商品交易所 (XDCE)
            'A': 'XDCE', 'B': 'XDCE', 'C': 'XDCE', 'CS': '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 = {}  # 主合约数据结构 - dict1
        self.hatch_patterns = {}  # 影线形态数据 - dict2
        
        # 投资组合管理
        self.portfolio_results = []  # 存储所有交易结果
        
        if self.verbose_logging:
            print("初始化期货烛台影线形态分析器")
            print(f"配置参数 - 历史数据回溯期: {self.historical_days}天, 前向分析期: {self.forward_days}天")
            print(f"影线与实体长度比率阈值: {self.hatch_to_body_ratio}, 相反影线比率阈值: {self.opposite_hatch_ratio}")
            print(f"最大盈利阈值: {self.max_revenue_threshold}, 最大亏损阈值: {self.max_loss_threshold}")
            print(f"监控窗口: {self.future_monitor_days}个交易日")
            print(f"最大分析合约数: {self.max_analysis_contracts}")
            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):
        """
        构建期货主力合约数据结构 (dict1)
        返回结构：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("使用主力合约获取逻辑...")
        
        # 获取要分析的期货品种列表
        if self.max_analysis_contracts > 0:
            symbol_list = list(self.exchange_map.keys())[:self.max_analysis_contracts]
        else:
            symbol_list = list(self.exchange_map.keys())
        
        # 获取主力合约历史数据
        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+15天)
        """
        if self.verbose_logging:
            print("\n=== 步骤2: 收集扩展时间范围的数据 ===")
        
        extended_data = {}
        
        for contract_code, period in self.contract_data.items():
            start_date, end_date = period
            
            # 计算扩展期限
            start_date_2 = start_date - datetime.timedelta(days=self.historical_days)
            end_date_2 = end_date + datetime.timedelta(days=self.forward_days)
            
            # if self.verbose_logging:
            #     print(f"获取 {contract_code} 数据")
            #     print(f"  原始期间: {start_date} 至 {end_date}")
            #     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:
                    extended_data[contract_code] = {
                        'data': data,
                        'original_start': start_date,
                        'original_end': end_date,
                        '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_body_thresholds(self, extended_data):
        """
        计算实体长度阈值：使用历史数据计算平均实体长度
        """
        if self.verbose_logging:
            print("\n=== 步骤3: 计算实体长度阈值 ===")
        
        threshold_data = {}
        
        for contract_code, contract_info in extended_data.items():
            data = contract_info['data']
            start_date = contract_info['original_start']
            start_date_2 = contract_info['extended_start']
            
            # if self.verbose_logging:
            #     print(f"计算 {contract_code} 的实体长度阈值")
            #     print(f"  历史期间: {start_date_2} 至 {start_date}")
            
            # 筛选历史数据：date >= start_date_2 且 date < start_date
            historical_mask = (data.index.date >= start_date_2) & (data.index.date < start_date)
            historical_data = data[historical_mask]
            
            if len(historical_data) == 0:
                if self.verbose_logging:
                    print(f"  历史数据无效，跳过")
                continue
            
            # 计算实体长度
            historical_data = historical_data.copy()
            historical_data['body_length'] = abs(historical_data['close'] - historical_data['open'])
            
            # 计算平均实体长度作为阈值
            body_threshold = historical_data['body_length'].mean()
            
            threshold_data[contract_code] = {
                'contract_info': contract_info,
                'body_threshold': body_threshold,
                'historical_days': len(historical_data)
            }
            
            # if self.verbose_logging:
            #     print(f"  实体长度阈值: {body_threshold:.4f} (基于{len(historical_data)}个历史交易日)")
        
        if self.verbose_logging:
            print("实体长度阈值计算完成")
        return threshold_data
    
    def identify_hatch_patterns(self, threshold_data):
        """
        影线形态识别：扫描符合条件的K线
        条件：实体长度 >= 阈值 且 影线长度 >= 1.2 × 实体长度
        """
        if self.verbose_logging:
            print(f"\n=== 步骤4: 识别影线形态 ===")
        
        pattern_dict = {}  # dict2结构
        
        for contract_code, threshold_info in threshold_data.items():
            contract_info = threshold_info['contract_info']
            data = contract_info['data']
            body_threshold = threshold_info['body_threshold']
            start_date = contract_info['original_start']  # 使用原始开始日期
            end_date = contract_info['original_end']
            
            # if self.verbose_logging:
            #     print(f"分析 {contract_code} 的影线形态")
            #     print(f"  扫描期间: {start_date} 至 {end_date}")
            #     print(f"  实体长度阈值: {body_threshold:.4f}")
            
            # 筛选扫描数据：从start_date到end_date（含）
            scan_mask = (data.index.date >= start_date) & (data.index.date <= end_date)
            scan_data = data[scan_mask]
            
            contract_patterns = {}
            pattern_count = 0
            
            for idx, (signal_date, row) in enumerate(scan_data.iterrows()):
                # 计算K线各项数值
                open_price = row['open']
                close_price = row['close']
                high_price = row['high']
                low_price = row['low']
                
                # 计算实体长度
                body_length = abs(close_price - open_price)
                
                # 检查实体长度是否满足阈值
                if body_length < body_threshold:
                    continue
                
                # 计算影线长度
                upper_hatch = high_price - max(open_price, close_price)
                lower_hatch = min(open_price, close_price) - low_price
                
                # 检查影线条件
                hatch_direction = None
                hatch_length = 0
                
                # 检查上影线条件：上影线长度符合要求 且 下影线长度小于实体长度的一半
                if (upper_hatch >= self.hatch_to_body_ratio * body_length and 
                    lower_hatch < self.opposite_hatch_ratio * body_length):
                    hatch_direction = 'up'
                    hatch_length = upper_hatch
                # 检查下影线条件：下影线长度符合要求 且 上影线长度小于实体长度的一半
                elif (lower_hatch >= self.hatch_to_body_ratio * body_length and 
                      upper_hatch < self.opposite_hatch_ratio * body_length):
                    hatch_direction = 'down'
                    hatch_length = lower_hatch
                
                # 如果满足影线条件，记录形态
                if hatch_direction is not None:
                    pattern_count += 1
                    
                    # dict3结构
                    pattern_info = {
                        'date': signal_date.strftime('%Y-%m-%d'),
                        'high': float(high_price),
                        'low': float(low_price),
                        'open': float(open_price),
                        'close': float(close_price),  # 标准收盘价
                        'body_len': float(body_length),
                        'hatch_dire': hatch_direction,
                        'hatch_len': float(hatch_length)
                    }
                    
                    contract_patterns[signal_date] = pattern_info
                    
                    # if self.verbose_logging:
                    #     opposite_hatch_len = lower_hatch if hatch_direction == 'up' else upper_hatch
                        # print(f"    发现形态: {signal_date.strftime('%Y-%m-%d')} {hatch_direction}影线 实体:{body_length:.4f} 影线:{hatch_length:.4f} 相反影线:{opposite_hatch_len:.4f}")
            
            if contract_patterns:
                pattern_dict[contract_code] = contract_patterns
            #     if self.verbose_logging:
            #         print(f"  检测到{pattern_count}个有效影线形态")
            # else:
            #     if self.verbose_logging:
            #         print(f"  未检测到有效影线形态")
        
        # if self.verbose_logging:
        #     total_patterns = sum(len(patterns) for patterns in pattern_dict.values())
        #     print(f"\n影线形态识别完成，共{total_patterns}个形态")
        
        self.hatch_patterns = pattern_dict
        return pattern_dict
    
    def analyze_future_performance(self, threshold_data, pattern_dict):
        """
        未来表现分析：分析影线形态对未来价格走势的预测能力
        交易方向：上影线做空，下影线做多
        """
        if self.verbose_logging:
            print(f"\n=== 步骤5: 分析未来表现 ===")
        
        enhanced_pattern_dict = {}
        
        for contract_code, contract_patterns in pattern_dict.items():
            # 获取对应的数据信息
            contract_info = threshold_data[contract_code]['contract_info']
            data = contract_info['data']
            end_date_2 = contract_info['extended_end']
            
            if self.verbose_logging:
                print(f"\n分析 {contract_code} 的未来表现")
                print(f"  监控窗口: {self.future_monitor_days}个交易日")
            
            enhanced_patterns = {}
            
            for signal_date, pattern_info in contract_patterns.items():
                hatch_direction = pattern_info['hatch_dire']
                standard_close = pattern_info['close']  # 标准收盘价
                print(f"形态日期：{signal_date.strftime('%Y-%m-%d')}， 标准收盘价: {standard_close}")
                
                # 在数据中找到信号日期的位置
                signal_idx = None
                for idx, (date, row) in enumerate(data.iterrows()):
                    if date.date() == signal_date.date():
                        signal_idx = idx
                        break
                
                if signal_idx is None:
                    if self.verbose_logging:
                        print(f"    未找到信号日期 {signal_date.strftime('%Y-%m-%d')} 在数据中")
                    continue
                
                # 分析未来N天的表现
                future_analysis = self._analyze_future_n_days(
                    data, signal_idx, hatch_direction, standard_close, 
                    self.future_monitor_days, end_date_2
                )
                
                # 计算最大收益和最大风险（原逻辑：与影线方向相反交易）
                max_revenue_info, min_revenue_info = self._calculate_max_revenue_risk(future_analysis, hatch_direction, standard_close)
                
                # 计算最大收益和最大风险（新逻辑：与影线方向相同交易）
                max_revenue_info_2, min_revenue_info_2 = self._calculate_max_revenue_risk_2(future_analysis, hatch_direction, standard_close)
                
                # if self.verbose_logging:
                #     print(f"      原逻辑最大收益: {max_revenue_info['revenue']:.4f} ({max_revenue_info['date']})")
                #     print(f"      原逻辑最大亏损: {min_revenue_info['revenue']:.4f} ({min_revenue_info['date']})")
                #     print(f"      新逻辑最大收益: {max_revenue_info_2['revenue']:.4f} ({max_revenue_info_2['date']})")
                #     print(f"      新逻辑最大亏损: {min_revenue_info_2['revenue']:.4f} ({min_revenue_info_2['date']})")

                # 计算conclusion字段（增强版，包含日损失检查）
                conclusion = self._calculate_conclusion(max_revenue_info, min_revenue_info, future_analysis, hatch_direction, standard_close)
                
                # 计算conclusion2字段（新逻辑：与影线方向相同交易）
                conclusion_2 = self._calculate_conclusion_2(max_revenue_info_2, min_revenue_info_2, future_analysis, hatch_direction, standard_close)
                
                # 计算新增字段
                body_direction, direction_consistency = self._calculate_direction_fields(pattern_info, hatch_direction)
                
                # 计算影线与实体长度比率
                hatch_to_body_ratio = pattern_info['hatch_len'] / pattern_info['body_len'] if pattern_info['body_len'] > 0 else 0
                
                # 计算移动平均线
                ma_values = self._calculate_moving_averages(data, signal_idx)
                
                # 计算基于收盘价与5K移动平均线关系的价格方向
                price_direction = self._calculate_price_direction(standard_close, ma_values['ma5'])
                
                # 计算趋势强度
                trend_strength = self._calculate_trend_strength(ma_values)
                
                # 动态仓位计算
                remaining_amount = self._calculate_dynamic_position(
                    conclusion, max_revenue_info['revenue'], self.initial_trade_amount
                )
                
                # 记录到投资组合（包含两种交易逻辑）
                self._add_portfolio_trade(
                    contract_code, 
                    signal_date.strftime('%Y-%m-%d'),
                    conclusion,
                    conclusion_2,
                    max_revenue_info['revenue'],
                    max_revenue_info_2['revenue'],
                    self.initial_trade_amount,
                    remaining_amount
                )
                
                # 增强pattern_info
                enhanced_pattern = pattern_info.copy()
                enhanced_pattern.update({
                    'max_revenue_date': max_revenue_info['date'],
                    'max_revenue': max_revenue_info['revenue'],
                    'min_revenue_date': min_revenue_info['date'],
                    'min_revenue': min_revenue_info['revenue'],
                    'max_revenue_date2': max_revenue_info_2['date'],
                    'max_revenue2': max_revenue_info_2['revenue'],
                    'min_revenue_date2': min_revenue_info_2['date'],
                    'min_revenue2': min_revenue_info_2['revenue'],
                    'conclusion': conclusion,
                    'conclusion2': conclusion_2,
                    'body_direction': body_direction,
                    'direction_consistency': direction_consistency,
                    'price_direction': price_direction,  # 新增：基于收盘价与5K移动平均线关系的价格方向
                    'trade_amount': float(self.initial_trade_amount),
                    'remaining_amount': float(remaining_amount),
                    'profit_loss': float(remaining_amount - self.initial_trade_amount),
                    'hatch_body_ratio': float(hatch_to_body_ratio),
                    'ma5': ma_values['ma5'],
                    'ma10': ma_values['ma10'],
                    'ma20': ma_values['ma20'],
                    'ma30': ma_values['ma30'],
                    'trend_strength': trend_strength
                })
                
                enhanced_patterns[signal_date] = enhanced_pattern
                
                if self.verbose_logging:
                    print(f"    {signal_date.strftime('%Y-%m-%d')} {hatch_direction}影线:")
                    print(f"      原逻辑 - 最大收益: {max_revenue_info['revenue']:.4f} ({max_revenue_info['date']})")
                    print(f"      原逻辑 - 最大亏损: {min_revenue_info['revenue']:.4f} ({min_revenue_info['date']})")
                    print(f"      原逻辑 - 结论: {conclusion}")
                    print(f"      新逻辑 - 最大收益: {max_revenue_info_2['revenue']:.4f} ({max_revenue_info_2['date']})")
                    print(f"      新逻辑 - 最大亏损: {min_revenue_info_2['revenue']:.4f} ({min_revenue_info_2['date']})")
                    print(f"      新逻辑 - 结论: {conclusion_2}")
                    print(f"      实体方向: {body_direction}")
                    print(f"      方向一致性: {direction_consistency}")
                    print(f"      价格方向: {price_direction} (收盘价vs5K线)")
            
            if enhanced_patterns:
                enhanced_pattern_dict[contract_code] = enhanced_patterns
                if self.verbose_logging:
                    print(f"  完成{len(enhanced_patterns)}个形态的未来表现分析")
        
        if self.verbose_logging:
            total_patterns = sum(len(patterns) for patterns in enhanced_pattern_dict.values())
            print(f"\n未来表现分析完成，共{total_patterns}个形态")
        
        return enhanced_pattern_dict
    
    def _analyze_future_n_days(self, data, signal_idx, hatch_direction, standard_close, monitor_days, end_date_2):
        """分析信号后未来N天的价格表现"""
        future_prices = []
        data_list = list(data.iterrows())
        
        for i in range(1, monitor_days + 1):
            future_idx = signal_idx + i
            if future_idx < len(data_list):
                future_date, future_row = data_list[future_idx]
                
                # 检查是否超出扩展结束日期
                if future_date.date() > end_date_2:
                    break
                
                future_prices.append({
                    'date': future_date.strftime('%Y-%m-%d'),
                    'high': float(future_row['high']),
                    'low': float(future_row['low'])
                })
            else:
                break
        
        return future_prices
    
    def _calculate_max_revenue_risk(self, future_analysis, hatch_direction, standard_close):
        """计算最大收益和最大风险（原逻辑：与影线方向相反交易）"""
        if not future_analysis:
            return {'date': None, 'revenue': 0.0}, {'date': None, 'revenue': 0.0}
        
        max_revenue = float('-inf')
        min_revenue = float('inf')
        max_revenue_date = None
        min_revenue_date = None
        
        for day_data in future_analysis:
            date = day_data['date']
            high = day_data['high']
            low = day_data['low']
            
            if hatch_direction == 'up':
                # 上影线做空：收益 = (标准收盘价 - 最低价) / 标准收盘价，风险 = (标准收盘价 - 最高价) / 标准收盘价
                revenue = (standard_close - low) / standard_close
                risk = (standard_close - high) / standard_close
            else:
                # 下影线做多：收益 = (最高价 - 标准收盘价) / 标准收盘价，风险 = (最低价 - 标准收盘价) / 标准收盘价
                revenue = (high - standard_close) / standard_close
                risk = (low - standard_close) / standard_close
            
            # 更新最大收益
            if revenue > max_revenue:
                max_revenue = round(revenue, 4)
                max_revenue_date = date
            
            # 更新最大风险（最大亏损，即最小收益的相反数）
            if risk < min_revenue:
                min_revenue = round(risk, 4)
                min_revenue_date = date
        
        return {'date': max_revenue_date, 'revenue': float(max_revenue)}, {'date': min_revenue_date, 'revenue': float(min_revenue)}
    
    def _calculate_max_revenue_risk_2(self, future_analysis, hatch_direction, standard_close):
        """计算最大收益和最大风险（新逻辑：与影线方向相同交易）"""
        if not future_analysis:
            return {'date': None, 'revenue': 0.0}, {'date': None, 'revenue': 0.0}
        
        max_revenue_2 = float('-inf')
        min_revenue_2 = float('inf')
        max_revenue_date_2 = None
        min_revenue_date_2 = None
        
        for day_data in future_analysis:
            date = day_data['date']
            high = day_data['high']
            low = day_data['low']
            
            if hatch_direction == 'up':
                # 上影线做多：收益 = (最高价 - 标准收盘价) / 标准收盘价，风险 = (最低价 - 标准收盘价) / 标准收盘价
                revenue_2 = (high - standard_close) / standard_close
                risk_2 = (low - standard_close) / standard_close
            else:
                # 下影线做空：收益 = (标准收盘价 - 最低价) / 标准收盘价，风险 = (标准收盘价 - 最高价) / 标准收盘价
                revenue_2 = (standard_close - low) / standard_close
                risk_2 = (standard_close - high) / standard_close
            
            # 更新最大收益
            if revenue_2 > max_revenue_2:
                max_revenue_2 = round(revenue_2, 4)
                max_revenue_date_2 = date
            
            # 更新最大风险（最大亏损，即最小收益的相反数）
            if risk_2 < min_revenue_2:
                min_revenue_2 = round(risk_2, 4)
                min_revenue_date_2 = date
        
        return {'date': max_revenue_date_2, 'revenue': float(max_revenue_2)}, {'date': min_revenue_date_2, 'revenue': float(min_revenue_2)}
    
    def _calculate_conclusion(self, max_revenue_info, min_revenue_info, future_analysis, hatch_direction, standard_close):
        """
        计算形态的结论：是否为有效的预测信号（增强版）
        
        逻辑：
        - 若满足以下条件则返回 "true"：
        1. （最大盈利日期 < 最大亏损日期）且（最大盈利 > max_revenue_threshold）
        2. 或者（最大盈利日期 > 最大亏损日期）且（最大盈利 > max_revenue_threshold）且（最大亏损 < max_loss_threshold）
        - 增强条件：如果在最大收益日期之前的任何一天，损失超过了MAX_LOSS_THRESHOLD，则返回 "false"
        - 否则返回 "false"
        """
        max_revenue_date = max_revenue_info['date']
        max_revenue = max_revenue_info['revenue']
        min_revenue_date = min_revenue_info['date']
        min_revenue = min_revenue_info['revenue']
        
        # 如果日期为空，返回false
        if max_revenue_date is None or min_revenue_date is None:
            return "false"
        
        # 转换日期字符串为datetime对象进行比较
        try:
            from datetime import datetime
            max_revenue_dt = datetime.strptime(max_revenue_date, '%Y-%m-%d')
            min_revenue_dt = datetime.strptime(min_revenue_date, '%Y-%m-%d')
        except:
            return "false"
        
        # 增强检查：检查最大收益日期之前是否有任何一天的损失超过阈值
        if future_analysis and self._check_early_excessive_loss(future_analysis, max_revenue_date, hatch_direction, standard_close):
            return "false"
        
        # 条件1：（最大盈利日期 < 最大亏损日期）且（最大盈利 > 阈值）
        condition1 = (max_revenue_dt < min_revenue_dt) and (max_revenue > self.max_revenue_threshold)
        
        # 条件2：（最大盈利日期 > 最大亏损日期）且（最大盈利 > 阈值）且（最大亏损 < 阈值）
        condition2 = (max_revenue_dt > min_revenue_dt) and (max_revenue > self.max_revenue_threshold) and (abs(min_revenue) < self.max_loss_threshold)
        
        return "true" if (condition1 or condition2) else "false"
    
    def _check_early_excessive_loss(self, future_analysis, max_revenue_date, hatch_direction, standard_close):
        """
        检查在最大收益日期之前是否有任何一天的损失超过阈值（原逻辑：与影线方向相反交易）
        
        参数:
            future_analysis: 未来价格分析数据
            max_revenue_date: 最大收益日期
            hatch_direction: 影线方向
            standard_close: 标准收盘价
            
        返回:
            bool: True表示存在早期过度损失
        """
        from datetime import datetime
        try:
            max_revenue_dt = datetime.strptime(max_revenue_date, '%Y-%m-%d')
        except:
            return False
        
        for day_data in future_analysis:
            day_date_str = day_data['date']
            try:
                day_dt = datetime.strptime(day_date_str, '%Y-%m-%d')
                
                # 只检查最大收益日期之前的日期
                if day_dt >= max_revenue_dt:
                    continue
                
                high = day_data['high']
                low = day_data['low']
                
                if hatch_direction == 'up':
                    # 上影线做空：亏损 = (标准收盘价 - 最高价) / 标准收盘价，当价格上涨时为负值（亏损）
                    loss_value = (standard_close - high) / standard_close
                else:
                    # 下影线做多：亏损 = (最低价 - 标准收盘价) / 标准收盘价，当价格下跌时为负值（亏损）
                    loss_value = (low - standard_close) / standard_close
                
                # 检查亏损是否超过阈值：亏损为负值，阈值为正值，所以检查 abs(亏损) > 阈值
                if loss_value < 0 and abs(loss_value) > self.max_loss_threshold:
                    return True
                    
            except:
                continue
        
        return False
    
    def _calculate_conclusion_2(self, max_revenue_info_2, min_revenue_info_2, future_analysis, hatch_direction, standard_close):
        """
        计算形态的结论2：是否为有效的预测信号（新逻辑：与影线方向相同交易）
        
        逻辑：
        - 若满足以下条件则返回 "true"：
        1. （最大盈利日期 < 最大亏损日期）且（最大盈利 > max_revenue_threshold）
        2. 或者（最大盈利日期 > 最大亏损日期）且（最大盈利 > max_revenue_threshold）且（最大亏损 < max_loss_threshold）
        - 增强条件：如果在最大收益日期之前的任何一天，损失超过了MAX_LOSS_THRESHOLD，则返回 "false"
        - 否则返回 "false"
        """
        max_revenue_date_2 = max_revenue_info_2['date']
        max_revenue_2 = max_revenue_info_2['revenue']
        min_revenue_date_2 = min_revenue_info_2['date']
        min_revenue_2 = min_revenue_info_2['revenue']
        
        # 如果日期为空，返回false
        if max_revenue_date_2 is None or min_revenue_date_2 is None:
            return "false"
        
        # 转换日期字符串为datetime对象进行比较
        try:
            from datetime import datetime
            max_revenue_dt_2 = datetime.strptime(max_revenue_date_2, '%Y-%m-%d')
            min_revenue_dt_2 = datetime.strptime(min_revenue_date_2, '%Y-%m-%d')
        except:
            return "false"
        
        # 增强检查：检查最大收益日期之前是否有任何一天的损失超过阈值（新逻辑）
        if future_analysis and self._check_early_excessive_loss_2(future_analysis, max_revenue_date_2, hatch_direction, standard_close):
            return "false"
        
        # 条件1：（最大盈利日期 < 最大亏损日期）且（最大盈利 > 阈值）
        condition1 = (max_revenue_dt_2 < min_revenue_dt_2) and (max_revenue_2 > self.max_revenue_threshold)
        
        # 条件2：（最大盈利日期 > 最大亏损日期）且（最大盈利 > 阈值）且（最大亏损 < 阈值）
        condition2 = (max_revenue_dt_2 > min_revenue_dt_2) and (max_revenue_2 > self.max_revenue_threshold) and (abs(min_revenue_2) < self.max_loss_threshold)
        
        return "true" if (condition1 or condition2) else "false"
    
    def _check_early_excessive_loss_2(self, future_analysis, max_revenue_date_2, hatch_direction, standard_close):
        """
        检查在最大收益日期之前是否有任何一天的损失超过阈值（新逻辑：与影线方向相同交易）
        
        参数:
            future_analysis: 未来价格分析数据
            max_revenue_date_2: 最大收益日期
            hatch_direction: 影线方向
            standard_close: 标准收盘价
            
        返回:
            bool: True表示存在早期过度损失
        """
        from datetime import datetime
        try:
            max_revenue_dt_2 = datetime.strptime(max_revenue_date_2, '%Y-%m-%d')
        except:
            return False
        
        for day_data in future_analysis:
            day_date_str = day_data['date']
            try:
                day_dt = datetime.strptime(day_date_str, '%Y-%m-%d')
                
                # 只检查最大收益日期之前的日期
                if day_dt >= max_revenue_dt_2:
                    continue
                
                high = day_data['high']
                low = day_data['low']
                
                if hatch_direction == 'up':
                    # 上影线做多：亏损 = (最低价 - 标准收盘价) / 标准收盘价，当价格下跌时为负值（亏损）
                    loss_value = (low - standard_close) / standard_close
                else:
                    # 下影线做空：亏损 = (标准收盘价 - 最高价) / 标准收盘价，当价格上涨时为负值（亏损）
                    loss_value = (standard_close - high) / standard_close
                
                # 检查亏损是否超过阈值：亏损为负值，阈值为正值，所以检查 abs(亏损) > 阈值
                if loss_value < 0 and abs(loss_value) > self.max_loss_threshold:
                    return True
                    
            except:
                continue
        
        return False
    
    def _calculate_direction_fields(self, pattern_info, hatch_direction):
        """
        计算body_direction和direction_consistency字段
        
        参数:
            pattern_info: K线形态信息
            hatch_direction: 影线方向 ('up' 或 'down')
        
        返回:
            tuple: (body_direction, direction_consistency)
        """
        open_price = pattern_info['open']
        close_price = pattern_info['close']
        
        # 计算body_direction：实体方向
        if close_price > open_price:
            body_direction = 'up'
        elif close_price < open_price:
            body_direction = 'down'
        else:
            body_direction = 'flat'  # 十字星情况
        
        # 计算direction_consistency：检查实体方向与影线方向是否相反
        if body_direction == 'flat':
            direction_consistency = 'false'  # 十字星情况无法判断一致性
        elif (body_direction == 'up' and hatch_direction == 'down') or \
             (body_direction == 'down' and hatch_direction == 'up'):
            direction_consistency = 'true'   # 相反，符合预期
        else:
            direction_consistency = 'false'  # 不相反
            
        return body_direction, direction_consistency
    
    def _calculate_price_direction(self, close_price, ma5_value):
        """
        计算基于收盘价与5K移动平均线关系的价格方向
        
        参数:
            close_price: 形态形成日的收盘价
            ma5_value: 5日移动平均线价格
            
        返回:
            str: 价格方向分类 ("上涨" 或 "下跌")
        """
        if ma5_value is None:
            return "数据不足"  # 如果MA5数据不足，返回数据不足
        
        if close_price >= ma5_value:
            return "上涨"
        else:
            return "下跌"
    
    def _calculate_dynamic_position(self, conclusion, max_revenue, trade_amount):
        """
        计算动态仓位调整后的剩余资金
        
        参数:
            conclusion: 交易结论 ('true' 为正确, 'false' 为错误)
            max_revenue: 最大盈利率
            trade_amount: 当前交易金额
            
        返回:
            float: 调整后的剩余资金
        """
        if conclusion == "false":  # 亏损交易（错误结果）
            remaining_amount = trade_amount * (1 - self.loss_rate)
            if self.verbose_logging:
                print(f"      亏损交易，剩余资金: {remaining_amount:.2f} (亏损率: {self.loss_rate})")
        else:  # 盈利交易（正确结果）
            remaining_amount = trade_amount * (1 + max_revenue - self.profit_drawdown_rate)
            if self.verbose_logging:
                print(f"      盈利交易，剩余资金: {remaining_amount:.2f} (最大利润: {max_revenue:.4f}, 回撤率: {self.profit_drawdown_rate})")
        
        return remaining_amount
    
    def _add_portfolio_trade(self, contract_code, date, conclusion, conclusion_2, max_revenue, max_revenue_2, trade_amount, remaining_amount):
        """
        添加交易记录到投资组合（支持两种交易逻辑）
        
        参数:
            contract_code: 合约代码
            date: 交易日期
            conclusion: 原逻辑交易结论
            conclusion_2: 新逻辑交易结论
            max_revenue: 原逻辑最大盈利率
            max_revenue_2: 新逻辑最大盈利率
            trade_amount: 交易金额
            remaining_amount: 剩余金额（基于原逻辑计算）
        """
        # 计算新逻辑的剩余金额
        remaining_amount_2 = self._calculate_dynamic_position(
            conclusion_2, max_revenue_2, trade_amount
        )
        
        trade_record = {
            'contract_code': contract_code,
            'date': date,
            'conclusion': conclusion,
            'conclusion2': conclusion_2,
            'max_revenue': max_revenue,
            'max_revenue2': max_revenue_2,
            'trade_amount': trade_amount,
            'remaining_amount': remaining_amount,
            'remaining_amount2': remaining_amount_2,
            'profit_loss': remaining_amount - trade_amount,
            'profit_loss2': remaining_amount_2 - trade_amount
        }
        self.portfolio_results.append(trade_record)
    
    def _calculate_moving_averages(self, data, signal_idx):
        """
        计算形态形成日的移动平均线
        
        参数:
            data: 价格数据DataFrame
            signal_idx: 信号日期在数据中的索引
            
        返回:
            dict: 包含各移动平均线的字典
        """
        if signal_idx < 29:  # 至少需要30天数据来计算30日移动平均
            return {'ma5': None, 'ma10': None, 'ma20': None, 'ma30': None}
        
        # 获取到信号日期为止的收盘价数据
        data_list = list(data.iterrows())
        closes = []
        for i in range(max(0, signal_idx - 29), signal_idx + 1):  # 取30天数据
            if i < len(data_list):
                _, row = data_list[i]
                closes.append(row['close'])
        
        if len(closes) < 30:
            return {'ma5': None, 'ma10': None, 'ma20': None, 'ma30': None}
        
        # 计算移动平均线（使用最后的N天数据）
        ma5 = sum(closes[-5:]) / 5 if len(closes) >= 5 else None
        ma10 = sum(closes[-10:]) / 10 if len(closes) >= 10 else None
        ma20 = sum(closes[-20:]) / 20 if len(closes) >= 20 else None
        ma30 = sum(closes[-30:]) / 30 if len(closes) >= 30 else None
        
        return {
            'ma5': round(ma5, 2) if ma5 is not None else None,
            'ma10': round(ma10, 2) if ma10 is not None else None,
            'ma20': round(ma20, 2) if ma20 is not None else None,
            'ma30': round(ma30, 2) if ma30 is not None else None
        }
    
    def _calculate_trend_strength(self, ma_values):
        """
        计算趋势强度评估
        
        参数:
            ma_values: 移动平均线字典
            
        返回:
            str: 趋势强度分类
        """
        ma5 = ma_values['ma5']
        ma10 = ma_values['ma10']
        ma20 = ma_values['ma20']
        ma30 = ma_values['ma30']
        
        # 如果有任何移动平均线为空，返回"数据不足"
        if None in [ma5, ma10, ma20, ma30]:
            return "数据不足"
        
        # 进行6项趋势比较
        comparisons = [
            ma5 >= ma10,  # 5K≥10K
            ma10 >= ma20,  # 10K≥20K
            ma20 >= ma30,  # 20K≥30K
            ma5 >= ma20,   # 5K≥20K
            ma10 >= ma30,  # 10K≥30K
            ma5 >= ma30    # 5K≥30K
        ]
        
        # 计算上涨趋势的数量
        uptrend_count = sum(comparisons)
        
        # 根据上涨趋势数量进行分类
        if uptrend_count == 6 or uptrend_count == 0:
            return "强趋势"
        elif uptrend_count == 3:
            return "震荡"
        elif uptrend_count == 5 or uptrend_count == 1:
            return "中趋势"
        elif uptrend_count == 4 or uptrend_count == 2:
            return "弱趋势"
        else:
            return "未知"  # 理论上不应该到这里
    
    def calculate_portfolio_summary(self):
        """
        计算投资组合整体表现汇总（支持两种交易逻辑对比）
        
        返回:
            dict: 投资组合汇总信息
        """
        if not self.portfolio_results:
            return {
                'total_trades': 0,
                'logic1': {
                    'successful_trades': 0,
                    'failed_trades': 0,
                    'success_rate': 0.0,
                    'total_profit_loss': 0.0,
                    'final_capital': self.initial_trade_amount,
                    'total_return': 0.0
                },
                'logic2': {
                    'successful_trades': 0,
                    'failed_trades': 0,
                    'success_rate': 0.0,
                    'total_profit_loss': 0.0,
                    'final_capital': self.initial_trade_amount,
                    'total_return': 0.0
                }
            }
        
        total_trades = len(self.portfolio_results)
        initial_capital = total_trades * self.initial_trade_amount
        
        # 原逻辑（与影线方向相反交易）统计
        successful_trades_1 = sum(1 for trade in self.portfolio_results if trade['conclusion'] == 'true')
        failed_trades_1 = total_trades - successful_trades_1
        success_rate_1 = (successful_trades_1 / total_trades) * 100 if total_trades > 0 else 0
        total_profit_loss_1 = sum(trade['profit_loss'] for trade in self.portfolio_results)
        final_capital_1 = initial_capital + total_profit_loss_1
        total_return_1 = (total_profit_loss_1 / initial_capital * 100) if initial_capital > 0 else 0
        
        # 新逻辑（与影线方向相同交易）统计
        successful_trades_2 = sum(1 for trade in self.portfolio_results if trade['conclusion2'] == 'true')
        failed_trades_2 = total_trades - successful_trades_2
        success_rate_2 = (successful_trades_2 / total_trades) * 100 if total_trades > 0 else 0
        total_profit_loss_2 = sum(trade['profit_loss2'] for trade in self.portfolio_results)
        final_capital_2 = initial_capital + total_profit_loss_2
        total_return_2 = (total_profit_loss_2 / initial_capital * 100) if initial_capital > 0 else 0
        
        portfolio_summary = {
            'total_trades': total_trades,
            'initial_capital': initial_capital,
            'logic1': {
                'name': '原逻辑（与影线方向相反交易）',
                'successful_trades': successful_trades_1,
                'failed_trades': failed_trades_1,
                'success_rate': round(success_rate_1, 2),
                'total_profit_loss': round(total_profit_loss_1, 2),
                'final_capital': round(final_capital_1, 2),
                'total_return': round(total_return_1, 2)
            },
            'logic2': {
                'name': '新逻辑（与影线方向相同交易）',
                'successful_trades': successful_trades_2,
                'failed_trades': failed_trades_2,
                'success_rate': round(success_rate_2, 2),
                'total_profit_loss': round(total_profit_loss_2, 2),
                'final_capital': round(final_capital_2, 2),
                'total_return': round(total_return_2, 2)
            }
        }
        
        if self.verbose_logging:
            print(f"\n=== 投资组合双向交易逻辑对比分析 ===")
            print(f"总交易次数: {total_trades}")
            print(f"初始资本: {initial_capital:.2f}")
            print(f"\n--- 原逻辑（与影线方向相反交易）---")
            print(f"成功交易: {successful_trades_1}")
            print(f"失败交易: {failed_trades_1}")
            print(f"成功率: {success_rate_1:.2f}%")
            print(f"总盈亏: {total_profit_loss_1:.2f}")
            print(f"最终资本: {final_capital_1:.2f}")
            print(f"总回报率: {total_return_1:.2f}%")
            print(f"\n--- 新逻辑（与影线方向相同交易）---")
            print(f"成功交易: {successful_trades_2}")
            print(f"失败交易: {failed_trades_2}")
            print(f"成功率: {success_rate_2:.2f}%")
            print(f"总盈亏: {total_profit_loss_2:.2f}")
            print(f"最终资本: {final_capital_2:.2f}")
            print(f"总回报率: {total_return_2:.2f}%")
            print(f"\n--- 逻辑对比 ---")
            print(f"成功率差异: {success_rate_2 - success_rate_1:.2f}% (新逻辑 - 原逻辑)")
            print(f"回报率差异: {total_return_2 - total_return_1:.2f}% (新逻辑 - 原逻辑)")
        
        return portfolio_summary
    
    def generate_csv_output(self, enhanced_pattern_dict):
        """
        生成CSV输出文件
        """
        if self.verbose_logging:
            print("\n=== 步骤6: 生成CSV输出文件 ===")
        
        # 准备CSV数据（按指定顺序排列字段）
        csv_data = []
        
        for contract_code, patterns in enhanced_pattern_dict.items():
            for signal_date, pattern_info in patterns.items():
                csv_data.append({
                    # 按更新的字段顺序：contract_code, date, conclusion, conclusion2, body_direction, direction_consistency, price_direction, body_len, hatch_len, hatch/body, close, open, high, low, max_revenue, max_revenue_date, min_revenue, min_revenue_date, max_revenue2, max_revenue_date2, min_revenue2, min_revenue_date2, profit_loss, remaining_amount, trade_amount
                    'contract_code': contract_code,
                    'date': pattern_info['date'],
                    'conclusion': pattern_info['conclusion'],
                    'conclusion2': pattern_info['conclusion2'],
                    'body_direction': pattern_info['body_direction'],
                    'direction_consistency': pattern_info['direction_consistency'],
                    'price_direction': pattern_info['price_direction'],  # 新增：基于收盘价与5K移动平均线关系的价格方向
                    'body_len': pattern_info['body_len'],
                    'hatch_len': pattern_info['hatch_len'],
                    'hatch/body': pattern_info['hatch_body_ratio'],
                    'close': pattern_info['close'],
                    'open': pattern_info['open'],
                    'high': pattern_info['high'],
                    'low': pattern_info['low'],
                    'max_revenue': pattern_info['max_revenue'],
                    'max_revenue_date': pattern_info['max_revenue_date'],
                    'min_revenue': pattern_info['min_revenue'],
                    'min_revenue_date': pattern_info['min_revenue_date'],
                    'max_revenue2': pattern_info['max_revenue2'],
                    'max_revenue_date2': pattern_info['max_revenue_date2'],
                    'min_revenue2': pattern_info['min_revenue2'],
                    'min_revenue_date2': pattern_info['min_revenue_date2'],
                    'profit_loss': pattern_info['profit_loss'],
                    'remaining_amount': pattern_info['remaining_amount'],
                    'trade_amount': pattern_info['trade_amount']
                })
        
        if csv_data:
            # 创建DataFrame并保存为CSV，严格按照用户指定的列顺序
            df = pd.DataFrame(csv_data)
            
            # 定义更新的确切列顺序（包含新的price_direction字段）
            column_order = [
                'contract_code', 'date', 'conclusion', 'conclusion2', 'body_direction', 
                'direction_consistency', 'price_direction', 'body_len', 'hatch_len', 'hatch/body', 
                'close', 'open', 'high', 'low', 'max_revenue', 'max_revenue_date', 
                'min_revenue', 'min_revenue_date', 'max_revenue2', 'max_revenue_date2', 
                'min_revenue2', 'min_revenue_date2', 'profit_loss', 'remaining_amount', 'trade_amount'
            ]
            
            # 重新排列DataFrame的列顺序
            df = df[column_order]
            
            timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
            filename = f'future_candlestick_hatch_analysis_{timestamp}.csv'
            df.to_csv(filename, index=False, encoding=self.output_encoding)
            
            if self.verbose_logging:
                print(f"分析结果已保存至: {filename}")
                print(f"总计 {len(csv_data)} 条记录")
            
            return filename
        else:
            if self.verbose_logging:
                print("没有数据可以输出")
            return None
    
    def run_complete_analysis(self):
        """执行完整的分析流程"""
        if self.verbose_logging:
            print("开始执行期货烛台影线形态分析")
            print("=" * 60)
        
        try:
            # 步骤1: 构建合约数据结构
            contract_structure = self.build_contract_structure()
            if not contract_structure:
                if self.verbose_logging:
                    print("未获取到有效合约，分析终止")
                return None
            
            # 步骤2: 收集扩展数据
            extended_data = self.collect_extended_data()
            if not extended_data:
                if self.verbose_logging:
                    print("未获取到有效数据，分析终止")
                return None
            
            # 步骤3: 计算实体长度阈值
            threshold_data = self.calculate_body_thresholds(extended_data)
            if not threshold_data:
                if self.verbose_logging:
                    print("未能计算有效阈值，分析终止")
                return None
            
            # 步骤4: 识别影线形态
            pattern_dict = self.identify_hatch_patterns(threshold_data)
            if not pattern_dict:
                if self.verbose_logging:
                    print("未检测到有效影线形态，分析终止")
                return None
            
            # 步骤5: 分析未来表现
            enhanced_pattern_dict = self.analyze_future_performance(threshold_data, pattern_dict)
            
            # 步骤6: 生成CSV输出
            output_filename = self.generate_csv_output(enhanced_pattern_dict)
            
            # 步骤7: 计算投资组合汇总
            portfolio_summary = self.calculate_portfolio_summary()
            
            # 分析汇总
            total_contracts = len(contract_structure)
            valid_contracts = len(threshold_data)
            total_patterns = sum(len(patterns) for patterns in enhanced_pattern_dict.values())
            
            # 计算模式有效性指标
            effective_patterns = 0
            for contract_patterns in enhanced_pattern_dict.values():
                for pattern_info in contract_patterns.values():
                    if pattern_info.get('conclusion') == 'true':
                        effective_patterns += 1
            
            effective_rate = (effective_patterns / total_patterns * 100) if total_patterns > 0 else 0
            
            if self.verbose_logging:
                print("\n" + "=" * 60)
                print("分析完成汇总:")
                print(f"总合约数: {total_contracts}")
                print(f"有效合约数: {valid_contracts}")
                print(f"检测形态数: {total_patterns}")
                print(f"有效形态数: {effective_patterns}")
                print(f"有效率: {effective_rate:.1f}%")
                print(f"输出文件: {output_filename}")
            
            return {
                'contract_structure': contract_structure,
                'threshold_data': threshold_data,
                'pattern_results': enhanced_pattern_dict,
                'output_filename': output_filename,
                'portfolio_summary': portfolio_summary,
                'summary': {
                    'total_contracts': total_contracts,
                    'valid_contracts': valid_contracts,
                    'total_patterns': total_patterns,
                    'effective_patterns': effective_patterns,
                    'effective_rate': effective_rate
                }
            }
            
        except Exception as e:
            if self.verbose_logging:
                print(f"分析过程中出现错误: {str(e)}")
                import traceback
                traceback.print_exc()
            return None


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

def run_candlestick_hatch_analysis(config=None):
    """运行期货烛台影线形态分析"""
    if config is None:
        config = CandlestickHatchConfig
    
    # 打印配置信息
    config.print_config()
    
    # 创建分析器并运行
    analyzer = FutureCandlestickHatchAnalyzer(config)
    results = analyzer.run_complete_analysis()
    return results

# 执行分析
if __name__ == "__main__":
    print("期货烛台影线形态双向交易分析工具")
    print("研究带有明显影线的K线对未来价格走势的预测能力")
    print("双向交易逻辑对比测试：")
    print("  原逻辑：上影线做空，下影线做多（与影线方向相反交易）")
    print("  新逻辑：上影线做多，下影线做空（与影线方向相同交易）")
    print("适用于聚宽在线研究平台")
    
    results = run_candlestick_hatch_analysis()
    
    if results:
        print("\n✅ 分析执行成功！")
        summary = results['summary']
        portfolio = results['portfolio_summary']
        print(f"📊 结果摘要:")
        print(f"   - 分析合约数: {summary['total_contracts']}")
        print(f"   - 有效数据合约: {summary['valid_contracts']}")
        print(f"   - 检测形态总数: {summary['total_patterns']}")
        print(f"   - 有效形态数: {summary['effective_patterns']}")
        print(f"   - 有效率: {summary['effective_rate']:.1f}%")
        print(f"   - 输出文件: {results['output_filename']}")
        print(f"\n💰 双向交易逻辑对比分析:")
        print(f"   - 总交易次数: {portfolio['total_trades']}")
        print(f"   - 初始资本: {portfolio['initial_capital']:.2f}")
        print(f"\n   📈 原逻辑（与影线方向相反交易）:")
        print(f"       成功率: {portfolio['logic1']['success_rate']:.2f}%")
        print(f"       总盈亏: {portfolio['logic1']['total_profit_loss']:.2f}")
        print(f"       总回报率: {portfolio['logic1']['total_return']:.2f}%")
        print(f"\n   📉 新逻辑（与影线方向相同交易）:")
        print(f"       成功率: {portfolio['logic2']['success_rate']:.2f}%")
        print(f"       总盈亏: {portfolio['logic2']['total_profit_loss']:.2f}")
        print(f"       总回报率: {portfolio['logic2']['total_return']:.2f}%")
        print(f"\n   🔍 逻辑对比:")
        success_diff = portfolio['logic2']['success_rate'] - portfolio['logic1']['success_rate']
        return_diff = portfolio['logic2']['total_return'] - portfolio['logic1']['total_return']
        print(f"       成功率差异: {success_diff:+.2f}% (新逻辑 - 原逻辑)")
        print(f"       回报率差异: {return_diff:+.2f}% (新逻辑 - 原逻辑)")
    else:
        print("\n❌ 分析执行失败，请检查错误信息")