如何在聚宽（JoinQuant）中构建小市值+流动性+涨停基因的周调仓选股策略？

策略设计思路

本策略旨在捕捉具有“涨停基因”的小市值活跃股票。通过结合基本面、流动性、历史价格异动以及风险控制，构建一个稳健的量化选股策略：

1. 基础过滤：剔除 ST、*ST 股票，剔除停牌股，确保交易安全性。
2. 小市值因子：筛选总市值最小的前 100 只股票，获取小市值溢价。
3. 流动性过滤：要求过去 20 个交易日的日均成交额达到一定标准（如大于 2000 万），避免买入无流动性的僵尸股。
4. 涨停基因筛选：在上述候选股中，筛选过去 20 个交易日内至少出现过 1 次涨停的股票。涨停代表资金关注度和短期爆发力。
5. 行业分散约束：为防止持仓过度集中于单一行业，限制同一申万一级行业买入的股票数量不超过 2 只。
6. 周调仓机制：每周第一个交易日开盘进行调仓，卖出不在选股池的股票，等权重买入新入选的股票。

聚宽策略源码实现

以下是基于聚宽 API 编写的完整策略代码。请在聚宽回测环境中运行：

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

def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉比error级别低的log
    log.set_level('order', 'error')
    
    # 策略参数设置
    g.limit_days = 20          # 考察过去多少个交易日的涨停记录和成交额
    g.min_amount = 20000000    # 20日日均成交额门槛（2000万）
    g.max_hold_count = 10      # 最大持仓股票数
    g.industry_limit = 2       # 单个行业最大持仓数
    
    # 设定每周第一个交易日运行调仓
    run_weekly(my_trade, weekday=1, time='09:30')

def before_trading_start(context):
    # 获取当前可交易的 A 股列表
    g.all_stocks = list(get_all_securities(['stock'], date=context.current_dt.date()).index)

def my_trade(context):
    # 1. 获取选股池
    target_list = get_stock_pool(context)
    log.info(f"今日选股池股票数量: {len(target_list)}")
    
    # 2. 行业分散过滤
    final_buy_list = apply_industry_limit(target_list, context.current_dt.date())
    log.info(f"行业分散后最终拟买入列表: {final_buy_list}")
    
    # 3. 执行调仓：卖出不在新买入列表中的持仓
    current_positions = list(context.portfolio.positions.keys())
    for stock in current_positions:
        if stock not in final_buy_list:
            order_target(stock, 0)
            log.info(f"卖出股票: {stock}")
            
    # 4. 买入新股票（等权重分配资金）
    if len(final_buy_list) > 0:
        # 计算可用资金
        available_cash = context.portfolio.available_cash
        # 确定需要新买入的股票
        to_buy = [s for s in final_buy_list if s not in context.portfolio.positions]
        if len(to_buy) > 0:
            value_per_stock = available_cash / len(to_buy)
            for stock in to_buy:
                order_target_value(stock, value_per_stock)
                log.info(f"买入股票: {stock}, 目标金额: {value_per_stock:.2f}")

def get_stock_pool(context):
    today = context.current_dt.date()
    
    # 1. 过滤 ST、退市、停牌股
    current_data = get_current_data()
    filter_stocks = []
    for stock in g.all_stocks:
        if not current_data[stock].is_st and not current_data[stock].paused:
            # 确保已上市
            info = get_security_info(stock)
            if info and (today - info.start_date).days > 365: # 上市满一年
                filter_stocks.append(stock)
                
    # 2. 获取市值数据，筛选市值最小的前 100 只
    q = query(
        valuation.code, valuation.market_cap
    ).filter(
        valuation.code.in_(filter_stocks)
    ).order_by(
        valuation.market_cap.asc()
    ).limit(100)
    df_valuation = get_fundamentals(q, date=today)
    small_cap_stocks = list(df_valuation['code'])
    
    # 3. 过滤流动性（20日日均成交额 > 2000万）及筛选历史涨停基因
    valid_stocks = []
    for stock in small_cap_stocks:
        # 获取过去 20 天的日线数据（包含收盘价、成交额、涨停价）
        hist = attribute_history(stock, count=g.limit_days, unit='1d', fields=['close', 'money', 'high_limit', 'pre_close'], skip_paused=True)
        if len(hist) < g.limit_days:
            continue
            
        # 流动性检查
        mean_money = hist['money'].mean()
        if mean_money < g.min_amount:
            continue
            
        # 涨停基因检查：判断过去 20 天内是否有收盘价等于涨停价的记录
        # 考虑前复权价格精度，使用接近判断
        is_limit_up = (hist['close'] >= hist['high_limit'] - 0.01)
        if is_limit_up.any():
            valid_stocks.append(stock)
            
    return valid_stocks[:g.max_hold_count]

def apply_industry_limit(stock_list, date):
    """行业分散约束：同一申万一级行业股票不超过 g.industry_limit 只"""
    if not stock_list:
        return []
        
    # 获取股票的行业分类
    industry_dict = get_industry(stock_list, date=date)
    
    selected_stocks = []
    industry_counter = {}
    
    for stock in stock_list:
        # 获取申万一级行业代码
        sw_info = industry_dict.get(stock, {}).get('sw_l1', None)
        sw_code = sw_info['industry_code'] if sw_info else 'unknown'
        
        current_count = industry_counter.get(sw_code, 0)
        if current_count < g.industry_limit:
            selected_stocks.append(stock)
            industry_counter[sw_code] = current_count + 1
            
        if len(selected_stocks) >= g.max_hold_count:
            break
            
    return selected_stocks

核心 API 函数解析

• get_fundamentals：用于获取股票的市值数据（valuation.market_cap），实现小市值股票的初筛。
• attribute_history：获取历史行情数据。通过提取 high_limit（涨停价）和 close（收盘价），精准识别个股在过去一段时间内是否触发过涨停。
• get_industry：获取股票所属的行业分类（如申万一级行业 sw_l1），配合计数器实现行业分散，降低行业集中度带来的系统性风险。
• run_weekly：定时运行函数，设定每周第一个交易日开盘时执行调仓，降低高频交易的摩擦成本。