MAPatternStrategy_v002.py

 - 新增均线穿越得分计算功能，设置均线穿越数量阈值
 - 历史趋势一致性的时候数据切片有问题
 - 趋势条件有问题，MA20和MA30也可以互换

.cursor/rules/intro.mdc

@@ -0,0 +1,14 @@
+---
+alwaysApply: true
+---
+
+# 概要
+这是一个策略仓库，不负责实际运行，所有代码都是在聚宽等线上平台运行的。
+
+# 目录简介
+ - 'API'是官方API的介绍文档
+ - 'Lib'则是所有策略存放的目录，其他策略的实现逻辑可用于参考
+
+# 重要信息
+ - 一般情况下是一个文件一整套逻辑，而不是将多个文件串联起来的传统编程套路
+ - 所以一般是专注于具体一个文件的内部逻辑，偶尔可以参考相同类型目录下的其他代码的实现逻辑

Lib/future/MAPatternStrategy_v002.py

@@ -60,6 +60,7 @@ def initialize(context):
     g.ma_distribution_lookback_days = 5  # MA5分布过滤回溯天数
     g.ma_distribution_min_ratio = 0.4  # MA5分布满足比例阈值
     g.enable_ma_distribution_filter = True  # 是否启用MA5分布过滤
+    g.ma_cross_threshold = 1  # 均线穿越数量阈值
     
     # 均线价差策略方案选择
     g.ma_gap_strategy_mode = 3  # 策略模式选择（1: 原方案, 2: 新方案, 3: 方案3）
@@ -87,6 +88,7 @@ def initialize(context):
     log.info(f"  MA5分布最低比例: {g.ma_distribution_min_ratio:.0%}")
     log.info(f"  启用MA5分布过滤: {g.enable_ma_distribution_filter}")
     log.info(f"  是否检查日内价差: {g.check_intraday_spread}")
+    log.info(f"  均线穿越阈值: {g.ma_cross_threshold}")
     log.info(f"  固定止损: {g.fixed_stop_loss_rate:.1%}")
     log.info(f"  均线跟踪止盈常规偏移: {g.ma_offset_ratio_normal:.1%}")
     log.info(f"  均线跟踪止盈收盘前偏移: {g.ma_offset_ratio_close:.1%}")
@@ -136,6 +138,7 @@ def initialize(context):
         'PP': {'has_night_session': True, 'margin_rate': {'long': 0.12, 'short': 0.12}, 'multiplier': 5, 'trading_start_time': '21:00'},
         'EB': {'has_night_session': True, 'margin_rate': {'long': 0.12, 'short': 0.12}, 'multiplier': 5, 'trading_start_time': '21:00'},
         'PG': {'has_night_session': True, 'margin_rate': {'long': 0.12, 'short': 0.12}, 'multiplier': 20, 'trading_start_time': '21:00'},
+        'PX': {'has_night_session': True, 'margin_rate': {'long': 0.1, 'short': 0.1}, 'multiplier': 5, 'trading_start_time': '21:00'},
         
         # 农产品
         'RM': {'has_night_session': True, 'margin_rate': {'long': 0.13, 'short': 0.13}, 'multiplier': 10, 'trading_start_time': '21:00'},
@@ -171,7 +174,6 @@ def initialize(context):
         'LR': {'has_night_session': True, 'margin_rate': {'long': 0.21, 'short': 0.21}, 'multiplier': 1, 'trading_start_time': '21:00'},
         'LG': {'has_night_session': True, 'margin_rate': {'long': 0.13, 'short': 0.13}, 'multiplier': 1, 'trading_start_time': '21:00'},
         'FB': {'has_night_session': True, 'margin_rate': {'long': 0.14, 'short': 0.14}, 'multiplier': 1, 'trading_start_time': '21:00'},
