Python量化实战：基于成交量三大择时因子的算法解析与回测验证

在量化投资领域，大多数技术指标都源于价格与成交量之间的关系。我们过去探讨了许多因子，但主要聚焦于价格维度的分析，对于成交量这一核心要素的深入挖掘相对较少。

然而在实际交易中，成交量往往被视为最真实的指标。它能有效反映标的的交易活跃度、市场参与者的换手意愿以及历史筹码的分布情况。正如市场老手常说的：价格或可被短暂操控，但成交量却是真金白银的博弈结果。因此，任何成熟的交易体系，都离不开对成交量因子的深度理解和应用。

你是否思考过成交量通常呈现哪些规律？深入了解后，你可能会发现以下特征：

1. 行情启动初期，成交量往往显著放大。
2. 价格见顶时，常伴随放量滞涨现象。
3. 主升浪阶段，通常呈现量价齐升的格局。
4. 下跌趋势中，则多见量价齐跌。

当然，对于经验丰富的交易者而言，这些规律并非一成不变，市场总是充满复杂性。这也引出了一系列值得探究的问题：放量达到多少才算有效？如何区分底部放量与高位派发？成交量可以衍生出哪些技术指标？它们又有哪些不同的应用场景？

本文将聚焦于成交量，系统梳理三种在个股、期货或指数上进行择时研究的经典方法，并提供完整的算法原理、实证回测与可执行的Python代码。

常见用法与实证

成交量均线（MAAMT）

1、算法
当成交量上穿其N日移动平均线时，产生买入信号；当成交量下穿其N日移动平均线时，产生卖出信号。

2、原理解析
从交易常识出发，放量可以分为底部放量和顶部放量。底部区域的成交量放大通常被视为积极的看涨信号。而顶部放量则可能意味着筹码松动，是一个偏空的信号。持续的缩量往往预示着市场活跃度下降，不利于趋势延续。

3、风险点
由于顶部放量可能是一个负面信号，单纯基于成交量上穿均线的MAAMT指标，有可能在行情见顶回落时遭遇较大回撤。这一现象在波动剧烈的个股上可能更为明显，而对于波动相对平缓的指数，表现或许会稳健一些。

4、实证结果分析
我们通过Python回测，观察该策略在不同标的上的表现。

中证1000指数 (000852.SH) 回测结果

沪深300指数 (000300.SH) 回测结果

贵州茅台 (600519.SH) 个股回测结果

5、总结
从以上三张回测图可以看出，成交量均线策略在宽基指数（如中证1000、沪深300）的择时上表现尚可，能够获得超越基准的正向收益。然而在个股（如贵州茅台）上的应用效果则较为一般，这提示我们需要结合标的特性来使用该指标。

6、Python代码实现
以下是生成买卖信号的核心代码，回测框架函数 backtest 将在文后统一展示。

def get_signals_ma(self,
                    df_vol,
                    n_days: int = 30, ):
    """生成买卖信号 - 使用向量化操作"""
    if df_vol is None:
        raise ValueError("输入数据为空！")

    data = df_vol.copy()

    # 计算成交量N日均线
    data['vol_ma'] = data['vol'].rolling(window=n_days, min_periods=1).mean()
    # 创建前一日的成交量均线列
    data['vol_ma_prev'] = data['vol_ma'].shift(1)
    data['vol_prev'] = data['vol'].shift(1)

    # 使用向量化操作生成信号
    # 买入信号：前一日成交量在均线下方，当日成交量在均线上方
    buy_condition = (data['vol_prev'] <= data['vol_ma_prev']) & (data['vol'] > data['vol_ma'])

    # 卖出信号：前一日成交量在均线上方，当日成交量在均线下方
    sell_condition = (data['vol_prev'] >= data['vol_ma_prev']) & (data['vol'] < data['vol_ma'])

    # 初始化信号列
    data['signal'] = 0

    # 设置买入信号
    data.loc[buy_condition, 'signal'] = 1

    # 设置卖出信号
    data.loc[sell_condition, 'signal'] = -1

    # 删除临时列
    data = data.drop(['vol_ma_prev', 'vol_prev'], axis=1)

    print(f"信号生成完成，共 {len(data[data['signal'] != 0])} 个交易信号")
    print(f"买入信号: {len(data[data['signal'] == 1])} 个")
    print(f"卖出信号: {len(data[data['signal'] == -1])} 个")

    return data.copy()

