聚宽平台如何编写带净值 MA20 风控的小市值动态调仓策略？（附 Python 源码）

聚宽小市值策略风控：净值 MA20 动态调仓实现方案

小市值策略（Small-Cap Strategy）在 A 股历史中具有极强的超额收益，但其波动大、回撤深的特点也让许多投资者望而却步。为了控制回撤，引入**策略净值均线风控（Equity Curve MA Filter）**是一种非常行之有效的系统性风控手段。

本方案将展示如何在聚宽（JoinQuant）平台中：

1. 跟踪中证 1000 或国证 2000 等中小盘指数，动态获取成分股。
2. 筛选出成分股中流通市值最小的 5 只股票进行调仓。
3. 每日计算策略净值，当**策略净值跌破 20 日均线（MA20）**时，触发风控一键清仓暂停交易。
4. 当策略净值连续 2 日站回 20 日均线时，恢复小市值选股与交易。

核心 API 介绍

• get_index_stocks(index_symbol, date)：获取指定指数在特定日期的成分股列表。
• get_fundamentals(query_object, date)：查询财务数据，用于获取股票的流通市值（valuation.circulating_market_cap）。
• set_option('use_real_price', True)：开启动态复权模式，确保回测与模拟交易价格真实。
• run_daily(func, time)：每日定时运行风控与交易逻辑。

策略 Python 源码

请在聚宽策略编辑框中复制以下完整代码进行回测：

# 导入聚宽函数库
import jqdata
import numpy as np
import pandas as pd

def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    
    # 策略参数设置
    g.index_code = '000852.XSHG' # 跟踪中证1000指数
    g.hold_num = 5               # 持股数量
    g.ma_window = 20             # 净值均线周期
    
    # 状态变量
    g.equity_history = []        # 记录每日历史净值
    g.is_trading = True          # 当前是否允许交易
    g.above_ma_days = 0          # 连续站上均线的天数
    
    # 运行时间设定
    # 每天 09:00 盘前进行风控状态判断
    run_daily(before_market_open, time='09:00')
    # 每天 09:30 开盘进行调仓交易
    run_daily(market_open, time='09:30')

def before_market_open(context):
    # 1. 记录前一日的策略总资产（净值）
    current_equity = context.portfolio.total_value
    g.equity_history.append(current_equity)
    
    # 保持历史净值长度，避免内存占用过大
    if len(g.equity_history) > g.ma_window * 2:
        g.equity_history.pop(0)
        
    # 2. 净值均线风控逻辑
    if len(g.equity_history) >= g.ma_window:
        # 计算 20 日净值均线
        ma20 = np.mean(g.equity_history[-g.ma_window:])
        last_equity = g.equity_history[-1]
        
        if g.is_trading:
            # 如果当前在交易，但净值跌破 MA20 -> 触发风控暂停交易
            if last_equity < ma20:
                g.is_trading = False
                g.above_ma_days = 0
                log.warn("【风控预警】策略净值跌破 MA20，触发风控，准备清仓暂停交易！")
        else:
            # 如果当前暂停交易，检查是否连续 2 日站回 MA20
            if last_equity > ma20:
                g.above_ma_days += 1
                if g.above_ma_days >= 2:
                    g.is_trading = True
                    log.info("【风控解除】策略净值连续 2 日站回 MA20，恢复小市值交易！")
            else:
                g.above_ma_days = 0
    else:
        # 历史数据不足 20 天时，默认允许交易
        g.is_trading = True

def market_open(context):
    # 1. 若触发风控暂停交易，则一键清仓空仓
    if not g.is_trading:
        for stock in list(context.portfolio.positions.keys()):
            if context.portfolio.positions[stock].closeable_amount > 0:
                order_target(stock, 0)
                log.info("【风控执行】卖出清仓: %s" % stock)
        return
        
    # 2. 正常交易状态下，执行小市值轮动调仓
    # 获取中证1000成分股
    stocks = get_index_stocks(g.index_code, date=context.current_dt)
    
    # 过滤停牌、ST、退市及上市未满1年的股票
    current_data = get_current_data()
    valid_stocks = []
    for s in stocks:
        if not current_data[s].paused and not current_data[s].is_st and '退' not in current_data[s].name:
            # 过滤上市未满1年的新股
            info = get_security_info(s)
            if info and (context.current_dt.date() - info.start_date).days > 365:
                valid_stocks.append(s)
                
    # 查询流通市值
    q = query(
        valuation.code,
        valuation.circulating_market_cap
    ).filter(
        valuation.code.in_(valid_stocks)
    ).order_by(
        valuation.circulating_market_cap.asc() # 按流通市值升序排列
    ).limit(g.hold_num)
    
    df = get_fundamentals(q, date=context.previous_date)
    target_list = list(df['code'])
    
    # 3. 调仓执行
    # 卖出不在目标持仓中的股票
    for stock in list(context.portfolio.positions.keys()):
        if stock not in target_list:
            order_target(stock, 0)
            log.info("调仓卖出: %s" % stock)
            
    # 买入目标股票
    position_count = len(context.portfolio.positions)
    if len(target_list) > 0:
        # 均分可用资金
        available_cash = context.portfolio.available_cash
        buy_cash = available_cash / max(1, (g.hold_num - position_count))
        for stock in target_list:
            if stock not in context.portfolio.positions:
                order_value(stock, buy_cash)
                log.info("调仓买入: %s" % stock)

策略逻辑深度解析

1. 中小盘指数跟踪：
   策略默认使用 000852.XSHG（中证 1000 指数）作为选股母池。中证 1000 指数代表了 A 股中流动性好、市值偏小的中小盘股票，是小市值策略极佳的选股池。你也可以将其修改为 399303.XSHE（国证 2000 指数）以进一步下沉市值。

2. 严格的股票过滤：
   小市值策略极易踩雷 ST 或退市股。代码中通过 get_current_data() 动态获取股票状态，剔除了停牌股、ST 股、退市整理期股，并过滤了上市未满 1 年的新股，避免次新股剧烈波动对策略的影响。

3. 净值 MA20 均线风控：

   • 避险机制：当市场发生系统性暴跌（如小市值踩踏）时，策略净值会迅速回落。一旦前一日收盘净值跌破 20 日均线，策略在次日开盘（09:30）立即执行一键清仓，锁定利润，空仓观望。
   • 恢复机制：空仓期间，策略净值（即持有的现金总额）保持平性。当市场企稳，20 日均线逐渐下行，策略净值重新站回 MA20 且连续 2 个交易日确认站稳后，策略重新激活，买入最新筛选出的小市值股票。这有效防止了均线附近的“反复打脸”震荡。