2505 动态调仓多因子策略 国九条小市值+搅屎棍选股策略2.1

策略核心思想

1. 多因子选股：综合 ROE、ROA 等基本面指标筛选优质股票
2. 动态仓位调整：根据市场波动情况动态调整持仓数量
3. 严格风控机制：设置多重止损条件保护资金安全
4. 季节性空仓：在特定月份（1 月、4 月）保持空仓

 

策略详细逻辑

1. 股票池筛选

• 基础股票池为中小板指数成分股（399101.XSHE）
• 多重过滤条件：
  • 过滤停牌、ST、退市股票
  • 排除科创板、北交所股票（代码以 68、8、4 开头）
  • 排除次新股（上市不满 375 天）
  • 排除当日涨停 / 跌停股票（新开仓时）
  • 审计意见过滤（排除近 3 年有非标准审计意见的股票）

2. 选股因子

• 核心财务指标：
  • ROE > 15%
  • ROA > 10%
• 市值因子：
  • 优先选择小市值股票（按市值升序排列）
• 最终筛选出 3 倍于目标持仓数量的候选股票

3. 动态仓位管理

• 根据中小板指数 10 日乖离率动态调整持仓数量：
  • 乖离率≥5%：持仓 2 只
  • 2%≤乖离率 < 5%：持仓 2 只
  • -2%≤乖离率 < 2%：持仓 3 只
  • -5%≤乖离率 <-2%：持仓 4 只
  • 乖离率 <-5%：持仓 5 只
• 在 1 月和 4 月完全空仓，持有货币 ETF（511880.XSHG）

4. 交易执行

• 每周一进行主调整（调仓频率可配置）
• 买入逻辑：
  • 等权重分配资金
  • 对科创板股票采用限价单处理
• 卖出逻辑：
  • 非目标持仓股票立即卖出
  • 昨日涨停股票特殊处理（观察至 14:00）
  • 多重止损条件：
    • 个股止损：亏损超过 9% 止损
    • 个股止盈：盈利超过 100% 止盈
    • 系统性止损：指数平均跌幅超过 5% 时清仓

5. 特殊处理

• 对昨日涨停股票：
  • 保留至下午 14:00
  • 14:00 后若打开涨停则卖出
• 审计意见缓存机制：
  • 每年 4 月最后一个交易日更新审计意见缓存
  • 劳动节后首个交易日清空缓存

策略特点

1. 动态适应性：根据市场波动自动调整风险暴露
2. 严格风控：多重止损机制保护资金安全
3. 基本面导向：以财务质量为核心选股标准
4. 季节性防御：在年报季等高风险时段空仓
5. 涨停板策略：对强势股采用特殊处理方式

风险提示

1. 小市值风险：偏好小市值股票可能放大波动
2. 调仓频率风险：周频调仓可能增加交易成本
3. 空仓期机会成本：1 月、4 月空仓可能错过行情
4. 财务数据滞后：基本面因子依赖历史数据

 

适用市场环境

• 市场波动适中的环境
• 基本面因子有效的市场
• 小盘股表现优于大盘股的市场
• 非 1 月、4 月的交易时段

参数配置建议

• g.pass_months：空仓月份设置
• g.stoploss_strategy：止损策略选择（1 - 个股止损，2 - 系统止损，3 - 两者结合）
• g.stoploss_limit：个股止损阈值
• g.stoploss_market：系统止损阈值
• g.highest：股价上限过滤
• g.period：调仓周期（当前为周频）

优化方向

1. 增加更多有效的选股因子
2. 优化动态仓位调整算法
3. 改进止损逻辑（如加入移动止盈）
4. 调整空仓月份设置
5. 优化交易执行算法（减少冲击成本）

from jqdata import *
from jqfactor import *
import numpy as np
import pandas as pd
from datetime import time
from jqdata import finance
from jqdata import get_trade_days

