2421 涨停突破分时阶梯止盈策略 年化123涨停弱转强竞价战法

涨停突破分时阶梯止盈策略

核心逻辑：

1. 标的筛选规则：

• 基础池：剔除次新股（上市 < 50 日）、ST 股、科创板 / 创业板股票、停牌股
• 动量筛选：选择昨日涨停（非一字板）且前两日未涨停的个股
• 流动性过滤：昨日成交额 > 7 亿、流通市值 > 70 亿、均价涨幅 > 7%
• 量压验证：左压天数计算（寻找近期最大成交量形成压力位后的突破）

2. 买入规则**（09:30）**：

• 集合竞价要求：成交量占比前日 3% 以上
• 合理高开：开盘价涨幅 0-6%（相对于昨日收盘）
• 等分仓位：可用资金平均分配至所有符合条件的标的

3. 动态卖出规则：

• 阶梯止盈止损：
  • 09:30：跌破成本价 3% 立即止损
  • 10:30：未盈利仓位清仓
  • 13:30：盈利不足 3% 仓位了结
• 尾盘统一清仓（14:50）：所有未封涨停板持仓了结
• 特殊处理：持仓股当日触及跌停时使用限价单排队卖出

核心指标：

• 均价涨幅 = (昨日成交额 / 成交量)/ 昨日收盘价 -1
• 左压天数：寻找最近 100 日内的成交量压力位突破
• 开价比值 = 开盘价 /(昨日涨停价 / 1.1)

 

策略特点：

1. 双重动量验证：结合昨日涨停动量与当日开盘动能
2. 流动性优先：通过成交额、市值过滤确保标的流动性
3. 动态风险控制：分时段逐步收紧止盈止损标准
4. 压力突破逻辑：通过左压天数验证突破有效性
5. 特殊行情规避：剔除连续涨停标的，降低接棒风险

策略代码

from jqdata import *
from jqfactor import *
from jqlib.technical_analysis import *
import datetime as dt
import pandas as pd



def initialize(context):
    set_option('use_real_price', True)
    log.set_level('system', 'error')

    run_daily(get_stock_list, '9:01')
    run_daily(buy, '09:30')
    run_daily(sell, '14:50')
    run_daily(sell_930, '9:30')
    run_daily(sell_1030, '10:30')
    run_daily(sell_1330, '13:30')

关键函数解锁后查看：

def sell(context):
    hold_list = list(context.portfolio.positions)
    current_data = get_current_data()
    
    for s in hold_list:
        if not (current_data[s].last_price == current_data[s].high_limit):
            if context.portfolio.positions[s].closeable_amount != 0:
                order_target_value(s, 0)
                print('卖出' + s)
                print('———————————————————————————————————')

def sell_930(context):
    hold_list = list(context.portfolio.positions)
    current_data = get_current_data()
    
    for s in hold_list:
        if not (current_data[s].last_price == current_data[s].high_limit):
            if context.portfolio.positions[s].closeable_amount != 0:
                if current_data[s].last_price < context.portfolio.positions[s].avg_cost*0.97:
                    # 如果跌，用限价单排板
                    if current_data[s].last_price == current_data[s].low_limit:
                        order_target_value(s, 0, LimitOrderStyle(current_data[s].low_limit))
                        print('930止损卖出' + s)
                        print('———————————————————————————————————')
                    # 未跌停，用市价单即刻买入
                    else:
                        order_target_value(s, 0, MarketOrderStyle())
                        print('930止损卖出' + s)
                        print('———————————————————————————————————')

