本文承接《QMT 之如何预警大盘走向（上）》，在上篇我们详解了大盘预警的核心逻辑与五大因子设计思路，本篇将基于迅投 QMT 的xtquant接口，给出完整可运行的 Python 代码实现，并对回测结果进行实战解读。

一、环境准备与基础配置

1.1 运行环境说明

本代码基于迅投 QMT 官方 Python 接口xtquant开发，可直接在 QMT 客户端内置 Python 环境、或配置了 xtquant 的本地 Python 环境中运行。核心依赖库如下：

• numpy：数值计算与矩阵运算
• pandas：结构化行情数据处理
• xtquant：QMT 官方行情与交易接口

1.2 全局参数配置

所有回测参数、因子阈值统一配置，严格对应上篇的因子定义，支持按需调优。

import numpy as np
import pandas as pd
from xtquant import xtdata
import datetime
from typing import Tuple, Dict

# ====================== 回测基础参数 ======================
START_DATE = "20260301"   # 回测起始日期
END_DATE = "20260630"     # 回测结束日期
SH_INDEX = "000001.SH"    # 上证指数
HS300_INDEX = "000300.SH" # 沪深300指数
ZZ1000_INDEX = "000852.SH"# 中证1000指数

# ====================== 因子阈值参数 ======================
# 趋势因子参数
TREND_MA20 = 20
TREND_MA60 = 60
TREND_RETURN_DAYS = 20

# 广度因子参数
BREADTH_MA20 = 20
BREADTH_MA60 = 60
BREADTH_MA20_THRESHOLD = 0.35  # MA20站上比例风险阈值
BREADTH_MA60_THRESHOLD = 0.50  # MA60站上比例风险阈值

# 风格因子参数
STYLE_RATIO_MA = 20
STYLE_COMPARE_DAYS = 5

# 波动率因子参数
VOL_SHORT = 10
VOL_LONG = 60
VOL_MULTIPLE = 1.2

# 成交额因子参数
AMOUNT_MA = 20
AMOUNT_MULTIPLE = 1.1

二、通用工具函数封装

先封装底层数据获取和通用指标计算函数，提升代码复用性与可维护性。

2.1 日线行情数据获取

通过 xtquant 下载并格式化指数 / 个股日线数据，统一处理时间索引与成交额单位。

def get_index_daily_data(code: str, start: str, end: str) -> pd.DataFrame:
    """
    获取指数/个股日线数据（前复权）
    :param code: 股票/指数代码
    :param start: 起始日期
    :param end: 结束日期
    :return: 格式化后的日线DataFrame
    """
    # 下载历史行情数据到本地
    xtdata.download_history_data(code, "1d", start, end)
    # 读取本地行情数据
    df = xtdata.get_local_data(
        field_list=['open', 'high', 'low', 'close', 'volume', 'amount'],
        stock_code=code,
        period="1d",
        start_time=start,
        end_time=end
    )
    # 数据格式化：重置索引、转换时间格式
    df = df.reset_index()
    df["time"] = pd.to_datetime(df["time"], unit='ms')
    df = df.set_index("time").sort_index()
    # 成交额单位转换为万元
    df["amount_wan"] = df["amount"] / 10000
    return df

2.2 沪深 300 成分股获取

市场广度因子需要遍历沪深 300 全成分股，通过板块接口获取最新成分股列表。

def get_hs300_constituents() -> list:
    """获取沪深300最新成分股列表"""
    xtdata.download_sector_data()
    constituents = xtdata.get_sector_component(HS300_INDEX)
    return [code for code in constituents[0]]

2.3 通用技术指标计算

封装移动平均线、年化波动率两个通用计算函数，供所有因子调用。

def calculate_ma(series: pd.Series, window: int) -> pd.Series:
    """计算移动平均线MA"""
    return series.rolling(window=window).mean()

def calculate_annual_vol(series: pd.Series, window: int) -> pd.Series:
    """计算年化波动率 = 日收益率滚动标准差 * sqrt(252)"""
    daily_return = series.pct_change()
    vol = daily_return.rolling(window=window).std() * np.sqrt(252)
    return vol