# 初始化函数
def initialize(context):
    set_option('avoid_future_data', True)
    set_benchmark('399101.XSHE')
    set_option('use_real_price', True)
    set_slippage(FixedSlippage(6/1000))
    set_order_cost(OrderCost(open_tax=0, close_tax=0.001, open_commission=2.5/10000, close_commission=2.5/10000, close_today_commission=0, min_commission=5), type='stock')
    log.set_level('order', 'error')
    log.set_level('system', 'error')
    log.set_level('strategy', 'debug')
    
    g.trading_signal = True
    g.run_stoploss = True
    g.filter_audit = True
    g.adjust_num = True
    g.hold_list = []
    g.yesterday_HL_list = []
    g.target_list = []
    g.pass_months = [1, 4]
    g.limitup_stocks = []
    g.min_mv = 10
    g.max_mv = 100
    g.stock_num = 5
    g.reason_to_sell = {}
    g.stoploss_strategy = 3
    g.stoploss_limit = 0.09
    g.stoploss_market = 0.05
    g.highest = 50
    g.etf = '511880.XSHG'
    g.day_count = 0
    g.period = 3
    
    run_daily(prepare_stock_list, '9:05')
    run_weekly(weekly_adjustment, 1, '10:30')
    run_daily(trade_afternoon, time='14:00', reference_security='399101.XSHE')
    run_daily(sell_stocks, time='9:55')
    run_daily(close_account, '14:50')
    run_daily(check_and_update_cache, '9:00')
    
    g.audit_cache = {}
    g.need_to_update_after_labor_day = False

# 准备股票池
def prepare_stock_list(context):
    g.hold_list = list(context.portfolio.positions.keys())
    g.limitup_stocks = []
    
    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)
        g.yesterday_HL_list = df[df['close'] == df['high_limit']]['code'].tolist()
    else:
        g.yesterday_HL_list = []
    
    g.trading_signal = today_is_between(context)

# 选股模块
def get_stock_list(context):
    final_list = []
    MKT_index = '399101.XSHE'
    ZXBZ_list = get_index_stocks(MKT_index)
    
    initial_list = filter_stocks(context, ZXBZ_list)
    
    q = query(valuation.code).filter(
        valuation.code.in_(initial_list),
        indicator.roe > 0.15,
        indicator.roa > 0.10
    ).order_by(valuation.market_cap.asc()).limit(g.stock_num * 3)
    
    df = get_fundamentals(q)
    
    if g.filter_audit:
        all_codes = df['code'].tolist()
        audit_results = {code: filter_audit(context, code) for code in all_codes}
        df['audit'] = df['code'].map(audit_results)
        df = df[df['audit']]
        log.info('去除掉了存在审计问题的股票{}只'.format(len(df[~df['audit']]) if not df.empty else 0))
    
    if not df.empty:
        final_list = df['code'].tolist()
        last_prices = history(1, unit='1d', field='close', security_list=final_list)
        final_list = [stock for stock in final_list if stock in g.hold_list or (last_prices[stock][-1] <= g.highest if stock in last_prices else True)]
    
    if not final_list:
        log.info('无适合股票，买入ETF')
        return [g.etf]
    else:
        return final_list

# 整体调整持仓
def weekly_adjustment(context):
    if g.trading_signal:
        g.day_count = 0
        if g.adjust_num:
            new_num = adjust_stock_num(context)
            g.stock_num = new_num
            log.info(f'持仓数量修改为{new_num}')
        g.target_list = get_stock_list(context)[:g.stock_num]
        log.info(str(g.target_list))
        
        sell_list = [stock for stock in g.hold_list if stock not in g.target_list and stock not in g.yesterday_HL_list]
        hold_list = [stock for stock in g.hold_list if stock in g.target_list or stock in g.yesterday_HL_list]
        log.info("卖出[%s]" % (str(sell_list)))
        log.info("已持有[%s]" % (str(hold_list)))
        
        for stock in sell_list:
            order_target_value(stock, 0)
        for stock in hold_list:
            indiv_value = context.portfolio.total_value / g.stock_num
            order_target_value(stock, indiv_value)
        
        buy_list = [stock for stock in g.target_list if stock not in g.hold_list]
        buy_security(context, buy_list, len(buy_list))
    else:
        buy_security(context, [g.etf], 1)
        log.info('该月份为空仓月份，持有银华日利ETF')

