如何使用 Mass Index 梅斯线判断行情反转

梅斯线（Mass Index，简称 MASS）是一种基于最高价和最低价波幅宽度的技术指标，主要用于预测趋势的反转。它并不直接指示价格的方向，而是通过衡量价格波动的带宽（High - Low）的变化来寻找“反转凸块”（Reversal Bulge）。

以下是关于梅斯线的原理说明及在聚宽（JoinQuant）平台上的策略实现代码。

1. 梅斯线（MASS）的基本原理

梅斯线基于一个假设：当价格波幅扩大到一定程度后，往往会发生行情的反转。

计算公式

1. 单日波幅：$Range = High - Low$
2. 平滑波幅：计算 Range 的 9 日指数移动平均（EMA），记为 $EMA_1$。
3. 二次平滑：计算 $EMA_1$ 的 9 日指数移动平均，记为 $EMA_2$。
4. 比率：$Ratio = EMA_1 / EMA_2$
5. MASS 值：将过去 25 天的 Ratio 进行累加。

交易信号：反转凸块（Reversal Bulge）

梅斯线最核心的用法是寻找**“反转凸块”**：

1. 形成凸块：MASS 指标先上升超过 27。
2. 确认反转：随后 MASS 指标跌破 26.5。
3. 方向确认：由于 MASS 只提示即将反转，不提示方向，因此必须结合一条趋势线（通常使用 9日 EMA 均线）来判断买卖方向。

• 买入信号：出现“反转凸块” + 股价上穿（或高于）9日 EMA 均线。
• 卖出信号：出现“反转凸块” + 股价下穿（或低于）9日 EMA 均线。

2. 聚宽策略实现代码

以下是一个完整的策略代码，实现了上述逻辑。该策略在每天开盘时计算 MASS 指标，当检测到“反转凸块”且趋势符合时进行交易。

# -*- coding: utf-8 -*-
import pandas as pd
import numpy as np

def initialize(context):
    """
    初始化函数
    """
    # 设定要操作的股票（示例：平安银行）
    g.security = '000001.XSHE'
    
    # 设定基准
    set_benchmark('000300.XSHG')
    
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # 设定手续费
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 每天开盘时运行
    run_daily(market_open, time='09:30')

def calculate_mass(security, data_window=60):
    """
    计算 MASS 指标
    """
    # 获取历史数据，需要足够长的数据来计算EMA和Sum
    # MASS(25) 需要 25天的sum，加上EMA的预热，取60天比较稳妥
    h = attribute_history(security, data_window, '1d', ['high', 'low', 'close'])
    
    # 1. 计算单日波幅 (High - Low)
    high_low_range = h['high'] - h['low']
    
    # 2. 计算 9日 EMA (EMA1)
    # pandas 的 ewm span=9 对应 9日 EMA
    ema1 = high_low_range.ewm(span=9, adjust=False).mean()
    
    # 3. 计算 EMA1 的 9日 EMA (EMA2)
    ema2 = ema1.ewm(span=9, adjust=False).mean()
    
    # 4. 计算比率 Ratio
    mass_ratio = ema1 / ema2
    
    # 5. 计算 MASS (25日累加)
    mass = mass_ratio.rolling(window=25).sum()
    
    # 计算 9日收盘价 EMA (用于判断方向)
    ema_close = h['close'].ewm(span=9, adjust=False).mean()
    
    return mass, ema_close, h['close']

