🚀 AI 一键生成 ptrade 策略代码

Ptrade 策略中能否在循环事件里不断获取股票实时价格?

ptrade | 发布时间: 2026/5/21
以下内容由 EasyQuant 生成。

问题描述

Ptrade在一个循环事件里 能否不断获取股票实时价格?

假设在一个事件循环里:
如:
run_daily(context, get_price, '09:44')
定义的get_price 函数,
然后get_price函数里面有一个死循环,不断地获取价格。

因为ptrade的行情切片 是每3秒更新的一次的,如果行情没更新,那么当前的价格也是过去最近的一个3s的价格。

现在问题是,在一个固定的时间里面,不断地读取价格函数,能获取到最新的价格吗 ?

解决方案

结论

强烈不建议run_daily 触发的函数中使用死循环来不断获取实时价格。虽然理论上你可能偶尔能拿到更新的价格,但在实际的 Ptrade 交易环境中,这种做法是极其危险且不可行的。

为什么不能用死循环?

  1. 阻塞策略主线程:Ptrade 是一个基于事件驱动的量化引擎。如果你在 run_daily 调用的函数里写了一个死循环(如 while True:),这个函数将永远不会执行完毕。这会导致整个策略的线程被阻塞,其他的核心事件(如 handle_dataon_order_response 委托回报、on_trade_response 成交回报等)都将无法被触发和执行,你的策略将彻底瘫痪。
  2. 触发 API 流控限制:券商柜台和 Ptrade 引擎对 API 的调用频率是有严格限制的。在一个没有 sleep 的死循环中疯狂调用 get_snapshot 或其他行情接口,极易触发系统的流量控制(流控),导致接口直接报错、返回 None,甚至可能导致你的交易账号被临时封禁或策略进程被系统强制杀掉。
  3. 无意义的性能消耗:正如你所说,A股的 Level-1 行情快照本身就是每 3 秒更新一次。在 3 秒内请求 1000 次和请求 1 次,得到的数据是完全一样的,死循环只会白白消耗服务器 CPU 资源。

正确的获取实时价格(高频)的方法

如果你需要以最高频率(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_datarun_interval 来实现高频的价格监控,这样既安全、稳定,又能保证策略的其他部分正常运转。