抢先布局！牛市中的量化打板策略

大家好，我是橙哥！今天我将为大家详细解读一个在牛市中非常有效的策略—— 首板低开打板策略 。这个策略的核心思想是利用 股票首次涨停后的低开机会进行买入，并在特定时间点进行止盈或止损卖出 。希望通过这篇文章，大家能够更好地理解这个策略的运作原理，并在实际操作中有所收获。 欢迎加入宽客邦量化俱乐部获取本文完整代码，我将在内部技术分享会中进行详细讲解。

策略概述

首先，让我们来了解一下这个策略的基本框架。首板低开打板策略主要分为以下几个步骤：

1. 选股条件：

• 选择昨日首次涨停的股票。
• 过滤掉连续涨停的股票。
• 选择相对位置较低的股票。
• 选择开盘价低于昨日收盘价一定比例的股票。

• 交易执行：

• 在开盘时买入符合条件的股票。
• 在特定时间点进行止盈或止损卖出。

代码解读

接下来，我们将逐段解读代码，帮助大家更好地理解策略的实现过程。

初始化函数

          
def initialize(context):  
    # 系统设置  
    set_option('use\_real\_price', True)  
    set_option('avoid\_future\_data', True)  
    log.set_level('system', 'error')  
    # 每日运行  
    run_daily(buy, '09:30')  # 9:25分知道开盘价后可以提前下单  
    run_daily(sell, '11:28')  
    run_daily(sell, '14:50')  

      

在策略的初始化阶段，我们进行了一些系统设置，并安排了每日的交易任务。具体来说：

• set_option('use_real_price', True)：设置使用真实价格进行交易。
• set_option('avoid_future_data', True)：避免使用未来数据。
• log.set_level('system', 'error')：设置日志级别为错误，减少不必要的日志输出。
• run_daily(buy, '09:30')：在每天的9:30执行买入操作。
• run_daily(sell, '11:28') 和 run_daily(sell, '14:50')：在每天的11:28和14:50执行卖出操作。

选股函数

          
def buy(context):  
    # 基础信息  
    date = transform_date(context.previous_date, 'str')  
    current_data = get_current_data()  
      
    # 昨日涨停列表  
    initial_list = prepare_stock_list(date)  
    hl_list = get_hl_stock(initial_list, date)  
      
    if len(hl_list) != 0:      
        # 获取非连板涨停的股票  
        ccd = get_continue_count_df(hl_list, date, 10)  
        lb_list = list(ccd.index)  
        stock_list = [s for s in hl_list if s not in lb_list]  
          
        # 计算相对位置  
        rpd = get_relative_position_df(stock_list, date, 60)  
        rpd = rpd[rpd['rp'] <= 0.5]  
        stock_list = list(rpd.index)  
          

      

在选股函数中，我们首先获取了昨日涨停的股票列表，并过滤掉连续涨停的股票。然后，我们计算这些股票的相对位置，并选择开盘价低于昨日收盘价一定比例的股票进行买入。具体步骤如下：

1. 获取昨日涨停列表：通过prepare_stock_list和get_hl_stock函数获取昨日涨停的股票列表。
2. 过滤非连板涨停的股票：通过get_continue_count_df函数计算连板数，并过滤掉连续涨停的股票。
3. 计算相对位置：通过get_relative_position_df函数计算股票的相对位置，并选择相对位置较低的股票。
4. 选择低开的股票：计算开盘价与昨日收盘价的比率，选择开盘价低于昨日收盘价一定比例的股票。
5. 买入操作：如果当前没有持仓，则按照等权重买入符合条件的股票。

卖出函数

          
def sell(context):  
    # 基础信息  
    date = transform_date(context.previous_date, 'str')  
    current_data = get_current_data()  
      
    # 根据时间执行不同的卖出策略  
    if str(context.current_dt)[-8:] == '11:28:00':  
        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 > context.portfolio.positions[s].avg_cost)):  
                print('止盈卖出', [get_security_info(s, date).display_name, s])  
                print('———————————————————————————————————')  
      
    if str(context.current_dt)[-8:] == '14:50:00':  
        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)):  
                print('止损卖出', [get_security_info(s, date).display_name, s])  
                print('———————————————————————————————————')  

      