def market_open(context):
    """
    每日交易逻辑
    """
    security = g.security
    
    # 获取计算好的指标序列
    mass_series, ema_close_series, close_series = calculate_mass(security)
    
    # 检查数据是否包含 NaN (刚上市或数据不足时不操作)
    if np.isnan(mass_series[-1]) or np.isnan(mass_series[-2]):
        return

    # --- 获取关键数值 ---
    # 当前 MASS 值
    current_mass = mass_series[-1]
    # 前一日 MASS 值
    prev_mass = mass_series[-2]
    
    # 当前收盘价 (昨收)
    current_close = close_series[-1]
    # 当前 9日 EMA
    current_ema = ema_close_series[-1]
    
    # 获取当前持仓
    position = context.portfolio.positions[security]
    
    # --- 判断“反转凸块” (Reversal Bulge) ---
    # 逻辑：之前大于27，现在跌破26.5。
    # 这里简化判断：昨日 > 26.5 且 今日 < 26.5，并且近期最高点曾超过27
    # 为了更严格，我们可以判断：昨日 MASS >= 26.5 且 今日 MASS < 26.5，且过去几天内有过 > 27 的情况
    
    # 简单判定逻辑：
    # 1. 穿越阈值：昨日在26.5上方，今日在26.5下方
    cross_down_threshold = (prev_mass >= 26.5) and (current_mass < 26.5)
    
    # 2. 凸块确认：过去5天内，最大值曾经超过27
    recent_max = mass_series[-5:].max()
    has_bulge = recent_max > 27
    
    is_reversal_signal = cross_down_threshold and has_bulge
    
    # --- 交易执行 ---
    
    if is_reversal_signal:
        log.info("检测到 MASS 反转凸块信号！当前MASS: %.2f" % current_mass)
        
        # 结合 EMA 判断方向
        if current_close > current_ema:
            # 价格在均线上方，看涨反转 -> 买入
            if position.total_amount == 0:
                order_value(security, context.portfolio.available_cash)
                log.info("趋势向上 (Price > EMA9)，执行买入: %s" % security)
                
        elif current_close < current_ema:
            # 价格在均线下方，看跌反转 -> 卖出
            if position.closeable_amount > 0:
                order_target(security, 0)
                log.info("趋势向下 (Price < EMA9)，执行卖出: %s" % security)
    
    # 止损逻辑 (可选)：如果亏损超过 10% 止损
    if position.total_amount > 0:
        if (current_close - position.avg_cost) / position.avg_cost < -0.10:
             order_target(security, 0)
             log.info("触发止损，卖出: %s" % security)

    # 记录数据用于画图
    record(MASS=current_mass, Threshold_High=27, Threshold_Low=26.5)

3. 代码关键点解析

1. 数据获取 (attribute_history):

   • 我们需要获取 high, low, close 数据。
   • data_window 设置为 60 天，因为 MASS 计算需要 25 天的累加求和，加上 EMA 计算需要一定的数据预热期，60天能保证数据的准确性。

2. MASS 计算逻辑:

   • 使用 pandas 的 ewm(span=9) 来计算指数移动平均，这比手动编写公式更高效且准确。
   • mass_ratio.rolling(window=25).sum() 实现了过去25天比率的累加。

3. 反转凸块 (Reversal Bulge) 的判定:

   • 代码中使用了 cross_down_threshold (跌破 26.5) 和 has_bulge (近期曾高于 27) 两个条件联合判断。这是为了捕捉 MASS 曲线先冲高再回落的形态。

4. 趋势过滤 (EMA Filter):

   • MASS 本身只告诉你“要变盘了”，但不告诉你往哪变。
   • 代码中引入了 ema_close (9日收盘价均线)。
   • 做多逻辑：MASS 发出反转信号 且 价格高于 9日 EMA。
   • 做空逻辑：MASS 发出反转信号 且 价格低于 9日 EMA。

4. 注意事项

• 适用场景：MASS 指标适合捕捉趋势末端的反转，在震荡市中可能会出现频繁的假信号。
• 参数调整：标准的参数是 (9, 25)，阈值是 27 和 26.5。对于不同的个股或 A 股市场整体，这些阈值可能需要根据回测结果进行微调（例如调整为 26 和 25.5）。
• 未来函数：本策略使用的是 attribute_history 获取截止到前一天的收盘数据，并在开盘时运行，避免了未来函数。

Q&A

Q: 为什么 MASS 指标需要配合 EMA 均线使用？
A: MASS 指标本质上是一个波动率指标，它衡量的是高低价差的带宽变化。当 MASS 很高时，说明近期波动率极大，这种状态不可持续，往往预示着行情反转。但它无法区分是“由跌转涨”还是“由涨转跌”，因此必须配合 EMA 均线来确认当前的主要趋势方向。

Q: 如何在聚宽中查看 MASS 指标的走势图？
A: 在代码中使用了 record(MASS=current_mass) 函数。回测结束后，在回测详情页面的收益曲线下方，会显示出 MASS 指标的历史走势曲线，方便您观察信号触发的时机。

Q: 可以将此策略用于分钟级别回测吗？
A: 可以。只需将 attribute_history 中的频率参数 '1d' 改为 '1m' 或其他分钟频率，并相应调整 run_daily 的运行频率即可。但需注意，分钟级别的波动噪音较大，MASS 的阈值（27/26.5）可能需要重新校准。