2374 低市值动量反转策略 主观辅助_形态识别_单边上涨策略优化v3 45406

策略核心思想

本策略为动量反转与趋势跟踪结合的策略，通过筛选低市值股票并结合盘中趋势、成交量、动量等指标，动态调整持仓，追求短期收益。策略核心在于：

1. 低市值筛选：选择市值低于160亿的股票，聚焦小盘股，捕捉其高波动性带来的超额收益。

2. 盘中趋势判断：通过OLS回归分析盘中价格走势，筛选出全天均匀上涨的股票。

3. 成交量放大：筛选成交量显著放大的股票（成交量大于过去5日均值的2倍），确保资金关注度高。

4. 动量反转：选择前一日动量（收益率）较低的股票，追求反转收益。

5. 涨停板管理：对昨日涨停的股票进行特殊处理，若盘中涨停打开则卖出，否则继续持有。

关键机制解析

1. 选股模块：

   • 初筛：剔除停牌、ST股、新股（上市不足250天）、创业板、科创板等高风险股票。

   • 低市值筛选：选择市值低于160亿的股票，聚焦小盘股。

   • 盘中趋势判断：通过OLS回归分析盘中价格走势，筛选出全天均匀上涨（R² > 0.8）且斜率（价格涨幅）为正的股票。

   • 成交量放大：筛选成交量大于过去5日均值2倍的股票，确保资金关注度高。

   • 动量反转：选择前一日动量（收益率）较低的股票，追求反转收益。

2. 交易模块：

   • 卖出逻辑：卖出不在当前选股列表中的持仓股票（除黄金ETF外）。

   • 买入逻辑：将可用资金等分买入选股列表中的股票，直至达到目标持仓数（g.arg_buy_max）。

   • 涨停板管理：对昨日涨停的股票，若盘中涨停打开则卖出，否则继续持有。

3. 风险控制：

   • 分散持仓：持仓股票数量固定为g.arg_buy_max（默认10只），分散单一股票风险。

   • 黄金ETF对冲：空闲资金买入黄金ETF（518880.XSHG），降低组合波动性。

   • 持仓时间限制：单只股票持仓时间不超过g.arg_hold_max（默认20天），避免长期持有风险。

策略特点

• 低市值聚焦：通过低市值筛选，捕捉小盘股的高波动性收益。

• 盘中趋势跟踪：通过OLS回归分析盘中价格走势，筛选出全天均匀上涨的股票。

• 动量反转：选择前一日动量较低的股票，追求反转收益。

• 涨停板管理：对昨日涨停的股票进行特殊处理，避免错失涨停收益。

---

策略代码

# 回测资金需要 1000000

# 导入函数库
from jqdata import *
import pandas as pd
import numpy as np
import talib as ta
import statsmodels.api as sm

# 本代码为加速版，n天一调仓而不是每天调仓，运行的更快

# 初始化函数

def initialize(context):
    ### 股票相关设定 ###
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    set_option('use_real_price', True)
    set_option("avoid_future_data", True)

    log.set_level('order', 'error')

    set_slippage(FixedSlippage(0.002))
    # 股票类每笔交易时的手续费是：买入时佣金万分之三，
    # 卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(close_tax=0.001,
                             open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')

    # 计数器
    g.counter = 0
    # 模型参数
    # 判断全天均匀上涨,OLS的R方最小值
    g.arg_rsquared_min = 0.8
    # 收盘价涨幅下限
    g.arg_close_rate = 0.052
    # 成交量回看天数
    g.arg_volume_days = 5
    # 成交量大于days均值的倍数
    g.arg_volume_multi = 2
    # 持仓股票的最大数目
    g.arg_buy_max = 10
    # 买入时要达到的比例
    g.buy_min_ratio = 0.4
    # 单只股票持仓时间
    g.arg_hold_max = 20
    # 持仓日数记录
    g.hold_days = g.arg_hold_max - 1

    # 选择股票
    g.stocks = []
    g.choice = 1000
    
    # 空闲资金买入黄金ETF
    # g.etf = ""
    g.etf = "518880.XSHG"
    # g.etf = "511010.XSHG"

    g.hold_list = []  # 当前持仓的全部股票
    g.yesterday_HL_list = []  # 记录持仓中昨日涨停的股票

    # 运行函数（reference_security为运行时间的参考标的；
    # 传入的标的只做种类区分，因此传入'000300.XSHG'或'510300.XSHG'是一样的）
    # 开盘前运行
    run_daily(before_market_open, time='before_open')
    run_daily(trade, time='14:55')
    run_daily(check_limit_up, time='14:00',
              reference_security='000852.XSHG')

