聚宽多策略组合：如何用 ETF 轮动、小市值和白马策略分配资金？

问题描述

聚宽多策略组合：如何用 ETF 轮动、小市值和白马策略分配资金？
如何在聚宽中把 ETF 轮动、小市值和白马股策略组合起来，使用动态资金上限、涨跌停过滤和止损规则管理各子策略？

解决方案

在量化交易中，单一策略往往难以应对所有的市场环境。通过将不同相关性的策略（如 ETF轮动、小市值 和 白马股 策略）进行组合，并利用聚宽（JoinQuant）的子账户功能进行资金隔离与动态管理，可以显著提升资金利用率并平滑收益曲线。

下面将为您详细介绍如何在聚宽中实现这一多策略组合系统。

一、核心设计思路

资金分配与子账户隔离：
使用聚宽的 set_subportfolios API，将初始资金划分为三个独立的子账户（SubPortfolio），分别对应：
- 子账户 0 (ETF轮动)：分配 30% 资金，用于配置低相关性的宽基或行业ETF。
- 子账户 1 (小市值)：分配 40% 资金，寻找市值最小且未停牌、非ST的股票。
- 子账户 2 (白马股)：分配 30% 资金，筛选高ROE、高成长性的绩优蓝筹股。
风控与过滤机制：
- 涨跌停过滤：在下单前，通过 get_current_data() 获取当前的涨跌停价（high_limit / low_limit），避免在涨停时追买或跌停时强卖。
- 动态止损：在每日运行中，监控持仓标的的浮动盈亏，若个股亏损超过设定阈值（如 -8%），则触发主动止损平仓。

二、聚宽多策略组合完整源码

请在聚宽回测环境中运行以下 Python 3 策略代码：

import pandas as pd
import numpy as np
from jqdata import *

def initialize(context):
    # 1. 基础设定
    set_benchmark('000300.XSHG')
    set_option('use_real_price', True) # 开启真实价格模式
    log.set_level('order', 'info')
    
    # 2. 初始化多子账户 (ETF轮动, 小市值, 白马股)
    init_cash = context.portfolio.starting_cash
    set_subportfolios([
        SubPortfolioConfig(cash=init_cash * 0.30, type='stock'), # pindex=0: ETF轮动
        SubPortfolioConfig(cash=init_cash * 0.40, type='stock'), # pindex=1: 小市值
        SubPortfolioConfig(cash=init_cash * 0.30, type='stock')  # pindex=2: 白马股
    ])
    
    # 3. 策略参数配置
    g.etf_pool = ['510300.XSHG', '510500.XSHG', '159915.XSHE'] # 沪深300, 中证500, 创业板
    g.stop_loss_threshold = -0.08 # 个人个股止损阈值 -8%
    
    # 4. 定时任务
    # 每天开盘前过滤停牌与ST
    run_daily(before_market_open, time='09:00')
    # 盘中每分钟进行止损监控
    run_daily(check_stop_loss, time='every_bar')
    # 盘中执行各子策略交易
    run_daily(trade_etf_rotation, time='14:50') # ETF轮动尾盘交易
    run_daily(trade_small_cap, time='09:30')    # 小市值开盘交易
    run_daily(trade_blue_chips, time='10:00')   # 白马股定时交易

def before_market_open(context):
    # 获取全市场股票，用于小市值和白马股筛选
    g.all_stocks = list(get_all_securities(['stock'], context.current_dt.date()).index)

def check_stop_loss(context):
    """ 盘中实时个股止损监控 """
    for pindex in [0, 1, 2]:
        sub_port = context.subportfolios[pindex]
        for stock, pos in sub_port.long_positions.items():
            # 计算个股盈亏比例
            if pos.total_amount > 0 and pos.avg_cost > 0:
                current_price = pos.price
                profit_rate = (current_price - pos.avg_cost) / pos.avg_cost
                if profit_rate <= g.stop_loss_threshold:
                    # 检查是否跌停，跌停则无法卖出
                    curr_data = get_current_data()
                    if current_price > curr_data[stock].low_limit:
                        order_target(stock, 0, pindex=pindex)
                        log.warning(f"[子账户{pindex}止损] 标的 {stock} 亏损达 {profit_rate:.2%}, 触发止损卖出")

def trade_etf_rotation(context):
    """ 子策略一：ETF 20日动量轮动 (pindex=0) """
    pindex = 0
    # 获取过去20天收盘价
    hist = history(20, '1d', 'close', security_list=g.etf_pool)
    # 计算20日涨幅
    momentum = (hist.iloc[-1] - hist.iloc[0]) / hist.iloc[0]
    best_etf = momentum.idxmax()
    
    curr_data = get_current_data()
    # 如果最强的ETF 20日涨幅大于0，且未停牌，则买入；否则空仓
    target_etf = best_etf if momentum[best_etf] > 0 else None
    
    sub_port = context.subportfolios[pindex]
    # 卖出非目标ETF
    for stock in list(sub_port.long_positions.keys()):
        if stock != target_etf and sub_port.long_positions[stock].closeable_amount > 0:
            if curr_data[stock].last_price > curr_data[stock].low_limit:
                order_target(stock, 0, pindex=pindex)
                
    # 买入目标ETF
    if target_etf and target_etf not in sub_port.long_positions:
        if not curr_data[target_etf].paused and curr_data[target_etf].last_price < curr_data[target_etf].high_limit:
            cash = sub_port.available_cash
            order_value(target_etf, cash, pindex=pindex)
            log.info(f"[ETF轮动] 买入最强动量ETF: {target_etf}")

