策略的核心是选择在低位、首次涨停后次日低开的股票,开盘时买入,上午有盈利就卖出,没有就拿到尾盘在卖出;策略优点在于选择的股票在低位,风险较低,并且刚启动首板,有大资金介入,第二天能快速跑路。
回测数据如下:

*回测数据只作测试用,不代表未来实际收益
上午开盘买入,上午有盈利就卖出,尾盘全部卖出
run_daily(buy, '09:30') # 开盘买入
run_daily(sell, '11:28') # 上午止盈
run_daily(sell, '14:50') # 下午止损
2、买入逻辑
买入过滤条件有三个,一是排除10天内连续涨停的股票,二是股票在60天内要在相对低位,三是要低开3%-4%
(1)基础过滤股票池,过滤科创北交、上市不超过250、停牌、ST股票
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
(2)排除10天内连续涨停的股票
# 获取非连板涨停的股票
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]
(2.1)计算连板数
# 计算连板数
def get_continue_count_df(hl_list, date, watch_days):
df = pd.DataFrame()
for d in range(2, watch_days+1): # 从2天开始,一直到观察周期结束
HLC = get_hl_count_df(hl_list, date, d) # 计算连板数
CHLC = HLC[HLC['count'] == d] # 筛选出连续d天涨停的股票
df = df.append(CHLC) # 记录连续d天涨停的股票
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
(2.2)计算涨停数
# 计算涨停数
def get_hl_count_df(hl_list, date, watch_days):
# 获取watch_days长度(观察周期)的行情数据
# 包含low(最低价)、close(收盘价)、high_limit(涨停价)三个字段
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 = [] # 普通涨停天数
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
(3)选取60天内在相对低位的股票
rpd = get_relative_position_df(stock_list, date, 60)
rpd = rpd[rpd['rp'] <= 0.5]
stock_list = list(rpd.index)
(3.1)计算相对位置
# 计算股票处于一段时间内相对位置
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()) # 计算一段时间内的最高价
low = df.groupby('code').apply(lambda df: df['low'].min()) # 计算一段时间内的最低价
result = pd.DataFrame()
result['rp'] = (close-low) / (high-low) # 计算相对位置
return result
else:
return pd.DataFrame(columns=['rp'])
(4)选取低开3%-4%的股票
df = get_price(stock_list, end_date=date, frequency='daily', fields=['close'], count=1, panel=False, fill_paused=False, skip_paused=True).set_index('code') if len(stock_list) != 0 else pd.DataFrame()
df['open_pct'] = [current_data[s].day_open/df.loc[s, 'close'] for s in stock_list]
df = df[(0.96 <= df['open_pct']) & (df['open_pct'] <= 0.97)]
3、卖出逻辑
卖出时段有两个,一是上午止盈卖出,二是下午只要是不涨停就全卖出
(1)上午不涨停,并且大于我的持仓成本就止盈卖出
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)):
order_target_value(s, 0)
print( '止盈卖出')
(2)不涨停就全卖出
if ((context.portfolio.positions[s].closeable_amount != 0) and (current_data[s].last_price < current_data[s].high_limit)):
order_target_value(s, 0)
print( '全部卖出')
这篇文章主要分享首版低开策略的思路,主要的逻辑在计算连板数和涨停数上面,量化交易的回测和实盘,可以在量化交易软件中进行,券商的量化交易软件免费并且申请也基本没有门槛,大家可以去体验下量化交易的魅力。
发布者:股市刺客,转载请注明出处:https://www.95sca.cn/archives/1352307
站内所有文章皆来自网络转载或读者投稿,请勿用于商业用途。如有侵权、不妥之处,请联系站长并出示版权证明以便删除。敬请谅解!