python - 彭博 API : how do I associate request with response on IntradayBarRequest?

Question

不幸的是，我在论坛上被引导相信(但不是 100% 确定)Bloomberg Desktop API 一次不允许多个 IntradayBarRequest 或 IntradayTickRequest，这不同于 HistoricalDataRequest 或订阅，其中多个允许同时请求。

因此这个问题可能没有实际意义，除非有人告诉我上述情况不正确

如果为真，那么处理下面问题的唯一方法是仅在处理完前一个请求后才发送每个新请求。

---

我正在使用 Python Bloomberg Desktop API 访问金融证券的订阅(实时更新)和历史每日数据。在这两种情况下，我都可以同时发送多个请求，当响应到来时(不一定按请求发送的顺序)，我可以使用 msg.getElement("securityData") 找出响应与哪个安全相关联.getElementAsString("security") 在历史数据的情况下，或者在订阅数据的情况下，通过使用 msg.correlationIds()[0].value() 查询我事先(在订阅请求时)设置的 correlationId .

但是我不知道如何为 IntradayBarResponse 请求执行此操作(并且 WAPI 文档没有帮助)。这些似乎没有可设置的correlationId，也没有上面的“securityData”字段。如果我发送多个 intradayBarRequests，我如何找出响应是针对哪种证券？

这是我的代码(改编自 API Python 示例)。

import blpapi  # interface to bloomberg
import time    # will need this for time parsing
from optparse import OptionParser
import pdb     # debugger, when necessary
import csv     # for csv reading
import string  # for string parsing
from pymongo import MongoClient
import inspect
from datetime import datetime
from bson.son import SON


def parseCmdLine():
    parser = OptionParser(description="Retrieve realtime data.")
    parser.add_option("-a",
                      "--ip",
                      dest="host",
                      help="server name or IP (default: %default)",
                      metavar="ipAddress",
                      default="localhost")
    parser.add_option("-p",
                      dest="port",
                      type="int",
                      help="server port (default: %default)",
                      metavar="tcpPort",
                      default=8194)
    parser.add_option("--me",
                      dest="maxEvents",
                      type="int",
                      help="stop after this many events (default: %default)",
                      metavar="maxEvents",
                      default=100000000000)
    parser.add_option("--mongohost",
                      dest="mongohost",
                      default="192.168.1.30")
    parser.add_option("--mongoport",
                      dest="mongoport",
                      type="int",
                      default=27017)

    (options, args) = parser.parse_args()

    return options