三、五大预警因子核心实现

每个因子独立封装为函数，输入行情数据，输出每日 0/1 信号（1 表示触发风险预警）。最终 5 个因子得分相加，得到 0-5 分的大盘风险预警总分。

3.1 趋势信号（Trend Signal）

核心逻辑：3 个条件同时满足时，判定为明确下跌趋势，触发预警得 1 分。

1. 收盘价低于 20 日均线
2. 20 日均线低于 60 日均线（均线空头排列）
3. 近 20 日累计收益率为负

def calculate_trend_signal(sh_df: pd.DataFrame) -> pd.Series:
    """
    趋势风险信号：3条件全满足得1分
    """
    df = sh_df.copy()
    # 计算长短周期均线
    df["ma20"] = calculate_ma(df["close"], TREND_MA20)
    df["ma60"] = calculate_ma(df["close"], TREND_MA60)
    # 计算20日区间收益率
    df["return_20d"] = df["close"].pct_change(TREND_RETURN_DAYS)
    
    # 三个风险触发条件
    cond1 = df["close"] < df["ma20"]       # 收盘价跌破MA20
    cond2 = df["ma20"] < df["ma60"]       # 均线空头排列
    cond3 = df["return_20d"] < 0          # 中期收益转负
    
    return (cond1 & cond2 & cond3).astype(int)

3.2 市场广度信号（Breadth Signal）

核心逻辑：基于沪深 300 全成分股统计市场赚钱效应，3 个条件同时满足时，判定为市场普跌、赚钱效应极差，触发预警得 1 分。

1. 成分股中站上 MA20 的比例低于 35%
2. 成分股中站上 MA60 的比例低于 50%
3. 当日创新低个股数量多于创新高个股数量

def calculate_breadth_signal(start: str, end: str) -> pd.Series:
    """
    广度风险信号：3条件全满足得1分（基于沪深300成分股）
    """
    stock_list = get_hs300_constituents()
    signal_dict = {}
    
    # 批量计算每只成分股的均线、新高新低标记
    for stock in stock_list:
        try:
            df = get_index_daily_data(stock, start, end)
            if len(df) < 60:
                continue
            # 计算均线
            df["ma20"] = calculate_ma(df["close"], BREADTH_MA20)
            df["ma60"] = calculate_ma(df["close"], BREADTH_MA60)
            # 标记是否站上均线
            df["above_ma20"] = (df["close"] > df["ma20"]).astype(int)
            df["above_ma60"] = (df["close"] > df["ma60"]).astype(int)
            # 标记60日新高/新低
            df["high_60d"] = df["high"].rolling(BREADTH_MA60).max()
            df["low_60d"] = df["low"].rolling(BREADTH_MA60).min()
            df["is_new_high"] = (df["high"] >= df["high_60d"]).astype(int)
            df["is_new_low"] = (df["low"] <= df["low_60d"]).astype(int)
            signal_dict[stock] = df
        except:
            continue
    
    # 按交易日统计全市场指标占比
    dates = pd.date_range(start=start, end=end, freq="B")
    breadth_df = pd.DataFrame(index=dates)
    
    above_ma20_ratio = []
    above_ma60_ratio = []
    new_high_cnt = []
    new_low_cnt = []
    
    for date in dates:
        valid_stocks = [s for s in signal_dict if date in signal_dict[s].index]
        if not valid_stocks:
            above_ma20_ratio.append(np.nan)
            above_ma60_ratio.append(np.nan)
            new_high_cnt.append(np.nan)
            new_low_cnt.append(np.nan)
            continue
        
        ma20_up = 0
        ma60_up = 0
        high_cnt = 0
        low_cnt = 0
        
        for stock in valid_stocks:
            row = signal_dict[stock].loc[date]
            ma20_up += row["above_ma20"]
            ma60_up += row["above_ma60"]
            high_cnt += row["is_new_high"]
            low_cnt += row["is_new_low"]
        
        total = len(valid_stocks)
        above_ma20_ratio.append(ma20_up / total)
        above_ma60_ratio.append(ma60_up / total)
        new_high_cnt.append(high_cnt)
        new_low_cnt.append(low_cnt)
    
    breadth_df["above_ma20_ratio"] = above_ma20_ratio
    breadth_df["above_ma60_ratio"] = above_ma60_ratio
    breadth_df["new_high"] = new_high_cnt
    breadth_df["new_low"] = new_low_cnt
    
    # 三个风险触发条件
    cond1 = breadth_df["above_ma20_ratio"] < BREADTH_MA20_THRESHOLD
    cond2 = breadth_df["above_ma60_ratio"] < BREADTH_MA60_THRESHOLD
    cond3 = breadth_df["new_low"] > breadth_df["new_high"]
    
    return (cond1 & cond2 & cond3).astype(int)