def trade_small_cap(context):
    """ 子策略二：小市值策略 (pindex=1) """
    pindex = 1
    # 1. 财务数据筛选：获取市值最小的前10只股票
    q = query(valuation.code, valuation.market_cap).filter(
        valuation.code.in_(g.all_stocks)
    ).order_by(valuation.market_cap.asc()).limit(50)
    df = get_fundamentals(q)
    
    # 2. 过滤ST、停牌、涨跌停
    curr_data = get_current_data()
    valid_stocks = []
    for stock in df['code']:
        if not curr_data[stock].is_st and not curr_data[stock].paused:
            # 过滤开盘即涨停或跌停的股票
            if curr_data[stock].day_open < curr_data[stock].high_limit and curr_data[stock].day_open > curr_data[stock].low_limit:
                valid_stocks.append(stock)
                if len(valid_stocks) == 5: # 最终持有5只
                    break
                    
    sub_port = context.subportfolios[pindex]
    # 3. 调仓：卖出不在新名单中的股票
    for stock in list(sub_port.long_positions.keys()):
        if stock not in valid_stocks:
            if curr_data[stock].last_price > curr_data[stock].low_limit:
                order_target(stock, 0, pindex=pindex)
                
    # 4. 买入新名单股票（等权重分配可用资金）
    sub_port = context.subportfolios[pindex] # 刷新账户状态
    if len(valid_stocks) > 0:
        target_value = sub_port.total_value / len(valid_stocks)
        for stock in valid_stocks:
            if curr_data[stock].last_price < curr_data[stock].high_limit:
                order_target_value(stock, target_value, pindex=pindex)

def trade_blue_chips(context):
    """ 子策略三：绩优白马股策略 (pindex=2) """
    pindex = 2
    # 1. 筛选高ROE(>15%)且净利润增长率(>10%)的白马股
    q = query(valuation.code).filter(
        valuation.code.in_(g.all_stocks),
        indicator.roe > 15,
        income.np_parent_company_owners_growth_rate > 10
    ).limit(10)
    df = get_fundamentals(q)
    target_list = list(df['code'])
    
    curr_data = get_current_data()
    sub_port = context.subportfolios[pindex]
    
    # 2. 调仓卖出
    for stock in list(sub_port.long_positions.keys()):
        if stock not in target_list:
            if curr_data[stock].last_price > curr_data[stock].low_limit:
                order_target(stock, 0, pindex=pindex)
                
    # 3. 等权重买入
    sub_port = context.subportfolios[pindex]
    if len(target_list) > 0:
        each_value = sub_port.total_value / len(target_list)
        for stock in target_list:
            if not curr_data[stock].paused and curr_data[stock].last_price < curr_data[stock].high_limit:
                order_target_value(stock, each_value, pindex=pindex)

三、关键API与技术解析

set_subportfolios 资金分配：
在 initialize 中，我们通过传入一个 SubPortfolioConfig 列表，将初始资金按 0.3 : 0.4 : 0.3 的比例分配给三个子账户。在后续的 order 系列函数中，必须指定 pindex 参数（如 pindex=1），否则订单会默认发送到第一个子账户，导致资金占用混乱。
get_current_data() 涨跌停与状态过滤：
在小市值和白马股策略中，频繁调仓容易遇到涨跌停。代码中通过 curr_data[stock].high_limit 和 curr_data[stock].low_limit 限制了买入和卖出动作。如果个股已经封死跌停，则不发送卖单，避免废单占用系统资源。
check_stop_loss 盘中动态止损：
通过 run_daily(check_stop_loss, time='every_bar')，策略在盘中每一分钟都会遍历三个子账户的所有持仓。利用 pos.avg_cost（持仓均价）与当前最新价计算浮动盈亏，一旦亏损突破 -8%，立即执行市价单平仓，实现多策略统一的底层风控。

问题描述

解决方案

一、 核心设计思路

二、 聚宽多策略组合完整源码

三、 关键API与技术解析

相关推荐

聚宽企业价值倍数(EV/EBITDA)选股策略

聚宽基本面：存货周转率选股

包含未来函数的量化策略能作为自选股池吗？ | JoinQuant 避坑指南

聚宽核心对象 Position 详解：精准管理持仓成本与数量

聚宽因子数据处理：使用 standardlize 函数进行 Z-Score 标准化

聚宽股指期货与现货套利策略实现

一、核心设计思路

二、聚宽多策略组合完整源码

三、关键API与技术解析