2407 追首板涨停 百日左压首板高开放量策略 回测数据：过去两年年化304% 48680

1. 基础筛选机制：

• 股票池构建：全市场非 ST / 非科创 / 非北交所股票
• 流动性要求：剔除上市不足 50 日新股及停牌个股
• 量能验证：昨日成交额 7-19 亿，流通市值 70-520 亿

2. 首板确认体系：

• 涨停基因：筛选昨日首次涨停且前两日无连板个股
• 集合竞价验证：高开 0-6.02%，竞量占比前日 3% 以上
• 均价支撑：昨日成交均价跌幅≤4%（排除恐慌抛压）

3. 技术面突破条件：

• 左压测试：突破百日高点且伴随成交量有效放大
• 量价配合：突破日量能高于前期压力位最大量能的 90%

4. 动态交易规则：

• 早盘介入：9:28 集合竞价结束前按高开幅度下单
• 分层止盈：11:28 前盈利个股获利了结
• 强制平仓：14:50 前未封板个股无论盈亏均离场

5. 特殊风控设计：

• 异常波动过滤：排除开盘涨幅超 6% 的过度高开个股
• 流动性保护：只交易主板标的，回避微小市值品种
• 筹码结构验证：百日量价关系排除弱势突破

策略代码

from jqdata import *
from jqfactor import *
import pandas as pd
from datetime import datetime,timedelta,date

################################### 初始化设置 #############################################
def initialize(context):
    set_option('use_real_price', True)
    log.set_level('system', 'error')
    set_option('avoid_future_data', True)

def after_code_changed(context):
    g.n_days_limit_up_list = []   #重新初始化列表
    unschedule_all() # 取消所有定时运行    
    # run_daily(get_stock_list, '9:05')
    run_daily(buy, '09:28')
    run_daily(sell, time='11:28', reference_security='000300.XSHG')
    run_daily(sell, time='14:50', reference_security='000300.XSHG')

## 定义股票池      
def set_stockpool(context):
    yesterday = context.previous_date 
    initial_list = get_all_securities('stock', yesterday).index.tolist()
    return initial_list

##################################  交易函数群 ##################################
def buy(context):
    current_data = get_current_data()
    qualified_stocks =  get_stock_list(context)   
    if qualified_stocks:
        value = context.portfolio.available_cash / len(qualified_stocks)
        for s in qualified_stocks:
            # 下单   #至少够买1手
            if context.portfolio.available_cash/current_data[s].last_price>100: 
                order_value(s, value, MarketOrderStyle(current_data[s].day_open))
                print('买入' + s)
                print('———————————————————————————————————')
                

def sell(context):
    stime = context.current_dt.strftime("%H%M")
    current_data = get_current_data()
    
    # 根据时间执行不同的卖出策略
    if stime == '1128':
        for s in list(context.portfolio.positions):  #上午有利润就跑
            if ((context.portfolio.positions[s].closeable_amount != 0) and (current_data[s].last_price < current_data[s].high_limit) and (current_data[s].last_price > 1*context.portfolio.positions[s].avg_cost)):#avg_cost当前持仓成本
                order_target_value(s, 0)
    elif stime == '1450':
        for s in list(context.portfolio.positions):
            if ((context.portfolio.positions[s].closeable_amount != 0) and (current_data[s].last_price < current_data[s].high_limit)):#closeable_amount可卖出的仓位
                order_target_value(s, 0)

####### 选股函数 #########

关键函数解锁后查看：

###################################  其它函数 ##################################

# 筛选出某一日涨停的股票
def get_hl_stock(stock_list, date1,days):
    if not stock_list:return []
    h_s = get_price(stock_list, end_date=date1, frequency='daily', fields=['close', 'high_limit', 'paused'],
                  count=days, panel=False, fill_paused=False, skip_paused=True
                  ).query('close==high_limit and paused==0').groupby('code').size()
    return h_s.index.tolist()

# 过滤函数
def filter_new_stock(initial_list, date, days=50):
    return [stock for stock in initial_list if get_security_info(stock).start_date < date - timedelta(days=days)]

def filter_st_paused_stock(initial_list, date):
    current_data = get_current_data()
    return [stock for stock in initial_list if not (
            current_data[stock].is_st or 
            current_data[stock].paused or 
            '退' in current_data[stock].name)]


def filter_kcbj_stock(initial_list):
    return [stock for stock in initial_list if stock[0] != '4' and stock[0] != '8' and stock[0] != '3' and stock[:2] != '68']


### end ###

2025-02-24