关键函数解锁后查看：

# 交易模块-自定义下单
def order_target_value_(security, value):
    try:
        if value == 0:
            pass
        else:
            log.debug("Order %s to value %f" % (security, value))
        return order_target_value(security, value)
    except Exception as e:
        log.error(f"下单 {security} 至价值 {value} 时出错: {e}")
        return None

# 交易模块-开仓
def open_position(security, value):
    try:
        order = order_target_value_(security, value)
        if order is not None and order.filled > 0:
            return True
        return False
    except Exception as e:
        log.error(f"开仓 {security} 价值为 {value} 时出错: {e}")
        return False

# 交易模块-平仓
def close_position(position):
    security = position.security
    try:
        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
    except Exception as e:
        log.error(f"平仓 {security} 时出错: {e}")
        return False

# 买入模块
def buy_security(context, target_list, num):
    position_count = len(context.portfolio.positions)
    target_num = num
    if target_num != 0:
        value = context.portfolio.total_value / target_num
        for stock in target_list:
            open_position(stock, value)
            log.info("买入[%s]（%s元）" % (stock, value))
            if len(context.portfolio.positions) == g.stock_num:
                break

# 判断今天是否跳过月份
def today_is_between(context):
    month = context.current_dt.month
    return False if month in g.pass_months else True

# 清仓后次日资金可转
def close_account(context):
    if not g.trading_signal:
        if g.hold_list and g.hold_list != [g.etf]:
            for stock in g.hold_list:
                position = context.portfolio.positions[stock]
                close_position(position)
                log.info("卖出[%s]" % (stock))

# 打印持仓信息
def print_position_info(context):
    for position in list(context.portfolio.positions.values()):
        securities = position.security
        cost = position.avg_cost
        price = position.price
        ret = 100 * (price / cost - 1)
        value = position.value
        amount = position.total_amount    
        print('代码:{}'.format(securities))
        print('成本价:{}'.format(format(cost, '.2f')))
        print('现价:{}'.format(price))
        print('收益率:{}%'.format(format(ret, '.2f')))
        print('持仓(股):{}'.format(amount))
        print('市值:{}'.format(format(value, '.2f')))
    print('———————————————————————————————————————分割线————————————————————————————————————————')

# 检查剩余金额并买入
def check_remain_amount(context):
    stoploss_list = []
    uplimit_list = []

    for key, value in g.reason_to_sell.items():
        if value == 'stoploss':
            stoploss_list.append(key)
        elif value == 'limitup':
            uplimit_list.append(key)
    empty_num = len(stoploss_list) + len(uplimit_list)
    addstock_num = len(uplimit_list)
    etf_num = len(stoploss_list)

    g.hold_list = list(context.portfolio.positions.keys())
    if len(g.hold_list) < g.stock_num:
        num_stocks_to_buy = min(addstock_num, g.stock_num - len(g.hold_list))
        target_list = [stock for stock in g.target_list if stock not in g.limitup_stocks][:num_stocks_to_buy]
        log.info('有余额可用' + str(round(context.portfolio.cash, 2)) + '元。买入' + str(target_list))
        buy_security(context, target_list, len(target_list))
        if etf_num != 0:
            log.info('有余额可用' + str(round(context.portfolio.cash, 2)) + '元。买入货币基金' + str(g.etf))
            buy_security(context, [g.etf], etf_num)
    g.reason_to_sell = {}

2025-03-28