3

不幸的是，我在论坛上被引导相信（但没有 100% 的确定性）Bloomberg Desktop API 一次不允许超过一个 IntradayBarRequest 或 IntradayTickRequest，这与允许多个同时请求的 HistoricalDataRequest 或 Subscriptions 不同。

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

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

我正在使用 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

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

4

3 回答 3

2

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

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

希望能帮助到你。

于 2013-08-03T15:16:54.510 回答
2

我不知道 python API，但我确实使用 Java API。

我仍在寻找一种将请求与响应相关联的方法，而不必执行低效的请求……等待响应……请求等。

正如您所发现的，您无法为 IntradayBarRequests 发送多个证券的查询，并且提要不包含证券 ID（例如股票代码）以轻松映射回您的查询。

最简单的解决方案是使用相关 ID。当您向会话提交请求时，您可以提供自己的相关 ID。下面的例子是用 Java 编写的，但我认为 python API 是相似的：

Session session = ...; //Bloomberg Session
CorrelationId cId = new CorrelationId(); //Unique ID
session.sendRequest(bbRequest, cId); //provide your own ID

然后，如果您使用异步会话，您会收到异步消息，其中包含指向原始 CorrelationId 的链接：

public void processEvent(Event event, Session session) {
    for (Message msg : event) {
        CorrelationID cId = msg.correlationID();
        //here you can link the result back to your original query
    }
}
于 2013-08-03T21:45:44.653 回答
0

我不相信您可以为刻度或条形数据下标。这是为什么... 对于定价数据，您会在每次更改值时得到响应。报价数据不只是这些价格的历史吗？对于条形数据，这不是通过按时间段（即 10 分钟、1 小时）分组删除了一些细节的价格历史记录。因此，如果您可以订阅其中任何一个，您应该何时收到彭博社的每条新回复？本质上，您可以创建自己的侦听器来进行分组，然后以您自己选择的频率从侦听器发出响应。

顺便说一句：我使用 CLR 与 python 中的 .net blp 对象对话。我的解决方案早于彭博对 Python 的任何支持，本质上是在做“无法做到”的事情。

于 2013-11-23T18:14:13.133 回答