-        'PX': {'has_night_session': True, 'margin_rate': {'long': 0.12, 'short': 0.12}, 'multiplier': 1, 'trading_start_time': '21:00'},
         'PM': {'has_night_session': True, 'margin_rate': {'long': 0.2, 'short': 0.2}, 'multiplier': 1, 'trading_start_time': '21:00'},
         'EC': {'has_night_session': False, 'margin_rate': {'long': 0.23, 'short': 0.23}, 'multiplier': 50, 'trading_start_time': '09:00'},
         'RR': {'has_night_session': True, 'margin_rate': {'long': 0.11, 'short': 0.11}, 'multiplier': 10, 'trading_start_time': '21:00'},
@@ -202,7 +204,7 @@ def initialize(context):
     
     # 策略品种选择策略配置
     # 方案1：全品种策略 - 考虑所有配置的期货品种
-    g.strategy_focus_symbols = []  # 空列表表示考虑所有品种
+    g.strategy_focus_symbols = ['J']  # 空列表表示考虑所有品种
     
     # 方案2：精选品种策略 - 只交易流动性较好的特定品种（如需使用请取消下行注释）
     # g.strategy_focus_symbols = ['RM', 'CJ', 'CY', 'JD', 'L', 'LC', 'SF', 'SI']
@@ -293,10 +295,12 @@ def check_ma_trend_and_open_gap(context):
     log.info(f"执行均线走势和开盘价差检查 - 时间: {context.current_dt}, 交易日: {current_trading_day}")
     log.info("=" * 60)
     
-    # 先检查换月移仓
+    # 换月移仓检查（在所有部分之前）
     position_auto_switch(context)
     
-    # 检查是否进入新交易日，必要时清空缓存
+    # ==================== 第一部分：基础数据获取 ====================
+    
+    # 步骤1：交易日检查和缓存清理
     if g.last_ma_trading_day != current_trading_day:
         if g.excluded_contracts:
             log.info(f"交易日切换至 {current_trading_day}，清空上一交易日的排除缓存")
@@ -304,10 +308,8 @@ def check_ma_trend_and_open_gap(context):
         g.ma_checked_underlyings = {}
         g.last_ma_trading_day = current_trading_day
 
-    # 获取当前时间
+    # 步骤2：获取当前时间和筛选可交易品种
     current_time = str(context.current_dt.time())[:5]  # HH:MM格式
-    
-    # 筛选可交易品种（根据交易开始时间判断）
     focus_symbols = g.strategy_focus_symbols if g.strategy_focus_symbols else list(g.futures_config.keys())
     tradable_symbols = []
     
@@ -320,13 +322,10 @@ def check_ma_trend_and_open_gap(context):
         should_trade = False
         
         if current_time == '21:05':
-            # 夜盘开盘：仅接受21:00开盘的品种
             should_trade = trading_start_time.startswith('21:00')
         elif current_time == '09:05':
-            # 日盘早盘：接受21:00和09:00开盘的品种
             should_trade = trading_start_time.startswith('21:00') or trading_start_time.startswith('09:00')
         elif current_time == '09:35':
-            # 日盘晚开：接受所有品种（21:00, 09:00, 09:30）
             should_trade = True
         
         if should_trade:
@@ -338,39 +337,38 @@ def check_ma_trend_and_open_gap(context):
     
     log.info(f"当前时间 {current_time} 开盘品种: {tradable_symbols}")
     
-    # 对每个品种执行均线和开盘价差检查
+    # 步骤3：对每个品种循环处理
     for symbol in tradable_symbols:
+        # 步骤3.1：检查是否已处理过
         if g.ma_checked_underlyings.get(symbol) == current_trading_day:
             log.info(f"{symbol} 已在交易日 {current_trading_day} 完成均线检查，跳过本次执行")
             continue
 
         try:
             g.ma_checked_underlyings[symbol] = current_trading_day
-            # 获取主力合约
+            
+            # 步骤3.2：获取主力合约
             dominant_future = get_dominant_future(symbol)
-            # log.debug(f"{symbol} 主力合约: {dominant_future}")
             if not dominant_future:
                 log.info(f"{symbol} 未找到主力合约，跳过")
                 continue
             
-            # 检查是否在排除缓存中（当日已检查过但不符合条件）
+            # 步骤3.3：检查排除缓存
             if dominant_future in g.excluded_contracts:
                 excluded_info = g.excluded_contracts[dominant_future]
                 if excluded_info['trading_day'] == current_trading_day:
-                    # log.debug(f"{symbol} 在排除缓存中（原因: {excluded_info['reason']}），跳过")
                     continue
                 else:
                     # 新的一天，从缓存中移除（会在after_market_close统一清理，这里也做兜底）
                     del g.excluded_contracts[dominant_future]
             
-            # 检查是否已有持仓
+            # 步骤3.4：检查是否已有持仓
             if check_symbol_prefix_match(dominant_future, set(g.trade_history.keys())):
                 log.info(f"{symbol} 已有持仓，跳过")
                 continue
             
+            # 步骤3.5：获取历史数据和前一交易日数据（合并优化）
             # 获取历史数据（需要足够计算MA30）
-            # 使用get_price获取数据，可以正确处理夜盘品种
-            # 注意：historical_data最后一行是昨天的数据，不包含今天的数据
             historical_data = get_price(dominant_future, end_date=context.current_dt, 
                                        frequency='1d', fields=['open', 'close', 'high', 'low'], 
                                        count=g.ma_historical_days)
@@ -379,6 +377,7 @@ def check_ma_trend_and_open_gap(context):
                 log.info(f"{symbol} 历史数据不足，跳过")
                 continue
 
+            # 获取前一交易日并在历史数据中匹配
             previous_trade_days = get_trade_days(end_date=current_trading_day, count=2)
             previous_trade_days = [normalize_trade_day_value(d) for d in previous_trade_days]
             previous_trading_day = None
@@ -391,9 +390,9 @@ def check_ma_trend_and_open_gap(context):
                 log.info(f"{symbol} 无法确定前一交易日，跳过")
                 continue
 
+            # 在历史数据中匹配前一交易日
             historical_dates = historical_data.index.date
             match_indices = np.where(historical_dates == previous_trading_day)[0]
-
             if len(match_indices) == 0:
                 earlier_indices = np.where(historical_dates < previous_trading_day)[0]
                 if len(earlier_indices) == 0:
@@ -401,36 +400,33 @@ def check_ma_trend_and_open_gap(context):
                     continue
                 match_indices = [earlier_indices[-1]]
 
+            # 提取截至前一交易日的数据，并一次性提取所有需要的字段
             data_upto_yesterday = historical_data.iloc[:match_indices[-1] + 1]
-            # log.debug(f"data_upto_yesterday: {data_upto_yesterday}")
             yesterday_data = data_upto_yesterday.iloc[-1]
             yesterday_close = yesterday_data['close']
+            yesterday_open = yesterday_data['open']
             
-            # 获取今天的开盘价（使用get_current_data API）
+            # 步骤3.6：获取当前价格数据
             current_data = get_current_data()[dominant_future]
             today_open = current_data.day_open
             
-            # log.info(f"  历史数据时间范围: {historical_data.index[0]} 至 {historical_data.index[-1]}")
+            # ==================== 第二部分：核心指标计算 ====================
             
-            # 计算昨天的均线值（使用截至前一交易日的数据）
+            # 步骤4：计算均线相关指标（合并优化）
             ma_values = calculate_ma_values(data_upto_yesterday, g.ma_periods)
             ma_proximity_counts = calculate_ma_proximity_counts(data_upto_yesterday, g.ma_periods, g.ma_pattern_lookback_days)
             
             log.info(f"{symbol}({dominant_future}) 均线检查:")
-            # log.debug(f"yesterday_data: {yesterday_data}")
-            # log.info(f"  昨收: {yesterday_close:.2f}, 今开: {today_open:.2f}")
-            # log.info(f"  昨日均线 - MA5: {ma_values['MA5']:.2f}, MA10: {ma_values['MA10']:.2f}, "
-            #         f"MA20: {ma_values['MA20']:.2f}, MA30: {ma_values['MA30']:.2f}")
             log.info(f"  均线贴近统计: {ma_proximity_counts}")