在卖出函数中，我们根据当前时间执行不同的卖出策略。具体来说：

1. 11:28止盈卖出：如果当前时间是11:28，并且股票价格高于买入成本价且低于涨停价，则进行止盈卖出。
2. 14:50止损卖出：如果当前时间是14:50，并且股票价格低于涨停价，则进行止损卖出。

日期处理函数

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

      

在日期处理函数中，我们定义了两个函数：

1. transform_date：将日期转换为不同的格式（字符串、datetime、date）。
2. get_shifted_date：根据自然日或交易日计算平移后的日期。

过滤函数

          
def filter\_new\_stock(initial\_list, date, days=250):  
    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)  
    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[: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  

      

在过滤函数中，我们定义了四个函数来过滤股票：

1. filter_new_stock：过滤掉上市时间不足250天的股票。
2. filter_st_stock：过滤掉ST股票。
3. filter_kcbj_stock：过滤掉科创板和北交所的股票。
4. filter_paused_stock：过滤掉停牌的股票。

每日初始股票池

          
def prepare\_stock\_list(date):  
    initial_list = get_all_securities('stock', date).index.tolist()  
    initial_list = filter_kcbj_stock(initial_list)    
    initial_list = filter_paused_stock(initial_list, date)  
    return initial_list  

      

在每日初始股票池函数中，我们首先获取所有股票，然后依次进行上述过滤操作，最终得到一个符合条件的股票列表。

筛选涨停股票

          
def get\_hl\_stock(initial\_list, date):  
    df = get_price(initial_list, end_date=date, frequency='daily', fields=['close', 'high', 'high\_limit'], count=1, panel=False, fill_paused=False, skip_paused=False)  
    hl_list = list(df.code)  
    return hl_list  

      

在筛选涨停股票函数中，我们获取昨日涨停的股票列表。具体来说：

1. 获取股票的收盘价、最高价和涨停价。
2. 去除停牌的股票。
3. 筛选出收盘价等于涨停价的股票。

计算涨停数和连板数

          
def get\_hl\_count\_df(hl\_list, date, watch\_days):  
    # 获取watch\_days的数据  
    df = get_price(hl_list, end_date=date, frequency='daily', fields=['low', 'close', 'high\_limit'], count=watch_days, panel=False, fill_paused=False, skip_paused=False)  
    df.index = df.code  
    # 计算涨停与一字涨停数，一字涨停定义为最低价等于涨停价  
    hl_count_list = []     
    # 创建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):     
    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  

      

在计算涨停数和连板数函数中，我们定义了两个函数：

1. get_hl_count_df：计算股票在一定天数内的涨停数和一字涨停数。
2. get_continue_count_df：计算股票的连板数，并返回连板数最多的股票。

计算股票相对位置

          
def get\_relative\_position\_df(stock\_list, date, watch\_days):  
    if len(stock_list) != 0:  
        df = get_price(stock_list, end_date=date, fields=['high', 'low', 'close'], count=watch_days, fill_paused=False, skip_paused=False, panel=False).dropna()  
        close = df.groupby('code').apply(lambda df: df.iloc[-1, -1])  
        high = df.groupby('code').apply(lambda df: df['high'].max())  
        result = pd.DataFrame()  
        result['rp'] = (close - low) / (high - low)  
        return result  
    else:  
        return pd.DataFrame(columns=['rp'])  

      

在计算股票相对位置函数中，我们计算股票在一定天数内的相对位置。具体来说：

1. 获取股票的最高价、最低价和收盘价。
2. 计算股票的相对位置，即收盘价与最低价和最高价的比率。

总结

通过以上代码的逐段解读，相信大家已经对这个首板低开打板策略有了更深入的了解。这个策略的核心思想是利用股票首次涨停后的低开机会进行买入，并在特定时间点进行止盈或止损卖出。虽然这个策略在某些市场环境下可能表现不错，但也存在较高的市场风险和技术风险。因此，在实际操作中，建议大家进行充分的回测和风险评估，谨慎实施。

希望这篇文章对大家有所帮助，如果有任何问题或建议，欢迎在评论区留言讨论！欢迎加入**「宽客邦量化俱乐部** 」 ，与专业投资者通行，共建财富自由之路！