def sell_1030(context):
    hold_list = list(context.portfolio.positions)
    current_data = get_current_data()
    
    for s in hold_list:
        if not (current_data[s].last_price == current_data[s].high_limit):
            if context.portfolio.positions[s].closeable_amount != 0:
                if current_data[s].last_price < context.portfolio.positions[s].avg_cost*1:
                    # 如果跌，用限价单排板
                    if current_data[s].last_price == current_data[s].low_limit:
                        order_target_value(s, 0, LimitOrderStyle(current_data[s].low_limit))
                        print('1030止损卖出' + s)
                        print('———————————————————————————————————')
                    # 未跌停，用市价单即刻买入
                    else:
                        order_target_value(s, 0, MarketOrderStyle())
                        print('1030止损卖出' + s)
                        print('———————————————————————————————————')

def sell_1330(context):
    hold_list = list(context.portfolio.positions)
    current_data = get_current_data()
    
    for s in hold_list:
        if not (current_data[s].last_price == current_data[s].high_limit):
            if context.portfolio.positions[s].closeable_amount != 0:
                if current_data[s].last_price < context.portfolio.positions[s].avg_cost*1.03:
                    # 如果跌，用限价单排板
                    if current_data[s].last_price == current_data[s].low_limit:
                        order_target_value(s, 0, LimitOrderStyle(current_data[s].low_limit))
                        print('1330止损卖出' + s)
                        print('———————————————————————————————————')
                    # 未跌停，用市价单即刻买入
                    else:
                        order_target_value(s, 0, MarketOrderStyle())
                        print('1330止损卖出' + s)
                        print('———————————————————————————————————')


# 处理日期相关函数
def transform_date(date, date_type):
    if type(date) == str:
        str_date = date
        dt_date = dt.datetime.strptime(date, '%Y-%m-%d')
        d_date = dt_date.date()
    elif type(date) == dt.datetime:
        str_date = date.strftime('%Y-%m-%d')
        dt_date = date
        d_date = dt_date.date()
    elif type(date) == dt.date:
        str_date = date.strftime('%Y-%m-%d')
        dt_date = dt.datetime.strptime(str_date, '%Y-%m-%d')
        d_date = date
    dct = {'str':str_date, 'dt':dt_date, 'd':d_date}
    return dct[date_type]

def get_shifted_date(date, days, days_type='T'):
    #获取上一个自然日
    d_date = transform_date(date, 'd')
    yesterday = d_date + dt.timedelta(-1)
    #移动days个自然日
    if days_type == 'N':
        shifted_date = yesterday + dt.timedelta(days+1)
    #移动days个交易日
    if days_type == 'T':
        all_trade_days = [i.strftime('%Y-%m-%d') for i in list(get_all_trade_days())]
        #如果上一个自然日是交易日，根据其在交易日列表中的index计算平移后的交易日        
        if str(yesterday) in all_trade_days:
            shifted_date = all_trade_days[all_trade_days.index(str(yesterday)) + days + 1]
        #否则，从上一个自然日向前数，先找到最近一个交易日，再开始平移
        else:
            for i in range(100):
                last_trade_date = yesterday - dt.timedelta(i)
                if str(last_trade_date) in all_trade_days:
                    shifted_date = all_trade_days[all_trade_days.index(str(last_trade_date)) + days + 1]
                    break
    return str(shifted_date)



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

def filter_st_stock(initial_list, date):
    str_date = transform_date(date, 'str')
    if get_shifted_date(str_date, 0, 'N') != get_shifted_date(str_date, 0, 'T'):
        str_date = get_shifted_date(str_date, -1, 'T')
    df = get_extras('is_st', initial_list, start_date=str_date, end_date=str_date, df=True)
    df = df.T
    df.columns = ['is_st']
    df = df[df['is_st'] == False]
    filter_list = list(df.index)
    return filter_list

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']

def filter_paused_stock(initial_list, date):
    df = get_price(initial_list, end_date=date, frequency='daily', fields=['paused'], count=1, panel=False, fill_paused=True)
    df = df[df['paused'] == 0]
    paused_list = list(df.code)
    return paused_list

# 一字
def filter_extreme_limit_stock(context, stock_list, date):
    tmp = []
    for stock in stock_list:
        df = get_price(stock, end_date=date, frequency='daily', fields=['low','high_limit'], count=1, panel=False)
        if df.iloc[0,0] < df.iloc[0,1]:
            tmp.append(stock)
    return tmp



