3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在早盘集合竞价阶段(9:15-9:25),撮合价格与匹配成交量的异动往往预示着全天资金的主攻方向。PTrade 平台提供了 get_trend_data 接口,方便投资者获取集中竞价期间的代码数据。
用于获取早盘集中竞价期间标的的成交量、加权价格、最新价等数据。
get_trend_data(date=None, stocks=None, market=None)
'YYYYMMDD'(如 '20230308')。回测/交易阶段默认当前交易日。'XSHG', 'XSHE')。不传参时默认返回沪深两市所有代码数据。返回以股票代码为 Key 的字典,结构如下:
{
'600570.SS': {
'time_stamp': 20230308092500000, # 时间戳
'hq_px': 42.50, # 最新价
'wavg_px': 42.48, # 加权价格
'business_amount': 1500000, # 总成交量(股)
'business_balance': 63720000.0, # 总成交额(元)
'amount': 0 # 持仓量
}
}
run_daily 运行竞价分析函数。business_amount(成交量)与历史平均开盘量进行对比。def initialize(context):
# 设置目标股票池
g.security = ['600570.SS', '000001.SZ', '600000.SS']
set_universe(g.security)
# 设定标志位
g.buy_list = []
# 在 9:25 获取集合竞价数据并筛选目标
run_daily(context, check_auction_volume, time='9:25')
def check_auction_volume(context):
g.buy_list = []
# 获取早盘集合竞价数据
trend_data = get_trend_data(stocks=g.security)
if not trend_data:
log.info("未获取到集合竞价数据")
return
for stock in g.security:
if stock in trend_data:
info = trend_data[stock]
auction_vol = info.get('business_amount', 0) # 竞价成交量
auction_px = info.get('hq_px', 0) # 竞价开盘价
# 获取过去5日平均成交量作为参考
hist = get_history(5, '1d', 'volume', security_list=stock)
if hist is None or hist.empty:
continue
avg_vol = hist['volume'].mean()
# 获取昨收价计算竞价涨幅
snapshot = get_snapshot(stock)
preclose = snapshot[stock].get('preclose_px', 0)
if preclose == 0:
continue
pct_change = (auction_vol - preclose) / preclose
# 条件:竞价放量(达到5日日均量的5%以上)且竞价高开2%~5%
if auction_vol > (avg_vol * 0.05) and 0.02 <= pct_change <= 0.05:
log.info(f"标的 {stock} 触发竞价突破:竞价量={auction_vol}, 涨幅={pct_change*100:.2f}%")
g.buy_list.append(stock)
def handle_data(context, data):
# 开盘后对筛选出的标的下单
if g.buy_list:
cash = context.portfolio.cash / len(g.buy_list)
for stock in g.buy_list:
if get_position(stock).amount == 0 and cash > 5000:
order_value(stock, cash)
log.info(f"开盘买入 {stock},金额:{cash}")
# 清空当日买入列表
g.buy_list = []
run_daily 的触发时间建议设置为 '9:25' 或 '9:26'。get_snapshot 时部分字段可能为 0,需对除零等异常做保护逻辑。