2424 季频质量低波涨停追踪策略 大容量低回撤价值投资-排除小市值因子

核心逻辑：

1. 因子选股模型：

title 因子权重 "盈利能力(ROIC+毛利率)" : 50
"低估值(营收/市值)" : 25
"低波动(120日方差)" : 25

• 筛选流程：
  1. 全市场剔除 ST / 次新股
  2. 因子标准化后综合评分排序
  3. 选取 Top40 且排除近 20 日持仓过的标的

2. 交易机制：

• 季度调仓：每年 1/4/7/10 月首个交易日执行
• 仓位管理：等权分配 40 只股票
• 特殊处理：
  • 涨停持仓股次日观察至 14:00，未封板则止盈
  • 历史持仓冷冻期 20 日，避免频繁交易

3. 风控体系：

• 零滑点假设下的理想化交易环境
• 涨停板特殊退出机制
• 低波动因子控制组合整体风险

核心指标：

• 质量得分 = (ROIC_TTM + 毛利率) - 市销率 - 120 日波动
• 涨停强度：收盘价 / 涨停价 ≥ 98%
• 冷冻周期：20 个交易日

 

策略特点：

1. 低频高效调仓：季度调仓降低换手率，适应机构资金规模
2. 因子正交设计：盈利 + 估值 + 波动三因子互不相关
3. 涨停板溢价捕获：允许涨停股继续持有至次日观察期
4. 防止过度交易：
   • 历史持仓冷冻机制
   • 严格季度调仓纪律
5. 组合分散特性：40 只等权持仓降低个股特异性风险

关键优化点：

1. 因子归一化处理：采用 MinMaxScaler 消除量纲差异
2. 交易成本极致压缩：0.012% 佣金 + 0 滑点设置
3. 特殊行情免疫：剔除最近 20 日所有持仓过的标的
4. 信息隔离机制：盘前 9:05 完成股票池准备，避免盘中干扰

策略代码

#导入函数库
from jqdata import *
from jqfactor import get_factor_values
from jqlib.technical_analysis import *
import numpy as np
import pandas as pd
import statsmodels.api as sm
import datetime as dt
from sklearn.preprocessing import MinMaxScaler

#初始化函数 
def initialize(context):
    # 设定基准
    set_benchmark('000905.XSHG')
    # 用真实价格交易
    set_option('use_real_price', True)
    # 打开防未来函数
    set_option("avoid_future_data", True)
    # 将滑点设置为0
    set_slippage(FixedSlippage(0))
    # 设置交易成本万分之1.2，不同滑点影响可在归因分析中查看
    set_order_cost(OrderCost(open_tax=0, close_tax=0.001, open_commission=0.00012, close_commission=0.00012, close_today_commission=0, min_commission=5),type='stock')
    # 过滤order中低于error级别的日志
    log.set_level('order', 'error')
    log.set_level('system', 'error')
    #初始化全局变量
    g.stock_num = 40
    g.limit_up_list = [] #记录持仓中涨停的股票
    g.hold_list = [] #当前持仓的全部股票
    g.history_hold_list = [] #过去一段时间内持仓过的股票
    g.not_buy_again_list = [] #最近买过且涨停过的股票一段时间内不再买入
    g.limit_days = 20 #不再买入的时间段天数
    g.target_list = [] #开盘前预操作股票池
    # 设置交易运行时间
    run_daily(prepare_stock_list, time='9:05', reference_security='000300.XSHG')
    run_monthly(adjust_position, 1, time='9:30', reference_security='000300.XSHG')
    run_daily(check_limit_up, time='14:00', reference_security='000300.XSHG') #检查持仓中的涨停股是否需要卖出
    run_monthly(print_position_info, 1, time='15:10', reference_security='000300.XSHG')

关键函数解锁后查看：

#1-3 准备股票池
def prepare_stock_list(context):
    #获取已持有列表
    g.hold_list= []
    for position in list(context.portfolio.positions.values()):
        stock = position.security
        g.hold_list.append(stock)
    #获取最近一段时间持有过的股票列表
    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:
        for stock in hold_list:
            temp_set.add(stock)
    g.not_buy_again_list = list(temp_set)
    #获取昨日涨停列表
    if g.hold_list != []:
        df = get_price(g.hold_list, end_date=context.previous_date, frequency='daily', fields=['close','high_limit'], count=1, panel=False, fill_paused=False)
        df = df[df['close'] == df['high_limit']]
        g.high_limit_list = list(df.code)
    else:
        g.high_limit_list = []


#1-5 整体调整持仓
def adjust_position(context):
    if context.previous_date.month not in [1,4,7,10]:
        return
    # 获取应买入列表 
    g.target_list = get_stock_list(context)
    #截取不超过最大持仓数的股票量
    g.target_list = g.target_list[:min(g.stock_num, len(g.target_list))]
    #调仓卖出
    for stock in g.hold_list:
        if (stock not in g.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 = len(g.target_list)
    if target_num > position_count:
        value = context.portfolio.cash / (target_num - position_count)
        for stock in g.target_list:
            if context.portfolio.positions[stock].total_amount == 0:
                if open_position(stock, value):
                    if len(context.portfolio.positions) == target_num:
                        break

#1-6 调整昨日涨停股票
def check_limit_up(context):
    now_time = context.current_dt
    if g.high_limit_list != []:
        #对昨日涨停股票观察到尾盘如不涨停则提前卖出，如果涨停即使不在应买入列表仍暂时持有
        for stock in g.high_limit_list:
            current_data = get_price(stock, end_date=now_time, frequency='1m', fields=['close','high_limit'], skip_paused=False, fq='pre', count=1, panel=False, fill_paused=True)
            if current_data.iloc[0,0] < current_data.iloc[0,1]:
                log.info("[%s]涨停打开，卖出" % (stock))
                position = context.portfolio.positions[stock]
                close_position(position)
            else:
                log.info("[%s]涨停，继续持有" % (stock))

#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 != 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 != None:
		if order.status == OrderStatus.held and order.filled == order.amount:
			return True
	return False

#4-1 打印每日持仓信息
def print_position_info(context):
	c = get_current_data()
	positions_dict = context.portfolio.positions
	for position in list(positions_dict.values()):
	    log.info("当前持仓：{0}:{1}, 市值：{2}, 盈利：{3}%, 建仓时间：{4}".format(c[position.security].name, position.security[:6], round(position.value,0), round((position.value-(position.avg_cost*position.total_amount))/(position.avg_cost*position.total_amount)*100,1), position.init_time))
	log.info('#########################################################################################\n\n')

2025-02-25