3.3 风格轮动信号（Style Signal）

核心逻辑：通过中证 1000 与沪深 300 的比值判断大小盘风格，2 个条件同时满足时，判定为小盘走弱、市场风险偏好下降，触发预警得 1 分。

1. 大小盘比值低于 20 日均线
2. 大小盘比值低于 5 日前的水平

def calculate_style_signal(zz1000_df: pd.DataFrame, hs300_df: pd.DataFrame) -> pd.Series:
    """
    风格风险信号：2条件全满足得1分
    """
    # 合并两个指数收盘价数据
    df = pd.DataFrame()
    df["zz1000_close"] = zz1000_df["close"]
    df["hs300_close"] = hs300_df["close"]
    df = df.dropna()
    
    # 计算大小盘相对强弱比值
    df["ratio"] = df["zz1000_close"] / df["hs300_close"]
    df["ratio_ma20"] = calculate_ma(df["ratio"], STYLE_RATIO_MA)
    df["ratio_5d_ago"] = df["ratio"].shift(STYLE_COMPARE_DAYS)
    
    # 两个风险触发条件
    cond1 = df["ratio"] < df["ratio_ma20"]   # 比值跌破均线
    cond2 = df["ratio"] < df["ratio_5d_ago"] # 比值持续走弱
    
    return (cond1 & cond2).astype(int)

3.4 波动率信号（Volatility Signal）

核心逻辑：短期波动率大幅放大往往伴随风险集中释放，满足 1 个条件即触发预警得 1 分。

• 10 日年化波动率 > 1.2 倍的 60 日年化波动率

def calculate_vol_signal(sh_df: pd.DataFrame) -> pd.Series:
    """
    波动率风险信号：1条件满足得1分
    """
    df = sh_df.copy()
    # 计算短长周期年化波动率
    df["vol_10d"] = calculate_annual_vol(df["close"], VOL_SHORT)
    df["vol_60d"] = calculate_annual_vol(df["close"], VOL_LONG)
    
    # 触发条件：短期波动率显著放大
    cond = df["vol_10d"] > (df["vol_60d"] * VOL_MULTIPLE)
    return cond.astype(int)

3.5 成交额信号（Amount Signal）

核心逻辑：下跌趋势中放量代表抛压加大，2 个条件同时满足时触发预警得 1 分。

1. 当日成交额 > 1.1 倍的 20 日均成交额
2. 当日收盘价低于 20 日均线

def calculate_amount_signal(sh_df: pd.DataFrame) -> pd.Series:
    """
    成交额风险信号：2条件全满足得1分
    """
    df = sh_df.copy()
    # 计算20日均成交额与价格均线
    df["amount_ma20"] = calculate_ma(df["amount_wan"], AMOUNT_MA)
    df["ma20"] = calculate_ma(df["close"], TREND_MA20)
    
    # 两个风险触发条件：放量 + 下跌趋势
    cond1 = df["amount_wan"] > (df["amount_ma20"] * AMOUNT_MULTIPLE)
    cond2 = df["close"] < df["ma20"]
    
    return (cond1 & cond2).astype(int)

四、主函数：信号合并与预警计算

