【源码分享】基于ptrade的双因子宏观择时策略回测版

在投资市场中，红利股和成长股都有各自的铁杆粉丝。红利股稳健但成长性有限，成长股爆发力强但波动较大。那么，能否根据宏观环境的变化，在红利和成长之间动态切换，既享受牛市的成长收益，又规避熊市的大幅回撤呢？

今天分享一个基于美债利率 + 市场波动的双因子择时策略，让你的资金在红利ETF和成长ETF之间智能轮动。

1.1 投资标的

策略在两只ETF之间轮动：

• 红利ETF（512890.SH）：中证红利指数ETF

• 成长ETF（159967.SZ）：创业板成长ETF

这两只ETF代表了A股市场的两种投资风格：

• 红利ETF：高股息率、低估值、防御性强

• 成长ETF：高成长性、高弹性、进攻性强

1.2 核心逻辑

策略基于两个宏观信号来判断市场环境：

信号1：利率信号（美国20年期国债收益率）

• 利率上行（加息） → 经济过热，防御为主 → 持有红利

• 利率下行（降息） → 经济刺激，进攻为主 → 持有成长

信号2：市场波动信号（沪深300波动率）

• 短期波动 > 长期波动 → 市场不确定性增加 → 持有红利

• 短期波动 < 长期波动 → 市场趋于稳定 → 持有成长

根据两个信号的组合，策略的持仓决策如下：

核心规则：只有在”降息+低波动”的环境下，才持有成长ETF；其他情况均持有红利ETF。

3.1 信号计算

利率信号：美国10年期国债利率与220个交易日前的利率进行比较：

# 当前利率 vs 220个交易日前的利率if 当前利率 / 220日前利率 > 1:    rate_signal = 1  # 加息else:    rate_signal = -1  # 降息

# 短期波动率（20日）vs 长期波动率（120日）short_var = 日收益率.rolling(20).var()long_var = 日收益率.rolling(120).var()if short_var > long_var:    benchmark_return_var = 1# 波动增加else:    benchmark_return_var = -1# 波动减少

3.2 调仓机制

调仓频率：每5个交易日检查一次（可调整）

调仓流程：

计算当前信号组合确定目标持仓（红利 or 成长）如果目标ETF已持有 → 不操作如果持有另一只ETF → 清仓并买入目标ETF满仓操作，资金利用率接近100%

集中持仓：只持有1只目标持仓

全仓操作：每日15:00（回测）计算目标持仓并调仓

交易成本：ETF免印花税，佣金万二，最低5元