EOM指标（简易波动指标）

1、算法
EOM (Ease of Movement) 简易波动指标的计算步骤如下：

1. 距离移动 = (当日最高价+当日最低价)/2 - (前一日最高价+前一日最低价)/2
2. 箱子比率 = (成交量 / 缩放因子) / (最高价 - 最低价) (缩放因子通常为100,000,000)
3. EOM = 距离移动 / 箱子比率
4. 交易信号：
   • EOM值上穿0轴：产生买入信号
   • EOM值下穿0轴：产生卖出信号

2、原理
EOM指标的核心思想在于衡量价格变动所需的“努力”程度。在成交量较小的情况下，价格出现大幅移动，表明市场趋势强劲，移动“容易”；反之，在成交量巨大的情况下，价格仅发生微小变动，则表明市场趋势疲弱，移动“困难”。

3、实证结果分析

沪深300指数EOM策略回测结果

中证1000指数EOM策略回测结果

4、总结
除了MAAMT和EOM，成交量还衍生出许多其他技术指标，如下表所示，感兴趣的读者可以基于提供的框架自行测试。

5、Python代码实现

    @staticmethod
    def calculate_eom(df_vol,
                       eom_period=14,
                       ma_period=9,
                       scale = 100000000):
        """计算EOM指标"""
        if df_vol is None:
            print("请先下载数据")
            return None

        data = df_vol.copy()

        # 1. 计算中点
        data['Midpoint'] = (data['high'] + data['low']) / 2

        # 2. 计算距离移动 (Distance Moved)
        # DM = 当日中点 - 前一日中点
        data['Midpoint_Prev'] = data['Midpoint'].shift(1)
        data['Distance_Moved'] = data['Midpoint'] - data['Midpoint_Prev']

        # 3. 计算箱子比率 (Box Ratio)
        # BR = (vol / Scale) / (high - low)
        # Scale通常为100000000，用于调整量纲

        data['Box_Ratio'] = (data['vol'] / scale) / (data['high'] - data['low'])

        # 避免除数为0的情况
        data['Box_Ratio'] = data['Box_Ratio'].replace([np.inf, -np.inf], 0)

        # 4. 计算原始EOM
        # EOM = Distance_Moved / Box_Ratio
        data['EOM_Raw'] = data['Distance_Moved'] / data['Box_Ratio']
        data['EOM_Raw'] = data['EOM_Raw'].replace([np.inf, -np.inf], 0)

        # 5. 计算EOM的N日简单移动平均
        data['EOM'] = data['EOM_Raw'].rolling(window=eom_period).mean()

        # 6. 计算EOM移动平均信号线
        data['EOM_MA'] = data['EOM'].rolling(window=ma_period).mean()

        # 清理临时列
        data = data.drop(['Midpoint_Prev'], axis=1)

        return data

    def calculate_eom_signals(self, df_vol: pd.DataFrame):
        """生成EOM交易信号"""
        if (df_vol is None) or df_vol.empty:
            raise ValueError('基础数据为空！')
        data = self.calculate_eom(df_vol)
        if data is None:
            raise ValueError('emo数据为空！')
        # 计算前一日的EOM值
        data['EOM_Prev'] = data['EOM'].shift(1)
        data['EOM_MA_Prev'] = data['EOM_MA'].shift(1)

        # 初始化信号列
        data['signal'] = 0

        # 信号1: EOM上穿0轴 (金叉)
        # 前一日EOM < 0 且 当日EOM > 0
        golden_cross = (data['EOM_Prev'] < 0) & (data['EOM'] > 0)
        data.loc[golden_cross, 'signal'] = 1

        # 信号2: EOM下穿0轴 (死叉)
        # 前一日EOM > 0 且 当日EOM < 0
        death_cross = (data['EOM_Prev'] > 0) & (data['EOM'] < 0)
        data.loc[death_cross, 'signal'] = -1

        # # 信号3: EOM上穿其移动平均线 (可选，更强烈的买入信号)
        # # 前一日EOM < EOM_MA 且 当日EOM > EOM_MA
        # eom_above_ma = (data['EOM_Prev'] < data['EOM_MA_Prev']) & (data['EOM'] > data['EOM_MA'])
        # data.loc[eom_above_ma, 'signal'] = 2  # 强烈买入信号
        #
        # # 信号4: EOM下穿其移动平均线 (可选，更强烈的卖出信号)
        # # 前一日EOM > EOM_MA 且 当日EOM < EOM_MA
        # eom_below_ma = (data['EOM_Prev'] > data['EOM_MA_Prev']) & (data['EOM'] < data['EOM_MA'])
        # data.loc[eom_below_ma, 'signal'] = -2  # 强烈卖出信号

        # 清理临时列
        data = data.drop(['EOM_Prev', 'EOM_MA_Prev'], axis=1)

        # 统计信号数量
        signals_summary = {
            '买入信号(上穿0轴)': len(data[data['signal'] == 1]),
            '卖出信号(下穿0轴)': len(data[data['signal'] == -1]),
            # '强烈买入信号(上穿MA)': len(data[data['signal'] == 2]),
            # '强烈卖出信号(下穿MA)': len(data[data['signal'] == -2]),
            '总信号数': len(data[data['signal'] != 0])
        }

        print("\nEOM信号统计:")
        for key, value in signals_summary.items():
            print(f"{key}: {value}个")

        return data

