3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
强烈不建议在 run_daily 触发的函数中使用死循环来不断获取实时价格。虽然理论上你可能偶尔能拿到更新的价格,但在实际的 Ptrade 交易环境中,这种做法是极其危险且不可行的。
run_daily 调用的函数里写了一个死循环(如 while True:),这个函数将永远不会执行完毕。这会导致整个策略的线程被阻塞,其他的核心事件(如 handle_data、on_order_response 委托回报、on_trade_response 成交回报等)都将无法被触发和执行,你的策略将彻底瘫痪。sleep 的死循环中疯狂调用 get_snapshot 或其他行情接口,极易触发系统的流量控制(流控),导致接口直接报错、返回 None,甚至可能导致你的交易账号被临时封禁或策略进程被系统强制杀掉。如果你需要以最高频率(3秒一次)监控价格并做出交易决策,Ptrade 提供了两个官方支持的机制:
tick_data 事件函数(推荐)tick_data 是 Ptrade 专门为 Tick 级别策略设计的事件函数,它在交易时间内每隔 3 秒自动执行一次,完美契合 A 股行情的更新频率。
def initialize(context):
g.security = '600570.SS'
set_universe(g.security)
# tick_data 每 3 秒自动触发一次
def tick_data(context, data):
security = g.security
# 获取当前 tick 的买一价
try:
current_price = eval(data[security]['tick']['bid_grp'][0])[1][0]
log.info("当前买一价: %s" % current_price)
# 你的交易逻辑
if current_price > 38.19:
# 注意:tick_data 中只能使用 order_tick 下单
order_tick(security, 100, 1)
except Exception as e:
log.error("获取价格异常: %s" % e)
def handle_data(context, data):
pass
run_interval 定时循环函数如果你不想用 tick_data,可以在 initialize 中使用 run_interval 注册一个按秒循环执行的函数。Ptrade 限制 run_interval 的最小时间间隔为 3 秒。
def initialize(context):
g.security = '600570.SS'
set_universe(g.security)
# 注册一个每 3 秒执行一次的函数
run_interval(context, check_price_func, seconds=3)
def check_price_func(context):
# 使用 get_snapshot 获取最新行情快照
snapshot = get_snapshot(g.security)
if snapshot and g.security in snapshot:
last_px = snapshot[g.security].get('last_px')
log.info("当前最新价: %s" % last_px)
# 你的交易逻辑
if last_px > 38.19:
order(g.security, 100)
def handle_data(context, data):
pass
在量化交易框架中,永远不要使用死循环来等待或轮询数据。请顺应框架的事件驱动机制,利用 tick_data 或 run_interval 来实现高频的价格监控,这样既安全、稳定,又能保证策略的其他部分正常运转。