# 导入函数库 from jqdata import * from jqdata import finance import pandas as pd import numpy as np from datetime import date, datetime, timedelta import re # 烛台影线形态反向交易策略 v002 # 基于烛台形态检测，采用两阶段仓位管理的交易策略 # 第一阶段：同时开立多空头寸（市场中性） # 第二阶段：亏损方平仓，保留盈利方 # 第三阶段：对剩余头寸应用标准止盈止损逻辑 # 设置以便完整打印 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('=' * 60) log.info('烛台影线形态交易策略 v002 初始化开始') log.info('策略类型: 两阶段仓位管理策略') log.info('=' * 60) ### 期货相关设定 ### # 设定账户为金融账户 set_subportfolios([SubPortfolioConfig(cash=context.portfolio.starting_cash, type='index_futures')]) # 期货类每笔交易时的手续费是: 买入时万分之0.23,卖出时万分之0.23,平今仓为万分之23 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.max_margin_per_position = 20000 # 单个标的最大持仓保证金（元） # 烛台形态检测参数（与研究文件保持一致） g.hatch_to_body_ratio = 1.2 # 影线与实体长度比率阈值 g.opposite_hatch_ratio = 0.5 # 相反方向影线与实体长度比率阈值 g.historical_days = 365 # 历史数据天数，用于计算平均实体长度阈值（与研究文件保持一致） # 全局阈值缓存（每月更新一次） g.monthly_body_threshold_cache = {} # 存储每月计算的实体长度阈值 g.last_threshold_update_month = None # 记录上次更新阈值的月份 # 止损止盈策略参数 g.fixed_stop_loss_rate = 0.01 # 固定止损比率（1%） g.trailing_stop_thresholds = [0.05, 0.10] # 动态追踪止损触发条件（5%, 10%） g.trailing_stop_rates = [0.01, 0.02, 0.03] # 动态追踪止损比率（1%, 2%, 3%） # 两阶段仓位管理参数 g.phase2_loss_threshold = 0.015 # 第二阶段亏损触发阈值（3%）- 当任一方向亏损达到此比例时平仓该方向 # 输出两阶段策略参数 log.info("两阶段仓位管理参数:") log.info(f" 第一阶段: 同时开立多空头寸（市场中性）") log.info(f" 第二阶段触发条件: 亏损方达到 {g.phase2_loss_threshold:.1%} 时平仓，保留盈利方") log.info(f" 第三阶段: 对剩余头寸应用标准止盈止损逻辑") log.info(f" - 固定止损: {g.fixed_stop_loss_rate:.1%}") log.info(f" - 动态追踪止损: {g.trailing_stop_rates}") # 开仓方向配置参数（v002版本已移除方向判断逻辑，保留变量以兼容其他代码） g.reverse_direction_symbols = [] # v002不再使用方向判断，同时开立多空头寸 # 期货品种完整配置字典 g.futures_config = { # 贵金属 'AU': {'has_night_session': True, 'margin_rate': {'long': 0.14, 'short': 0.14}, 'multiplier': 1000}, 'AG': {'has_night_session': True, 'margin_rate': {'long': 0.14, 'short': 0.14}, 'multiplier': 15}, # 有色金属 'CU': {'has_night_session': True, 'margin_rate': {'long': 0.09, 'short': 0.09}, 'multiplier': 5}, 'AL': {'has_night_session': True, 'margin_rate': {'long': 0.09, 'short': 0.09}, 'multiplier': 5}, 'ZN': {'has_night_session': True, 'margin_rate': {'long': 0.09, 'short': 0.09}, 'multiplier': 5}, 'PB': {'has_night_session': True, 'margin_rate': {'long': 0.09, 'short': 0.09}, 'multiplier': 5}, 'NI': {'has_night_session': True, 'margin_rate': {'long': 0.12, 'short': 0.12}, 'multiplier': 1}, 'SN': {'has_night_session': True, 'margin_rate': {'long': 0.12, 'short': 0.12}, 'multiplier': 1}, 'SS': {'has_night_session': True, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 5}, # 黑色系 'RB': {'has_night_session': True, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 10}, 'HC': {'has_night_session': True, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 10}, 'I': {'has_night_session': True, 'margin_rate': {'long': 0.1, 'short': 0.1}, 'multiplier': 100}, 'JM': {'has_night_session': True, 'margin_rate': {'long': 0.22, 'short': 0.22}, 'multiplier': 100}, 'J': {'has_night_session': True, 'margin_rate': {'long': 0.22, 'short': 0.22}, 'multiplier': 60}, # 能源化工 'SP': {'has_night_session': True, 'margin_rate': {'long': 0.1, 'short': 0.1}, 'multiplier': 10}, 'FU': {'has_night_session': True, 'margin_rate': {'long': 0.08, 'short': 0.08}, 'multiplier': 10}, 'BU': {'has_night_session': True, 'margin_rate': {'long': 0.04, 'short': 0.04}, 'multiplier': 10}, 'RU': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 10}, 'BR': {'has_night_session': True, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 5}, 'SC': {'has_night_session': True, 'margin_rate': {'long': 0.12, 'short': 0.12}, 'multiplier': 1000}, 'NR': {'has_night_session': True, 'margin_rate': {'long': 0.13, 'short': 0.13}, 'multiplier': 10}, 'LU': {'has_night_session': True, 'margin_rate': {'long': 0.15, 'short': 0.15}, 'multiplier': 10}, 'LC': {'has_night_session': False, 'margin_rate': {'long': 0.1, 'short': 0.1}, 'multiplier': 1}, # 化工 'FG': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 20}, 'TA': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, 'MA': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 10}, 'SA': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 20}, 'L': {'has_night_session': True, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 5}, 'V': {'has_night_session': True, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 5}, 'EG': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 10}, 'PP': {'has_night_session': True, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 5}, 'EB': {'has_night_session': True, 'margin_rate': {'long': 0.12, 'short': 0.12}, 'multiplier': 5}, 'PG': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 20}, # 农产品 'RM': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 10}, 'OI': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 10}, 'CF': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, 'SR': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 10}, 'PF': {'has_night_session': True, 'margin_rate': {'long': 0.1, 'short': 0.1}, 'multiplier': 5}, 'C': {'has_night_session': True, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 10}, 'CS': {'has_night_session': True, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 10}, 'CY': {'has_night_session': True, 'margin_rate': {'long': 0.15, 'short': 0.15}, 'multiplier': 5}, 'A': {'has_night_session': True, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 10}, 'B': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 10}, 'M': {'has_night_session': True, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 10}, 'Y': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 10}, 'P': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 10}, # 无夜盘品种 'IF': {'has_night_session': False, 'margin_rate': {'long': 0.08, 'short': 0.08}, 'multiplier': 300}, 'IH': {'has_night_session': False, 'margin_rate': {'long': 0.08, 'short': 0.08}, 'multiplier': 300}, 'IC': {'has_night_session': False, 'margin_rate': {'long': 0.08, 'short': 0.08}, 'multiplier': 200}, 'IM': {'has_night_session': False, 'margin_rate': {'long': 0.08, 'short': 0.08}, 'multiplier': 200}, 'AP': {'has_night_session': False, 'margin_rate': {'long': 0.08, 'short': 0.08}, 'multiplier': 10}, 'CJ': {'has_night_session': False, 'margin_rate': {'long': 0.09, 'short': 0.09}, 'multiplier': 5}, 'PK': {'has_night_session': False, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, 'JD': {'has_night_session': False, 'margin_rate': {'long': 0.07, 'short': 0.07}, 'multiplier': 10}, 'LH': {'has_night_session': False, 'margin_rate': {'long': 0.1, 'short': 0.1}, 'multiplier': 16} } # 策略品种选择策略配置 # 方案1：全品种策略 - 考虑所有配置的期货品种 # g.strategy_focus_symbols = [] # 空列表表示考虑所有品种 # 方案2：精选品种策略 - 只交易流动性较好的特定品种（如需使用请取消下行注释） g.strategy_focus_symbols = ['RM', 'CJ', 'CY', 'JD', 'L', 'LC', 'SF', 'SI'] log.info(f"品种选择策略: {'全品种策略（覆盖所有配置品种）' if not g.strategy_focus_symbols else '精选品种策略（' + str(len(g.strategy_focus_symbols)) + '个品种）'}") # 交易记录和数据存储 g.trade_history = {} g.candlestick_signals = {} # 存储烛台形态信号 g.daily_data_cache = {} # 存储历史日线数据缓存 g.minute_data_cache = {} # 存储今日分钟数据缓存 g.body_threshold_cache = {} # 存储实体长度阈值缓存 # 保证金比例管理 g.margin_rate_history = {} # 保证金比例变化历史记录 g.today_trades = [] # 当日交易记录 # 定时任务设置 # 每月第1个交易日更新阈值 run_monthly(monthly_update_thresholds, 1, 'before_open', reference_security='IF1808.CCFX') # 夜盘开始 - 仅止损止盈检查 run_daily(main_trading_stop_only, time='21:05:00', reference_security='IF1808.CCFX') run_daily(main_trading_stop_only, time='21:35:00', reference_security='IF1808.CCFX') run_daily(main_trading_stop_only, time='22:05:00', reference_security='IF1808.CCFX') run_daily(main_trading_stop_only, time='22:35:00', reference_security='IF1808.CCFX') # 日盘开始 - 仅止损止盈检查 run_daily(main_trading_stop_only, time='09:05:00', reference_security='IF1808.CCFX') run_daily(main_trading_stop_only, time='09:35:00', reference_security='IF1808.CCFX') run_daily(main_trading_stop_only, time='10:05:00', reference_security='IF1808.CCFX') run_daily(main_trading_stop_only, time='10:35:00', reference_security='IF1808.CCFX') run_daily(main_trading_stop_only, time='11:05:00', reference_security='IF1808.CCFX') run_daily(main_trading_stop_only, time='11:25:00', reference_security='IF1808.CCFX') run_daily(main_trading_stop_only, time='13:35:00', reference_security='IF1808.CCFX') run_daily(main_trading_stop_only, time='14:05:00', reference_security='IF1808.CCFX') # 收盘前 - 14:55进行完整交易检查，14:35仅止损止盈 run_daily(main_trading_stop_only, time='14:35:00', reference_security='IF1808.CCFX') # 完整交易检查 - 仅在14:55执行（任务1, 2, 3, 4, 5, 6, 7） run_daily(main_trading_complete, time='14:55:00', reference_security='IF1808.CCFX') # 收盘后 run_daily(after_market_close, time='15:30:00', reference_security='IF1808.CCFX') ############################ 主程序执行函数 ################################### def monthly_update_thresholds(context): """每月更新实体长度阈值""" log.info("=" * 60) log.info("每月阈值更新开始") log.info("=" * 60) current_month = context.current_dt.strftime('%Y-%m') # 检查是否需要更新 if g.last_threshold_update_month == current_month: log.info(f"本月阈值已更新，跳过更新") return # 获取所有需要计算阈值的品种 focus_symbols = g.strategy_focus_symbols if g.strategy_focus_symbols else list(g.futures_config.keys()) updated_count = 0 for symbol in focus_symbols: try: # 获取主力合约 dominant_future = get_dominant_future(symbol) if not dominant_future: continue # 获取365天历史数据 data = attribute_history(dominant_future, g.historical_days, '1d', ['open', 'close', 'high', 'low', 'volume'], df=True) if data is not None and len(data) > 30: # 确保有足够数据 # 计算实体长度阈值 data['body_length'] = abs(data['close'] - data['open']) body_threshold = data['body_length'].mean() # 存储到全局阈值缓存 g.monthly_body_threshold_cache[dominant_future] = body_threshold updated_count += 1 log.info(f"更新 {symbol}({dominant_future}) 实体长度阈值: {body_threshold:.4f}") except Exception as e: log.warning(f"更新{symbol}阈值时出错: {str(e)}") continue # 更新最后更新月份 g.last_threshold_update_month = current_month log.info(f"阈值更新完成，共更新 {updated_count} 个品种") log.info("=" * 60) def main_trading_stop_only(context): """仅进行止损止盈检查的交易时段""" # 只进行止损止盈检查 task_7_check_stop_loss_profit(context) def main_trading_complete(context): """14:55 完整交易检查 - 执行任务1, 2, 3, 4, 5, 6, 7""" log.info("=" * 60) log.info("14:55 烛台影线形态反向交易策略 - 完整交易检查") log.info("=" * 60) # 任务1: 获取所有可交易品种 task_1_get_tradable_futures(context) # 任务2: 获取历史数据并处理阈值（包括单独计算缺失阈值） task_2_load_historical_data_and_thresholds(context) # 任务3: 获取今日分钟数据 task_3_update_realtime_data(context) # 任务4: 检测烛台形态 task_4_detect_candlestick_patterns(context) # 任务5: 检查开仓条件（简化版） filtered_signals = task_5_check_opening_conditions(context) # 任务6: 执行交易 if filtered_signals: task_6_execute_trades(context, filtered_signals) # 任务7: 检查止损止盈 task_7_check_stop_loss_profit(context) # 任务8: 检查换月移仓 task_8_check_position_switch(context) ############################ 核心任务函数 ################################### def task_1_get_tradable_futures(context): """任务1: 获取所有可交易品种（分白天和晚上）""" log.info("执行任务1: 获取可交易品种") current_time = str(context.current_dt.time())[:2] # 从策略关注列表中筛选可交易品种 focus_symbols = g.strategy_focus_symbols if g.strategy_focus_symbols else list(g.futures_config.keys()) potential_icon_list = [] night_session_symbols = [] day_only_symbols = [] # 预先分类所有品种 for symbol in focus_symbols: if get_futures_config(symbol, 'has_night_session', False): night_session_symbols.append(symbol) else: day_only_symbols.append(symbol) if current_time in ('21', '22'): # 夜盘时间：只考虑有夜盘的品种 potential_icon_list = night_session_symbols[:] log.info(f"夜盘时间，有夜盘交易的品种: {len(night_session_symbols)}个") log.info(f"夜盘可交易品种: {potential_icon_list[:10]}{'...' if len(potential_icon_list) > 10 else ''}") # 显示前10个，避免日志过长 else: # 日盘时间：所有品种都可以交易（包括有夜盘的和只有日盘的） potential_icon_list = focus_symbols[:] log.info(f"日盘时间，全部品种: {len(focus_symbols)}个（含夜盘品种: {len(night_session_symbols)}个，日盘专属品种: {len(day_only_symbols)}个）") if day_only_symbols: log.info(f"日盘专属品种: {day_only_symbols}") log.info(f"日盘可交易品种: {potential_icon_list[:10]}{'...' if len(potential_icon_list) > 10 else ''}") # 显示前10个，避免日志过长 potential_future_list = [] for symbol in potential_icon_list: dominant_future = get_dominant_future(symbol) if dominant_future: potential_future_list.append(dominant_future) # 过滤掉已有持仓的品种 existing_positions = set(g.trade_history.keys()) potential_future_list = [f for f in potential_future_list if not check_symbol_prefix_match(f, existing_positions)] # 存储到全局变量 g.tradable_futures = potential_future_list log.info(f"最终可交易期货品种数量: {len(potential_future_list)}") return potential_future_list def task_2_load_historical_data_and_thresholds(context): """任务2: 获取历史数据并处理阈值（包括单独计算缺失阈值）""" log.info("执行任务2: 加载历史数据并处理阈值") if not hasattr(g, 'tradable_futures'): return for future_code in g.tradable_futures: try: # 获取30天历史数据（用于分钟数据合并） data = attribute_history(future_code, 30, '1d', ['open', 'close', 'high', 'low', 'volume'], df=True) if data is not None and len(data) > 0: # 排除今天的数据 today = context.current_dt.date() data = data[data.index.date < today] g.daily_data_cache[future_code] = data # 处理阈值：优先使用月度阈值，缺失时单独计算 if future_code in g.monthly_body_threshold_cache: # 使用月度阈值缓存 body_threshold = g.monthly_body_threshold_cache[future_code] g.body_threshold_cache[future_code] = body_threshold # log.info(f"已缓存 {future_code} 历史数据 {len(data)} 条，使用月度阈值: {body_threshold:.4f}") else: # 单独计算缺失的阈值 log.info(f"{future_code} 未找到月度阈值，开始单独计算") body_threshold = calculate_individual_threshold(context, future_code) if body_threshold is not None: g.body_threshold_cache[future_code] = body_threshold log.info(f"已为 {future_code} 单独计算阈值: {body_threshold:.4f}") else: log.warning(f"{future_code} 无法计算阈值，跳过") except Exception as e: log.warning(f"加载{future_code}历史数据时出错: {str(e)}") continue log.info(f"历史数据缓存完成，共缓存 {len(g.daily_data_cache)} 个品种") def calculate_individual_threshold(context, future_code): """为单个工具计算实体长度阈值并自动缓存到月度阈值中""" try: log.info(f"为 {future_code} 单独计算365天历史阈值") # 获取365天历史数据 data = attribute_history(future_code, g.historical_days, '1d', ['open', 'close', 'high', 'low', 'volume'], df=True) if data is not None and len(data) > 30: # 确保有足够数据 # 排除今天的数据 today = context.current_dt.date() data = data[data.index.date < today] # 计算实体长度阈值 data['body_length'] = abs(data['close'] - data['open']) body_threshold = data['body_length'].mean() # 🚀 优化1：将计算结果自动添加到月度阈值缓存中 g.monthly_body_threshold_cache[future_code] = body_threshold log.info(f"✅ {future_code} 单独计算完成，阈值: {body_threshold:.4f}，已添加到月度缓存") log.info(f"当前月度缓存包含 {len(g.monthly_body_threshold_cache)} 个品种阈值") return body_threshold else: log.warning(f"{future_code} 历史数据不足，无法计算阈值") return None except Exception as e: log.warning(f"为 {future_code} 单独计算阈值时出错: {str(e)}") return None def task_3_update_realtime_data(context): """任务3: 获取今日分钟数据""" # log.info("执行任务3: 更新实时数据") # 收集需要更新数据的品种 update_symbols = set() # 添加可交易品种 if hasattr(g, 'tradable_futures') and g.tradable_futures: update_symbols.update(g.tradable_futures) # 添加持仓品种（用于止损止盈）- 去掉_long/_short后缀 if hasattr(g, 'trade_history') and g.trade_history: for key in g.trade_history.keys(): # 去掉可能的_long或_short后缀 clean_symbol = key.replace('_long', '').replace('_short', '') update_symbols.add(clean_symbol) if not update_symbols: return for future_code in update_symbols: try: # 获取今日分钟数据 minute_data = get_today_minute_data(context, future_code) if minute_data is None: continue # 获取历史数据 historical_data = g.daily_data_cache.get(future_code) if historical_data is None: # 为持仓品种临时获取历史数据 try: data = attribute_history(future_code, g.historical_days, '1d', ['open', 'close', 'high', 'low', 'volume'], df=True) if data is not None and len(data) > 0: today = context.current_dt.date() data = data[data.index.date < today] # 计算实体长度阈值 data['body_length'] = abs(data['close'] - data['open']) body_threshold = data['body_length'].mean() g.daily_data_cache[future_code] = data g.body_threshold_cache[future_code] = body_threshold historical_data = data log.info(f"为持仓品种 {future_code} 临时获取历史数据 {len(data)} 条") except Exception as e: log.warning(f"获取{future_code}历史数据失败: {str(e)}") continue if historical_data is None: continue # 将今日数据合并为日线数据 today_data = aggregate_minute_to_daily(minute_data) if today_data is not None: # 合并历史数据和今日数据 combined_data = pd.concat([historical_data, today_data], sort=False) g.minute_data_cache[future_code] = combined_data except Exception as e: log.warning(f"更新{future_code}实时数据时出错: {str(e)}") continue def task_4_detect_candlestick_patterns(context): """任务4: 检测烛台形态""" # log.info("执行任务4: 检测烛台形态") candlestick_signals = [] # 获取已持仓的品种列表 existing_positions = set(g.trade_history.keys()) for future_code, data in g.minute_data_cache.items(): try: # 检查是否已有相似持仓，如果有则跳过分析 if check_symbol_prefix_match(future_code, existing_positions): continue # 获取实体长度阈值 body_threshold = g.body_threshold_cache.get(future_code) if body_threshold is None: continue # 检查最新的烛台形态 latest_pattern = check_latest_candlestick_pattern(data, future_code, body_threshold) if latest_pattern: candlestick_signals.append(latest_pattern) log.info(f"{future_code} 发现烛台形态: {latest_pattern['hatch_direction']}影线") except Exception as e: log.warning(f"检测{future_code}烛台形态时出错: {str(e)}") continue # 存储到全局变量 g.candlestick_signals = candlestick_signals if len(candlestick_signals) > 0: log.info(f"烛台形态检测完成，发现信号 {len(candlestick_signals)} 个") return candlestick_signals def task_5_check_opening_conditions(context): """任务5: 检查开仓条件（简化版）""" log.info("执行任务5: 检查开仓条件（简化版）") if not hasattr(g, 'candlestick_signals') or not g.candlestick_signals: log.info("没有检测到烛台信号") return [] log.info(f"检测到 {len(g.candlestick_signals)} 个烛台信号，开始筛选") filtered_signals = [] for i, signal in enumerate(g.candlestick_signals): log.info(f"处理信号 {i+1}: {signal['symbol']} {signal['hatch_direction']}影线") # 仅检查是否已有相似持仓（避免重复开仓同一品种） if check_symbol_prefix_match(signal['symbol'], set(g.trade_history.keys())): log.info(f"{signal['symbol']} 已有相似持仓，跳过") continue # 简化版：只要形态满足条件就接受，无额外验证 filtered_signals.append(signal) log.info(f"{signal['symbol']} 烛台形态满足条件，接受交易信号") log.info(f"开仓条件检查完成，满足条件 {len(filtered_signals)} 个") return filtered_signals def task_6_execute_trades(context, filtered_signals): """任务6: 执行交易（两阶段策略版 - 同时开立多空头寸）""" log.info("执行任务6: 执行交易（两阶段策略版）") if not filtered_signals: log.info("没有满足条件的交易信号") return log.info(f"开始执行 {len(filtered_signals)} 个交易信号") for i, signal in enumerate(filtered_signals): symbol = signal['symbol'] log.info(f"执行交易 {i+1}: {symbol} {signal['hatch_direction']}影线形态") log.info(f"{symbol} 第一阶段: 同时开立多空头寸（市场中性）") try: # 计算目标手数（使用long方向计算，多空手数相同） target_hands = calculate_target_hands(context, symbol, 'long') log.info(f"计算目标手数: {symbol} 多空各 {target_hands} 手") if target_hands > 0: # 同时开立多头和空头头寸 log.info(f"开始同时下单: {symbol} 多头 {target_hands}手 + 空头 {target_hands}手") success = open_dual_position(context, symbol, target_hands, signal) if success: log.info(f"✅ 成功开立双向头寸 {symbol}, 多空各 {target_hands} 手（第一阶段）") else: log.warning(f"❌ 双向开仓失败 {symbol}") else: log.warning(f"{symbol} 计算目标手数为0，跳过") except Exception as e: log.warning(f"{symbol} 交易执行出错: {str(e)}") continue log.info("交易执行完成") def task_8_check_position_switch(context): """任务8: 检查换月移仓""" # log.info("执行任务8: 检查换月移仓") switch_result = position_auto_switch(context) if switch_result: log.info(f"执行了 {len(switch_result)} 次移仓换月") for result in switch_result: log.info(f"移仓: {result['before']} -> {result['after']}") def task_7_check_stop_loss_profit(context): """任务7: 检查止损止盈（带交易时间验证）""" # log.info("执行任务7: 检查止损止盈") # 获取当前时间 current_time = context.current_dt.strftime('%H:%M') current_hour = int(current_time[:2]) # 判断是否为夜盘时间（21:00-23:00 和 00:00-02:30） is_night_session = (current_hour >= 21) or (current_hour <= 2) # 遍历所有持仓进行止损止盈检查 subportfolio = context.subportfolios[0] long_positions = list(subportfolio.long_positions.values()) short_positions = list(subportfolio.short_positions.values()) closed_count = 0 skipped_count = 0 for position in long_positions + short_positions: security = position.security underlying_symbol = security.split('.')[0][:-4] # 检查交易时间适配性 has_night_session = get_futures_config(underlying_symbol, 'has_night_session', False) # 如果是夜盘时间，但品种不支持夜盘交易，则跳过 if is_night_session and not has_night_session: skipped_count += 1 # log.info(f"跳过夜盘时间的日间品种: {underlying_symbol} ({current_time})") continue # 执行止损止盈检查 if check_stop_loss_profit(context, position): closed_count += 1 if closed_count > 0: log.info(f"执行了 {closed_count} 次止损止盈") if skipped_count > 0: log.info(f"夜盘时间跳过 {skipped_count} 个日间品种的止损止盈检查") ############################ 数据处理辅助函数 ################################### def check_has_night_session(underlying_symbol): """检查品种是否有夜盘""" return get_futures_config(underlying_symbol, 'has_night_session', False) def get_today_minute_data(context, future_code): """获取今日分钟数据""" try: # 判断该品种是否有夜盘 underlying_symbol = future_code.split('.')[0][:-4] has_night_session = check_has_night_session(underlying_symbol) end_time = context.current_dt # 获取足够的历史分钟数据 minute_data = attribute_history(future_code, count=800, # 获取足够多的数据 unit='1m', fields=['open', 'close', 'high', 'low', 'volume'], df=True) if minute_data is None or len(minute_data) == 0: return None # 提取所有日期（年月日维度） minute_data['date'] = minute_data.index.date unique_dates = sorted(minute_data['date'].unique()) if has_night_session: # 有夜盘的品种：需要找到前一交易日的21:00作为今日开盘起点 today_date = end_time.date() # 找到今天之前的最后一个交易日 previous_trading_dates = [d for d in unique_dates if d < today_date] if not previous_trading_dates: return minute_data previous_trading_date = max(previous_trading_dates) # 找到前一交易日21:00:00的数据作为开盘起点 previous_day_data = minute_data[minute_data['date'] == previous_trading_date] night_21_data = previous_day_data[previous_day_data.index.hour == 21] if len(night_21_data) > 0: # 从前一交易日21:00开始的所有数据 start_time = night_21_data.index[0] # 21:00:00的时间点 filtered_data = minute_data[minute_data.index >= start_time] return filtered_data.drop(columns=['date']) else: # 备选方案：使用今天9:00开始的数据 today_data = minute_data[minute_data['date'] == today_date] day_9_data = today_data[today_data.index.hour >= 9] if len(day_9_data) > 0: return day_9_data.drop(columns=['date']) else: return minute_data.drop(columns=['date']) else: # 没有夜盘的品种：从今天9:00:00开始 today_date = end_time.date() today_data = minute_data[minute_data['date'] == today_date] # 找到今天9:00:00开始的数据 day_9_data = today_data[today_data.index.hour >= 9] if len(day_9_data) > 0: return day_9_data.drop(columns=['date']) else: return today_data.drop(columns=['date']) if len(today_data) > 0 else minute_data.drop(columns=['date']) except Exception as e: log.warning(f"获取{future_code}今日分钟数据时出错: {str(e)}") return None def aggregate_minute_to_daily(minute_data): """将分钟数据聚合为日数据""" try: if minute_data is None or len(minute_data) == 0: return None # 获取今日日期（使用最后一条数据的日期作为今日日期） today_date = minute_data.index[-1].date() # 聚合为日数据 daily_data = pd.DataFrame({ 'open': [minute_data['open'].iloc[0]], # 今日交易开始时的开盘价 'close': [minute_data['close'].iloc[-1]], # 当前收盘价，实时更新 'high': [minute_data['high'].max()], 'low': [minute_data['low'].min()], 'volume': [minute_data['volume'].sum()] }, index=[pd.Timestamp(today_date)]) return daily_data except Exception as e: log.warning(f"聚合分钟数据时出错: {str(e)}") return None def check_latest_candlestick_pattern(data, future_code, body_threshold): """检查最新的烛台形态""" if len(data) < 1: return None # 获取最新一天的数据 today = data.iloc[-1] # 检查烛台形态 pattern_result = check_candlestick_pattern_single_day(today, body_threshold) if not pattern_result: return None return { 'symbol': future_code, 'date': today.name, 'hatch_direction': pattern_result['hatch_direction'], 'open': today['open'], 'close': today['close'], 'high': today['high'], 'low': today['low'], 'body_length': pattern_result['body_length'], 'hatch_length': pattern_result['hatch_length'] } def check_candlestick_pattern_single_day(row, body_threshold): """检查单日烛台形态是否符合影线条件""" 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: return None # 计算影线长度 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 >= g.hatch_to_body_ratio * body_length and lower_hatch < g.opposite_hatch_ratio * body_length): hatch_direction = 'up' hatch_length = upper_hatch # 检查下影线条件：下影线长度符合要求 且 上影线长度小于实体长度的一半 elif (lower_hatch >= g.hatch_to_body_ratio * body_length and upper_hatch < g.opposite_hatch_ratio * body_length): hatch_direction = 'down' hatch_length = lower_hatch # 如果满足影线条件，返回形态信息 if hatch_direction is not None: return { 'hatch_direction': hatch_direction, 'body_length': body_length, 'hatch_length': hatch_length } return None ############################ 动态保证金率调整函数 ################################### def detect_and_update_margin_rates(context, symbol, actual_hands, cash_change, direction): """ 检测并更新保证金率配置 参数: symbol: 期货合约代码 actual_hands: 实际成交手数 cash_change: 实际资金变化（保证金使用） direction: 交易方向 """ try: underlying_symbol = symbol.split('.')[0][:-4] if underlying_symbol not in g.futures_config: return current_price = get_current_data()[symbol].last_price multiplier = get_multiplier(underlying_symbol) # 计算实际保证金率 contract_value = current_price * multiplier * actual_hands actual_margin_rate = cash_change / contract_value if contract_value > 0 else 0 # 获取配置中的保证金率 config_margin_rate = g.futures_config[underlying_symbol]['margin_rate'][direction] # 计算差异百分比 rate_diff_percentage = abs(actual_margin_rate - config_margin_rate) / config_margin_rate * 100 if config_margin_rate > 0 else 100 # 如果差异超过10%，更新配置 if rate_diff_percentage > 10: log.info(f"🔍 检测到{underlying_symbol}保证金率差异: 配置={config_margin_rate:.3f}, 实际={actual_margin_rate:.3f}, 差异={rate_diff_percentage:.1f}%") # 更新配置 g.futures_config[underlying_symbol]['margin_rate'][direction] = actual_margin_rate # 如果是双向更新（多空保证金率通常相同） if g.futures_config[underlying_symbol]['margin_rate']['long'] == g.futures_config[underlying_symbol]['margin_rate']['short']: g.futures_config[underlying_symbol]['margin_rate']['long'] = actual_margin_rate g.futures_config[underlying_symbol]['margin_rate']['short'] = actual_margin_rate log.info(f"✅ 已更新{underlying_symbol}保证金率为 {actual_margin_rate:.3f} (双向)") else: log.info(f"✅ 已更新{underlying_symbol} {direction}保证金率为 {actual_margin_rate:.3f}") except Exception as e: log.warning(f"检测保证金率时出错 {symbol}: {str(e)}") ############################ 机会性仓位调整函数 ################################### def opportunistic_position_increase(context, symbol, direction, signal): """ 机会性仓位调整：在保证金使用量低于限制时自动增加仓位 参数: symbol: 期货合约代码 direction: 交易方向 signal: 交易信号 """ try: if symbol not in g.trade_history: return False underlying_symbol = symbol.split('.')[0][:-4] current_price = get_current_data()[symbol].last_price margin_rate = get_margin_rate(underlying_symbol, direction) multiplier = get_multiplier(underlying_symbol) # 获取当前持仓信息 current_actual_margin = g.trade_history[symbol]['actual_margin'] current_hands = g.trade_history[symbol]['actual_hands'] # 计算剩余保证金容量 max_margin = g.max_margin_per_position remaining_margin_capacity = max_margin - current_actual_margin # 检查是否有足够的剩余容量增加至少1手 single_hand_margin = current_price * multiplier * margin_rate if remaining_margin_capacity >= single_hand_margin: # 计算可以增加的最大手数 additional_hands = int(remaining_margin_capacity / single_hand_margin) # 同时考虑可用资金限制 available_cash = context.portfolio.available_cash * g.usage_percentage max_hands_by_cash = int(available_cash / single_hand_margin) # 取较小值 additional_hands = min(additional_hands, max_hands_by_cash) if additional_hands >= 1: log.info(f"🚀 机会性增仓机会: {symbol} 当前{current_hands}手(保证金:{current_actual_margin:.0f}), 可增加{additional_hands}手") # 计算新的目标手数 new_target_hands = current_hands + additional_hands # 执行增仓 order = order_target(symbol, new_target_hands, side=direction) if order and order.filled > 0: # 计算实际增仓使用的保证金 cash_after = context.portfolio.available_cash additional_margin_used = order.filled * single_hand_margin # 估算使用的保证金 # 更新交易记录 new_total_hands = current_hands + order.filled new_total_margin = current_actual_margin + additional_margin_used g.trade_history[symbol]['actual_hands'] = new_total_hands g.trade_history[symbol]['target_hands'] = new_target_hands g.trade_history[symbol]['actual_margin'] = new_total_margin log.info(f"✅ 成功增仓 {symbol} {order.filled}手, 总手数: {new_total_hands}手, 总保证金: {new_total_margin:.0f}") # 记录增仓交易 g.today_trades.append({ 'security': symbol, 'underlying_symbol': underlying_symbol, 'direction': direction, 'order_amount': order.filled, 'order_price': order.avg_cost if order.avg_cost else order.price, 'cash_change': additional_margin_used, 'time': context.current_dt, 'trade_type': 'increase' # 标记为增仓 }) return True return False except Exception as e: log.warning(f"机会性增仓时出错 {symbol}: {str(e)}") return False ############################ 交易执行函数 ################################### def open_dual_position(context, security, target_hands, signal): """同时开立多空头寸（第一阶段）""" try: underlying_symbol = security.split('.')[0][:-4] # 记录交易前的可用资金 cash_before = context.portfolio.available_cash # 先开多头 log.info(f"开立多头头寸: {security} {target_hands}手") order_long = order_target(security, target_hands, side='long') if order_long is None or order_long.filled == 0: log.warning(f"多头开仓失败: {security}") return False # 记录多头开仓后的资金 cash_after_long = context.portfolio.available_cash cash_change_long = cash_before - cash_after_long # 再开空头 log.info(f"开立空头头寸: {security} {target_hands}手") order_short = order_target(security, target_hands, side='short') if order_short is None or order_short.filled == 0: log.warning(f"空头开仓失败: {security}，需要平掉多头") # 如果空头开仓失败，平掉已开的多头 close_position(context, security, 'long') return False # 记录空头开仓后的资金 cash_after_short = context.portfolio.available_cash cash_change_short = cash_after_long - cash_after_short # 获取订单价格和数量 long_price = order_long.avg_cost if order_long.avg_cost else order_long.price short_price = order_short.avg_cost if order_short.avg_cost else order_short.price long_amount = order_long.filled short_amount = order_short.filled # 记录当日交易 g.today_trades.append({ 'security': security, 'underlying_symbol': underlying_symbol, 'direction': 'long', 'order_amount': long_amount, 'order_price': long_price, 'cash_change': cash_change_long, 'time': context.current_dt, 'phase': 1 }) g.today_trades.append({ 'security': security, 'underlying_symbol': underlying_symbol, 'direction': 'short', 'order_amount': short_amount, 'order_price': short_price, 'cash_change': cash_change_short, 'time': context.current_dt, 'phase': 1 }) # 生成唯一的配对ID pair_id = f"{security}_{context.current_dt.strftime('%Y%m%d_%H%M%S')}" # 记录多头交易信息 long_key = f"{security}_long" g.trade_history[long_key] = { 'entry_price': long_price, 'target_hands': target_hands, 'actual_hands': long_amount, 'actual_margin': cash_change_long, 'direction': 'long', 'entry_time': context.current_dt, 'signal_info': signal, 'max_profit': 0.0, 'max_profit_price': long_price, 'phase': 1, # 第一阶段：双向持仓 'pair_id': pair_id, # 配对ID 'paired_position': f"{security}_short" # 配对的另一方 } # 记录空头交易信息 short_key = f"{security}_short" g.trade_history[short_key] = { 'entry_price': short_price, 'target_hands': target_hands, 'actual_hands': short_amount, 'actual_margin': cash_change_short, 'direction': 'short', 'entry_time': context.current_dt, 'signal_info': signal, 'max_profit': 0.0, 'max_profit_price': short_price, 'phase': 1, # 第一阶段：双向持仓 'pair_id': pair_id, # 配对ID 'paired_position': f"{security}_long" # 配对的另一方 } log.info(f"成功开立双向头寸 - 多头: {long_amount}手@{long_price:.2f}(保证金:{cash_change_long:.0f}), " f"空头: {short_amount}手@{short_price:.2f}(保证金:{cash_change_short:.0f})") return True except Exception as e: log.warning(f"双向开仓失败 {security}: {str(e)}") return False def open_position(context, security, target_hands, direction, signal): """开仓（优化版：使用order_target按手数开仓）""" try: # 记录交易前的可用资金 cash_before = context.portfolio.available_cash # 使用order_target按手数开仓，自动处理持仓差额 order = order_target(security, target_hands, side=direction) if order is not None and order.filled > 0: # 记录交易后的可用资金 cash_after = context.portfolio.available_cash # 计算实际资金变化 cash_change = cash_before - cash_after # 获取订单价格和数量 order_price = order.avg_cost if order.avg_cost else order.price order_amount = order.filled # 记录当日交易 underlying_symbol = security.split('.')[0][:-4] g.today_trades.append({ 'security': security, 'underlying_symbol': underlying_symbol, 'direction': direction, 'order_amount': order_amount, 'order_price': order_price, 'cash_change': cash_change, 'time': context.current_dt }) # 记录交易信息，包含止损止盈相关信息 g.trade_history[security] = { 'entry_price': order_price, 'target_hands': target_hands, 'actual_hands': order_amount, 'actual_margin': cash_change, 'direction': direction, 'entry_time': context.current_dt, 'signal_info': signal, 'max_profit': 0.0, # 初始化最大利润 'max_profit_price': order_price # 记录最大利润时的价格 } # 🚀 优化2：动态保证金率调整 detect_and_update_margin_rates(context, security, order_amount, cash_change, direction) # 🚀 优化3：机会性仓位调整 opportunistic_position_increase(context, security, direction, signal) return True except Exception as e: log.warning(f"开仓失败 {security}: {str(e)}") return False def close_position(context, security, direction): """平仓（优化版：使用order_target平仓到0手）""" try: # 使用order_target平仓到0手，自动处理持仓清零 order = order_target(security, 0, side=direction) if order is not None and order.filled > 0: underlying_symbol = security.split('.')[0][:-4] # 记录当日交易（平仓） g.today_trades.append({ 'security': security, 'underlying_symbol': underlying_symbol, 'direction': direction, 'order_amount': -order.filled, # 负数表示平仓 'order_price': order.avg_cost if order.avg_cost else order.price, 'cash_change': 0, # 平仓不计算保证金变化 'time': context.current_dt }) log.info(f"平仓成功 - 品种: {underlying_symbol}, 平仓手数: {order.filled}") # 从交易历史中移除 if security in g.trade_history: del g.trade_history[security] return True except Exception as e: log.warning(f"平仓失败 {security}: {str(e)}") return False def check_stop_loss_profit(context, position): """检查止损止盈（支持两阶段策略）""" security = position.security # 尝试匹配trade_history中的key（可能是原始security或带_long/_short后缀的） trade_key = None if security in g.trade_history: trade_key = security else: # 检查是否有带方向后缀的key for key in g.trade_history.keys(): if key.startswith(security + '_'): if g.trade_history[key]['direction'] == position.side: trade_key = key break if trade_key is None: return False trade_info = g.trade_history[trade_key] direction = trade_info['direction'] entry_price = trade_info['entry_price'] current_price = position.price phase = trade_info.get('phase', 3) # 默认第三阶段（兼容旧数据） # 计算当前盈亏比率 if direction == 'long': profit_rate = (current_price - entry_price) / entry_price else: profit_rate = (entry_price - current_price) / entry_price # 更新最大利润记录 if profit_rate > trade_info['max_profit']: trade_info['max_profit'] = profit_rate trade_info['max_profit_price'] = current_price g.trade_history[trade_key] = trade_info # 第一阶段：双向持仓，检查是否有一方亏损达到阈值 if phase == 1: # 只有亏损达到阈值才平仓 if profit_rate <= -g.phase2_loss_threshold: log.info(f"【第二阶段触发】{security} {direction} 亏损达到阈值") log.info(f"亏损率: {profit_rate:.3%}, 触发阈值: {-g.phase2_loss_threshold:.3%}, 成本价: {entry_price:.2f}, 当前价格: {current_price:.2f}") # 平仓亏损方向 close_position(context, security, direction) # 更新配对的另一方进入第三阶段 paired_key = trade_info.get('paired_position') if paired_key and paired_key in g.trade_history: g.trade_history[paired_key]['phase'] = 3 log.info(f"【进入第三阶段】{paired_key} 继续持有，应用标准止盈止损逻辑") return True # 第一阶段不进行其他止损止盈检查 return False # 第三阶段：单向持仓，应用原有的标准止盈止损逻辑 # 检查固定止损 if profit_rate <= -g.fixed_stop_loss_rate: log.info(f"【第三阶段】触发固定止损 {security} {direction}, 当前亏损率: {profit_rate:.3%}, 成本价: {entry_price:.2f}, 当前价格: {current_price:.2f}") close_position(context, security, direction) return True # 检查动态追踪止盈 max_profit = trade_info['max_profit'] if max_profit > 0: # 确定追踪止损比率 if max_profit <= g.trailing_stop_thresholds[0]: # ≤5% trailing_rate = g.trailing_stop_rates[0] # 2% elif max_profit <= g.trailing_stop_thresholds[1]: # 5%-10% trailing_rate = g.trailing_stop_rates[1] # 3% else: # >10% trailing_rate = g.trailing_stop_rates[2] # 4% # 检查是否触发追踪止损 profit_drawdown = max_profit - profit_rate if profit_drawdown >= trailing_rate: log.info(f"【第三阶段】触发动态追踪止损 {security} {direction}") log.info(f"最大利润: {max_profit:.3%}, 当前利润: {profit_rate:.3%}, 回撤: {profit_drawdown:.3%}, 触发阈值: {trailing_rate:.3%}") close_position(context, security, direction) return True return False ############################ 辅助函数 ################################### def determine_trading_direction(symbol, hatch_direction): """ 确定开仓方向（支持可配置的正向/反向逻辑） 参数: symbol: 期货合约代码 (如 'JD2510.XDCE') hatch_direction: 影线方向 ('up' 或 'down') 返回: (direction, logic_description): 交易方向和逻辑说明的元组 """ # 提取基础品种代码 underlying_symbol = symbol.split('.')[0][:-4] # 如 'JD2510.XDCE' -> 'JD' # 检查是否在反向逻辑列表中 use_reverse_logic = underlying_symbol in g.reverse_direction_symbols if use_reverse_logic: # 反向逻辑：上影线做空，下影线做多（当前逻辑） if hatch_direction == 'up': direction = 'short' logic_description = f"反向逻辑：{underlying_symbol}上影线做空" else: direction = 'long' logic_description = f"反向逻辑：{underlying_symbol}下影线做多" else: # 正向逻辑：上影线做多，下影线做空（新逻辑） if hatch_direction == 'up': direction = 'long' logic_description = f"正向逻辑：{underlying_symbol}上影线做多" else: direction = 'short' logic_description = f"正向逻辑：{underlying_symbol}下影线做空" return direction, logic_description def get_futures_config(underlying_symbol, config_key=None, default_value=None): """获取期货品种配置信息的辅助函数""" if underlying_symbol not in g.futures_config: if config_key and default_value is not None: return default_value return {} if config_key is None: return g.futures_config[underlying_symbol] return g.futures_config[underlying_symbol].get(config_key, default_value) def get_margin_rate(underlying_symbol, direction, default_rate=0.10): """获取保证金比例的辅助函数""" return g.futures_config.get(underlying_symbol, {}).get('margin_rate', {}).get(direction, default_rate) def get_multiplier(underlying_symbol, default_multiplier=10): """获取合约乘数的辅助函数""" return g.futures_config.get(underlying_symbol, {}).get('multiplier', default_multiplier) def calculate_target_hands(context, security, direction): """计算目标开仓手数（优化版：直接返回手数用于order_target）""" current_price = get_current_data()[security].last_price underlying_symbol = security.split('.')[0][:-4] # 使用保证金比例 margin_rate = get_margin_rate(underlying_symbol, direction) multiplier = get_multiplier(underlying_symbol) # 计算单手保证金 single_hand_margin = current_price * multiplier * margin_rate # 还要考虑可用资金限制 available_cash = context.portfolio.available_cash * g.usage_percentage # 根据单个标的最大持仓保证金限制计算开仓数量 max_margin = g.max_margin_per_position if single_hand_margin <= max_margin: # 如果单手保证金不超过最大限制，计算最大可开仓手数 max_hands = int(max_margin / single_hand_margin) max_hands_by_cash = int(available_cash / single_hand_margin) # 取两者较小值 actual_hands = min(max_hands, max_hands_by_cash) # 确保至少开1手 actual_hands = max(1, actual_hands) # 实际保证金 actual_margin = single_hand_margin * actual_hands log.info(f"单手保证金: {single_hand_margin:.0f}, 目标开仓手数: {actual_hands}, 预计保证金: {actual_margin:.0f}") # 直接返回手数，用于order_target() return actual_hands else: # 如果单手保证金超过最大限制，默认开仓1手 actual_hands = 1 actual_margin = single_hand_margin * actual_hands log.info(f"单手保证金: {single_hand_margin:.0f} 超过最大限制: {max_margin}, 默认开仓1手, 预计保证金: {actual_margin:.0f}") # 直接返回手数，用于order_target() return actual_hands def check_sufficient_capital(context, symbol): """检查资金是否充足""" try: current_price = get_current_data()[symbol].last_price underlying_symbol = symbol.split('.')[0][:-4] margin_rate = get_margin_rate(underlying_symbol, 'long') multiplier = get_multiplier(underlying_symbol) # 计算单手保证金 single_hand_margin = current_price * multiplier * margin_rate # 使用实际可用资金（考虑资金使用比例） available_cash = context.portfolio.available_cash * g.usage_percentage # 检查是否有足够资金开仓至少1手 return available_cash >= single_hand_margin except: return False def check_price_and_liquidity(signal): """检查价格合理性和流动性""" try: # 简单的合理性检查 symbol = signal['symbol'] current_data = get_current_data()[symbol] log.info(f"{symbol} 价格检查 - 当前价: {current_data.last_price:.2f}, 涨停: {current_data.high_limit:.2f}, 跌停: {current_data.low_limit:.2f}, 成交量: {current_data.volume}") # 检查是否处于涨跌停 if current_data.last_price <= current_data.low_limit: log.warning(f"{symbol} 价格触及跌停板 ({current_data.last_price:.2f} <= {current_data.low_limit:.2f})，跳过") return False if current_data.last_price >= current_data.high_limit: log.warning(f"{symbol} 价格触及涨停板 ({current_data.last_price:.2f} >= {current_data.high_limit:.2f})，跳过") return False # 检查成交量是否充足 if current_data.volume <= 0: log.warning(f"{symbol} 无成交量 ({current_data.volume})，跳过") return False log.info(f"{symbol} 价格和流动性检查通过") return True except Exception as e: log.warning(f"{signal['symbol']} 价格检查时出错: {str(e)}") return False def check_symbol_prefix_match(symbol, hold_symbols): """检查是否有相似的持仓品种（支持带_long/_short后缀的key）""" symbol_prefix = symbol[:-9] for hold_symbol in hold_symbols: # 移除可能的_long或_short后缀 clean_hold_symbol = hold_symbol.replace('_long', '').replace('_short', '') hold_symbol_prefix = clean_hold_symbol[:-9] if len(clean_hold_symbol) > 9 else clean_hold_symbol if symbol_prefix == hold_symbol_prefix: return True return False def after_market_close(context): """收盘后运行函数""" log.info(str('函数运行时间(after_market_close):'+str(context.current_dt.time()))) # 只有当天有交易时才打印统计信息 if g.today_trades: print_daily_trading_summary(context) # 清空当日交易记录 g.today_trades = [] log.info('##############################################################') def print_daily_trading_summary(context): """打印当日交易汇总""" if not g.today_trades: return log.info("\n=== 当日交易汇总 ===") total_margin = 0 for trade in g.today_trades: if trade['order_amount'] > 0: # 开仓 log.info(f"开仓 {trade['underlying_symbol']} {trade['direction']} {trade['order_amount']}手 " f"价格:{trade['order_price']:.2f} 保证金:{trade['cash_change']:.0f}") total_margin += trade['cash_change'] else: # 平仓 log.info(f"平仓 {trade['underlying_symbol']} {trade['direction']} {abs(trade['order_amount'])}手 " f"价格:{trade['order_price']:.2f}") log.info(f"当日保证金占用: {total_margin:.0f}") log.info("==================\n") ########################## 自动移仓换月函数 ################################# def position_auto_switch(context, pindex=0, switch_func=None, callback=None): """ 期货自动移仓换月。默认使用市价单进行开平仓。 """ 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[A-Z]{1,})", symbol) if not match: raise ValueError("未知期货标的: {}".format(symbol)) else: dominant = get_dominant_future(match.groupdict()["underlying_symbol"]) cur = get_current_data() symbol_last_price = cur[symbol].last_price dominant_last_price = cur[dominant].last_price log.info(f'当前持仓合约: {symbol}, 当前主力合约: {dominant}') if dominant > symbol: for positions_ in (subportfolio.long_positions, subportfolio.short_positions): if symbol not in positions_.keys(): continue else : p = positions_[symbol] if switch_func is not None: switch_func(context, pindex, p, dominant) else: amount = p.total_amount # 跌停不能开空和平多，涨停不能开多和平空。 if p.side == "long": symbol_low_limit = cur[symbol].low_limit dominant_high_limit = cur[dominant].high_limit if symbol_last_price <= symbol_low_limit: log.warning("标的{}跌停，无法平仓。移仓换月取消。".format(symbol)) continue elif dominant_last_price >= dominant_high_limit: log.warning("标的{}涨停，无法开仓。移仓换月取消。".format(dominant)) continue else: log.info("进行移仓换月: ({0},long) -> ({1},long)".format(symbol, dominant)) order_old = order_target(symbol, 0, side='long') if order_old != None and order_old.filled > 0: order_new = order_target(dominant, amount, side='long') if order_new != None and order_new.filled > 0: switch_result.append({"before": symbol, "after": dominant, "side": "long"}) # 换月中的买卖都成功了，则增加新的记录去掉旧的记录 # 检查是否有带_long后缀的key old_key = f"{symbol}_long" if f"{symbol}_long" in g.trade_history else symbol new_key = f"{dominant}_long" if "_long" in old_key else dominant if old_key in g.trade_history: g.trade_history[new_key] = g.trade_history[old_key] del g.trade_history[old_key] else: log.warning("标的{}交易失败，无法开仓。移仓换月失败。".format(dominant)) if p.side == "short": symbol_high_limit = cur[symbol].high_limit dominant_low_limit = cur[dominant].low_limit if symbol_last_price >= symbol_high_limit: log.warning("标的{}涨停，无法平仓。移仓换月取消。".format(symbol)) continue elif dominant_last_price <= dominant_low_limit: log.warning("标的{}跌停，无法开仓。移仓换月取消。".format(dominant)) continue else: log.info("进行移仓换月: ({0},short) -> ({1},short)".format(symbol, dominant)) order_old = order_target(symbol, 0, side='short') if order_old != None and order_old.filled > 0: order_new = order_target(dominant, amount, side='short') if order_new != None and order_new.filled > 0: switch_result.append({"before": symbol, "after": dominant, "side": "short"}) # 换月中的买卖都成功了，则增加新的记录去掉旧的记录 # 检查是否有带_short后缀的key old_key = f"{symbol}_short" if f"{symbol}_short" in g.trade_history else symbol new_key = f"{dominant}_short" if "_short" in old_key else dominant if old_key in g.trade_history: g.trade_history[new_key] = g.trade_history[old_key] del g.trade_history[old_key] else: log.warning("标的{}交易失败，无法开仓。移仓换月失败。".format(dominant)) if callback: callback(context, pindex, p, dominant) return switch_result