def main():
    options = parseCmdLine()

    # connect to MongoDB MONGO MONGO MONGO MONGO ----------------
    print "Connecting to MongoDB"
    print options.mongohost
    print options.mongoport
    client = MongoClient(options.mongohost, options.mongoport) # connect to MongoDB
    db = client.bb # connect to the DB database
    bbsecs = db.bbsecs
    bbticks = db.bbticks
    # now get the securities list


    # Fill SessionOptions
    sessionOptions = blpapi.SessionOptions()
    sessionOptions.setServerHost(options.host)
    sessionOptions.setServerPort(options.port)

    print "connecting to Bloomberg"
    print "Connecting to %s:%d" % (options.host, options.port)

    # Create a Session
    session = blpapi.Session(sessionOptions)

    # Start a Session
    if not session.start():
        print "Failed to start session."
        return

    # open the market data subscription service
    if not session.openService("//blp/mktbar"):
        print "Failed to open //blp/mktbar"
        return
    if not session.openService("//blp/refdata"):
        print "Failed to open //blp/refdata"
        return



    # now startup the subscription list
    # Now open the secs.dat file and read it, append each to subscription list
    maxtimes = bbticks.aggregate([{'$group': {'_id':'$ticker', 'maxtime':{'$max': '$time'}}}]) # get the last updates by ticker
    refDataService = session.getService("//blp/refdata") # start the ref
    for i in maxtimes["result"]:
        ticker = i["_id"]
        tstamp = i["maxtime"]
        request = refDataService.createRequest("IntradayBarRequest")
        request.set("security", ticker)
        request.set("eventType", "TRADE")
        request.set("interval", 1)
        request.set("startDateTime", tstamp)
        request.set("endDateTime", datetime.now())
        print "Sending Request:", ticker
        session.sendRequest(request)

    subscriptions = blpapi.SubscriptionList()
    secdic = dict() # a new dictionary
    for post in bbsecs.find():
        print(post["ticker"])
        # subscribe tick
        #subscriptions.add(str(post["ticker"]), "LAST_PRICE", [], blpapi.CorrelationId("TICK:" + str(post["ticker"])))
        #subscribe 1 minute bars
        subscriptions.add("//blp/mktbar/ticker/"+str(post["ticker"]), 
                          "LAST_PRICE", 
                          "interval=1.0",
                          blpapi.CorrelationId(str(post["ticker"])))
        # setup the dictionary
        secdic[post["bbsecnum"]] = post["ticker"]
    if not session.openService("//blp/refdata"):
        print "Failed to open //blp/refdata"
        return
    # now subscribe
    session.subscribe(subscriptions)


    # HISTORICALHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHH
    # Obtain previously opened service
    #refDataService = session.getService("//blp/refdata")
    # Create and fill the request for the historical data
    #request = refDataService.createRequest("HistoricalDataRequest")
    #for post in bbsecs.find():  
    #    request.getElement("securities").appendValue(str(post["ticker"]))
    #request.getElement("fields").appendValue("LAST_PRICE")
    #request.set("periodicityAdjustment", "ACTUAL")
    #request.set("periodicitySelection", "DAILY")
    #request.set("startDate", "20100101")
    #request.set("endDate", "20121231")
    #request.set("maxDataPoints", 2000)
    #print "Sending Request:", request
    # Send the request
    #session.sendRequest(request)
    #hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhh


    try:
        # Process received events
        eventCount = 0
        while(True):
            # We provide timeout to give the chance to Ctrl+C handling:
            event = session.nextEvent(500)
            for msg in event:
                if event.eventType() == blpapi.Event.SUBSCRIPTION_STATUS:
                    #print "%s - %s" % (msg.correlationIds()[0].value(), msg)
                    print "subscription status"
                elif event.eventType() == blpapi.Event.SUBSCRIPTION_DATA:
                    key = msg.correlationIds()[0].value()
                    if msg.messageType() == "MarketBarStart":
                        open = msg.getElementAsFloat("OPEN")
                        high = msg.getElementAsFloat("HIGH")
                        low = msg.getElementAsFloat("LOW")
                        close = msg.getElementAsFloat("CLOSE")
                        btstamp = msg.getElementAsDatetime("TIME")
                        tstamp = datetime.now()
                        print "bar", key, close, tstamp
                        bbticks.insert({"type": "BAR", "ticker": key, "value": close, \
                                "open": open, "high": high, "low": low, "close": close, \
                                "time": tstamp})
                    elif msg.messageType() == "MarketBarUpdate":
                        close = msg.getElementAsFloat("CLOSE")
                        #print "tick", close,
                        #bbticks.insert({"type": "TICK", "ticker": key, "value": close, "time": tstamp})

                    #if etype == "TRADE":
                    #    if msg.hasElement("LAST_TRADE"):
                    #        key = msg.correlationIds()[0].value(),
                    #        keytype = key[:(key.index(":"))]
                    #        key = key[(key.index(":") + 1):]
                    #        value = msg.getElementAsString("LAST_TRADE")
                    #        timestamp = msg.getElementAsDatetime("TRADE_UPDATE_STAMP_RT")
                    #        print key, value, 
                    #        bbticks.insert({"ticker": key, "value": value, "timestamp": timestamp})

                else:
                    if msg.messageType() == "HistoricalDataResponse":
                        securityData = msg.getElement("securityData")
                        security = securityData.getElementAsString("security")
                        fieldDataArray = securityData.getElement("fieldData")
                        for j in range(0, fieldDataArray.numValues()):
                            fieldData = fieldDataArray.getValueAsElement(j)
                            field = fieldData.getElement(0)
                            tstamp = field.getValueAsDatetime()
                            tstamp = datetime(tstamp.year, tstamp.month, tstamp.day)
                            field = fieldData.getElement(1)
                            close = field.getValueAsFloat()
                            #print "history", security, close, 
                            #bbticks.insert({"type": "DAILY", "ticker": security, "value": close, "close": close, \
                            #        "time": tstamp})
                    elif msg.messageType() == "IntradayBarResponse":
                        print "IntradayBarResponse"
                        data = msg.getElement("barData").getElement("barTickData")
                        numvals = data.numValues()
                        print numvals
                        if numvals > 0:
                            print data.getValueAsElement(1).getElement(1).getValueAsFloat()



            if event.eventType() == blpapi.Event.SUBSCRIPTION_DATA:
                eventCount += 1
                if eventCount >= options.maxEvents:
                    break
    finally:
        # Stop the session
        session.stop()

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print "Ctrl+C pressed. Stopping..."

我查看了 eidData，它是空的，即使我要求返回它也是如此。我关注的是货币，而不是股票，因此不需要交换权利。

> (Pdb) print eid._Element__dataHolder IntradayBarResponse = {
>     barData = {
>         eidData[] = {
>         }
>         barTickData[] = {
>             barTickData = {
>                 time = 2013-08-02T18:36:00.000
>                 open = 4.233100
>                 high = 4.233600
>                 low = 4.233100
>                 close = 4.233400
>                 volume = 0
>                 numEvents = 119
>                 value = 0.000000
>             }
>             barTickData = {
>                 time = 2013-08-02T18:37:00.000
>                 open = 4.233400
>                 high = 4.233700
>                 low = 4.233100
>                 close = 4.233500
>                 volume = 0
>                 numEvents = 96
>                 value = 0.000000
>             }
>             barTickData = {
>                 time = 2013-08-02T18:38:00.000
>                 open = 4.233500
>                 high = 4.233600
>                 low = 4.233300
>                 close = 4.233500
>                 volume = 0
>                 numEvents = 135
>                 value = 0.000000
>             }
>             barTickData = {
>                 time = 2013-08-02T18:39:00.000

我仍在寻找一种将请求与响应相关联的方法，而不必执行低效请求....等待响应....请求等。但是，我不确定彭博社是否提供此功能。这种限制似乎也存在于历史报价数据中。

Answer 1

Thomas，您应该始终使用响应的 correlationId 来查找它与哪个请求相关联。 IIRC、引用数据和历史数据请求支持多种证券，但市场数据和盘中柱可能每个请求只有一种证券。

这就是为什么在 refdata 和历史响应中有额外的“securityData”字段。对于市场数据和日内柱线，correlationId 足以识别请求并因此确保安全性。

希望对您有所帮助。