#coding:utf-8"""红利成长择时策略 - PTRADE 平台版说明：- 原始策略使用美债利率 + 基准ETF波动来择时，在PTRADE上我们使用基准ETF的长周期价格作为利率代理（若你有平台可用的利率数据，可替换fetch_us_rate部分）。- 每 g.period 天调仓一次，在两只ETF之间切换：红利ETF vs 成长ETF。- 严格使用平台的全局对象 `g.` 和 `log.info()` 打印日志，避免使用 print。依赖：PTRADE运行时提供的全局对象和函数，例如：- set_universe(), get_history(), get_positions(), get_position(), order(), order_value(), order_target_value(), context.portfolio"""import pandas as pdimport numpy as npfrom datetime import timedeltaimport tushare as ts# 帮助函数：信号中文标签（便于日志中显示）defrate_label(x):    return'加息'if x == 1else ('降息'if x == -1else'数据不足')defvar_label(x):    return'波动增加'if x == 1else ('波动减少'if x == -1else'数据不足')# 策略参数与全局对象definitialize(context):    """策略初始化 - 在PTRADE运行时被调用一次"""    # 全局参数    g.etf_pool = ['512890.SS', '159967.SZ']  # [红利ETF, 成长ETF]    g.hs_300_etf = '510300.SS'                # 基准ETF    g.period = 5                              # 轮仓周期（交易日）    g.day_counter = 1                         # 计数器    # tushare pro client (请替换为有效token)    try:        g.pro = ts.pro_api('86e174db47e551e2ce3e6900e3d4a2e1e9c7814801f9ad65a46f378b')    except Exception as e:        log.info('初始化 tushare 失败：{}'.format(e))    # 调仓配置    g.max_alloc_pct = 1                    # 可用资金最大使用比例    # 初始化持仓/买卖列表    g.buy_list = []    g.sell_list = []    # 设置股票池（可选，确保ETF能交易）    try:        set_universe(g.etf_pool + [g.hs_300_etf])    except Exception as e:        log.info('set_universe 失败：{}'.format(str(e)))    # 日志    log.info('=' * 60)    log.info('初始化 红利成长择时策略')    log.info('ETF池: {}'.format(g.etf_pool))    log.info('基准ETF: {}'.format(g.hs_300_etf))    log.info('调仓周期: {} 天'.format(g.period))    log.info('初始化完成')    log.info('=' * 60)    set_commission(commission_ratio =0.0002, min_commission=5.0, type='ETF')defhandle_data(context, data):    """每个bar调用一次：按照 g.period 调仓"""    # 调仓频率控制    ifnot (g.day_counter == 1or g.day_counter % g.period == 0):        g.day_counter += 1        return    log.info('-' * 60)    log.info('开始选股/择时与调仓 (day_counter={})'.format(g.day_counter))    # 打印当前资产    try:        total_asset = context.portfolio.portfolio_value        available_cash = context.portfolio.cash        position_value = total_asset - available_cash        log.info('调仓前总资产: {:.2f}, 持仓市值: {:.2f}, 可用资金: {:.2f}'.format(total_asset, position_value, available_cash))    except Exception as e:        log.info('获取资产信息失败: {}'.format(str(e)))    # 获取历史数据（至少需要 250 天来计算 long-term comparisons）    required_days_long = 250    required_days_short = 130    try:        # get_history(count, unit, field, security, fq)        # 获取基准ETF的最近 required_days_long 个日线收盘价        hs_df = get_history(required_days_long, '1d', 'close', g.hs_300_etf, fq='pre', include=False)    except Exception as e:        log.info('获取基准ETF历史数据失败: {}'.format(str(e)))        g.day_counter += 1        return    if hs_df isNoneorlen(hs_df) < required_days_short:        log.info('历史数据不足，跳过本次调仓（需要>= {} 条，当前={}）'.format(required_days_short, 0if hs_df isNoneelselen(hs_df)))        g.day_counter += 1        return    # 计算复权收盘价序列与日收益    try:        # 有些平台返回DataFrame columns: ['close']，兼容处理        ifisinstance(hs_df, dict):            # 如果返回dict（多只股票），取第一只            hs_close = list(hs_df.values())[0]['close']        else:            hs_close = hs_df['close']        hs_close = hs_close.astype(float)        return1 = hs_close.pct_change()        # short variance: 20-day var of 1-day returns; long var: 120-day var        short_var = return1.rolling(20).var()        long_var = return1.rolling(120).var()        # benchmark_return_var 信号        benchmark_return_var = 1if short_var.iloc[-1] > long_var.iloc[-1] else -1        # rate signal：通过 tushare 拉取美国国债利率数据（us_tltr），使用 ltc 字段计算        try:            # 以 context.current_dt 为参考日期，向前取 365 天作为窗口            end_dt = context.current_dt            end_date = end_dt.strftime('%Y%m%d')            start_date = (end_dt - timedelta(days=365)).strftime('%Y%m%d')            rate_df = g.pro.us_tltr(start_date=start_date, end_date=end_date)        except Exception as e:            log.info('获取利率数据失败: {}'.format(str(e)))            rate_df = None        # 数据预处理        if rate_df isnotNone:            rate_df['date'] = pd.to_datetime(rate_df['date'])            rate_df.sort_values(by="date", ascending=True, inplace=True)            rate_df.reset_index(drop=True, inplace=True)        # 确保 hs_df_proc 的 trade_date 也是 datetime 类型        ifisinstance(hs_df, dict):            hs_df_proc = list(hs_df.values())[0].copy()        else:            hs_df_proc = hs_df.copy()        # 处理 trade_date        if'trade_date'in hs_df_proc.columns:            hs_df_proc['trade_date'] = pd.to_datetime(hs_df_proc['trade_date'])        else:            try:                hs_df_proc = hs_df_proc.reset_index()                hs_df_proc['trade_date'] = pd.to_datetime(hs_df_proc.iloc[:,0])            except Exception:                hs_df_proc['trade_date'] = pd.to_datetime(pd.Series([pd.NaT]*len(hs_df_proc)))        # 标准化 close 列名        if'close'in hs_df_proc.columns:            hs_df_proc['hfq_close'] = hs_df_proc['close'].astype(float)        elif'hfq_close'in hs_df_proc.columns:            hs_df_proc['hfq_close'] = hs_df_proc['hfq_close'].astype(float)        else:            hs_df_proc['hfq_close'] = hs_df_proc.iloc[:,1].astype(float)        hs_df_proc.sort_values(by='trade_date', inplace=True)        hs_df_proc.reset_index(drop=True, inplace=True)        hs_df_proc['return1'] = hs_df_proc['hfq_close'].pct_change()        # 合并数据（与QMT一致）        signal_data = pd.merge(hs_df_proc, rate_df, left_on="trade_date", right_on="date")        # 利率和市场波动信号（在merge后的signal_data上计算，与QMT一致）        signal_data['rate_signal'] = np.where(signal_data['ltc']/signal_data['ltc'].shift(220)>1,1,-1)        signal_data['benchmark_return_var'] = np.where(signal_data['return1'].rolling(20).var()>signal_data['return1'].rolling(120).var(),1,-1)        # 获取操作当日信号：使用最新的历史数据        try:            rate_signal = signal_data['rate_signal'].iloc[-1]            benchmark_return_var = signal_data['benchmark_return_var'].iloc[-1]        except Exception as e:            log.info('获取信号失败: {}'.format(e))            rate_signal = -2            benchmark_return_var = -2    except Exception as e:        log.info('计算信号失败: {}'.format(str(e)))        g.day_counter += 1        return    # 读取当日信号，附带中文标签（加息/降息，波动增加/波动减少）    try:        log.info('rate_signal={}({}) , benchmark_return_var={}({})'.format(rate_signal, rate_label(rate_signal), benchmark_return_var, var_label(benchmark_return_var)))    except Exception:        log.info('rate_signal={}, benchmark_return_var={}'.format(rate_signal, benchmark_return_var))    # 决策逻辑：与原策略一致    buy_target = None    try:        if (rate_signal == 1and benchmark_return_var == 1) or (rate_signal == 1and benchmark_return_var == -1) or (rate_signal == -1and benchmark_return_var == 1):            buy_target = g.etf_pool[0]  # 红利ETF        elif rate_signal == -1and benchmark_return_var == -1:            buy_target = g.etf_pool[1]  # 成长ETF        else:            # 异常或中性情况，不调仓            log.info('信号中性或异常，不调仓（rate_signal={}, benchmark={}）'.format(rate_signal, benchmark_return_var))            g.day_counter += 1            return    except Exception as e:        log.info('决策出错: {}'.format(str(e)))        g.day_counter += 1        return    # 获取当前持仓    try:        positions = get_positions()  # 返回 dict: {security: positionobj}    except Exception as e:        log.info('获取持仓失败: {}'.format(str(e)))        g.day_counter += 1        return    # 满仓策略：如果目标买入的股票已在持仓中，无需调仓    # 由于持仓代码格式可能不同（如 '600570.SS' vs '600570.XSHG'），需要规范化比较    defnormalize_code(code):        """统一股票代码格式，提取主代码部分"""        if'.'in code:            return code.split('.')[0]        return code    target_base = normalize_code(buy_target)    position_codes_base = [normalize_code(sec) for sec in positions.keys()]    if target_base in position_codes_base:        log.info('目标ETF {} 已在持仓中，无需调仓'.format(buy_target))        g.day_counter += 1        log.info('调仓流程完成')        log.info('-' * 60)        return    # 组建买入/卖出列表    g.buy_list = [buy_target]    g.sell_list = []    # 卖出所有持仓（因为目标不在持仓中，需要清仓后买入）    for sec in positions:        g.sell_list.append(sec)    log.info('目标买入ETF: {}'.format(g.buy_list))    log.info('需要卖出的ETF: {}'.format(g.sell_list))    # 执行卖出：只清仓非目标标的（如果目标已在持仓中，不会被卖出）    for sec in g.sell_list:        try:            log.info('发送卖出指令: {}'.format(sec))            # 将目标持仓调整为0股            order_target_value(sec, 0)            log.info('已发送卖出指令: {}'.format(sec))        except Exception as e:            log.info('卖出指令失败 {}: {}'.format(sec, str(e)))    # 调仓目标：将总资产调整至目标ETF（无论是否已持有，都调整至满仓）    target = g.buy_list[0]    try:        # 总资产 = 现金 + 持仓市值（PTRADE 使用 portfolio_value 字段）        total_value = context.portfolio.portfolio_value        # 目标市值：按总资产计算（满仓）        target_value = total_value * g.max_alloc_pct        if target_value <= 0:            log.info('总资产为0或负，跳过调仓')        else:            # 检查当前是否已持有目标标的            if target in positions:                current_value = positions[target].value ifhasattr(positions[target], 'value') else0                log.info('当前已持有 {} , 当前市值: {:.2f} , 目标市值: {:.2f} (总资产: {:.2f})'.format(target, current_value, target_value, total_value))            else:                log.info('准备买入 {} , 目标市值: {:.2f} 元 (总资产: {:.2f})'.format(target, target_value, total_value))            # 使用按目标市值下单接口（order_target_value）调仓至目标市值            # 如果已有持仓，会自动计算差额进行加仓或减仓；如果没有则全部买入            order_target_value(target, target_value)            log.info('已发送调仓指令: {} , 目标市值: {:.2f}'.format(target, target_value))    except Exception as e:        log.info('调仓指令失败 {}: {}'.format(target, str(e)))    g.day_counter += 1    # 打印调仓后资产    try:        total_asset = context.portfolio.portfolio_value        available_cash = context.portfolio.cash        position_value = total_asset - available_cash        log.info('调仓后总资产: {:.2f}, 持仓市值: {:.2f}, 可用资金: {:.2f}'.format(total_asset, position_value, available_cash))    except Exception as e:        log.info('获取调仓后资产信息失败: {}'.format(str(e)))    log.info('调仓流程完成')    log.info('-' * 60)# 可选辅助函数（如果需要更复杂的数据获取，可替换为get_market_data_ex或平台对应函数）deffetch_us_rate_proxy():    """占位：如果有平台利率数据，可以在这里实现拉取并返回Series"""    returnNone# 结束

1、需要注册tushare账号，账号得120积分（可通过捐款获取，大概12块钱），然后登录tushare网站，右上角个人主页–接口TOKEN，获取tushare的token：

2、所在的券商ptrade要能支持tushare，目前了解到国盛（ptrade可以访问外网）、湘财（ptrade本身支持），其他券商如果支持，也可以私信我。

本策略仅供学习参考，不构成投资建议。市场有风险，回测收益不代表未来表现，实盘需谨慎。

本策略思想来自于grid老师，特此感谢！