Bloomberg Data Integration

[StumpCore]

Hi Folks,

I am currently working on a very basic and rudimental implementation of Bloomberg Live Data into the TradingNode for live trading. My end goal is to just receive the live data for specific tickers and let my strategies work on new incoming BarType Data for now.
I tried to implement the necessary files for the python adapter only, since there is no complete bloomberg library in rust as of now.
The following error occurs when I use '.run_async()' which is located in the SandboxExecutionClient I am using.

'NoneType' object has no attribute 'value'
Traceback (most recent call last):
  File "D:\dev\algo_trading\at_backtest\src\at_backtest\nt_trader.py", line 82, in run_live_bbg
    await self._live_node.run_async()
  File "D:\dev\algo_trading\at_backtest\.venv\Lib\site-packages\nautilus_trader\live\node.py", line 349, in run_async
    await self.kernel.start_async()
  File "D:\dev\algo_trading\at_backtest\.venv\Lib\site-packages\nautilus_trader\system\kernel.py", line 1026, in start_async
    if not await self._await_execution_reconciliation():
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\dev\algo_trading\at_backtest\.venv\Lib\site-packages\nautilus_trader\system\kernel.py", line 1337, in _await_execution_reconciliation
    if not await self._exec_engine.reconcile_execution_state(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\dev\algo_trading\at_backtest\.venv\Lib\site-packages\nautilus_trader\live\execution_engine.py", line 1728, in reconcile_execution_state
    result = self._reconcile_execution_mass_status(mass_status)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\dev\algo_trading\at_backtest\.venv\Lib\site-packages\nautilus_trader\live\execution_engine.py", line 1865, in _reconcile_execution_mass_status
    self._adjust_mass_status_fills(mass_status)
  File "D:\dev\algo_trading\at_backtest\.venv\Lib\site-packages\nautilus_trader\live\execution_engine.py", line 1985, in _adjust_mass_status_fills
    adjusted_results = adjust_fills_for_partial_window(
                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\dev\algo_trading\at_backtest\.venv\Lib\site-packages\nautilus_trader\live\reconciliation.py", line 624, in adjust_fills_for_partial_window
    pyo3_mass_status = mass_status.to_pyo3()
                       ^^^^^^^^^^^^^^^^^^^^^
  File "D:\dev\algo_trading\at_backtest\.venv\Lib\site-packages\nautilus_trader\execution\reports.py", line 1213, in to_pyo3
    account_id=nautilus_pyo3.AccountId(self.account_id.value),
                                       ^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'value'

Below you find my creation of the node:

        catalog_config = DataCatalogConfig(
            path=self.catalog_path, fs_protocol="file", name="historical_data"
        )

        node_config = TradingNodeConfig(
            trader_id=self._trader_id,
            logging=LoggingConfig(log_level=self._logging_lvl),
            catalogs=[catalog_config],
            data_clients={
                "SIM_CASH": BbgLiveDataClientConfig(
                    account_currency="EUR",
                    base_currency="EUR",
                    external_clients=[ClientId("SIM_CASH")],
                    instrument_provider=BbgInstrumentProviderConfig(
                        load_all=True,
                    ),
                )
            },
            exec_clients={
                "SIM_CASH": SandboxExecutionClientConfig(
                    venue="SIM_CASH",
                    base_currency="EUR",
                    oms_type="NETTING",
                    account_type="CASH",
                    starting_balances=["1_000_000 EUR"],
                    instrument_provider=BbgInstrumentProviderConfig(
                        load_all=True,
                    ),
                )
            },
        )
        node = TradingNode(config=node_config)

        strat_combs = []
        ticker_map = {}
        inst_ids = []

        for instrument, wrangler, _ticks in self.portfolio.hist_tpl:
            instr_id = instrument.id
            bar_type = wrangler.bar_type

            # Adding the strategy list
            strat_combs += self.strategy_ft.create_configs(
                instrument_id=instr_id, bar_type=bar_type
            )

            # Adding the Ticker List
            ticker = instr_id.symbol.value
            ticker_map[ticker] = bar_type
            inst_ids.append(instr_id)

        strategies = self.strategy_ft.create_strategies(strat_combs)
        node.trader.add_strategies(strategies)
     
        node.add_data_client_factory("SIM_CASH", factory=BbgLiveDataClientFactory)
        node.add_exec_client_factory("SIM_CASH", factory=SandboxLiveExecClientFactory)

        _inst = self.catalog.instruments()
        ids = []
        for i in _inst:
            node.cache.add_instrument(i)
            ids.append(i.id)

        node.build()

The following code pulls the data from BBG successfully and keeps spinning if I remove the ExecFactory or change naming conventions:

class BbgLiveDataMarketClient(LiveMarketDataClient):
    """Example `LiveMarketDataClient` showing the overridable abstract methods."""

    def __init__(
        self,
        loop: asyncio.AbstractEventLoop,
        client: ClientId,
        msgbus: Any,
        config: Any,  # This will be your DataClientConfig
        cache: Any,
        clock: Any,
        instrument_provider: Any,
    ):
        super().__init__(
            loop=loop,
            client_id=client,
            venue=Venue("SIM_CASH"),
            msgbus=msgbus,
            cache=cache,
            clock=clock,
            config=config,
            instrument_provider=instrument_provider,
        )
        self.tickers = [inst.raw_symbol.value for inst in cache.instruments()]
        self.instrument_ids = [inst.id for inst in cache.instruments()]
        self.mapping = {}

    def _start(self):
        print("Starting")
        for i in self.instrument_ids:
            ticker = i.symbol.value
            bar_type = BarType.from_str(f"{i}-1-DAY-LAST-EXTERNAL")
            self.mapping[ticker] = bar_type

    async def _connect(self) -> None:
        print("Connected")
        self._loop_task = self.create_task(self._run_loop())
        self._set_connected(True)

    async def _disconnect(self) -> None:
        print("Disconnected")

    async def _run_loop(self) -> None:
        ticker_bars: dict[str, dict[str, Any]] = {}
        fields = HIST_FIELDS
        data_type = DataType(Bar)

        try:
            async for update in blp.stream(
                tickers=self.tickers, flds=fields, interval=1
            ):
                ticker = update.get("TICKER")
                if not ticker:
                    continue

                # Data processing logic remains the same
                current_values = ticker_bars.get(
                    ticker,
                    {
                        "OPEN": 0.0,
                        "HIGH": 0.0,
                        "LOW": 0.0,
                        "LAST_PRICE": 0.0,
                        "VOLUME": 0.0,
                    },
                )

                # Update our local state with whatever fields came in this tick
                for key in current_values:
                    if key in update and update[key] is not None:
                        current_values[key] = update[key]

                ticker_bars[ticker] = current_values

                # Extract for validation
                o, h, l, c = (
                    current_values["OPEN"],
                    current_values["HIGH"],
                    current_values["LOW"],
                    current_values["LAST_PRICE"],
                )

                if (o >= l and o <= h) and (c >= l and c <= h) and (h >= l):
                    bar = Bar(
                        bar_type=self.mapping.get(ticker),
                        open=Price.from_str(f"{o}"),
                        high=Price.from_str(f"{h}"),
                        low=Price.from_str(f"{l}"),
                        close=Price.from_str(f"{c}"),
                        volume=Quantity.from_str(f"{current_values['VOLUME']}"),
                        ts_event=dt_to_unix_nanos(
                            datetime.datetime.now(datetime.timezone.utc)
                        ),
                        ts_init=dt_to_unix_nanos(
                            datetime.datetime.now(datetime.timezone.utc)
                        ),
                    )
                    print(f"{bar}")

        except Exception as e:
            self._log.error(f"Bloomberg stream error: {e}")

Unfortunately, due to the limitations of bloomberg I cannot share a small examples for you to test the issue.
I can confirm the following:

• .build() works without any error
• all instruments are loaded and available in the cache
• all accounts (currently only EUR) are created during the .build() phase
• the strategies are loaded and on_start() is called
• the BbgLiveDataClientFactory is called correctly and the data flow starts

Please apologize the messy code structure, since I am currently just trying to establish a rudimental version before moving on with improved readability and coding standards.

Thank you.

PS: Text File of Complete Output.
error_nautilus_trader(1).txt

[StumpCore]

I was able to solve the issue by wrapping the SandboxExecutionClient and adding a new self._set_account_id() to the init

class SandboxLiveExecClientFactory(SandboxLiveExecClientFactory):
    @staticmethod
    def create(
        loop: asyncio.AbstractEventLoop,
        name: str,
        config: SandboxExecutionClientConfig,
        portfolio: PortfolioFacade,
        msgbus: MessageBus,
        cache: Cache,
        clock: LiveClock,
    ) -> SandboxExecutionClient:
        exec_client = BbgLiveExecutionClient(
            loop=loop,
            clock=clock,
            portfolio=portfolio,
            msgbus=msgbus,
            cache=cache,
            config=config,
        )
        return exec_client


class BbgLiveExecutionClient(SandboxExecutionClient):
    def __init__(
        self,
        loop: asyncio.AbstractEventLoop,
        portfolio: PortfolioFacade,
        msgbus: MessageBus,
        cache: Cache,
        clock: LiveClock,
        config: SandboxExecutionClientConfig,
    ) -> None:
        super().__init__(
            loop,
            portfolio,
            msgbus,
            cache,
            clock,
            config,
        )

        # Adding to avoid error during creation
        self._set_account_id(AccountId(f"{config.venue}-001"))

This is not the go to solution, but maybe it helps somebody in the future tackling Bloomberg.