策略逻辑说明：涨停板竞价策略

策略名称

策略核心思想

1. 涨停板筛选：选择近期有涨停表现的股票
2. 竞价强度分析：通过分析集合竞价阶段的量价关系，筛选出强势股票
3. 科创板特殊处理：对科创板股票采用不同的交易处理方式

策略详细逻辑

1. 股票池筛选

• 初始股票池可选择全 A 股或特定指数成分股
• 基础筛选条件：
  • 过滤停牌股票
  • 过滤 ST 股票
  • 过滤退市股票
  • 过滤上市不满 30 天的新股
  • 仅保留创业板和科创板股票（代码以 3、68、69 开头）

2. 涨停板筛选

• 筛选出在指定周期内（可配置）有涨停表现的股票
• 可设置涨停次数门槛（如至少 1 次涨停）

3. 竞价分析

• 在集合竞价阶段（9:24-9:25）收集以下数据：
  • 价格变化率（p_ratio）：竞价结束价 / 竞价开始价
  • 成交量变化率（v_ratio）：竞价结束量 / 竞价开始量
  • 竞价总金额（money）
  • 开盘价 / 昨日收盘价（open_close）
  • 开盘价 / 昨日最高价（open_high）
• 筛选标准：
  • 竞价金额大于 1000 万
  • 成交量放大超过 2.5 倍
  • 价格上升超过 94%
  • 开盘价高于昨日收盘价 5% 以上

4. 交易执行

• 在开盘时买入通过筛选的股票
• 采用等分资金的方式买入（50% 总资金均分）
• 对科创板股票采用限价单（上浮 10%）处理
• 持仓管理：
  • 盘中不卖出涨停股票
  • 对非涨停股票立即卖出

策略特点

1. 专注强势股：通过涨停板和竞价强度双重筛选，锁定市场强势股票
2. 短线交易：当日买入，当日根据涨停情况决定是否持有
3. 科创板适配：针对科创板交易规则进行特殊处理
4. 量化筛选：通过严格的量化标准筛选股票，减少主观判断

风险提示

1. 追高风险：高开股票可能存在回落风险
2. 流动性风险：涨停板股票可能出现流动性不足
3. 市场风格风险：在震荡市或下跌市中表现可能不佳
4. 科创板风险：科创板股票波动更大，风险更高

适用市场环境

• 市场整体处于上升趋势
• 短线资金活跃，涨停板效应明显
• 市场波动较大的环境

参数配置建议

• g.index：股票池选择（全 A 股或特定指数）
• g.begin_times/g.end_times：竞价分析时间段
• 涨停筛选周期和次数
• 竞价筛选标准（金额、量价变化率等）
• 买入资金比例（当前设置为 50%）

优化方向

1. 增加更多技术指标筛选
2. 优化卖出时机判断
3. 加入止损机制
4. 根据不同市场环境动态调整参数

 

---

 

from jqdata import *
from jqlib.technical_analysis  import *
from sklearn.linear_model import LinearRegression
from jqfactor import get_factor_values
import numpy as np
import pandas as pd 
import time
import gc

# 初始化函数，设定基准等等
def after_code_changed(context):
    # 输出内容到日志 log.info()
    log.info('初始函数开始运行且全局只运行一次')
    unschedule_all()

    # 过滤掉order系列API产生的比error级别低的log
    # log.set_level('order', 'error')
    set_params()    #1 设置策略参数
    set_variables() #2 设置中间变量
    set_backtest()  #3 设置回测条件

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

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

    #测试不同时间买入
    run_daily(market_open, time='9:30')
    #run_daily(market_run, time='9:30')
    run_daily(market_run, time='14:55')
      # 收盘时运行
    #run_daily(market_close, time='9:30')
      # 收盘后运行
    #run_daily(after_market_close, time='20:00')


