如何通过Bloomberg BLPAPI持续获取USD/JPY的高低价？

问题描述 #

使用Bloomberg Python API（blpapi）开发实时金融数据应用时，发现仅能在初始订阅响应（init paint/Summary事件）中获取USD/JPY的高低价（HIGH/LOW），后续实时更新仅返回买卖价（BID/ASK），无法持续获取高低价的变动更新。

原实现代码如下：

from blpapi import SubscriptionList, SessionOptions, Session

def create_session():
    session_options = SessionOptions()
    session_options.setServerHost('localhost')
    session_options.setServerPort(8194)
    session = Session(session_options)
    session.start()
    session.openService('//blp/mktdata')
    return session

def subscribe_to_data(session):
    subscription_list = SubscriptionList()
    subscription_list.add('USDJPY Curncy', ["BID","ASK","HIGH","LOW", ""], blpapi.CorrelationId('USDJPY'))
    session.subscribe(subscription_list)

def process_message(msg):
    if msg.hasElement('BID') and msg.hasElement('ASK'):
        bid = msg.getElementAsFloat('BID')
        ask = msg.getElementAsFloat('ASK')
    if msg.hasElement('HIGH') and msg.hasElement('LOW'):
        high = msg.getElementAsFloat('HIGH')
        low = msg.getElementAsFloat('LOW')
        
    # 更新应用状态
    update_gui(bid, ask, high, low)

def main():
    session = create_session()
    subscribe_to_data(session)
    while True:
        event = session.nextEvent()
        for msg in event:
            process_message(msg)

if __name__ == "__main__":
    main()

解决方案 #

核心修改点是替换订阅的高低价字段，并配置更新间隔：

• 原HIGH/LOW字段仅在初始响应返回，需改用支持实时更新的字段：PRICE_HIGH_RT（实时最高价）、HIGH_TDY（当日最高价）、PX_HIGH_ASK_RT（实时卖价最高价）、PX_HIGH_BID_RT（实时买价最高价），对应实时最低价可使用PRICE_LOW_RT、LOW_TDY等字段
• 订阅时添加interval={更新间隔}参数，控制实时数据推送频率

修改后的完整实现代码：

import blpapi

class RealTimePricing:

    def __init__(self, security="USDJPY Curncy", interval=0.5):
        self.security = security  # 目标币种，默认设为USDJPY
        self.interval = interval  # 数据更新间隔（秒）
        self.session = None

    def start_session(self):
        options = blpapi.SessionOptions()
        options.setServerHost("localhost")
        options.setServerPort(8194)

        self.session = blpapi.Session(options)

        if not self.session.start():
            print("会话启动失败。")
            return False

        if not self.session.openService("//blp/mktdata"):
            print("无法打开//blp/mktdata服务。")
            return False

        return True

    def subscribe(self):
        subscription_list = blpapi.SubscriptionList()
        # 订阅包含实时高低价的字段，并设置更新间隔
        subscription_list.add(self.security, "BID,ASK,HIGH,LOW,PRICE_HIGH_RT,HIGH_TDY,PX_HIGH_ASK_RT,PX_HIGH_BID_RT,PRICE_LOW_RT,LOW_TDY,PX_LOW_ASK_RT,PX_LOW_BID_RT", 
                              f"interval={self.interval}", blpapi.CorrelationId(self.security))
        self.session.subscribe(subscription_list)

    def process_message(self, msg):
        # 处理所有高低价相关字段
        fields = ["HIGH", "LOW", "PRICE_HIGH_RT", "HIGH_TDY", "PX_HIGH_ASK_RT", "PX_HIGH_BID_RT",
                  "PRICE_LOW_RT", "LOW_TDY", "PX_LOW_ASK_RT", "PX_LOW_BID_RT"]
        for field in fields:
            if msg.hasElement(field):
                value = msg.getElementAsFloat(field)
                print(f"{field}: {value}")

    def event_loop(self):
        try:
            while True:
                event = self.session.nextEvent(500)
                # 仅处理订阅数据和状态事件
                if event.eventType() in [blpapi.Event.SUBSCRIPTION_DATA, blpapi.Event.SUBSCRIPTION_STATUS]:
                    for msg in event:
                        self.process_message(msg)
        except KeyboardInterrupt:
            print("正在退出...")

if __name__ == "__main__":
    pricing = RealTimePricing()
    if pricing.start_session():
        pricing.subscribe()
        pricing.event_loop()

关键说明 #

• 不同的高低价字段对应不同的统计维度（如当日整体、实时买卖价维度），可根据业务需求选择订阅
• interval参数控制数据推送频率，可根据实际需求调整（单位：秒）
• 处理消息时需判断字段是否存在，避免因部分字段未更新导致的异常

内容的提问来源于stack exchange，提问作者Py_trader