+            
+            # 检查均线贴近计数
             proximity_sum = ma_proximity_counts.get('MA5', 0) + ma_proximity_counts.get('MA10', 0)
             if proximity_sum < g.ma_proximity_min_threshold:
                 log.info(f"  {symbol}({dominant_future}) ✗ 均线贴近计数不足，MA5+MA10={proximity_sum} < {g.ma_proximity_min_threshold}，跳过")
-                g.excluded_contracts[dominant_future] = {
-                    'reason': 'ma_proximity',
-                    'trading_day': current_trading_day
-                }
+                add_to_excluded_contracts(dominant_future, 'ma_proximity', current_trading_day)
                 continue
             
+            # 步骤5：计算极端趋势天数
             extreme_above_count, extreme_below_count = calculate_extreme_trend_days(
                 data_upto_yesterday,
                 g.ma_periods,
@@ -447,29 +443,20 @@ def check_ma_trend_and_open_gap(context):
                 log.info(
                     f"  {symbol}({dominant_future}) ✗ 极端趋势多空同时出现且 min(A,B)={min_extreme} ≥ {filter_threshold}，跳过"
                 )
-                g.excluded_contracts[dominant_future] = {
-                    'reason': 'ma_extreme_trend',
-                    'trading_day': current_trading_day
-                }
+                add_to_excluded_contracts(dominant_future, 'ma_extreme_trend', current_trading_day)
                 continue
 
-            # 判断均线走势（使用新的灵活模式检查）
+            # 步骤6：判断均线走势
             direction = None
             if check_ma_pattern(ma_values, 'long'):
                 direction = 'long'
-                # log.info(f"  {symbol}({dominant_future}) 均线走势判断: 多头排列")
             elif check_ma_pattern(ma_values, 'short'):
                 direction = 'short'
-                # log.info(f"  {symbol}({dominant_future}) 均线走势判断: 空头排列")
             else:
-                # log.info(f"  均线走势判断: 不符合多头或空头排列，跳过")
-                # 将不符合条件的合约加入排除缓存
-                g.excluded_contracts[dominant_future] = {
-                    'reason': 'ma_trend',
-                    'trading_day': current_trading_day
-                }
+                add_to_excluded_contracts(dominant_future, 'ma_trend', current_trading_day)
                 continue
             
+            # 步骤7：检查MA5分布过滤
             if g.enable_ma_distribution_filter:
                 distribution_passed, distribution_stats = check_ma5_distribution_filter(
                     data_upto_yesterday,
@@ -488,13 +475,10 @@ def check_ma_trend_and_open_gap(context):
                     log.info(
                         f"  {symbol}({dominant_future}) ✗ MA5分布过滤未通过（{reason}）"
                     )
-                    g.excluded_contracts[dominant_future] = {
-                        'reason': 'ma5_distribution',
-                        'trading_day': current_trading_day
-                    }
+                    add_to_excluded_contracts(dominant_future, 'ma5_distribution', current_trading_day)
                     continue
             
-            # 检查历史均线模式一致性
+            # 步骤8：检查历史均线模式一致性
             consistency_passed, consistency_ratio = check_historical_ma_pattern_consistency(
                 historical_data, direction, g.ma_pattern_lookback_days, g.ma_pattern_consistency_threshold
             )
@@ -502,24 +486,19 @@ def check_ma_trend_and_open_gap(context):
             if not consistency_passed:
                 log.info(f"  {symbol}({dominant_future}) ✗ 历史均线模式一致性不足 "
                         f"({consistency_ratio:.1%} < {g.ma_pattern_consistency_threshold:.1%})，跳过")
-                g.excluded_contracts[dominant_future] = {
-                    'reason': 'ma_consistency',
-                    'trading_day': current_trading_day
-                }
+                add_to_excluded_contracts(dominant_future, 'ma_consistency', current_trading_day)
                 continue
             else:
                 log.info(f"  {symbol}({dominant_future}) ✓ 历史均线模式一致性检查通过 "
                         f"({consistency_ratio:.1%} >= {g.ma_pattern_consistency_threshold:.1%})")
             
-            # 计算开盘价差比例
+            # 步骤9：计算开盘价差并检查（合并优化）
             open_gap_ratio = (today_open - yesterday_close) / yesterday_close
-            
             log.info(f"  开盘价差检查: 昨收 {yesterday_close:.2f}, 今开 {today_open:.2f}, "
                     f"价差比例 {open_gap_ratio:.2%}")
             
             # 检查开盘价差是否符合方向要求
             gap_check_passed = False
-            
             if g.ma_gap_strategy_mode == 1:
                 # 方案1：多头检查上跳，空头检查下跳
                 if direction == 'long' and open_gap_ratio >= g.ma_open_gap_threshold:
@@ -538,18 +517,10 @@ def check_ma_trend_and_open_gap(context):
                     gap_check_passed = True
             
             if not gap_check_passed:
-                # log.info(f"  ✗ 开盘价差不符合方案{g.ma_gap_strategy_mode} {direction}方向要求，跳过")
-                # 将不符合条件的合约加入排除缓存
-                g.excluded_contracts[dominant_future] = {
-                    'reason': 'open_gap',
-                    'trading_day': current_trading_day
-                }
+                add_to_excluded_contracts(dominant_future, 'open_gap', current_trading_day)
                 continue
             
-            # 获取前一日开盘价（用于方案3）
-            yesterday_open = yesterday_data['open']
-            
-            # 将通过检查的品种加入候选列表
+            # 步骤10：将通过检查的品种加入候选列表
             g.daily_ma_candidates[dominant_future] = {
                 'symbol': symbol,
                 'direction': direction,
@@ -672,6 +643,12 @@ def check_open_and_stop(context):
                         log.info(f"  ✗ 方案3缺少前一日开盘或收盘价数据")
                 
                 if should_open:
+                    ma_values = candidate_info.get('ma_values') or {}
+                    cross_score = calculate_ma_cross_score(open_price, current_price, ma_values, direction)
+                    log.info(f"  均线穿越得分: {cross_score}, 阈值: {g.ma_cross_threshold}")
+                    if cross_score < g.ma_cross_threshold:
+                        log.info(f"  ✗ 均线穿越得分不足，跳过开仓")
+                        continue
                     # 执行开仓
                     log.info(f"  准备开仓: {symbol} {direction}")
                     target_hands = calculate_target_hands(context, dominant_future, direction)
@@ -904,6 +881,33 @@ def calculate_ma_values(data, periods):
     return ma_values
 
 
+def calculate_ma_cross_score(open_price, current_price, ma_values, direction):
+    """根据开盘价与当前价统计多周期均线穿越得分"""
+    if not ma_values:
+        return 0
+    assert direction in ('long', 'short')
+    score = 0
+    for period in g.ma_periods:
+        key = f'MA{period}'
+        ma_value = ma_values.get(key)
+        if ma_value is None:
+            continue
+        cross_up = open_price < ma_value and current_price > ma_value
+        cross_down = open_price > ma_value and current_price < ma_value
+        if not (cross_up or cross_down):
+            continue
+        if direction == 'long':
+            delta = 1 if cross_up else -1
+        else:
+            delta = -1 if cross_up else 1
+        score += delta
+        log.debug(
+            f"  均线穿越[{key}] - 开盘 {open_price:.2f}, 当前 {current_price:.2f}, "
+            f"均线 {ma_value:.2f}, 方向 {direction}, 增量 {delta}, 当前得分 {score}"
+        )
+    return score
+
+
 def calculate_ma_proximity_counts(data, periods, lookback_days):
     """统计近 lookback_days 天收盘价贴近各均线的次数"""
     proximity_counts = {f'MA{period}': 0 for period in periods}
@@ -1022,14 +1026,20 @@ def check_ma_pattern(ma_values, direction):
     
     if direction == 'long':
         # 多头模式：MA30 <= MA20 <= MA10 <= MA5 或 MA30 <= MA20 <= MA5 <= MA10
+        # 或者：MA20 <= MA30 <= MA10 <= MA5 或 MA20 <= MA30 <= MA5 <= MA10
         pattern1 = (ma30 <= ma20 <= ma10 <= ma5)
         pattern2 = (ma30 <= ma20 <= ma5 <= ma10)
-        return pattern1 or pattern2
+        pattern3 = (ma20 <= ma30 <= ma10 <= ma5)
+        pattern4 = (ma20 <= ma30 <= ma5 <= ma10)
+        return pattern1 or pattern2 or pattern3 or pattern4
     elif direction == 'short':
         # 空头模式：MA10 <= MA5 <= MA20 <= MA30 或 MA5 <= MA10 <= MA20 <= MA30
+        # 或者：MA10 <= MA5 <= MA30 <= MA20 或 MA5 <= MA10 <= MA30 <= MA20
         pattern1 = (ma10 <= ma5 <= ma20 <= ma30)
         pattern2 = (ma5 <= ma10 <= ma20 <= ma30)
-        return pattern1 or pattern2
+        pattern3 = (ma10 <= ma5 <= ma30 <= ma20)
+        pattern4 = (ma5 <= ma10 <= ma30 <= ma20)
+        return pattern1 or pattern2 or pattern3 or pattern4
     else:
         return False
 
@@ -1051,22 +1061,32 @@ def check_historical_ma_pattern_consistency(historical_data, direction, lookback
     
     match_count = 0
     total_count = lookback_days
+    # log.debug(f"历史均线模式一致性检查: {direction}, 检查过去{lookback_days}天的数据")
+    # log.debug(f"历史数据: {historical_data}")
     
     # 检查过去lookback_days天的均线模式
     for i in range(lookback_days):
         # 获取倒数第(i+1)天的数据（i=0时是昨天，i=1时是前天，依此类推）
         end_idx = -(i + 1)
-        if end_idx == -1:
+        # 获取这一天的具体日期
+        date = historical_data.index[end_idx].date()
+        # 获取到该天（包括该天）为止的所有数据
+        if i == 0:
             data_slice = historical_data
         else:
-            data_slice = historical_data.iloc[:end_idx]
+            data_slice = historical_data.iloc[:-i]
         
         # 计算该天的均线值
+        # log.debug(f"对于倒数第{i+1}天，end_idx: {end_idx}，日期: {date}，计算均线值: {data_slice}")
         ma_values = calculate_ma_values(data_slice, g.ma_periods)
+        # log.debug(f"end_idx: {end_idx}，日期: {date}，倒数第{i+1}天的均线值: {ma_values}")
         
         # 检查是否符合模式
         if check_ma_pattern(ma_values, direction):
             match_count += 1
+            # log.debug(f"日期: {date}，对于倒数第{i+1}天，历史均线模式一致性检查: {direction} 符合模式")
+        # else:
+            # log.debug(f"日期: {date}，对于倒数第{i+1}天，历史均线模式一致性检查: {direction} 不符合模式")
     
     consistency_ratio = match_count / total_count
     passed = consistency_ratio >= consistency_threshold
@@ -1196,6 +1216,12 @@ def get_multiplier(underlying_symbol, default_multiplier=10):
     """获取合约乘数的辅助函数"""
     return g.futures_config.get(underlying_symbol, {}).get('multiplier', default_multiplier)
 
+def add_to_excluded_contracts(dominant_future, reason, current_trading_day):
+    """将合约添加到排除缓存"""
+    g.excluded_contracts[dominant_future] = {
+        'reason': reason,
+        'trading_day': current_trading_day
+    }
 
 def has_reached_trading_start(current_dt, trading_start_time_str, has_night_session=False):
     """判断当前是否已到达合约允许交易的起始时间"""
@@ -1235,6 +1261,7 @@ def calculate_target_hands(context, security, direction):
     multiplier = get_multiplier(underlying_symbol)
     
     # 计算单手保证金
+    log.debug(f"计算单手保证金: {current_price:.2f} * {multiplier:.2f} * {margin_rate:.2f} = {current_price * multiplier * margin_rate:.2f}")
     single_hand_margin = current_price * multiplier * margin_rate
     
     # 还要考虑可用资金限制