#1 设置策略参数
def set_params():
    #设置全局参数
    g.index ='all'          #all-zz-300-500-1000，single-个股信号诊断
    
    g.begin_times = ' 09:24:00'
    g.end_times =  ' 09:25:10'


#2 设置中间变量
def set_variables():
    #暂时未用，测试用全池
    g.stocknum = 0              #单日买入数，0-代表全取,

    
#3 设置回测条件
def set_backtest():
    ## 设定g.index作为基准
    if g.index == 'all':
        set_benchmark('000001.XSHG')
    else:
        set_benchmark(g.index)
        
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    set_option("avoid_future_data", True)
    #显示所有列
    pd.set_option('display.max_columns', None)
    #显示所有行
    pd.set_option('display.max_rows', None)
    log.set_level('order', 'error')    # 设置报错等级
    
## 开盘前运行函数
def before_market_open(context):
    # 输出运行时间
    log.info('------------------------美好的一天开始了------------------------')
    log.info('函数运行时间(before_market_open)：'+str(context.current_dt.time()))
    #0，预置全局参数
    today_date = context.current_dt.date()
    lastd_date = context.previous_date
    all_data = get_current_data()
    g.poollist = []
    g.sell_list =[]
        
    num1,num2,num3,num4,num5,num6=0,0,0,0,0,0    #用于过程追踪

    #0，构建基准指数票池，三去+去新
    start_time = time.time()
    if g.index =='all':
        stocklist = list(get_all_securities(['stock']).index)   #取all
    elif g.index == 'zz':
        stocklist = get_index_stocks('000300.XSHG', date = None) + get_index_stocks('000905.XSHG', date = None) + get_index_stocks('000852.XSHG', date = None)
    else:
        stocklist = get_index_stocks(g.index, date = None)
    
    num1 = len(stocklist)    
    stocklist = [stockcode for stockcode in stocklist if not all_data[stockcode].paused]
    stocklist = [stockcode for stockcode in stocklist if not all_data[stockcode].is_st]
    stocklist = [stockcode for stockcode in stocklist if'退' not in all_data[stockcode].name]
    stocklist = [stockcode for stockcode in stocklist if (today_date-get_security_info(stockcode).start_date).days>30]
    stocklist = [stockcode for stockcode in stocklist if stockcode[0] == '3' or stockcode[:2] == '68' or stockcode[:2] == '69']
    num2 = len(stocklist)

    end_time = time.time()
    print('Step0,基准%s,原始%d只,四去后共%d只,构建耗时:%.1f 秒' % (g.index,num1,num2,end_time-start_time))
    
    #1,昨日涨停票
    start_time = time.time()
    g.poollist = get_up_filter_jiang(context,stocklist,lastd_date,1,0,1)
    
    end_time = time.time()
    print('Step1,周期内有涨停共%d只,构建耗时:%.1f 秒' % (len(g.poollist),end_time-start_time))
    #log.info(g.poollist)

关键函数解锁后查看：

        
## 早盘时运行函数
def market_open(context):
    log.info('函数运行时间(market_open):'+str(context.current_dt.time()))
    
    if len(g.auct_list) ==0:
        log.info('今日无买信')
        return
    else:
        log.info('*****今日买信共%d只*****:' % len(g.auct_list))
        log.info(g.auct_list)

    total_value = context.portfolio.total_value
    buy_cash = 0.5*total_value/len(g.auct_list)
    for stockcode in g.auct_list:
        if stockcode in list(context.portfolio.positions.keys()):
            continue
        buy_stock(context,stockcode,buy_cash)
    
    return

## 收盘时运行函数
def market_run(context):
    log.info('函数运行时间(market_run):'+str(context.current_dt.time()))
    today_date = context.current_dt.date()
    lastd_date = context.previous_date
    current_data = get_current_data()
    
    for stockcode in context.portfolio.positions:
        if current_data[stockcode].paused == True:
            continue
        if context.portfolio.positions[stockcode].closeable_amount ==0:
            continue

        #非停出
        if current_data[stockcode].last_price != current_data[stockcode].high_limit:
            log.info('非涨停即出%s' % stockcode)
            sell_stock(context,stockcode,0)
            continue
        