主函数负责串联全流程：获取基础数据、计算各因子信号、合并得到每日预警总分、输出最终结果。

def main():
    print("=" * 50)
    print(f"开始计算五因子风险预警 | 时间区间: {START_DATE} - {END_DATE}")
    print("=" * 50)
    
    # 1. 获取三大指数基础行情数据
    print("\n>>> 获取指数行情数据...")
    sh_df = get_index_daily_data(SH_INDEX, START_DATE, END_DATE)
    hs300_df = get_index_daily_data(HS300_INDEX, START_DATE, END_DATE)
    zz1000_df = get_index_daily_data(ZZ1000_INDEX, START_DATE, END_DATE)
    
    # 2. 逐个计算五大因子风险信号
    print("\n1. 计算趋势信号...")
    trend_sig = calculate_trend_signal(sh_df)
    
    print("2. 计算广度信号(遍历成分股，耗时较长，请耐心等待)...")
    breadth_sig = calculate_breadth_signal(START_DATE, END_DATE)
    
    print("3. 计算风格信号...")
    style_sig = calculate_style_signal(zz1000_df, hs300_df)
    
    print("4. 计算波动率信号...")
    vol_sig = calculate_vol_signal(sh_df)
    
    print("5. 计算成交额信号...")
    amount_sig = calculate_amount_signal(sh_df)
    
    # 3. 合并所有信号，计算风险总分
    result = pd.DataFrame()
    result["trend_signal"] = trend_sig
    result["breadth_signal"] = breadth_sig
    result["style_signal"] = style_sig
    result["vol_signal"] = vol_sig
    result["amount_signal"] = amount_sig
    # 风险总分：0-5分，分数越高市场风险越大
    result["warning_score"] = result.sum(axis=1)
    # 附加上证指数收盘价便于对照
    result["close"] = sh_df["close"]
    
    # 4. 数据清洗：去除无效空值
    result = result.dropna()
    
    # 5. 控制台输出结果
    print("\n" + "=" * 50)
    print("五因子大盘风险预警结果（每日）")
    print("=" * 50)
    print(result.round(4))
    
    return result

if __name__ == "__main__":
    warning_result = main()

五、回测结果与实战解读

我们以上证指数为标的，对历史行情进行回测，得到预警时序结果如下：

回测图分为上下两部分：

• 上半部分：上证指数日线收盘价走势
• 下半部分：每日预警总分（0-5 分），以及五个因子各自的触发状态

核心解读

1. 信号累积比单日分数更重要：量化预警的核心价值不是单日拿到 5 分才警惕，而是观察信号如何逐个触发、分数逐步抬升的过程。当分数从 0 分逐步上涨到 3 分以上时，往往是风险持续累积的阶段，是提前降仓的关键窗口。
2. 多维度交叉验证降噪：单一因子触发可能是市场噪音，但当趋势、广度、波动率多个维度同时发出信号，市场进入下跌行情的概率会大幅提升。
3. 提前预警特性：这套体系的设计目标是在下跌初期捕捉信号，而非下跌已经发生后才确认，给仓位调整留出充足的决策时间。

六、总结与实盘使用建议

1. 开箱即用：将上述代码完整复制到 QMT 的 Python 编辑器中，配置好日期参数即可直接运行；广度因子需要遍历 300 只个股，计算耗时稍长属于正常现象。
2. 阈值自定义优化：所有因子的周期、阈值都可以根据自身风险偏好调整，保守型投资者可降低阈值，更早触发风险预警。
3. 实盘运行方案：可将脚本设置为每日收盘后自动运行，输出当日预警分数，作为次日仓位调整的核心参考依据。
4. 扩展方向：可在此基础上加入北向资金、股指期货升贴水、行业轮动等更多因子，也可以根据预警分数设计对应的动态仓位管理策略。

量化预警的意义，从来不是事后证明 “我预判对了”，而是把模糊的 “市场不对劲” 拆解成有数据、有时间点、可验证的决策依据。

风险提示：本文仅为量化策略技术分享，不构成任何投资建议。市场有风险，投资需谨慎。