# 导入函数库 from jqdata import * from jqdata import finance import pandas as pd import numpy as np from datetime import date, datetime, timedelta import re # 多均线穿越突破策略 v001 # 基于K线实体穿越多条均线的交易策略 # 设置以便完整打印 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')]) # 期货类每笔交易时的手续费是: 买入时万分之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.min_cross_mas = 3 # 最少穿越均线数量 g.max_margin_per_position = 20000 # 单个标的最大持仓保证金（元） g.price_deviation_min = 0.005 # 价格偏离临界线最小比例（0.5%） g.price_deviation_max = 0.01 # 价格偏离临界线最大比例（1%） # 均线交叉数量检查相关参数 g.max_ma_crosses = 4 # 最大允许的均线交叉数量 g.ma_cross_check_days = 10 # 检查均线交叉的天数 # 止损止盈策略参数 g.gap_ratio_threshold = 0.002 # 跳空比例：0.2% g.market_close_times = ["14:55:00"] # 盘尾时间 g.profit_thresholds = [5000, 15000] # 价格分区 g.stop_ratios = [0.0025, 0.005, 0.01, 0.02] # 止盈止损比例：[0.25%, 0.5%, 1%, 2%] # 期货品种完整配置字典 g.futures_config = { # 贵金属 'AU': {'has_night_session': True, 'margin_rate': {'long': 0.04, 'short': 0.04}, 'multiplier': 1000}, 'AG': {'has_night_session': True, 'margin_rate': {'long': 0.04, 'short': 0.04}, 'multiplier': 15}, # 有色金属 'CU': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, 'AL': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, 'ZN': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, 'PB': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, 'NI': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 1}, 'SN': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 1}, 'SS': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, # 黑色系 'RB': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 10}, 'HC': {'has_night_session': True, 'margin_rate': {'long': 0.04, 'short': 0.04}, '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}, 'AO': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 20}, '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}, 'BC': {'has_night_session': True, 'margin_rate': {'long': 0.13, 'short': 0.13}, 'multiplier': 5}, # 化工 '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}, 'CY': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, 'SH': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 30}, 'PX': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, # 农产品 '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}, '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}, 'PR': {'has_night_session': True, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 10}, 'AD': {'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}, 'EC': {'has_night_session': False, 'margin_rate': {'long': 0.12, 'short': 0.12}, 'multiplier': 50}, 'SF': {'has_night_session': False, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, 'SM': {'has_night_session': False, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, 'UR': {'has_night_session': False, 'margin_rate': {'long': 0.09, 'short': 0.09}, 'multiplier': 20}, 'AP': {'has_night_session': False, 'margin_rate': {'long': 0.08, 'short': 0.08}, 'multiplier': 10}, 'CJ': {'has_night_session': False, 'margin_rate': {'long': 0.07, 'short': 0.07}, '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.08, 'short': 0.08}, 'multiplier': 5}, 'LH': {'has_night_session': False, 'margin_rate': {'long': 0.1, 'short': 0.1}, 'multiplier': 16}, 'SI': {'has_night_session': False, 'margin_rate': {'long': 0.13, 'short': 0.13}, 'multiplier': 5}, 'LC': {'has_night_session': False, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 1}, 'PS': {'has_night_session': False, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5}, 'LG': {'has_night_session': False, 'margin_rate': {'long': 0.05, 'short': 0.05}, 'multiplier': 5} } # 当前策略关注的标的列表（可以根据需要调整，为空则考虑所有品种） g.strategy_focus_symbols = ['RM'] # 如果关注列表为空，则使用所有配置的品种 if not g.strategy_focus_symbols: g.strategy_focus_symbols = list(g.futures_config.keys()) log.info("策略关注品种列表为空，将考虑所有配置的品种") # 打印配置摘要 log.info(f"策略关注品种总数: {len(g.strategy_focus_symbols)}") # 交易记录和数据存储 g.trade_history = {} g.ma_cross_signals = {} # 存储多均线穿越信号 g.daily_data_cache = {} # 存储历史日线数据缓存 g.minute_data_cache = {} # 存储今日分钟数据缓存 g.ma_data_cache = {} # 存储均线数据缓存 g.ma_cross_filtered_futures = {} # 存储通过均线交叉检查的品种（每日缓存） g.gap_check_results = {} # 存储跳空检查结果（每日缓存） # 保证金比例管理（g.futures_config中的保证金比例会根据实际交易校准） g.margin_rate_history = {} # 保证金比例变化历史记录 g.today_trades = [] # 当日交易记录 # 定时任务设置 - 根据新的时间安排 # 夜盘开始 run_daily(main_trading_21_05, time='21:05:00', reference_security='IF1808.CCFX') run_daily(main_trading_regular, time='21:35:00', reference_security='IF1808.CCFX') run_daily(main_trading_regular, time='22:05:00', reference_security='IF1808.CCFX') run_daily(main_trading_regular, time='22:35:00', reference_security='IF1808.CCFX') # 日盘开始 run_daily(main_trading_09_05, time='09:05:00', reference_security='IF1808.CCFX') run_daily(main_trading_regular, time='09:35:00', reference_security='IF1808.CCFX') run_daily(main_trading_regular, time='10:05:00', reference_security='IF1808.CCFX') run_daily(main_trading_regular, time='10:35:00', reference_security='IF1808.CCFX') run_daily(main_trading_regular, time='11:05:00', reference_security='IF1808.CCFX') run_daily(main_trading_regular, time='11:25:00', reference_security='IF1808.CCFX') run_daily(main_trading_regular, time='13:35:00', reference_security='IF1808.CCFX') run_daily(main_trading_regular, time='14:05:00', reference_security='IF1808.CCFX') # 收盘前 run_daily(main_trading_before_close, time='14:35:00', reference_security='IF1808.CCFX') run_daily(main_trading_before_close, time='14:55:00', reference_security='IF1808.CCFX') # 收盘后 run_daily(after_market_close, time='15:30:00', reference_security='IF1808.CCFX') ############################ 主程序执行函数 ################################### def main_trading_21_05(context): """21:05 执行任务1, 2, 3, 4, 5, 6, 9""" log.info("-" * 50) log.info("21:05 多均线穿越突破策略 - 夜盘开始") log.info("执行任务: 1, 2, 3, 4, 5, 6, 9") log.info("-" * 50) # 任务1: 获取所有可交易品种 task_1_get_tradable_futures(context) # 任务2: 获取历史数据并保存到内存 task_2_load_historical_data(context) # 任务3: 检查均线交叉数量，过滤掉交叉过多的品种 task_3_check_ma_crosses(context) # 任务4: 获取今日分钟数据并合并均线数据 task_4_update_realtime_data(context) # 任务5: 判断上穿下穿情况 task_5_analyze_ma_crosses(context) # 任务6: 检查开仓条件 filtered_signals = task_6_check_opening_conditions(context) # 如果任务6有满足条件的，执行任务7开仓 if filtered_signals: task_7_execute_trades(context, filtered_signals) # 任务9: 检查止损止盈 task_9_check_stop_loss_profit(context) def main_trading_09_05(context): """09:05 执行任务1, 2, 3, 4, 5, 6, 9""" log.info("-" * 50) log.info("09:05 多均线穿越突破策略 - 日盘开始") log.info("执行任务: 1, 2, 3, 4, 5, 6, 9") log.info("-" * 50) # 任务1: 获取所有可交易品种 task_1_get_tradable_futures(context) # 任务2: 获取历史数据并保存到内存 task_2_load_historical_data(context) # 任务3: 检查均线交叉数量，过滤掉交叉过多的品种 task_3_check_ma_crosses(context) # 任务4: 获取今日分钟数据并合并均线数据 task_4_update_realtime_data(context) # 任务5: 判断上穿下穿情况 task_5_analyze_ma_crosses(context) # 任务6: 检查开仓条件 filtered_signals = task_6_check_opening_conditions(context) # 如果任务6有满足条件的，执行任务7开仓 if filtered_signals: task_7_execute_trades(context, filtered_signals) # 任务9: 检查止损止盈 task_9_check_stop_loss_profit(context) def main_trading_regular(context): """常规时间执行任务4, 5, 6, 9""" current_time = context.current_dt.strftime('%H:%M') # log.info(f"----- {current_time} 常规检查 -----") # log.info("执行任务: 4, 5, 6, 9") # 任务4: 获取今日分钟数据并合并均线数据 task_4_update_realtime_data(context) # 任务5: 判断上穿下穿情况 task_5_analyze_ma_crosses(context) # 任务6: 检查开仓条件 filtered_signals = task_6_check_opening_conditions(context) # 如果任务6有满足条件的，执行任务7开仓 if filtered_signals: task_7_execute_trades(context, filtered_signals) # 任务9: 检查止损止盈 task_9_check_stop_loss_profit(context) def main_trading_before_close(context): """收盘前执行任务4, 5, 6, 8, 9""" current_time = context.current_dt.strftime('%H:%M') # log.info(f"----- {current_time} 收盘前检查 -----") # log.info("执行任务: 4, 5, 6, 8, 9") # 任务4: 获取今日分钟数据并合并均线数据 task_4_update_realtime_data(context) # 任务5: 判断上穿下穿情况 task_5_analyze_ma_crosses(context) # 任务6: 检查开仓条件 filtered_signals = task_6_check_opening_conditions(context) # 如果任务6有满足条件的，执行任务7开仓 if filtered_signals: task_7_execute_trades(context, filtered_signals) # 任务8: 检查换月移仓 task_8_check_position_switch(context) # 任务9: 检查止损止盈 task_9_check_stop_loss_profit(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 = [] if current_time in ('21', '22'): # 夜盘时间：只考虑有夜盘的品种 for symbol in focus_symbols: if get_futures_config(symbol, 'has_night_session', False): potential_icon_list.append(symbol) log.info(f"夜盘时间，可交易品种: {potential_icon_list}") else: # 日盘时间：所有关注的品种都可以交易 potential_icon_list = focus_symbols[:] log.info(f"日盘时间，可交易品种: {potential_icon_list}") 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(context): """任务2: 获取所有可交易品种今天之前所需的数据，并保存到内存中""" # log.info("执行任务2: 加载历史数据到内存") if not hasattr(g, 'tradable_futures'): return for future_code in g.tradable_futures: try: # 获取50天历史数据（用于计算均线） data = attribute_history(future_code, 50, '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 # log.info(f"已缓存 {future_code} 历史数据 {len(data)} 条") except Exception as e: log.warning(f"加载{future_code}历史数据时出错: {str(e)}") continue log.info(f"历史数据缓存完成，共缓存 {len(g.daily_data_cache)} 个品种") def task_3_check_ma_crosses(context): """任务3: 检查均线交叉数量，过滤掉交叉过多的品种""" # log.info("执行任务3: 检查均线交叉数量") if not hasattr(g, 'tradable_futures'): return # 存储通过均线交叉检查的品种 today_date = context.current_dt.date() if today_date not in g.ma_cross_filtered_futures: g.ma_cross_filtered_futures[today_date] = [] filtered_futures = [] for future_code in g.tradable_futures: try: # 获取历史数据 historical_data = g.daily_data_cache.get(future_code) if historical_data is None or len(historical_data) < 30: log.warning(f"{future_code} 历史数据不足，跳过") continue # 计算移动平均线 data_with_ma = historical_data.copy() data_with_ma['MA5'] = data_with_ma['close'].rolling(window=5).mean() data_with_ma['MA10'] = data_with_ma['close'].rolling(window=10).mean() data_with_ma['MA20'] = data_with_ma['close'].rolling(window=20).mean() data_with_ma['MA30'] = data_with_ma['close'].rolling(window=30).mean() # 检查均线交叉数量 ma_crosses = count_ma_crosses(data_with_ma, g.ma_cross_check_days) # 如果交叉数量在允许范围内，保留该品种 if ma_crosses <= g.max_ma_crosses: filtered_futures.append(future_code) log.info(f"{future_code} 通过均线交叉检查，交叉数量: {ma_crosses}") else: log.info(f"{future_code} 均线交叉过多，被过滤，交叉数量: {ma_crosses} > {g.max_ma_crosses}") except Exception as e: log.warning(f"检查{future_code}均线交叉时出错: {str(e)}") continue # 更新可交易品种列表 g.tradable_futures = filtered_futures g.ma_cross_filtered_futures[today_date] = filtered_futures log.info(f"均线交叉检查完成，剩余可交易品种数量: {len(g.tradable_futures)}") if len(g.tradable_futures) > 0: log.info(f"通过检查的品种: {g.tradable_futures}") return filtered_futures def task_4_update_realtime_data(context): """任务4: 获取所有可交易品种和持仓品种今天所需的分钟数据作为今天数据，和2中的数据合并出最新的均线数据，并保存到内存中""" log.info("执行任务4: 更新实时数据和均线") # 收集需要更新数据的品种 update_symbols = set() # 添加可交易品种 if hasattr(g, 'tradable_futures') and g.tradable_futures: update_symbols.update(g.tradable_futures) # 添加持仓品种（用于止损止盈） if hasattr(g, 'trade_history') and g.trade_history: update_symbols.update(g.trade_history.keys()) if not update_symbols: log.info("没有需要更新的品种") return today_date = context.current_dt.date() for future_code in update_symbols: # log.info(f"任务4 future_code: {future_code}") try: # 获取今日分钟数据 minute_data = get_today_minute_data(context, future_code) # log.info(f"minute_data: {minute_data}") 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, 50, '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 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 # 检查跳空（只在第一次获取今日数据时检查） if future_code not in g.gap_check_results: gap_result = check_gap_opening(historical_data, minute_data) g.gap_check_results[future_code] = gap_result if gap_result['has_gap']: log.info(f"{future_code} 检测到跳空开盘，跳空比例: {gap_result['gap_ratio']:.3%}") # 合并数据并计算均线 combined_data = combine_and_calculate_ma(historical_data, minute_data) if combined_data is not None: g.ma_data_cache[future_code] = combined_data # log.info(f"已更新 {future_code} 均线数据") except Exception as e: log.warning(f"更新{future_code}实时数据时出错: {str(e)}") continue log.info(f"实时数据更新完成，共更新 {len(g.ma_data_cache)} 个品种（可交易: {len(g.tradable_futures) if hasattr(g, 'tradable_futures') else 0}，持仓: {len(g.trade_history) if hasattr(g, 'trade_history') else 0}）") def task_5_analyze_ma_crosses(context): """任务5: 根据交易品种的今天数据和均线数据，判断是否出现了上穿或者下穿的情况""" # log.info("执行任务5: 分析均线穿越") ma_cross_signals = [] # 获取已持仓的品种列表 existing_positions = set(g.trade_history.keys()) for future_code, data in g.ma_data_cache.items(): log.info(f"future_code: {future_code}") try: # 检查是否已有相似持仓，如果有则跳过分析 if check_symbol_prefix_match(future_code, existing_positions): # log.(f"{future_code} 已有相似持仓，跳过均线穿越分析") continue # 检查最新的多均线穿越 latest_cross = check_latest_multi_ma_cross(data, future_code) if latest_cross: ma_cross_signals.append(latest_cross) log.info(f"{future_code} 发现多均线穿越: {latest_cross['direction']}") log.info(f"开盘价格: {latest_cross['open']}, 当前价格: {latest_cross['close']}, 临界线: {latest_cross['critical_ma_name']}({latest_cross['critical_ma_value']:.2f}), 穿越数量: {latest_cross['crossed_count']}") log.info(f"MA5: {latest_cross['ma5']}, MA10: {latest_cross['ma10']}, MA20: {latest_cross['ma20']}, MA30: {latest_cross['ma30']}") except Exception as e: log.warning(f"分析{future_code}均线穿越时出错: {str(e)}") continue # 存储到全局变量 g.ma_cross_signals = ma_cross_signals if len(ma_cross_signals) > 0: log.info(f"均线穿越分析完成，发现信号 {len(ma_cross_signals)} 个") log.debug(f"ma_cross_signals: {ma_cross_signals}") return ma_cross_signals def task_6_check_opening_conditions(context): """任务6: 针对那些有上穿和下穿的情况检查是否满足开仓条件""" # log.info("执行任务6: 检查开仓条件") if not hasattr(g, 'ma_cross_signals'): return [] filtered_signals = [] for signal in g.ma_cross_signals: # 检查是否已有相似持仓 if check_symbol_prefix_match(signal['symbol'], set(g.trade_history.keys())): log.warning(f"{signal['symbol']} 已有相似持仓，跳过") continue # 检查价格位置合理性 if not check_price_position(signal): log.warning(f"{signal['symbol']} 价格位置不合理，跳过") continue filtered_signals.append(signal) log.info(f"{signal['symbol']} 满足开仓条件") if len(filtered_signals) > 0: log.info(f"开仓条件检查完成，满足条件 {len(filtered_signals)} 个") return filtered_signals def task_7_execute_trades(context, filtered_signals): """任务7: 针对满足开仓条件的标的，计算购买的金额，并发起交易请求""" # log.info("执行任务7: 执行交易") for signal in filtered_signals: symbol = signal['symbol'] direction = 'long' if signal['direction'] == 'up' else 'short' # 检查资金充足性 if not check_sufficient_capital(context, symbol): log.warning(f"{symbol} 资金不足，跳过开仓") continue # 计算开仓金额 order_value = calculate_order_value(context, symbol, direction) log.info(f"计算开仓金额: {symbol} {direction} 金额: {order_value}") # 如果计算出的金额为0，说明资金不足，跳过开仓 if order_value <= 0: log.warning(f"资金不足，跳过开仓 {symbol} {direction}") continue # 执行开仓 success = open_position(context, symbol, order_value, direction, signal) if success: # 获取实际保证金（只有在开仓成功时才能获取） actual_margin = g.trade_history[symbol]['actual_margin'] # 获取成交价格 actual_price = g.trade_history[symbol]['entry_price'] log.info(f"成功开仓 {symbol} {direction}, 成交价格: {actual_price:.2f}, 金额: {order_value}, 实际保证金: {actual_margin:.0f}") else: log.warning(f"开仓失败 {symbol} {direction}") 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']}") # else: # log.info("无需移仓换月") def task_9_check_stop_loss_profit(context): """任务9: 针对已经持仓的标的，检查是否止损或止盈""" # log.info("执行任务9: 检查止损止盈") # 遍历所有持仓进行止损止盈检查 subportfolio = context.subportfolios[0] long_positions = list(subportfolio.long_positions.values()) short_positions = list(subportfolio.short_positions.values()) closed_count = 0 for position in long_positions + short_positions: if check_stop_loss_profit(context, position): closed_count += 1 if closed_count > 0: log.info(f"执行了 {closed_count} 次止损止盈") # else: # log.info("无需止损止盈") ############################ 数据处理辅助函数 ################################### 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 log.debug(f"原始分钟数据范围: {minute_data.index[0]} 到 {minute_data.index[-1]}") # 提取所有日期（年月日维度） minute_data['date'] = minute_data.index.date unique_dates = sorted(minute_data['date'].unique()) log.debug(f"数据包含的日期: {unique_dates}") if has_night_session: # 有夜盘的品种：需要找到前一交易日的21:00作为今日开盘起点 today_date = end_time.date() # 找到今天之前的最后一个交易日 previous_trading_dates = [d for d in unique_dates if d < today_date] log.debug(f"夜盘标的 today_date: {today_date}, previous_trading_dates: {previous_trading_dates}") if not previous_trading_dates: log.warning(f"找不到{future_code}的前一交易日数据") return minute_data previous_trading_date = max(previous_trading_dates) log.debug(f"前一交易日: {previous_trading_date}") # 找到前一交易日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] log.debug(f"夜盘品种，从{start_time}开始，数据量: {len(filtered_data)}") return filtered_data.drop(columns=['date']) else: log.warning(f"找不到{future_code}前一交易日21:00的数据") # 备选方案：使用今天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: log.debug(f"日盘品种，从今天9:00开始，数据量: {len(day_9_data)}") return day_9_data.drop(columns=['date']) else: log.warning(f"找不到{future_code}今天9:00的数据") 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 combine_and_calculate_ma(historical_data, minute_data): """合并历史数据和分钟数据，计算均线""" try: # 将分钟数据聚合为日数据 today_data = aggregate_minute_to_daily(minute_data) if today_data is None: return None # 合并历史数据和今日数据 combined_data = pd.concat([historical_data, today_data]) # 计算移动平均线 combined_data['MA5'] = combined_data['close'].rolling(window=5).mean() combined_data['MA10'] = combined_data['close'].rolling(window=10).mean() combined_data['MA20'] = combined_data['close'].rolling(window=20).mean() combined_data['MA30'] = combined_data['close'].rolling(window=30).mean() return combined_data except Exception as e: log.warning(f"合并数据和计算均线时出错: {str(e)}") return None def aggregate_minute_to_daily(minute_data): """将分钟数据聚合为日数据 注意：传入的minute_data已经是经过过滤的今日交易数据 - 对于有夜盘的品种：数据从前一交易日21:00开始 - 对于无夜盘的品种：数据从当日9:00开始 开盘价(open)是第一个分钟K线的开盘价（今日交易开盘价） 收盘价(close)是当前最新的分钟K线收盘价，会随时间更新 """ try: if minute_data is None or len(minute_data) == 0: return None # 获取今日日期（使用最后一条数据的日期作为今日日期） today_date = minute_data.index[-1].date() # log.debug(f"聚合数据范围: {minute_data.index[0]} 到 {minute_data.index[-1]}") # log.debug(f"开盘价数据: {minute_data['open'].iloc[0]:.2f}") # 聚合为日数据 # 开盘价：今日交易开始时的第一个分钟K线的开盘价（固定不变） # 收盘价：当前最新分钟K线的收盘价（实时更新） # 最高价：所有分钟K线中的最高价 # 最低价：所有分钟K线中的最低价 # 成交量：所有分钟K线成交量的总和 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)]) # log.debug(f"聚合后的日数据: 开盘价={daily_data['open'].iloc[0]:.2f}, 收盘价={daily_data['close'].iloc[0]:.2f}") return daily_data except Exception as e: log.warning(f"聚合分钟数据时出错: {str(e)}") return None def check_gap_opening(historical_data, minute_data): """ 检查开盘是否跳空 :param historical_data: 历史日线数据 :param minute_data: 今日分钟数据 :return: 跳空检查结果字典 """ try: if historical_data is None or len(historical_data) == 0 or minute_data is None or len(minute_data) == 0: return {'has_gap': False, 'gap_ratio': 0.0} # 获取前一交易日收盘价 previous_close = historical_data['close'].iloc[-1] # 获取今日开盘价 today_open = minute_data['open'].iloc[0] # 计算跳空比例 gap_ratio = abs(today_open - previous_close) / previous_close # 判断是否跳空 has_gap = gap_ratio >= g.gap_ratio_threshold return { 'has_gap': has_gap, 'gap_ratio': gap_ratio, 'previous_close': previous_close, 'today_open': today_open } except Exception as e: log.warning(f"检查跳空开盘时出错: {str(e)}") return {'has_gap': False, 'gap_ratio': 0.0} ############################ 原有函数保持不变 ################################### def check_latest_multi_ma_cross(data, future_code): """检查最新的多均线穿越情况""" if len(data) < 2: return None # 获取最新两天的数据 today = data.iloc[-1] yesterday = data.iloc[-2] log.debug(f"today: {today}") log.debug(f"yesterday: {yesterday}") # 检查多均线穿越 cross_result = check_multi_ma_cross_single_day(today) if not cross_result: return None # TODO：验证穿越的有效性，暂时感觉没用先注释了 # 验证穿越的有效性 # if not validate_ma_cross(today, yesterday): # return None return { 'symbol': future_code, 'date': today.name, 'direction': cross_result['direction'], 'open': today['open'], 'close': today['close'], 'high': today['high'], 'low': today['low'], 'ma5': today['MA5'], 'ma10': today['MA10'], 'ma20': today['MA20'], 'ma30': today['MA30'], 'critical_ma_name': cross_result['critical_ma_name'], 'critical_ma_value': cross_result['critical_ma_value'], 'crossed_count': cross_result['crossed_count'] } def check_multi_ma_cross_single_day(row): """检查单日K线是否穿越了至少3条均线，并返回临界线信息""" open_price = row['open'] close_price = row['close'] ma_values = [('MA5', row['MA5']), ('MA10', row['MA10']), ('MA20', row['MA20']), ('MA30', row['MA30'])] log.debug(f"open_price: {open_price}, close_price: {close_price}, ma_values: {ma_values}") # 检查是否有NaN值 if pd.isna(row['MA5']) or pd.isna(row['MA10']) or pd.isna(row['MA20']) or pd.isna(row['MA30']): return None # 如果开盘价和收盘价相等，不可能有穿越 if open_price == close_price: return None # 1. 统计开盘价和均线的高低关系 open_above_count = 0 # 开盘价高于均线的数量 open_below_count = 0 # 开盘价低于均线的数量 open_above_mas = [] # 开盘价高于的均线列表 open_below_mas = [] # 开盘价低于的均线列表 for ma_name, ma_value in ma_values: if open_price > ma_value: open_above_count += 1 open_above_mas.append((ma_name, ma_value)) elif open_price < ma_value: open_below_count += 1 open_below_mas.append((ma_name, ma_value)) # 2. 统计收盘价和均线的高低关系 close_above_count = 0 # 收盘价高于均线的数量 close_below_count = 0 # 收盘价低于均线的数量 close_above_mas = [] # 收盘价高于的均线列表 close_below_mas = [] # 收盘价低于的均线列表 for ma_name, ma_value in ma_values: if close_price > ma_value: close_above_count += 1 close_above_mas.append((ma_name, ma_value)) elif close_price < ma_value: close_below_count += 1 close_below_mas.append((ma_name, ma_value)) # 3. 计算穿越情况 # 上穿：收盘价高于的数量比开盘价高于的数量增加了 upward_cross_count = close_above_count - open_above_count # 下穿：收盘价低于的数量比开盘价低于的数量增加了 downward_cross_count = close_below_count - open_below_count log.debug(f"开盘价高于均线数量: {open_above_count}, 收盘价高于均线数量: {close_above_count}") log.debug(f"开盘价低于均线数量: {open_below_count}, 收盘价低于均线数量: {close_below_count}") log.debug(f"上穿数量: {upward_cross_count}, 下穿数量: {downward_cross_count}") # 4. 判断是否满足最少穿越条件并找到临界线 if upward_cross_count >= g.min_cross_mas: # 上穿：找到被穿越的均线中值最大的一条作为临界线 # 被穿越的均线是那些开盘价低于但收盘价高于的均线 crossed_mas = [] crossed_ma_names = [] for ma_name, ma_value in ma_values: if open_price <= ma_value and close_price > ma_value: crossed_mas.append((ma_name, ma_value)) crossed_ma_names.append(ma_name) if len(crossed_mas) > 0: # 找到值最大的被穿越均线 critical_ma = max(crossed_mas, key=lambda x: x[1]) return { 'direction': 'up', 'critical_ma_name': critical_ma[0], 'critical_ma_value': critical_ma[1], 'crossed_count': upward_cross_count, 'crossed_ma_names': crossed_ma_names # 添加被穿越的均线名称列表 } elif downward_cross_count >= g.min_cross_mas: # 下穿：找到被穿越的均线中值最小的一条作为临界线 # 被穿越的均线是那些开盘价高于但收盘价低于的均线 crossed_mas = [] crossed_ma_names = [] for ma_name, ma_value in ma_values: if open_price >= ma_value and close_price < ma_value: crossed_mas.append((ma_name, ma_value)) crossed_ma_names.append(ma_name) if len(crossed_mas) > 0: # 找到值最小的被穿越均线 critical_ma = min(crossed_mas, key=lambda x: x[1]) return { 'direction': 'down', 'critical_ma_name': critical_ma[0], 'critical_ma_value': critical_ma[1], 'crossed_count': downward_cross_count, 'crossed_ma_names': crossed_ma_names # 添加被穿越的均线名称列表 } return None def validate_ma_cross(today, yesterday): """验证均线穿越的有效性""" # 检查均线排列是否合理 ma_today = [today['MA5'], today['MA10'], today['MA20'], today['MA30']] ma_yesterday = [yesterday['MA5'], yesterday['MA10'], yesterday['MA20'], yesterday['MA30']] # 简单的均线排列检查 # 可以根据需要添加更复杂的验证逻辑 return True def check_sufficient_capital(context, symbol): """检查资金是否充足""" try: # 计算单手保证金 single_hand_margin = calculate_required_margin(context, symbol) # 使用实际可用资金（考虑资金使用比例） available_cash = context.portfolio.available_cash * g.usage_percentage # 检查是否有足够资金开仓至少1手 return available_cash >= single_hand_margin except: return False def check_price_position(signal): """检查价格位置的合理性""" close = signal['close'] critical_ma_value = signal['critical_ma_value'] # 检查收盘价与临界线的偏离度 deviation = abs(close - critical_ma_value) / critical_ma_value # 偏离度需要在合理范围内：大于最小偏离度且小于最大偏离度 if deviation < g.price_deviation_min: log.warning(f"价格偏离度过小: {deviation:.3%} < {g.price_deviation_min:.1%}") return False elif deviation > g.price_deviation_max: log.warning(f"价格偏离度过大: {deviation:.3%} > {g.price_deviation_max:.1%}") return False else: log.info(f"价格偏离度合理: {g.price_deviation_min:.1%} <= {deviation:.3%} <= {g.price_deviation_max:.1%}") return True ############################ 交易执行函数 ################################### def open_position(context, security, value, direction, signal): """开仓""" try: # 记录交易前的可用资金 cash_before = context.portfolio.available_cash order = order_target_value(security, value, side=direction) log.debug(f"order: {order}") 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] multiplier = get_multiplier(underlying_symbol) # 单笔保证金 = 资金变化 / 数量 single_margin = cash_change / order_amount if order_amount > 0 else 0 # 实际保证金比例 = 单笔保证金 / (订单价格 * 合约乘数) contract_value = order_price * multiplier actual_margin_rate = single_margin / contract_value if contract_value > 0 else 0 # 校准保证金比例（只有变化大于1%时才更新） current_rate = get_margin_rate(underlying_symbol, direction) rate_change = abs(actual_margin_rate - current_rate) / current_rate if current_rate > 0 else 0 if rate_change > 0.01: # 变化大于1% # 记录保证金比例变化历史 history_key = f"{underlying_symbol}_{direction}" if history_key not in g.margin_rate_history: g.margin_rate_history[history_key] = [] g.margin_rate_history[history_key].append({ 'date': context.current_dt.date(), 'time': context.current_dt.time(), 'old_rate': current_rate, 'new_rate': actual_margin_rate, 'change_pct': rate_change * 100, 'security': security }) # 直接更新配置字典中的保证金比例 if underlying_symbol in g.futures_config: g.futures_config[underlying_symbol]['margin_rate'][direction] = actual_margin_rate log.debug(f"保证金比例校准: {underlying_symbol}_{direction} {current_rate:.4f} -> {actual_margin_rate:.4f} (变化{rate_change*100:.1f}%)") # 记录当日交易 g.today_trades.append({ 'security': security, # 交易标的 'underlying_symbol': underlying_symbol, # 标的字母 'direction': direction, # 方向 'order_amount': order_amount, # 开仓数量 'order_price': order_price, # 开仓金额 'cash_change': cash_change, # 现金变化 'actual_margin_rate': actual_margin_rate, # 实际保证金率 'time': context.current_dt # 成交日期 }) # 记录交易信息 g.trade_history[security] = { 'entry_price': order_price, # 成交价格 'position_value': value, # 开仓金额 'actual_margin': cash_change, # 实际保证金 'direction': direction, # 方向 'entry_time': context.current_dt, # 开仓时间 'signal_info': signal # 信号信息 } log.debug(f"开仓成功 - 品种: {underlying_symbol}, 手数: {order_amount}, 订单价格: {order_price:.2f}") log.debug(f"资金变化: {cash_change:.0f}, 实际保证金比例: {actual_margin_rate:.4f}") return True except Exception as e: log.warning(f"开仓失败 {security}: {str(e)}") return False def close_position(context, security, direction): """平仓""" try: order = order_target_value(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, # 平仓不计算保证金变化 'actual_margin_rate': 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 if security not in g.trade_history: return False trade_info = g.trade_history[security] # 跟踪均线止损止盈 tracking_stop_result = check_tracking_ma_stop(context, security, position, trade_info) if tracking_stop_result: return True return False def check_tracking_ma_stop(context, security, position, trade_info): """ 检查跟踪均线止损止盈 :param context: 上下文对象 :param security: 标的代码 :param position: 持仓对象 :param trade_info: 交易信息 :return: 是否触发止损止盈 """ try: direction = trade_info['direction'] entry_price = trade_info['entry_price'] current_price = position.price # 获取最新的均线数据 if security not in g.ma_data_cache: return False ma_data = g.ma_data_cache[security] if len(ma_data) == 0: return False latest_data = ma_data.iloc[-1] # 获取四条均线价格和今日最高最低价 ma5 = latest_data['MA5'] ma10 = latest_data['MA10'] ma20 = latest_data['MA20'] ma30 = latest_data['MA30'] today_high = latest_data['high'] today_low = latest_data['low'] # 检查是否有NaN值 if pd.isna(ma5) or pd.isna(ma10) or pd.isna(ma20) or pd.isna(ma30): return False # 获取开仓时被穿越的均线信息 signal_info = trade_info.get('signal_info', {}) crossed_ma_names = signal_info.get('crossed_ma_names', ['MA5', 'MA10', 'MA20', 'MA30']) # 根据方向确定止损均线，需要过滤掉不符合条件的均线 mas = [ma5, ma10, ma20, ma30] ma_names = ['MA5', 'MA10', 'MA20', 'MA30'] # 第一步：只考虑被穿越的均线 crossed_mas = [] crossed_ma_prices = [] for i, ma_name in enumerate(ma_names): if ma_name in crossed_ma_names: crossed_mas.append(ma_name) crossed_ma_prices.append(mas[i]) log.debug(f"开仓时被穿越的均线: {crossed_ma_names}") if direction == 'long': # 多仓：在被穿越的均线中，选择价格低于今日最高价的均线 valid_mas = [] valid_ma_names = [] for i, ma_name in enumerate(crossed_mas): ma_price = crossed_ma_prices[i] if ma_price <= today_high: valid_mas.append(ma_price) valid_ma_names.append(ma_name) if not valid_mas: # 如果被穿越的均线都高于今日最高价，使用被穿越均线中的最低价 if crossed_ma_prices: stop_ma_price = min(crossed_ma_prices) stop_ma_name = crossed_mas[crossed_ma_prices.index(stop_ma_price)] log.warning(f"多仓被穿越均线都高于今日最高价{today_high:.2f}，使用被穿越均线中最低的") else: # 如果没有被穿越均线信息，使用所有均线中的最低价 stop_ma_price = min(mas) stop_ma_name = ma_names[mas.index(stop_ma_price)] log.warning(f"多仓无被穿越均线信息，使用所有均线中最低的") else: # 多仓跟踪符合条件的被穿越均线中价格最高的 stop_ma_price = max(valid_mas) stop_ma_name = valid_ma_names[valid_mas.index(stop_ma_price)] else: # 空仓：在被穿越的均线中，选择价格高于今日最低价的均线 valid_mas = [] valid_ma_names = [] for i, ma_name in enumerate(crossed_mas): ma_price = crossed_ma_prices[i] if ma_price >= today_low: valid_mas.append(ma_price) valid_ma_names.append(ma_name) if not valid_mas: # 如果被穿越的均线都低于今日最低价，使用被穿越均线中的最高价 if crossed_ma_prices: stop_ma_price = max(crossed_ma_prices) stop_ma_name = crossed_mas[crossed_ma_prices.index(stop_ma_price)] log.warning(f"空仓被穿越均线都低于今日最低价{today_low:.2f}，使用被穿越均线中最高的") else: # 如果没有被穿越均线信息，使用所有均线中的最高价 stop_ma_price = max(mas) stop_ma_name = ma_names[mas.index(stop_ma_price)] log.warning(f"空仓无被穿越均线信息，使用所有均线中最高的") else: # 空仓跟踪符合条件的被穿越均线中价格最低的 stop_ma_price = min(valid_mas) stop_ma_name = valid_ma_names[valid_mas.index(stop_ma_price)] log.debug(f"今日高低价: {today_high:.2f}/{today_low:.2f}, 被穿越均线: {crossed_ma_names}, 选择止损均线: {stop_ma_name}({stop_ma_price:.2f})") log.debug(f"所有均线: MA5={ma5:.2f}, MA10={ma10:.2f}, MA20={ma20:.2f}, MA30={ma30:.2f}") # 计算止损均线的收益水平 underlying_symbol = security.split('.')[0][:-4] multiplier = get_multiplier(underlying_symbol) position_value = abs(position.total_amount) * stop_ma_price * multiplier if direction == 'long': ma_profit = (stop_ma_price - entry_price) * multiplier * abs(position.total_amount) else: ma_profit = (entry_price - stop_ma_price) * multiplier * abs(position.total_amount) # 获取跳空信息 gap_info = g.gap_check_results.get(security, {'has_gap': False}) has_gap = gap_info['has_gap'] # 判断是否盘尾 current_time = context.current_dt.strftime('%H:%M:%S') is_market_close = current_time in g.market_close_times # 确定止损比例 stop_ratio = get_tracking_stop_ratio(ma_profit, has_gap, is_market_close) # 计算止损价格 if direction == 'long': stop_price = stop_ma_price * (1 - stop_ratio) should_stop = current_price <= stop_price else: stop_price = stop_ma_price * (1 + stop_ratio) should_stop = current_price >= stop_price if should_stop: log.info(f"触发跟踪均线止损 {security} {direction}") log.info(f"止损均线价格: {stop_ma_price:.2f}, 方向: {direction}, 收益: {ma_profit:.0f}, 跳空: {has_gap}, 盘尾: {is_market_close}") log.info(f"止损比例: {stop_ratio:.4f}, 止损价格: {stop_price:.2f}, 当前价格: {current_price:.2f}") close_position(context, security, direction) return True return False except Exception as e: log.warning(f"检查跟踪均线止损时出错 {security}: {str(e)}") return False def get_tracking_stop_ratio(ma_profit, has_gap, is_market_close): """ 根据均线收益、跳空情况、盘中盘尾确定止损比例 :param ma_profit: 止损均线的收益水平 :param has_gap: 是否跳空 :param is_market_close: 是否盘尾 :return: 止损比例 """ # 根据收益水平分区 if ma_profit < g.profit_thresholds[0]: # 收益 < 5000 if is_market_close: # 盘尾 if has_gap: return g.stop_ratios[0] # 0.25% else: return g.stop_ratios[1] # 0.5% else: # 盘中 if has_gap: return g.stop_ratios[1] # 0.5% else: return g.stop_ratios[2] # 1% elif ma_profit < g.profit_thresholds[1]: # 5000 <= 收益 < 15000 if is_market_close: # 盘尾 return g.stop_ratios[1] # 0.5% else: # 盘中 return g.stop_ratios[2] # 1% else: # 收益 >= 15000 if is_market_close: # 盘尾 return g.stop_ratios[2] # 1% else: # 盘中 return g.stop_ratios[3] # 2% ############################ 辅助函数 ################################### def get_futures_config(underlying_symbol, config_key=None, default_value=None): """ 获取期货品种配置信息的辅助函数 :param underlying_symbol: 品种符号，如 'AU', 'PF' 等 :param config_key: 配置键，如 'multiplier', 'has_night_session' 等 :param default_value: 默认值 :return: 配置值或整个配置字典 """ 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): """ 获取保证金比例的辅助函数 :param underlying_symbol: 品种符号 :param direction: 方向 'long' 或 'short' :param default_rate: 默认保证金比例 :return: 保证金比例 """ return g.futures_config.get(underlying_symbol, {}).get('margin_rate', {}).get(direction, default_rate) def get_multiplier(underlying_symbol, default_multiplier=10): """ 获取合约乘数的辅助函数 :param underlying_symbol: 品种符号 :param default_multiplier: 默认合约乘数 :return: 合约乘数 """ return g.futures_config.get(underlying_symbol, {}).get('multiplier', default_multiplier) def calculate_order_value(context, security, direction): """计算开仓金额""" 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 log.debug(f"可用资金: {available_cash:.0f}") # 根据单个标的最大持仓保证金限制计算开仓数量 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) # 实际保证金金额 actual_margin = current_price * multiplier * margin_rate * actual_hands # 计算订单金额 order_value = actual_margin log.debug(f"单手保证金: {single_hand_margin:.0f}, 最大手数(保证金限制): {max_hands}, 最大手数(资金限制): {max_hands_by_cash}, 实际开仓手数: {actual_hands}, 实际保证金: {actual_margin:.0f}") else: # 如果单手保证金超过最大限制，默认开仓1手 actual_hands = 1 actual_margin = current_price * multiplier * margin_rate * actual_hands # 计算订单金额 order_value = actual_margin log.debug(f"单手保证金: {single_hand_margin:.0f} 超过最大限制: {max_margin}, 默认开仓1手, 实际保证金: {actual_margin:.0f}") log.debug(f"计算结果 - 品种: {underlying_symbol}, 开仓手数: {actual_hands}, 订单金额: {order_value:.0f}, 实际保证金: {actual_margin:.0f}") return order_value def calculate_required_margin(context, symbol): """计算所需保证金""" 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) return current_price * multiplier * margin_rate 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 after_market_close(context): """收盘后运行函数""" log.info(str('函数运行时间(after_market_close):'+str(context.current_dt.time()))) # 只有当天有交易时才打印统计信息 if g.today_trades: print_daily_trading_summary(context) print_margin_rate_changes(context) # 清空当日交易记录 g.today_trades = [] log.info('##############################################################') def print_daily_trading_summary(context): """打印当日交易汇总""" if not g.today_trades: return log.debug("\n=== 当日交易汇总 ===") total_margin = 0 for trade in g.today_trades: if trade['order_amount'] > 0: # 开仓 log.debug(f"开仓 {trade['underlying_symbol']} {trade['direction']} {trade['order_amount']}手 " f"价格:{trade['order_price']:.2f} 保证金:{trade['cash_change']:.0f} " f"比例:{trade['actual_margin_rate']:.4f}") total_margin += trade['cash_change'] else: # 平仓 log.debug(f"平仓 {trade['underlying_symbol']} {trade['direction']} {abs(trade['order_amount'])}手 " f"价格:{trade['order_price']:.2f}") log.debug(f"当日保证金占用: {total_margin:.0f}") log.debug("==================\n") def print_margin_rate_changes(context): """打印保证金比例变化记录""" if not g.margin_rate_history: return log.debug("\n=== 保证金比例变化记录 ===") for key, history in g.margin_rate_history.items(): log.debug(f"{key}:") for record in history: log.debug(f" {record['date']} {record['time']}: {record['old_rate']:.4f} -> {record['new_rate']:.4f} " f"(变化{record['change_pct']:.1f}%)") log.debug("========================\n") ########################## 自动移仓换月函数 ################################# def position_auto_switch(context,pindex=0,switch_func=None, callback=None): """ 期货自动移仓换月。默认使用市价单进行开平仓。 :param context: 上下文对象 :param pindex: 子仓对象 :param switch_func: 用户自定义的移仓换月函数. 函数原型必须满足: func(context, pindex, previous_dominant_future_position, current_dominant_future_symbol) :param callback: 移仓换月完成后的回调函数。 函数原型必须满足: func(context, pindex, previous_dominant_future_position, current_dominant_future_symbol) :return: 发生移仓换月的标的。类型为列表。 """ 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)) raise ValueError("Unknow target: {}".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'current_hold_symbol: {symbol}, current_main_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)) log.warning("Can't close {} position due to the limit up. Cancelling the exchange.".format(symbol)) continue elif dominant_last_price >= dominant_high_limit: # log.warning("标的{}涨停，无法开仓。移仓换月取消。".format(symbol)) log.warning("Can't close {} position due to the limit down. Cancelling the exchange.".format(symbol)) continue else: # log.info("进行移仓换月: ({0},long) -> ({1},long)".format(symbol, dominant)) log.info("Start the exchange: ({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"}) # 换月中的买卖都成功了，则增加新的记录去掉旧的记录 g.trade_history[dominant] = g.trade_history[symbol] del g.trade_history[symbol] else: # log.warning("标的{}交易失败，无法开仓。移仓换月取消。".format(domaint)) log.warning("Trade of {} failed, no new positions. Cancelling the exchange.".format(dominant)) # 换月中的买成功了，卖失败了，则在换月记录里增加新的记录，在交易记录里去掉旧的记录 log.warning(f'换月多头失败{dominant}, {g.trade_history[symbol]}') g.change_fail_history[domaint] = g.trade_history[symbol] del g.trade_history[symbol] if callback: callback(context, pindex, p, 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)) log.warning("Can't close {} position due to the limit up. Cancelling the exchange.".format(symbol)) continue elif dominant_last_price <= dominant_low_limit: # log.warning("标的{}跌停，无法开仓。移仓换月取消。".format(symbol)) log.warning("Can't close {} position due to the limit down. Cancelling the exchange.".format(symbol)) continue else: # log.info("进行移仓换月: ({0},short) -> ({1},short)".format(symbol, dominant)) log.info("Start the exchange: ({0},short) -> ({1},short)".format(symbol, dominant)) order_old = order_target(symbol,0,side='short') if order_old != None and order_old.filled > 0: log.info(f'换月做空{dominant},数量位{amount}') 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"}) # 换月中的买卖都成功了，则增加新的记录去掉旧的记录 g.trade_history[dominant] = g.trade_history[symbol] del g.trade_history[symbol] else: # log.warning("标的{}交易失败，无法开仓。移仓换月取消。".format(dominant)) log.warning("Trade of {} failed, no new positions. Cancelling the exchange.".format(dominant)) # 换月中的买成功了，卖失败了，则在换月记录里增加新的记录，在交易记录里去掉旧的记录 log.warning(f'换月空头失败{dominant}, {g.trade_history[symbol]}') g.change_fail_history[dominant] = g.trade_history[symbol] log.warning(f'失败记录{g.change_fail_history}') del g.trade_history[symbol] # order_target(symbol,0,side='short') # order_target(dominant,amount,side='short') # switch_result.append({"before": symbol, "after": dominant, "side": "short"}) if callback: callback(context, pindex, p, dominant) return switch_result def count_ma_crosses(data, days): """ 判断过去指定天数内4条MA均线的交叉次数 :param data: 包含MA5, MA10, MA20, MA30的DataFrame :param days: 检查的天数 :return: 总交叉次数 """ if len(data) < days + 1: return 0 # 获取最近days天的数据 recent_data = data.iloc[-days-1:] ma5 = recent_data['MA5'].values ma10 = recent_data['MA10'].values ma20 = recent_data['MA20'].values ma30 = recent_data['MA30'].values # 检查是否有NaN值 if np.any(np.isnan([ma5, ma10, ma20, ma30])): return 0 # 计算各均线间的交叉次数 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]))]) total_crosses = cross_5_10 + cross_5_20 + cross_5_30 + cross_10_20 + cross_10_30 + cross_20_30 log.debug(f'总交叉数: {total_crosses}, MA5-MA10: {cross_5_10}, MA5-MA20: {cross_5_20}, MA5-MA30: {cross_5_30}, MA10-MA20: {cross_10_20}, MA10-MA30: {cross_10_30}, MA20-MA30: {cross_20_30}') return total_crosses