# 开盘前运行函数

def before_market_open(context):
    # if g.hold_days < g.arg_hold_max:
    #     return
    
    yesterday = context.previous_date

    # 获取已持有列表
    g.hold_list = [s for s in context.portfolio.positions]

    # 获取昨日涨停列表
    if g.hold_list != []:
        df = get_price(g.hold_list, end_date=yesterday, frequency='daily', fields=[
                       'close', 'high_limit'], count=1, panel=False, fill_paused=False)
        df = df[df['close'] == df['high_limit']]
        g.yesterday_HL_list = list(df.code)
    else:
        g.yesterday_HL_list = []

    # 股票池
    # stocks = get_all_securities(date=yesterday).index.tolist()

    # 获取股票代码并按市值排序
    fundamentals_data = get_fundamentals(query(valuation.code, valuation.market_cap).order_by(
        valuation.market_cap.asc()).filter(valuation.market_cap < 160).limit(g.choice))
    stocks = list(fundamentals_data['code'])

    # 各种过滤
    current_data = get_current_data()
    stocks = [s for s in stocks if not current_data[s].paused
                and not current_data[s].is_st
                and 'N' not in current_data[s].name
                and '退' not in current_data[s].name
                and current_data[s].low_limit < current_data[s].day_open < current_data[s].high_limit and s[0] != '4' and s[0] != '8' and s[:2] != '68' \ # and s[:2] != '30' \ and s not in g.hold_list] g.stocks = [stock for stock in stocks if yesterday - get_security_info(stock).start_date > datetime.timedelta(days=250)]

    log.info('股票数量{}'.format(len(g.stocks)))

关键函数解锁后查看：

# 分钟运行函数
def trade(context):
    g.hold_days += 1
    if g.hold_days < g.arg_hold_max:
        return
    
    stocks = get_stocks(context)

    # 逐股票操作：卖出，先卖再买才有钱
    if context.portfolio.positions:
        for stock in context.portfolio.positions.keys():
            # if stock in stocks:
            #     continue
            
            if stock == g.etf:
                continue

            order_target_value(stock, 0)
    
    # 低于个数就不买
    count_min = int(g.buy_min_ratio * g.arg_buy_max)
    if len(stocks) < count_min: if g.etf: order_value(g.etf, context.portfolio.available_cash) return val = context.portfolio.total_value / \ (g.arg_buy_max - len(context.portfolio.positions)) for stock in stocks: if len(context.portfolio.positions) >= g.arg_buy_max:
            break
        
        # 资金不够卖出黄金ETF
        if g.etf and context.portfolio.available_cash < val and g.etf in context.portfolio.positions: order_value(g.etf, -val) # 因为可能会隔天重复选股，不要紧，有钱的话继续买一份，之后按照最后一天选择的日期卖出 order_value(stock, val) if len(context.portfolio.positions) > 0:
        g.hold_days = 0

    # 空闲资金买入黄金ETF
    if g.etf:
        order_value(g.etf, context.portfolio.available_cash)


def check_limit_up(context):
    now_time = context.current_dt

    # 对昨日涨停股票观察到尾盘如不涨停则提前卖出，如果涨停即使不在应买入列表仍暂时持有
    for stock in g.yesterday_HL_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))
            order_value(stock, 0)
        else:
            log.info("[%s]涨停，继续持有" % (stock))
            
    if g.etf:
        order_value(g.etf, context.portfolio.available_cash)
                

2025-02-23