策略概述
选股条件：
选择昨日首次涨停的股票。 过滤掉连续涨停的股票。 选择相对位置较低的股票。 选择开盘价低于昨日收盘价一定比例的股票。
交易执行：
在开盘时买入符合条件的股票。 在特定时间点进行止盈或止损卖出。
代码解读
接下来，我们将逐段解读代码，帮助大家更好地理解策略的实现过程。
初始化函数
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)
在选股函数中，我们首先获取了昨日涨停的股票列表，并过滤掉连续涨停的股票。然后，我们计算这些股票的相对位置，并选择开盘价低于昨日收盘价一定比例的股票进行买入。具体步骤如下：
获取昨日涨停列表：通过 prepare_stock_list和get_hl_stock函数获取昨日涨停的股票列表。过滤非连板涨停的股票：通过 get_continue_count_df函数计算连板数，并过滤掉连续涨停的股票。计算相对位置：通过 get_relative_position_df函数计算股票的相对位置，并选择相对位置较低的股票。选择低开的股票：计算开盘价与昨日收盘价的比率，选择开盘价低于昨日收盘价一定比例的股票。 买入操作：如果当前没有持仓，则按照等权重买入符合条件的股票。
卖出函数
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('———————————————————————————————————')
在卖出函数中，我们根据当前时间执行不同的卖出策略。具体来说：
11:28止盈卖出：如果当前时间是11:28，并且股票价格高于买入成本价且低于涨停价，则进行止盈卖出。 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)
在日期处理函数中，我们定义了两个函数：
transform_date：将日期转换为不同的格式（字符串、datetime、date）。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
在过滤函数中，我们定义了四个函数来过滤股票：
filter_new_stock：过滤掉上市时间不足250天的股票。filter_st_stock：过滤掉ST股票。filter_kcbj_stock：过滤掉科创板和北交所的股票。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
在筛选涨停股票函数中，我们获取昨日涨停的股票列表。具体来说：
获取股票的收盘价、最高价和涨停价。 去除停牌的股票。 筛选出收盘价等于涨停价的股票。
计算涨停数和连板数
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
在计算涨停数和连板数函数中，我们定义了两个函数：
get_hl_count_df：计算股票在一定天数内的涨停数和一字涨停数。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'])
在计算股票相对位置函数中，我们计算股票在一定天数内的相对位置。具体来说：
获取股票的最高价、最低价和收盘价。 计算股票的相对位置，即收盘价与最低价和最高价的比率。