VMACD_MTM算法（成交量MACD动量）

1、算法流程

1. 计算基于成交量的MACD指标（VMACD）。
2. 对VMACD序列进行Z-score标准化处理。
3. 计算标准化后VMACD的动量（VMACD_MTM）。
4. 根据阈值生成交易信号。

2、交易规则

• 当日 VMACD_MTM > 阈值T：信号为 1（开仓/持仓）
• 当日 VMACD_MTM < -阈值T：信号为 -1（平仓/空仓）
• 当日 VMACD_MTM 介于 [-T, T] 之间：保持上一交易日的状态

3、实证结果分析

沪深300指数VMACD_MTM策略回测结果

中证1000指数VMACD_MTM策略回测结果

4、总结
整体来看，VMACD_MTM也是一个表现不错的择时指标。当然，本文展示的结果未经过任何参数优化。值得注意的是，过度拟合历史数据的优化可能降低策略的泛化能力，读者可以尝试滚动窗口优化等稳健方法。

5、Python代码实现

class VolMacdMtm:
    def __init__(self, n1: int = 12, n2: int = 26, n3: int = 9, n_window: int = 60, threshold: float = 1.0):
        """
        初始化 VMACD_MTM 参数

        Parameters:
        -----------
        n1: int
            VMACD 快速移动平均窗口，默认 12
        n2: int
            VMACD 慢速移动平均窗口，默认 26
        n3: int
            VMACD DEA 计算窗口，默认 9
        n_window: int
            VMACD_MTM 计算窗口，默认 60
        threshold: float
            交易信号阈值 T，默认 1.0
        """
        self.n1 = n1
        self.n2 = n2
        self.n3 = n3
        self.n_window = n_window
        self.threshold = threshold

    @staticmethod
    def calculate_ema(data: pd.Series, window: int) -> pd.Series:
        """计算指数移动平均"""
        return data.ewm(span=window, adjust=False).mean()

    # 计算vmacd
    def calculate_vmacd(self, volume: pd.Series) -> pd.DataFrame:
        """
        计算 VMACD 指标

        Parameters:
        -----------
        volume: pd.Series
            成交量序列

        Returns:
        --------
        pd.DataFrame: 包含 V_DIF, V_DEA, VMACD 的数据框
        """
        # 计算 V_DIF
        ema_fast = self.calculate_ema(volume, self.n1)
        ema_slow = self.calculate_ema(volume, self.n2)
        v_dif = ema_fast - ema_slow

        # 计算 V_DEA
        v_dea = self.calculate_ema(v_dif, self.n3)

        # 计算 VMACD
        vmacd = (v_dif - v_dea) * 2

        return pd.DataFrame({
            'V_DIF': v_dif,
            'V_DEA': v_dea,
            'VMACD': vmacd
        })

    # 标准化vmacd
    def standardize_vmacd(self, vmacd: pd.Series) -> pd.Series:
        """
        对 VMACD 进行 Z-score 标准化

        Parameters:
        -----------
        vmacd: pd.Series
            VMACD 序列

        Returns:
        --------
        pd.Series: 标准化后的 VMACD
        """
        # 滚动计算均值和标准差
        rolling_mean = vmacd.rolling(window=self.n_window, min_periods=1).mean()
        rolling_std = vmacd.rolling(window=self.n_window, min_periods=1).std()

        # Z-score 标准化
        vmacd_std = (vmacd - rolling_mean) / rolling_std

        return vmacd_std

    def calculate_vmacd_mtm(self, vmacd_std: pd.Series) -> pd.Series:
        """
        计算 VMACD_MTM 指标

        Parameters:
        -----------
        vmacd_std: pd.Series
            标准化后的 VMACD 序列

        Returns:
        --------
        pd.Series: VMACD_MTM 序列
        """
        # 计算 VMACD_diff (标准化后)
        vmacd_diff = vmacd_std.diff()

        # 计算 VMACD_MTM (近N个交易日的VMACD_diff累计和)
        vmacd_mtm = vmacd_diff.rolling(window=self.n_window).sum()

        return vmacd_mtm

    def calculate_vmacd_mtm_signals(self, vmacd_mtm: pd.Series) -> pd.Series:
        """
        生成交易信号

        规则:
        - 当日 VMACD_MTM > T: 信号为 1 (持仓)
        - 当日 VMACD_MTM < -T: 信号为 -1 (空仓)
        - 当日 VMACD_MTM 在 [-T, T] 之间: 保持上一状态

        Parameters:
        -----------
        vmacd_mtm: pd.Series
            VMACD_MTM 序列

        Returns:
        --------
        pd.Series: 交易信号序列 (1: 做多, -1: 空仓, 0: 初始状态)
        """
        signals = pd.Series(0, index=vmacd_mtm.index)
        prev_signal = 0  # 初始状态为空仓

        for i in range(len(vmacd_mtm)):
            if pd.isna(vmacd_mtm.iloc[i]):
                signals.iloc[i] = 0
                continue

            if vmacd_mtm.iloc[i] > self.threshold:
                signals.iloc[i] = 1
                prev_signal = 1
            elif vmacd_mtm.iloc[i] < -self.threshold:
                signals.iloc[i] = -1
                prev_signal = -1
            else:
                # 保持上一状态
                signals.iloc[i] = prev_signal

        # 为了跟回测框架保持一致，这里需要仅仅保持第一次的交易信号其他为0
        signals = pd.Series(np.where(signals!=signals.shift(1), signals, 0), index=signals.index)
        return signals

    def calculate_all(self, df: pd.DataFrame, volume_col: str = 'vol') -> pd.DataFrame:
        """
        计算完整的 VMACD_MTM 指标和信号

        Parameters:
        -----------
        df: pd.DataFrame
            包含成交量等数据的数据框，需要有日期索引
        volume_col: str
            成交量列名

        Returns:
        --------
        pd.DataFrame: 包含所有指标和信号的数据框
        """
        # 计算 VMACD
        vmacd_df = self.calculate_vmacd(df[volume_col])

        # 标准化 VMACD
        vmacd_std = self.standardize_vmacd(vmacd_df['VMACD'])

        # 计算 VMACD_MTM
        vmacd_mtm = self.calculate_vmacd_mtm(vmacd_std)

        # 生成信号
        signals = self.calculate_vmacd_mtm_signals(vmacd_mtm)

        # 合并结果
        result_df = pd.DataFrame({
            'vol': df[volume_col],
            'V_DIF': vmacd_df['V_DIF'],
            'V_DEA': vmacd_df['V_DEA'],
            'VMACD': vmacd_df['VMACD'],
            'VMACD_std': vmacd_std,
            'VMACD_MTM': vmacd_mtm,
            'threshold_buy': [self.threshold] * len(df[volume_col]),
            'threshold_sell': [-self.threshold] * len(df[volume_col]),
            'signal': signals
        })

        result_df = pd.merge(result_df, df, how='inner',left_index=True, right_index=True)
        return result_df

总结

本文系统地介绍了三种基于成交量的经典择时指标：成交量均线（MAAMT）、简易波动指标（EOM）以及成交量MACD动量（VMACD_MTM）。我们不仅阐述了其算法原理，还提供了在主要指数上的实证回测结果和可直接运行的Python代码。

从回测结果来看，成交量确实是一个值得深入挖掘的有效因子。值得注意的是，本文介绍的指标和思路可以作为构建更复杂策略的“原材料”。开发者可以根据自身的风险偏好和交易标的，对这些基础因子进行改良，或者将其与其他技术因子、基本面因子相结合，构建出更具竞争力的多因子择时策略。更多关于量化策略的讨论与实践，欢迎关注云栈社区。