## 收盘时运行函数
def market_close(context):
    log.info('函数运行时间(market_close):'+str(context.current_dt.time()))

        
## 收盘后运行函数
def after_market_close(context):
    log.info(str('函数运行时间(after_market_close):'+str(context.current_dt.time())))


"""
---------------------------------函数定义-主要策略-----------------------------------------------
"""
#蒋的方法，N天M涨停过滤
def get_up_filter_jiang(context,stocklist,check_date,check_duration,up_num,direction):
    # 输出运行时间
    log.info('-函数运行时间(get_up_filter_jiang)：'+str(context.current_dt.time()))
    #0，预置，今天是D日
    all_data = get_current_data()
    poollist=[]
    
    if len(stocklist)==0:
        log.info("输入为空")
        return poollist
    
    # 交易日历
    trd_days = get_trade_days(end_date=check_date, count=check_duration)  # array[datetime.date]
    s_trd_days = pd.Series(range(len(trd_days)), index=trd_days)  # Series[index:交易日期，value:第几个交易日]
    back_date = trd_days[0]
    
    #2，形态过滤，一月内两次以上涨停(盘中过10%也算)
    start_time = time.time()
    # 取数
    df_price = get_price(stocklist,end_date=check_date,frequency='1d',fields=['pre_close','open','close','high','high_limit','low_limit','paused']
    ,skip_paused=False,fq='pre',count=check_duration,panel=False,fill_paused=True)
    
    # 过滤出涨停的股票，按time索引
    df_up = df_price[(df_price.close == df_price.high_limit) & (df_price.paused == 0)].set_index('time')
    # 标注出df_up中的time对应的是第几个交易日(ith)
    df_up['ith'] = s_trd_days
    
    code_set = set(df_up.code.values)
    if direction ==1:
        poollist =[stockcode for stockcode in code_set if ((len(df_up[df_up.code ==stockcode]) > up_num))]
    elif direction ==-1:
        poollist =[stockcode for stockcode in code_set if ((len(df_up[df_up.code ==stockcode]) < up_num))]
    else:
        poollist =[stockcode for stockcode in code_set if ((len(df_up[df_up.code ==stockcode]) == up_num))]
    
    end_time = time.time()
    #log.info('---%d天(%s--%s)%d次涨停过滤出%d只标的,构建耗时:%.1f 秒' % (check_duration,back_date,check_date,up_num,len(poollist),end_time-start_time))        
    #log.info(poollist)

    return poollist
    

"""
---------------------------------函数定义-次要过滤-----------------------------------------------
"""

"""
---------------------------------函数定义-辅助函数-----------------------------------------------
"""
##买入函数
def buy_stock(context,stockcode,cash):
    today_date = context.current_dt.date()
    current_data = get_current_data()
    
    if stockcode[0:3] == '688':
        last_price = current_data[stockcode].last_price
        if order_target_value(stockcode,cash,MarketOrderStyle(1.1*last_price)) != None: #科创板需要设定限值
            log.info('%s买入%s' % (today_date,stockcode))
    else:
        if order_target_value(stockcode, cash) != None:
            log.info('%s买入%s' % (today_date,stockcode))
            
##卖出函数
def sell_stock(context,stockcode,cash):
    today_date = context.current_dt.date()
    current_data = get_current_data()
    
    if stockcode[0:3] == '688':
        last_price = current_data[stockcode].last_price
        if order_target_value(stockcode,cash,MarketOrderStyle(0.9*last_price)) != None: #科创板需要设定限值
            log.info('%s卖出%s' % (today_date,stockcode))
    else:
        if order_target_value(stockcode,cash) != None:
            log.info('%s卖出%s' % (today_date,stockcode))