2350 多因子成长股月轮动策略 一种新思路，超额376%！ 40560 量化策略 自动化交易策略代码

策略名称：多因子成长股月轮动策略

---

策略说明：

本策略基于多因子模型，结合成长股轮动和涨停板管理逻辑，通过筛选质量因子优异、成长性强的股票，并在每月调仓时动态管理持仓。策略的核心思想是捕捉成长股的长期潜力，同时通过涨停板监控控制短期风险。

---

策略核心逻辑：

1. 多因子选股模型：

   • 因子选择：

     • 扣非利润比（衡量盈利质量）

     • 短期/长期预期利润增长率（评估成长性）

     • 净利润增长率（反映盈利能力）

     • 5年盈利增长率（衡量长期成长性）

   • 线性加权：通过历史回测优化的权重（0.2, 0.4, 0.2, 0.2）对四因子进行加权，生成综合评分。

   • 筛选规则：

     • 选取评分最高的前8%股票，剔除科创板、ST股和停牌股。

     • 按流通市值升序排序，优先选择小市值标的。

2. 动态调仓机制：

   • 月度调仓：每月第一个交易日开盘时，卖出非目标股（保留昨日涨停股），买入新筛选的股票。

   • 市值控制：持仓数量固定为8只，分散配置成长股。

   • 涨停监控：每日14:00检查昨日涨停股，若今日未封板则卖出。

3. 风险控制：

   • 硬性过滤：

     • 排除科创板、ST股、上市未满375天次新股。

     • 剔除停牌、涨停/跌停股。

   • 趋势过滤：通过20日均线斜率（OLS回归）剔除下跌趋势明显的股票。

   • 防重复买入：标记20天内已操作股票，避免短期重复交易。

---

策略特点：

• 因子驱动：结合质量类与成长类因子，兼顾盈利质量与成长潜力。

• 小市值偏好：通过流通市值排序，捕捉小盘成长股超额收益。

• 低频调仓：月度主调仓+日度辅助交易，平衡交易频率与冲击成本。

• 自动化风控：涨停板管理与实时持仓监控，降低极端行情风险。

---

适用场景：

• 市场环境：适用于成长股表现优异的牛市或结构性行情。

• 资金规模：适合中小资金（1亿以内），小市值策略容量有限。

• 投资者类型：适合追求长期收益、能承受中高波动的量化投资者。

---

风险提示：

1. 小市值风险：小盘股流动性较低，极端行情下易出现大幅波动。

2. 因子失效：若市场风格转向价值股，策略可能阶段性跑输基准。

3. 涨停板依赖：涨停股次日溢价效应不稳定，需动态跟踪市场情绪。

4. 数据依赖：依赖财务数据时效性，定期报告期可能存在信息滞后。

---

代码结构：

模块	函数	功能说明
初始化	initialize	设定基准、交易参数，定义全局变量，绑定定时任务
数据准备	prepare_stock_list	更新持仓列表，识别昨日涨停股，预加载当日目标股池
核心选股	get_stock_list	多因子评分、市值排序、基本面过滤
月度调仓	monthly_adjustment	主调仓逻辑，卖出非目标股，买入新标的，推送交易信号
日常交易	check_limit_up	尾盘涨停监控，破板则卖出
风控模块	filter_*	5类过滤规则（停牌/ST/次新/涨跌停等）
交易执行	open_position/close_position	按等金额分配买入目标股，避免单一个股过度集中

---

优化建议：

1. 因子迭代：

   • 引入动量因子（如20日收益率）或波动率因子，提升组合弹性。

   • 测试换手率、机构持股比例等另类因子，优化评分模型。

2. 仓位管理：

   • 动态调整持仓数量（如根据市场波动率缩放）。

   • 加入止损机制（如个股回撤超10%强制平仓）。

3. 交易优化：

   • 使用VWAP/TWAP算法拆单，降低冲击成本。

   • 结合Level2盘口数据，优化涨停板判断逻辑。

4. 风控增强：

   • 设置最大回撤阈值，触发后自动降仓。

   • 监控市场广度指标（如涨跌停家数），极端市况下切换保守模式。

