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

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 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                 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. 计算股票的相对位置，即收盘价与最低价和最高价的比率。

总结