# 每日初始股票池
def prepare_stock_list(date): 
    initial_list = get_all_securities('stock', date).index.tolist()
    initial_list = filter_kcbj_stock(initial_list)
    initial_list = filter_new_stock(initial_list, date)
    initial_list = filter_st_stock(initial_list, date)
    initial_list = filter_paused_stock(initial_list, date)
    return initial_list


# 计算左压天数
def calculate_zyts(s, context):
    high_prices = attribute_history(s, 101, '1d', fields=['high'], skip_paused=True)['high']
    prev_high = high_prices.iloc[-1]
    zyts_0 = next((i-1 for i, high in enumerate(high_prices[-3::-1], 2) if high >= prev_high), 100)
    zyts = zyts_0 + 5
    return zyts


# 筛选出某一日涨停的股票
def get_hl_stock(initial_list, date):
    df = get_price(initial_list, end_date=date, frequency='daily', fields=['close','high_limit'], count=1, panel=False, fill_paused=False, skip_paused=False)
    df = df.dropna() #去除停牌
    df = df[df['close'] == df['high_limit']]
    hl_list = list(df.code)
    return hl_list
    
# 筛选曾涨停
def get_ever_hl_stock(initial_list, date):
    df = get_price(initial_list, end_date=date, frequency='daily', fields=['high','high_limit'], count=1, panel=False, fill_paused=False, skip_paused=False)
    df = df.dropna() #去除停牌
    df = df[df['high'] == df['high_limit']]
    hl_list = list(df.code)
    return hl_list

# 计算涨停数
def get_hl_count_df(hl_list, date, watch_days):
    # 获取watch_days的数据
    df = get_price(hl_list, end_date=date, frequency='daily', fields=['close','high_limit','low'], count=watch_days, panel=False, fill_paused=False, skip_paused=False)
    df.index = df.code
    #计算涨停与一字涨停数，一字涨停定义为最低价等于涨停价
    hl_count_list = []
    extreme_hl_count_list = []
    for stock in hl_list:
        df_sub = df.loc[stock]
        hl_days = df_sub[df_sub.close==df_sub.high_limit].high_limit.count()
        extreme_hl_days = df_sub[df_sub.low==df_sub.high_limit].high_limit.count()
        hl_count_list.append(hl_days)
        extreme_hl_count_list.append(extreme_hl_days)
    #创建df记录
    df = pd.DataFrame(index=hl_list, data={'count':hl_count_list, 'extreme_count':extreme_hl_count_list})
    return df

# 计算连板数
def get_continue_count_df(hl_list, date, watch_days):
    df = pd.DataFrame()
    for d in range(2, watch_days+1):
        HLC = get_hl_count_df(hl_list, date, d)
        CHLC = HLC[HLC['count'] == d]
        df = df.append(CHLC)
    stock_list = list(set(df.index))
    ccd = pd.DataFrame()
    for s in stock_list:
        tmp = df.loc[[s]]
        if len(tmp) > 1:
            M = tmp['count'].max()
            tmp = tmp[tmp['count'] == M]
        ccd = ccd.append(tmp)
    if len(ccd) != 0:
        ccd = ccd.sort_values(by='count', ascending=False)    
    return ccd

# 计算昨涨幅
def get_index_increase_ratio(index_code, context):
    # 获取指数昨天和前天的收盘价
    close_prices = attribute_history(index_code, 2, '1d', fields=['close'], skip_paused=True)
    if len(close_prices) < 2:
        return 0  # 如果数据不足，返回0
    day_before_yesterday_close = close_prices['close'][0]
    yesterday_close = close_prices['close'][1]
    
    # 计算涨幅
    increase_ratio = (yesterday_close - day_before_yesterday_close) / day_before_yesterday_close
    return increase_ratio