2378 国盛多因子系列选股体系和基于Lasso的收益预测模型策略（9大类共59个因子）多因子综合评分轮动python量化策略

1. 九维度因子体系：

• 构建波动性 / 成长 / 反转 / 杠杆 / 红利 / 价值 / 流动性 / 盈利 / 质量 9 大类因子池
• 每类精选核心指标（如成长类包含单季度 EPS / 净利润 / 经营利润 / 现金流增速）
• 采用分层赋权法：每类因子等权重，同类子因子等权重
• 动态筛选各维度前 10% 个股进行综合评分

2. 动态评分机制：

• 波动性：选取 1 月期隐含波动率分位 ivr_1m
• 价值因子：综合 BP/CFP/SP/ 价值残差四项指标
• 质量因子：考量三费营收比 / 资产周转率 / 固定资产占比
• 每周计算个股综合得分，选取 TOP10 构建组合

3. 交易风控体系：

• 严格过滤次新股（上市 < 375 天）
• 排除 ST/*ST / 退市 / 科创板 / 北交所股票
• 剔除 7 日内存在停牌的个股
• 等金额分配持仓（单只占比 10%）
• 每周一开盘完成调仓

4. 特殊处理机制：

• 涨停股自动排除（9:40 价非涨停）
• 异常值三重校验（因子值 ±3σ 截断）
• 流动性保障（20 日成交额前 80%）
• 动态手续费适配（区分不同年份费率）

策略代码

# 回测资金 1000000

# 导入函数库
from jqdata import *
import numpy as np
import pandas as pd
from six import BytesIO, StringIO
import json 

def set_params():
    # g.benchmark = '000300.XSHG' # (HS300) 
    g.benchmark = '000001.XSHG' # (SZZS)
    g.buy_list = []
    g.stock_num = 10   # 持仓最大股票数
    g.sample_windows = 252 # 样本窗口长度
    
# 初始化函数，设定基准等等
def initialize(context):
    set_params()
    
    set_benchmark(g.benchmark)                  # 设定沪深300作为基准
    set_option('use_real_price', True)          # 开启动态复权模式(真实价格)
    log.info('初始函数开始运行且全局只运行一次')# 输出内容到日志 log.info()
    log.set_level('order', 'error')             # 过滤掉order系列API产生的比error级别低的log
    
    ## 运行函数（reference_security为运行时间的参考标的；传入的标的只做种类区分，因此传入'000300.XSHG'或'510300.XSHG'是一样的）
    run_weekly(before_market_open,  weekday=1, time='before_open', reference_security='000300.XSHG')  # 开盘前运行
    run_weekly(market_open,         weekday=1, time='9:31',        reference_security='000300.XSHG')  # 开盘时运行
    run_weekly(after_market_close,  weekday=1, time='after_close', reference_security='000300.XSHG')  # 收盘后运行

def set_trader(dt):
    # 设置交易费率
    set_slippage(FixedSlippage(0)) # 将滑点设置为0
    # 根据不同的时间段设置手续费
    if dt>datetime.datetime(2013,1, 1):
        # 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
        set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    elif dt>datetime.datetime(2011,1, 1):
        set_order_cost(OrderCost(close_tax=0.001, open_commission=0.001, close_commission=0.001, min_commission=5), type='stock')

关键函数解锁后查看：

## 开盘时运行函数
def market_open(context):
    log.info('函数运行时间(market_open):'+str(context.current_dt.time()))
    
    hold_list=list(context.portfolio.positions.keys()) 
    log.info('g.hold_list:' + str(hold_list))
    buy_list = g.buy_list
    log.info('buy_list:'+ str(buy_list)) 
    
    for s in hold_list:
        if s not in buy_list:
            order_target_value(s, 0)
    
    capital_unit = context.portfolio.total_value / len(buy_list)
    #capital_unit = context.portfolio.available_cash / len(buy_list)
    log.info('capital_unit:' + str(capital_unit))
    for s in buy_list:
        order_target_value(s, capital_unit)

## 收盘后运行函数
def after_market_close(context):
    log.info(str('函数运行时间(after_market_close):'+str(context.current_dt.time())))
    #得到当天所有成交记录
    trades = get_trades()
    for _trade in trades.values():
        log.info('成交记录：'+str(_trade))
    log.info('一天结束')
    log.info('##############################################################')

# 过滤【次新 + 科创北交 + ST退市 + 停牌】
def filter_stocks(all_security_df, yesterday, current_dt):
    all_security_df['index'] = all_security_df.index

    # 过滤次新股
    filter_fn = lambda t: (yesterday - t) >= datetime.timedelta(days=375)
    all_security_df = all_security_df[all_security_df['start_date'].apply(filter_fn)]

    # 过滤科创北交
    filter_fn = lambda t: t[0] != '4' and t[0] != '8' and t[:2] != '68'
    all_security_df = all_security_df[all_security_df['index'].apply(filter_fn)]

    # 过滤ST及其他具有退市标签的股票
    st_df = get_extras('is_st', all_security_df.index, start_date=current_dt, end_date=current_dt, df=True).T
    st_list = st_df[st_df==True].dropna().index
    filter_fn = lambda t: ("ST" in t or "*" in t or "退" in t)
    all_security_df = all_security_df[(~all_security_df['index'].isin(st_list)) & (~all_security_df['display_name'].apply(filter_fn))]
    
    # 过滤停牌
    susp_df = get_price(all_security_df.index.tolist(), \
                        end_date=current_dt, count=1, frequency='daily', \
                        fields='paused', panel=False)
    unsusp_stocks = susp_df[susp_df['paused'] < 1]["code"].tolist() # 得到当日未停牌股票代码list:
    days=7 # 过滤出前7天没有停牌过的Stocks
    feasible_stocks=[s for s in unsusp_stocks if sum(attribute_history(s, days, unit='1d',fields=('paused'),skip_paused=False))[0]==0]
    return feasible_stocks

2025-02-23