JUKUAN/Lib/future/TrendBreakoutStrategy_v003.py

# 导入函数库
from jqdata import *
from jqdata import finance
import pandas as pd
from datetime import date, datetime, timedelta
import re

# 趋势突破交易策略 v003
# 基于均线趋势突破的期货交易策略 - 修复版本

# 设置以便完整打印 DataFrame
pd.set_option('display.max_rows', None)
pd.set_option('display.max_columns', None)
pd.set_option('display.width', None)
pd.set_option('display.max_colwidth', 20)

## 初始化函数
def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    log.info('趋势突破交易策略初始化开始')

    ### 期货相关设定 ###
    # 设定账户为金融账户
    set_subportfolios([SubPortfolioConfig(cash=context.portfolio.starting_cash, type='index_futures')])
    # 期货类每笔交易时的手续费
    set_order_cost(OrderCost(open_commission=0.000023, close_commission=0.000023, close_today_commission=0.0023), type='index_futures')
    # 设置期货交易的滑点
    set_slippage(StepRelatedSlippage(2))
    
    # 初始化全局变量
    g.usage_percentage = 0.8  # 最大持股数量
    g.default_days = 10  # 判断趋势线的最小天数
    g.continuous_days_length = 5  # 破趋势后的观察天数
    g.change_direction_days = 5  # 检查均线穿过的天数范围
    g.crossed_symbols_history = {}  # 用于存储过去几天的穿越信息
    g.trade_history = {}  # 初始化交易记录
    g.change_fail_history = {}  # 初始化换月建仓失败的记录
    g.high_low_ma_relations = {}
    
    # 定义默认的保证金比例
    g.default_margin_rates = {
        'long': {'A': 0.07, 'AG': 0.04, 'AL': 0.05, 'AO': 0.05, 'AP': 0.08, 'AU': 0.04, 'B': 0.05,
        'BC': 0.13, 'BR': 0.07, 'BU': 0.04, 'C': 0.07, 'CF': 0.05, 'CJ': 0.07, 'CS': 0.07,
        'CU': 0.05, 'CY': 0.05, 'EB': 0.12, 'EC': 0.12, 'EG': 0.05, 'FG': 0.05, 'FU': 0.08,
        'HC': 0.04, 'I': 0.1, 'J': 0.22, 'JD': 0.08, 'JM': 0.22, 
        'L': 0.07, 'LC': 0.05, 'LH': 0.1, 'LR': 0.05, 'LU': 0.15, 'M': 0.07, 'MA': 0.05, 'NI': 0.05, 'NR': 0.13, 'OI': 0.05,
        'P': 0.05, 'PB': 0.05, 'PF': 0.1, 'PG': 0.05, 'PK': 0.05,
        'PP': 0.07, 'RB': 0.05, 'RI': 0.05, 'RM': 0.05, 'RU': 0.05,
        'SA': 0.05, 'SC': 0.12, 'SF': 0.05, 'SH': 0.05, 'SI': 0.13, 'SM': 0.05, 'SN': 0.05, 'SP': 0.1, 'SR': 0.05,
        'SS': 0.05, 'TA': 0.05, 'UR': 0.09, 'V': 0.07,
        'Y': 0.05, 'ZC': 0.05, 'ZN': 0.05}, 
        'short': {'A': 0.07, 'AG': 0.04, 'AL': 0.05, 'AO': 0.05, 'AP': 0.08, 'AU': 0.04, 'B': 0.05,
        'BC': 0.13, 'BR': 0.07, 'BU': 0.04, 'C': 0.07, 'CF': 0.05, 'CJ': 0.07, 'CS': 0.07,
        'CU': 0.05, 'CY': 0.05, 'EB': 0.12, 'EC': 0.12, 'EG': 0.05, 'FG': 0.05, 'FU': 0.08,
        'HC': 0.04, 'I': 0.1, 'J': 0.22, 'JD': 0.08, 'JM': 0.22, 
        'L': 0.07, 'LC': 0.05, 'LH': 0.1, 'LR': 0.05, 'LU': 0.15, 'M': 0.07, 'MA': 0.05, 'NI': 0.05, 'NR': 0.13, 'OI': 0.05,
        'P': 0.05, 'PB': 0.05, 'PF': 0.1, 'PG': 0.05, 'PK': 0.05,
        'PP': 0.07, 'RB': 0.05, 'RI': 0.05, 'RM': 0.05, 'RU': 0.05,
        'SA': 0.05, 'SC': 0.12, 'SF': 0.05, 'SH': 0.05, 'SI': 0.13, 'SM': 0.05, 'SN': 0.05, 'SP': 0.1, 'SR': 0.05,
        'SS': 0.05, 'TA': 0.05, 'UR': 0.09, 'V': 0.07,
        'Y': 0.05, 'ZC': 0.05, 'ZN': 0.05}
    }
    
    g.multiplier = {
        'A': 10, 'AG': 15, 'AL': 5, 'AO': 20, 'AP': 10, 'AU': 1000, 'B': 10,
        'BC': 5, 'BR': 5, 'BU': 10, 'C': 10, 'CF': 5, 'CJ': 5, 'CS': 10,
        'CU': 5, 'CY': 5, 'EB': 5, 'EC': 50, 'EG': 10, 'FG': 20, 'FU': 10,
        'HC': 10, 'I': 100, 'J': 60, 'JD': 5, 'JM': 100, 
        'L': 5, 'LC': 1, 'LH': 16, 'LR': 0.05, 'LU': 10, 'M': 10, 'MA': 10, 'NI': 1, 'NR': 10, 'OI': 10,
        'P': 10, 'PB': 5, 'PF': 5, 'PG': 20, 'PK': 5,
        'PP': 5, 'RB': 10, 'RI': 0.05, 'RM': 10, 'RU': 10,
        'SA': 20, 'SC': 1000, 'SF': 5, 'SH': 30, 'SI': 5, 'SM': 5, 'SN': 1, 'SP': 10, 'SR': 10,
        'SS': 5, 'TA': 5, 'UR': 20, 'V': 5,
        'Y': 10, 'ZC': 0.05, 'ZN': 5
    }
    
    # 临时止损检查
    run_daily(loss_control, time='21:15:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='21:45:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='22:15:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='22:45:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='09:15:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='09:45:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='10:15:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='10:45:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='11:15:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='13:15:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='13:45:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='14:15:00', reference_security='IF1808.CCFX')
    run_daily(loss_control, time='14:56:00', reference_security='IF1808.CCFX')

    # 收盘前运行
    run_daily(before_market_close, time='14:55', reference_security='IF8888.CCFX')

############################ 主程序中执行函数 ###################################

def before_market_close(context):
    """收盘前主策略执行"""
    print("-" * 20 + "New day ending！" + "-" * 20)
    if len(g.trade_history) > 0:
        print(f'当前交易记录: {g.trade_history}, 换月失败记录: {g.change_fail_history}')
    
    # 1. 检查趋势
    trend_symbols, daily_data_info = check_trend(context)
    print_list_elements("趋势标的", trend_symbols)
    
    # 2. 检查影线穿越
    crossed_symbols = check_shadow_cross(context, trend_symbols, daily_data_info)
    print_list_elements("穿越标的", crossed_symbols)
    
    # 3. 检查买入条件
    buy_symbols = check_buy_condition(context, crossed_symbols)
    print_list_elements("买入标的", buy_symbols)
    
    # 4. 执行交易
    subportfolio = context.subportfolios[0]
    hold_symbols = set(subportfolio.long_positions.keys()) | set(subportfolio.short_positions.keys())
    
    for symbol, line_label, line_type, direction in buy_symbols:
        if check_symbol_prefix_match(symbol, hold_symbols):
            print(f'已有相似持仓 {symbol}，跳过交易')
        else: 
            print(f'无相似持仓 {symbol}，执行交易')
            value_to_invest = calculate_order_value(context, symbol, direction)
            open_position(symbol, value_to_invest, direction, line_label)

def loss_control(context):
    """止损控制"""
    print("-" * 20 + "止损控制" + "-" * 20)
    
    # 检查是否有正常的夜盘
    now = context.current_dt.time()
    now_hour = now.hour
    if now_hour >= 21:
        test_future = get_dominant_future("A")
        if test_future:
            test_data = attribute_history(test_future, 1, '1m', 'close')
            test_hour = test_data.index[0].hour
            if test_hour <= 15:
                print(f'最近的数据小时为: {test_hour}，该夜盘不存在，直接停止')
                return
    
    # 先把换月买入失败的重新买入
    if len(g.change_fail_history) > 0:
        print(f'检查换月失败的{g.change_fail_history}')
        for symbol in list(g.change_fail_history.keys()):
            direction = g.change_fail_history[symbol]['direction']
            value_to_invest = calculate_order_value(context, symbol, direction)
            success = open_position(symbol, value_to_invest, direction, g.change_fail_history[symbol].get("line_label", "unknown"))
            if success:
                del g.change_fail_history[symbol]
    
    # 检查损失
    target_time = datetime.strptime('14:55:00', '%H:%M:%S').time()
    potential_future_list = get_potential_future_list(context)
    
    # 遍历所有持仓
    subportfolio = context.subportfolios[0]
    all_positions = list(subportfolio.long_positions.values()) + list(subportfolio.short_positions.values())
    
    for position in all_positions:
        if position.security in potential_future_list:
            # 检查固定止损
            if check_loss_for_close(context, position, position.side):
                continue
            
            # 检查基于动态跟踪线的平仓条件
            if now > target_time:
                check_ma_for_close(context, position, 0.01, 4)
            else:
                check_ma_for_close(context, position, 0.003, 4)

############################ 交易模块 ###################################

def order_target_value_(security, value, direction):
    """自定义下单"""
    if value == 0:
        log.debug("平仓 %s" % (security))
    else:
        log.debug("下单 %s 金额 %f" % (security, value))
    return order_target_value(security, value, side=direction)

def open_position(security, value, direction, line_label):
    """开仓"""
    order = order_target_value_(security, value, direction)
    if order is not None and order.filled > 0:
        print(f'成功开仓 {security} 方向 {direction} 线标 {line_label}')
        g.trade_history[security] = {
            'entry_price': get_current_data()[security].last_price,
            'position_value': value,
            'direction': direction,
            'line_label': line_label,
            'finish_time': order.finish_time
        }
        return True
    return False

def close_position(position, direction):
    """平仓"""
    security = position.security
    order = order_target_value_(security, 0, direction)
    if order is not None:
        if order.status == OrderStatus.held and order.filled == order.amount:
            # 如果成功平仓，从交易历史中移除该标的
            if security in g.trade_history:
                del g.trade_history[security]
            return True
    return False

############################ 策略核心函数 ###################################

def get_potential_future_list(context):
    """获取可交易的期货品种列表"""
    potential_night_list = ['NI', 'CF', 'PF', 'Y', 'M', 'B', 'SN', 'RM']
    potential_day_list = ['JD', 'UR']
    
    if str(context.current_dt.time())[:2] in ('21', '22'):
        potential_icon_list = potential_night_list
    else:
        potential_icon_list = potential_day_list + potential_night_list
    
    potential_future_list = []
    for i in potential_icon_list:
        dominant_future = get_dominant_future(i)
        if dominant_future:
            potential_future_list.append(dominant_future)
    
    return potential_future_list

def check_trend(context):
    """检查品类的主连是否满足形成趋势的条件"""
    print("-" * 20 + "检查趋势" + "-" * 20)
    trend_symbols = []
    daily_data_info = {}
    
    # 获取可以交易的所有标的主连
    potential_future_list = get_potential_future_list(context)
    potential_future_list = [item for item in potential_future_list if item not in g.trade_history.keys()]
    
    # 针对所有标的需要的基础数据
    ma_crosses_dict = {}
    for symbol in potential_future_list:
        # 获取50天的收盘价数据
        close_data = attribute_history(symbol, 50, '1d', ['close', 'high', 'low', 'open'], df=True)
        close_series = close_data['close']

        # 计算移动平均线
        ma5 = close_series.rolling(window=5).mean()
        ma10 = close_series.rolling(window=10).mean()
        ma20 = close_series.rolling(window=20).mean()
        ma30 = close_series.rolling(window=30).mean()

        trend_info = {'symbol': symbol, 'trend_lines': []}
        daily_data = []

        # 初始化连续天数计数器
        continuous_days = {
            'above_ma5': [0] * g.continuous_days_length,
            'below_ma5': [0] * g.continuous_days_length,
            'above_ma10': [0] * g.continuous_days_length,
            'below_ma10': [0] * g.continuous_days_length,
            'above_ma20': [0] * g.continuous_days_length,
            'below_ma20': [0] * g.continuous_days_length,
            'above_ma30': [0] * g.continuous_days_length,
            'below_ma30': [0] * g.continuous_days_length
        }
        
        # 获取连续在某一个均线上或者下的天数
        for i in range(len(close_series)):
            update_continuous_days(continuous_days, close_series[i], ma5[i], ma10[i], ma20[i], ma30[i], g.continuous_days_length)

            # 收集每日数据
            day_data = {
                'date': close_data.index[i].date(),
                'close': close_series[i],
                'high': close_data['high'][i],
                'low': close_data['low'][i],
                'open': close_data['open'][i],
                'ma5': ma5[i],
                'ma10': ma10[i],
                'ma20': ma20[i],
                'ma30': ma30[i],
                'continuous_above_ma5': continuous_days['above_ma5'].copy(),
                'continuous_below_ma5': continuous_days['below_ma5'].copy(),
                'continuous_above_ma10': continuous_days['above_ma10'].copy(),
                'continuous_below_ma10': continuous_days['below_ma10'].copy(),
                'continuous_above_ma20': continuous_days['above_ma20'].copy(),
                'continuous_below_ma20': continuous_days['below_ma20'].copy(),
                'continuous_above_ma30': continuous_days['above_ma30'].copy(),
                'continuous_below_ma30': continuous_days['below_ma30'].copy()
            }
            daily_data.append(day_data)
        
        daily_data_info[symbol] = daily_data
        
        # 检查过去一定天数内均线的相交次数
        ma_crosses_dict[symbol] = count_ma_crosses(daily_data_info[symbol], g.change_direction_days)
                
        # 检查哪些均线会形成什么类型的趋势
        for ma_type in ['ma5', 'ma10', 'ma20', 'ma30']:
            above_days = continuous_days[f'above_{ma_type}']
            below_days = continuous_days[f'below_{ma_type}']
        
            above_condition = any(day >= g.default_days for day in above_days)
            below_condition = any(day >= g.default_days for day in below_days)
        
            if above_condition:
                trend_info['trend_lines'].append((ma_type, 'support'))
        
            if below_condition:
                trend_info['trend_lines'].append((ma_type, 'resistance'))

        if trend_info['trend_lines']:
            trend_symbols.append(trend_info)
    
    # 去除掉在一段时间内均线过于频繁交叉的对象
    valid_trend_symbols = []
    
    for trend_info in trend_symbols:
        symbol = trend_info["symbol"]
        ma_crosses = ma_crosses_dict[symbol]
        
        if ma_crosses <= 3:
            valid_trend_symbols.append(trend_info)
                
    return valid_trend_symbols, daily_data_info

def check_shadow_cross(context, trend_symbols, daily_data_info):
    """检查影线穿越"""
    print("-" * 20 + "检查影线穿越" + "-" * 20)
    
    # 检查并进行换月
    switch_result = position_auto_switch(context)
    if switch_result:
        print(f'换月结果: {switch_result}')
    
    today_crossed_symbols = {}
    
    for trend_info in trend_symbols:
        symbol = trend_info['symbol']
        
        # 获取昨天的数据
        yesterday_data = daily_data_info[symbol][-1]
        
        # 临时存储当前标的的穿越信息
        symbol_crosses = []

        for line_label, line_type in trend_info['trend_lines']:
            yesterday_day = yesterday_data['date']
            close = yesterday_data['close']
            high = yesterday_data['high']
            low = yesterday_data['low']
            ma_value = yesterday_data[line_label]
            close_ma_difference = (close - ma_value) / ma_value
            
            # 删除指定的键
            yesterday_data_copy = yesterday_data.copy()
            keys_to_remove = ['continuous_above_ma5', 'continuous_below_ma5', 'continuous_above_ma10', 'continuous_below_ma10', 'continuous_above_ma20', 'continuous_below_ma20', 'continuous_above_ma30', 'continuous_below_ma30']
            for key in keys_to_remove:
                yesterday_data_copy.pop(key, None)

            # 判断上一个交易日完整的数据中是否穿越
            if (line_type == 'support' and low < ma_value and close_ma_difference <= 0.0005) or (line_type == 'resistance' and high > ma_value and close_ma_difference >= -0.0005):
                symbol_crosses.append({
                    'date': yesterday_day,
                    'symbol': symbol,
                    'line_label': line_label,
                    'line_type': line_type,
                    'direction': 'long' if line_type == 'support' else 'short',
                    'latest_data': yesterday_data_copy
                })
                print(f"{symbol} 穿越 {line_label} ({line_type})")

        # 从符合条件的趋势线中选择数字较大的
        if symbol_crosses:
            best_cross = max(symbol_crosses, key=lambda x: int(''.join(filter(str.isdigit, x['line_label']))))
            update_crossed_symbols_history(context, best_cross)
            
    # 针对出现趋势变化的检查均线顺序、破趋势天数、收盘价和均线的最高点关系
    for symbol, records in g.crossed_symbols_history.items():
        latest_record = max(records, key=lambda x: x['date'])
        
        # 检查各种条件
        relation_check = check_ma_relations(context, symbol)
        duration_check, ma_close = check_cross_details(context, symbol)
        
        if relation_check and duration_check and ma_close:
            print(f'{symbol}满足所有条件，加入购买清单')
            today_crossed_symbols[symbol] = [latest_record]
            
    return [item for sublist in today_crossed_symbols.values() for item in sublist]

def check_buy_condition(context, crossed_symbols):
    """检查买入条件"""
    print("-" * 20 + "检查买入条件" + "-" * 20)
    buy_symbols = []

    for crossed_symbol in crossed_symbols:
        symbol = crossed_symbol['symbol']
        line_label = crossed_symbol['line_label']
        line_type = crossed_symbol['line_type']
        
        # 获取最新数据
        latest_data = crossed_symbol['latest_data']
        ma_value = latest_data[line_label]
        close_price = latest_data['close']
        today_change = (latest_data['close'] - latest_data['open']) / latest_data['open']

        if line_label != "ma5":
            if line_type == 'support' and 1.005 * ma_value <= close_price <= 1.02 * ma_value and today_change >= -0.002:
                buy_symbols.append((symbol, line_label, 'support', 'long'))
            elif line_type == 'resistance' and 0.98 * ma_value <= close_price <= 0.995 * ma_value and today_change <= 0.002:
                buy_symbols.append((symbol, line_label, 'resistance', 'short'))

    return buy_symbols

############################ 辅助函数 ###################################

def calculate_order_value(context, security, direction):
    """计算可以用于交易的金额"""
    current_price = get_current_data()[security].last_price
    underlying_symbol = security.split('.')[0][:-4]

    # 获取保证金比例
    margin_rate = g.default_margin_rates.get(direction, {}).get(underlying_symbol, 0.10)
    # 获取合约乘数
    multiplier = g.multiplier.get(underlying_symbol, 10)
    # 计算单手保证金
    single_hand_margin = current_price * multiplier * margin_rate

    # 根据单手保证金决定购买手数
    if single_hand_margin <= 20000:
        total_margin = 20000
    else:
        total_margin = single_hand_margin

    return total_margin

def update_continuous_days(continuous_days, close, ma5, ma10, ma20, ma30, length):
    """更新连续天数"""
    for key in ['above_ma5', 'below_ma5', 'above_ma10', 'below_ma10', 'above_ma20', 'below_ma20', 'above_ma30', 'below_ma30']:
        condition_met = False
        if ((key == 'above_ma5' and close > ma5) or
            (key == 'below_ma5' and close < ma5) or
            (key == 'above_ma10' and close > ma10) or
            (key == 'below_ma10' and close < ma10) or
            (key == 'above_ma20' and close > ma20) or
            (key == 'below_ma20' and close < ma20) or
            (key == 'above_ma30' and close > ma30) or
            (key == 'below_ma30' and close < ma30)):
            condition_met = True
        
        if condition_met:
            continuous_days[key].insert(0, continuous_days[key][0] + 1)
        else:
            continuous_days[key].insert(0, 0)
        continuous_days[key] = continuous_days[key][:length]

def count_ma_crosses(future_data, days):
    """计算均线交叉次数"""
    recent_data = future_data[-days:]
    ma5 = [day['ma5'] for day in recent_data]
    ma10 = [day['ma10'] for day in recent_data]
    ma20 = [day['ma20'] for day in recent_data]
    ma30 = [day['ma30'] for day in recent_data]
    
    cross_5_10 = sum([1 for i in range(1, len(ma5)) if ((ma5[-i] > ma10[-i] and ma5[-i-1] < ma10[-i-1]) or 
                                                      (ma5[-i] < ma10[-i] and ma5[-i-1] > ma10[-i-1]))])
    cross_5_20 = sum([1 for i in range(1, len(ma5)) if ((ma5[-i] > ma20[-i] and ma5[-i-1] < ma20[-i-1]) or 
                                                      (ma5[-i] < ma20[-i] and ma5[-i-1] > ma20[-i-1]))])
    cross_5_30 = sum([1 for i in range(1, len(ma5)) if ((ma5[-i] > ma30[-i] and ma5[-i-1] < ma30[-i-1]) or 
                                                      (ma5[-i] < ma30[-i] and ma5[-i-1] > ma30[-i-1]))])
    cross_10_20 = sum([1 for i in range(1, len(ma10)) if ((ma10[-i] > ma20[-i] and ma10[-i-1] < ma20[-i-1]) or 
                                                      (ma10[-i] < ma20[-i] and ma10[-i-1] > ma20[-i-1]))])
    cross_10_30 = sum([1 for i in range(1, len(ma10)) if ((ma10[-i] > ma30[-i] and ma10[-i-1] < ma30[-i-1]) or 
                                                      (ma10[-i] < ma30[-i] and ma10[-i-1] > ma30[-i-1]))])
    cross_20_30 = sum([1 for i in range(1, len(ma20)) if ((ma20[-i] > ma30[-i] and ma20[-i-1] < ma30[-i-1]) or 
                                                      (ma20[-i] < ma30[-i] and ma20[-i-1] > ma30[-i-1]))])
    
    return cross_5_10 + cross_5_20 + cross_5_30 + cross_10_20 + cross_10_30 + cross_20_30

def update_crossed_symbols_history(context, new_record):
    """更新穿越历史记录"""
    symbol = new_record['symbol']
    
    if symbol not in g.crossed_symbols_history:
        g.crossed_symbols_history[symbol] = []
    
    g.crossed_symbols_history[symbol].append(new_record)

def check_ma_relations(context, symbol):
    """检查MA均线之间的关系"""
    if symbol not in g.crossed_symbols_history:
        return False
        
    latest_record = g.crossed_symbols_history[symbol][-1]
    line_label = latest_record['line_label'].lower()
    line_type = latest_record['line_type']
    MA_values = latest_record['latest_data']
    
    MA5 = MA_values['ma5']
    MA10 = MA_values['ma10']
    MA20 = MA_values['ma20']
    MA30 = MA_values['ma30']
    
    conditions = {
        ('ma10', 'resistance'): MA5 > MA10 > MA20 > MA30,
        ('ma10', 'support'): MA30 > MA20 > MA10 > MA5,
        ('ma20', 'resistance'): (MA30 >= MA20 * 0.999 and MA5 >= MA10 * 0.999) or (MA30 > MA20 > MA10 > MA5),
        ('ma20', 'support'): (MA20 >= MA30 * 0.999 and MA10 >= MA5 * 0.999) or (MA5 > MA10 > MA20 > MA30),
        ('ma30', 'resistance'): (MA30 >= MA20 * 0.999 and MA5 >= MA10 * 0.999) or (MA30 > MA20 > MA10 > MA5),
        ('ma30', 'support'): (MA20 >= MA30 * 0.999 and MA10 >= MA5 * 0.999) or (MA5 > MA10 > MA20 > MA30),
    }
    
    return conditions.get((line_label, line_type), False)

def check_cross_details(context, symbol):
    """检查穿越详情"""
    if symbol not in g.crossed_symbols_history:
        return False, False
        
    all_records = g.crossed_symbols_history[symbol]
    if not all_records:
        return False, False

    # 检查穿越天数
    first_day = all_records[0]['date']
    today = context.current_dt.date()
    all_days = get_trade_days(first_day, today)
    duration_length = len(all_days)
    cross_duration = duration_length <= 6

    # 检查价格与均线的关系
    ma_close = False
    for record in all_records:
        line_label = record['line_label']
        ma_price = record['latest_data'][line_label]
        close = record['latest_data']['close']
        ma_close_rate = abs((close - ma_price) / ma_price)
        if ma_close_rate <= 0.02:
            ma_close = True

    return cross_duration, ma_close

def check_loss_for_close(context, position, direction, initial_loss_limit=-4000, loss_increment_per_day=200):
    """检查固定止损"""
    if position.security not in g.trade_history:
        return False
        
    trade_info = g.trade_history.get(position.security, {})
    finish_time = trade_info.get('finish_time')
    
    # 获取持仓天数
    holding_days = 0
    if finish_time:
        finish_date = finish_time.date()
        current_date = context.current_dt.date()
        all_trade_days = get_all_trade_days()
        holding_days = sum((finish_date <= d <= current_date) for d in all_trade_days)
        
    # 调整损失限制
    adjusted_loss_limit = initial_loss_limit + holding_days * loss_increment_per_day
    
    # 计算盈亏
    multiplier = position.value / (position.total_amount * position.price)
    if direction == 'long':
        revenue = multiplier * (position.price - position.acc_avg_cost) * position.total_amount
    else:
        revenue = multiplier * (position.acc_avg_cost - position.price) * position.total_amount

    if revenue < adjusted_loss_limit:
        close_position(position, g.trade_history[position.security]['direction'])
        return True

    return False

def check_ma_for_close(context, position, offset_ratio, days_for_adjustment):
    """根据均线止损"""
    security = position.security
    if security not in g.trade_history:
        return False
    
    trade_info = g.trade_history.get(position.security, {})
    finish_time = trade_info.get('finish_time')
    line_label = trade_info.get('line_label')
    
    # 获取持仓天数
    holding_days = 0
    if finish_time:
        finish_date = finish_time.date()
        current_date = context.current_dt.date()
        all_trade_days = get_all_trade_days()
        holding_days = sum((finish_date <= d <= current_date) for d in all_trade_days)
    
    # 计算变化率
    today_price = get_current_data()[position.security].last_price
    avg_daily_change_rate = calculate_average_daily_change_rate(position.security)
    historical_data = attribute_history(position.security, 1, '1d', ['close'])
    yesterday_close = historical_data['close'].iloc[-1]
    today_change_rate = abs((today_price - yesterday_close) / yesterday_close)

    # 选择止损均线
    close_line = None
    if today_change_rate >= 1.5 * avg_daily_change_rate:
        close_line = 'ma5'
    elif holding_days <= days_for_adjustment:
        close_line = line_label
    else:
        close_line = 'ma5' if today_change_rate >= 1.2 * avg_daily_change_rate else 'ma10'
    
    # 计算MA值并进行平仓判断
    ma_values = calculate_ma_values(position.security, [5, 10, 20, 30])
    adjusted_ma_value = ma_values[close_line] * (1 + offset_ratio if position.side == 'short' else 1 - offset_ratio)

    if (position.side == 'long' and today_price < adjusted_ma_value) or \
       (position.side == 'short' and today_price > adjusted_ma_value):
        close_position(position, g.trade_history[position.security]['direction'])
        return True

    return False

def calculate_average_daily_change_rate(security, days=30):
    """计算日均变化率"""
    historical_data = attribute_history(security, days + 1, '1d', ['close'])
    daily_change_rates = abs(historical_data['close'].pct_change()).iloc[1:]
    return daily_change_rates.mean()

def calculate_ma_values(security, ma_periods):
    """计算MA值"""
    historical_data = attribute_history(security, max(ma_periods), '1d', ['close'])
    today_price = get_current_data()[security].last_price
    close_prices = historical_data['close'].tolist() + [today_price]
    ma_values = {f'ma{period}': sum(close_prices[-period:]) / period for period in ma_periods}
    return ma_values

def check_symbol_prefix_match(symbol, hold_symbols):
    """检查是否有相似持仓"""
    symbol_prefix = symbol[:-9]
    
    for hold_symbol in hold_symbols:
        hold_symbol_prefix = hold_symbol[:-9]
        if symbol_prefix == hold_symbol_prefix:
            return True
    return False

def print_list_elements(title, elements):
    """打印列表元素"""
    print(f"{title}:")
    for item in elements:
        print(item)

############################ 移仓换月函数 ###################################

def position_auto_switch(context, pindex=0):
    """期货自动移仓换月"""
    import re
    subportfolio = context.subportfolios[pindex]
    symbols = set(subportfolio.long_positions.keys()) | set(subportfolio.short_positions.keys())
    switch_result = []
    
    for symbol in symbols:
        match = re.match(r"(?P<underlying_symbol>[A-Z]{1,})", symbol)
        if not match:
            continue
            
        dominant = get_dominant_future(match.groupdict()["underlying_symbol"])
        if not dominant or dominant <= symbol:
            continue
            
        cur = get_current_data()
        symbol_last_price = cur[symbol].last_price
        dominant_last_price = cur[dominant].last_price
        
        for positions_ in (subportfolio.long_positions, subportfolio.short_positions):
            if symbol not in positions_.keys():
                continue
                
            position = positions_[symbol]
            amount = position.total_amount
            side = position.side
            
            # 检查涨跌停限制
            if side == "long":
                symbol_low_limit = cur[symbol].low_limit
                dominant_high_limit = cur[dominant].high_limit
                if symbol_last_price <= symbol_low_limit or dominant_last_price >= dominant_high_limit:
                    continue
            else:
                symbol_high_limit = cur[symbol].high_limit
                dominant_low_limit = cur[dominant].low_limit
                if symbol_last_price >= symbol_high_limit or dominant_last_price <= dominant_low_limit:
                    continue
            
            # 执行移仓换月
            order_old = order_target(symbol, 0, side=side)
            if order_old is not None and order_old.filled > 0:
                order_new = order_target(dominant, amount, side=side)
                if order_new is not None and order_new.filled > 0:
                    # 换月成功，更新交易记录
                    if symbol in g.trade_history:
                        g.trade_history[dominant] = g.trade_history[symbol]
                        del g.trade_history[symbol]
                    switch_result.append({"before": symbol, "after": dominant, "side": side})
                    print(f"移仓换月成功: {symbol} -> {dominant}")
                else:
                    # 换月失败，记录失败信息
                    if symbol in g.trade_history:
                        g.change_fail_history[dominant] = g.trade_history[symbol]
                        del g.trade_history[symbol]
                    print(f"移仓换月失败: {symbol} -> {dominant}")
    
    return switch_result