使用异步 API 调用将 excel 功能移植到 BLPAPI/xbbg

Question

你好——我正在尝试将一些 Excel 工作表移植到一个我可以即时运行的 python 脚本中。我正在查询 Bloomberg 数据，基本上我想做的是在特定日期获取特定 ISIN 的历史出价和要价。我正在使用 xbbg 库查询 API 并在 Jupyter Notebook 中工作。数据存储在 Pandas 数据框中。

这是我尝试建模的非常简单的 Excel 功能：

我可以毫无问题地为单个订单项提取此数据。但是，我正在尝试针对 25,000 个指数成分股/日期运行此公式（在 Excel 中，您只需拖放公式即可）。我有允许并返回正确数据的代码，但是运行这么多行需要几个小时。我相信这是因为代码在继续下一个标识符之前等待每个 API 调用解决。

我一直在寻找这个问题的解决方案，即使用一个名为 asyncio 的库。我确实有可以使用语法的代码片段，但它仍然花费太多时间。这是我用于提取此数据的异步版本的代码：

async def bbg_data(isin, start_date, end_date):
    isin = '/isin/' + isin + '@BVAL'
    previous_close_date = pd.to_datetime(end_date).strftime('%Y%m%d')
    px_bid = blp.bdh(tickers=isin, flds=['PX_BID'], start_date=previous_close_date, end_date=previous_close_date)
    if not px_bid.empty:
        px_bid = px_bid.iloc[0][0]
    else:
        px_bid = "N/A"
    px_ask = blp.bdh(tickers=isin, flds=['PX_ASK'], start_date=previous_close_date, end_date=previous_close_date)
    if not px_ask.empty:
        px_ask = px_ask.iloc[0][0]
    else:
        px_ask = "N/A"
    return px_bid, px_ask

async def main():
    tasks = [asyncio.create_task(bbg_data(isin, start_date, end_date)) for isin, start_date, end_date in ttd_sample[['ISIN', 'PreviousClose', 'PreviousClose']].values]
    results = await asyncio.gather(*tasks)
    ttd_sample['previous_bid_price'] = [result[0] for result in results]
    ttd_sample['previous_ask_price'] = [result[1] for result in results]

await main()

我对这个问题有几个问题，希望有 BLPAPI/asyncio 经验的人：

1. 我对 xbbg 库的依赖让我很头疼吗？我是否能够仅使用 blpapi 一次解决这个问题？如果是这样，对于初学者程序员有什么好的教程吗？

2. 我编写 asyncio 代码的方式有问题吗？我认为我很接近，但我不确定它是否真的是异步的。

任何关于如何解决这个问题的建议将不胜感激。

Answer 1

每个

bdh()

请求的设置成本都很高。获取一系列日期，然后选择您想要的日期可能会更快。例如，这里只有 one 调用

bdh()

而不是 10:

from xbbg import blp
from datetime import date
import pandas as pd

pairs = [['DE0001102325 Govt',date(2023,4,3)],
         ['DE0001134922 Govt',date(2023,3,15)],
         ['DE0001134922 Govt',date(2023,4,19)],
         ['DE0001102358 Govt',date(2023,2,14)],
         ['DE0001102366 Govt',date(2023,4,12)] ]

tickers = [p[0] for p in pairs]
dates = [p[1] for p in pairs]

fields = ['PX_BID','PX_ASK']

min_date = min(dates)
max_date = max(dates)

df = blp.bdh(set(tickers),fields,min_date,max_date)
   
data=[]
index=[]
for [ticker,dt] in pairs:
    data.append([df[ticker].loc[dt][f] for f in fields])
    index.append((ticker,dt))

dfResult = pd.DataFrame(data,index=index,columns=fields)

print(dfResult)

随着输出：

                                  PX_BID   PX_ASK
(DE0001102325 Govt, 2023-04-03)   99.xxx   99.xxx
(DE0001134922 Govt, 2023-03-15)  102.xxx  103.xxx
(DE0001134922 Govt, 2023-04-19)  102.xxx  102.xxx
(DE0001102358 Govt, 2023-02-14)   98.xxx   98.xxx
(DE0001102366 Govt, 2023-04-12)   97.xxx   97.xxx

注意：代码可以用不同的日期重复，通过使用

set

只将不同的代码传递给

bdh()

调用。根据

min_date

和

max_date

之间的差异，将大量行情代码分成更小的日期范围更小的块可能更有效。 25,000 可能太多而无法在一次调用中发送（您冒着连接超时的风险），因此也许从 100 个 isin-date 对的块开始。