---

总结：

本策略通过多因子模型精选成长股，结合月频调仓与实时风控，在控制风险的前提下捕捉小盘成长股超额收益。策略逻辑清晰，模块化设计便于迭代，适合作为量化多头策略的核心组件。实盘前需重点测试因子稳定性与小市值流动性风险。

策略代码

# 标题：新策略，一种新思路，超额376%！

import statsmodels.api as sm
from jqdata import *
from jqfactor import get_factor_values


# 初始化函数
def initialize(context):
    # 设定基准
    set_benchmark('000905.XSHG')
    # 用真实价格交易
    set_option('use_real_price', True)
    # 打开防未来函数
    set_option("avoid_future_data", True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')

    # 初始化全局变量
    g.stock_num = 8#持股数
    g.limit_days = 20#用来检查最近20天内列表中有涨停的股票

    g.hold_list = []
    g.history_hold_list = []
    g.not_buy_again_list = []
    # 设置交易时间，每天运行
    run_daily(prepare_stock_list, time='9:05', reference_security='000300.XSHG')
    #每天获取g.hold_list已持仓列表、g.not_buy_again_list20天内买过的接下来不买的股的列表、g.high_limit_list昨日涨停列表
    #run_weekly(weekly_adjustment, weekday=1, time='9:30', reference_security='000300.XSHG')
    run_monthly(monthly_adjustment, monthday=1, time='9:30', reference_security='000300.XSHG')
    #每周获取股票列表，去除最近20天内曾经涨停过的和曾经买过的股，去掉下跌趋势明显的
    run_daily(check_limit_up, time='14:00', reference_security='000300.XSHG')
    #把昨日涨停今天没涨停的股票卖出

关键函数解锁：

# 1-3 准备股票池
def prepare_stock_list(context):
    # 获取已持有列表
    g.hold_list = list(context.portfolio.positions)

    # 获取最近一段时间持有过的股票列表
    g.history_hold_list.append(g.hold_list)
    if len(g.history_hold_list) >= g.limit_days:
        g.history_hold_list = g.history_hold_list[-g.limit_days:]
    #
    temp_set = set()
    for hold_list in g.history_hold_list:
        temp_set = temp_set.union(set(hold_list))#通过set.union()去重
    #
    g.not_buy_again_list = list(temp_set)#不买最近20天买过的股票

    # 获取持仓的昨日涨停列表
    g.high_limit_list = []
    if g.hold_list:
        df = get_price(g.hold_list, end_date=context.previous_date, frequency='daily',
                       fields=['close', 'high_limit', 'paused'],
                       count=1, panel=False)
        g.high_limit_list = df.query('close==high_limit and paused==0')['code'].tolist()#paused为0表示不停牌


# 1-4 整体调整持仓  
def monthly_adjustment(context):
    # type: (Context) -> None
    # 获取应买入列表
    target_list = get_stock_list(context)#获取股票列表
    #
    target_list = filter_paused_stock(target_list)
    target_list = filter_limit_stock(context, target_list)

    # target_list中，去除最近20天内曾经涨停过的和曾经买过的股
    recent_limit_up_list = get_recent_limit_up_stock(context, target_list, g.limit_days)
    black_list = list(set(g.not_buy_again_list).intersection(set(recent_limit_up_list)))
    target_list = [stock for stock in target_list if stock not in black_list]

    # 最近20天的MA20的斜率，去掉下跌趋势明显的，即斜率<-2的
    h_ma = history(20 + 20, '1d', 'close', target_list).rolling(window=20).mean().iloc[20:]#df.rolling(window=?).mean()将DF的最近几个窗口（或值）进行滚动求平均
    #上面取最后20行
    X = np.arange(len(h_ma))#生成0、1、...19的数组
    tmp_target_list = []
    for stock in target_list:
        MA_N_Arr = h_ma[stock].values#得到每个股票最近20天的MA20数值
        MA_N_Arr = MA_N_Arr - MA_N_Arr[0]  # 截距归零，理解成标准化
        slope = round(sm.OLS(MA_N_Arr, X).fit().params[0] * 100, 1)# Statsmodels 中 OLS 回归功能sm.OLS(因变量,自变量)，在 OLS之后调用拟合函数 fit()，
        #才进行回归运算，并且得到RegressionResultsWrapper结果，它包含了这组数据进行回归拟合的结果摘要。调用 params 可以查看计算出的回归系数 b0,b1,…,bn。
        #params[0]是为了去除列表，取具体值。sm.OLS(Y,X).fit().summary()可以看总体回归情况
        #print("测试%s"%sm.OLS(MA_N_Arr, X).fit().summary())
        remove_it = False
        if slope < -2:
            if stock not in g.hold_list:
                print('{}下降趋势明显，切勿开仓'.format(stock))
                remove_it = True
        if not remove_it:
            tmp_target_list.append(stock)
    #
    target_list = tmp_target_list
    # 调仓：不在列表，昨日未涨停的持仓票卖出。
    for stock in g.hold_list:
        if (stock not in target_list) and (stock not in g.high_limit_list):
            log.info("卖出[%s]" % stock)
            position = context.portfolio.positions[stock]
            close_position(position)
        else:
            log.info("已持有[%s]" % stock)

    position_count = len(context.portfolio.positions)
    target_num = g.stock_num
    if target_num > position_count:
        value =  context.portfolio.available_cash / (target_num - position_count)
        for stock in target_list:
            if stock not in context.portfolio.positions:
                if open_position(stock, value):
                    if len(context.portfolio.positions) >= g.stock_num:
                        break


# 1-5 调整昨日涨停股票
def check_limit_up(context):
    current_data = get_current_data()
    if g.high_limit_list:#if list: list非0非空，则为true
        for stock in g.high_limit_list:
            if current_data[stock].last_price < current_data[stock].high_limit:
                log.info("[%s]涨停打开，卖出" % stock)
                position = context.portfolio.positions[stock]
                close_position(position)
            else:
                log.info("[%s]涨停，继续持有" % stock)


# 2-1 过滤停牌股票
def filter_paused_stock(stock_list):
    current_data = get_current_data()
    return [stock for stock in stock_list if not current_data[stock].paused]


# 2-2 过滤ST及其他具有退市标签的股票
def filter_st_stock(stock_list):
    current_data = get_current_data()
    return [stock for stock in stock_list if not (
            current_data[stock].is_st or
            'ST' in current_data[stock].name or
            '*' in current_data[stock].name or
            '退' in current_data[stock].name)]


# 2-3 获取最近rencent_days个交易日内有涨停的股票列表
def get_recent_limit_up_stock(context, stock_list, recent_days):
    # type: (Context, list, int) -> list
    yesterday = context.previous_date
    h = get_price(stock_list, end_date=yesterday, frequency='daily', fields=['close', 'high_limit', 'paused'],
                  count=recent_days, panel=False)
    s_limit = h.query('close==high_limit and paused==0').groupby('code')['high_limit'].count()
    return s_limit.index.tolist()


# 2-4 过滤涨停的股票
def filter_limit_stock(context, stock_list):
    # type: (Context, list) -> list
    current_data = get_current_data()
    holdings = list(context.portfolio.positions)
    return [stock for stock in stock_list if (stock in holdings) or
            current_data[stock].low_limit < current_data[stock].last_price < current_data[stock].high_limit]


# 2-6 过滤科创板
def filter_kcb_stock(stock_list):
    return [stock for stock in stock_list if not stock.startswith('68')]


# 3-1 交易模块-自定义下单
def order_target_value_(security, value):
    if value == 0:
        log.debug("Selling out %s" % security)
    else:
        log.debug("Order %s to value %f" % (security, value))
    return order_target_value(security, value)


# 3-2 交易模块-开仓
def open_position(security, value):
    _order = order_target_value_(security, value)
    if _order is not None and _order.filled > 0:
        return True
    return False


# 3-3 交易模块-平仓
def close_position(position):
    security = position.security
    _order = order_target_value_(security, 0)  # 可能会因停牌失败
    if _order is not None:
        if _order.status == OrderStatus.held and _order.filled == _order.amount:
            return True
    return False

2025-02-23