API enhancements

[dikshantctin]

Can we have a package to maintain the WebSocket subscriptions , and fetching data made simpler with web sockets for multiple securities and similar for rest api's ,

I've been trying to do the same using async and using multiprocessing to make it faster on my end but cannot come up with a scaleable general utility for my use case.

[dikshantctin]

Assume my usecase for 20000 conids.

[Voyz]

Hi @dikshantctin thanks for sharing your request.

I'm sorry but I'm not entirely sure what are you trying to describe here. Are you making a feature request? Or asking for advice? If you could expand on your example use case in more detail I'd appreciate it. Outline what is the bottleneck or pain point in the current API that you're experiencing and how you'd imagine it improved.

[dikshantctin]

import os
import signal
import sys
import time
import redis
import json
from multiprocessing import Process

from ibind import IbkrWsKey, IbkrWsClient, ibind_logs_initialize
from utils import DBUtils, Logger

logger = Logger.getLogger(name)

Initialize logs (no need to log to a file)

ibind_logs_initialize(log_to_file=True)

fetching conids from a list of 30000 symbols
conids = [list of securities
]

Function to split conids into batches

def batch_conids(conids, num_batches):
batch_size = len(conids) // num_batches
return [conids[i * batch_size: (i + 1) * batch_size] for i in range(num_batches)]

Split the conids into batches

conid_batches = batch_conids(conids, 10)

Fields to request market data for

market_data_fields = ['55', '71', '84', '86', '88', '85', '87', '7295', '7296', '70']

Function to process each batch of conids and subscribe to market data

def websocket_subscription_for_batch(conids_batch, redis_key):
ws_client = IbkrWsClient(cacert=os.getenv('IBIND_CACERT', False)) # Each process initializes its own client
ws_client.start()

redis_client = redis.Redis(host='localhost', port=6379, db=0)  # Initialize Redis in each process
market_queue_accessors = ws_client.new_queue_accessor(IbkrWsKey.MARKET_DATA)

# Create subscription requests for each conid in the batch
requests = [{'channel': f'md+{conid}', 'data': {"fields": market_data_fields}} for conid in conids_batch]

def stop(_, _1):
    for request in requests:
        ws_client.unsubscribe(**request)
    ws_client.shutdown()

signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)

for request in requests:
    while not ws_client.subscribe(**request):
        time.sleep(1)

# Process market data as long as the WebSocket client is running
while ws_client.running:
    try:
        while not market_queue_accessors.empty():
            data = market_queue_accessors.get()

            # Serialize the market data as JSON and push it to Redis
            json_data = json.dumps(data)
            redis_client.rpush(redis_key, json_data)
    except KeyboardInterrupt:
        logger.info(f"Shutting down market data processing for batch: {redis_key}")
        break

if name == "main":
# Create processes, each processing a batch of conids
processes = []
for i, conids_batch in enumerate(conid_batches):
redis_key = f'market_data_batch_{i + 1}' # Unique Redis key for each batch
process = Process(target=websocket_subscription_for_batch, args=(conids_batch, redis_key))
processes.append(process)

# Start all processes
for process in processes:
    process.start()

# Wait for all processes to complete
for process in processes:
    process.join()

Let meknow if you need help understanding what i am trying to achieve

[Voyz]

@dikshantctin I'm sorry, but I won't be able to understand your request if you only provide the code without deeper written description. Also, please use proper code formatting when sharing code.

[dikshantctin]

I see that in you usage examples you have are trying to make requests for a few con id's . How should i move forward if i have to use your package to build a scalable solution to work with a universe of securities.

For example: consider i am building a portfolio,
Now that i have the portfolio, I want to fetch latest prices for it.
I can't understand how i can use Ibind to give me live prices for that, and use it within my pandas dataframe of portfolio.

Also, If you see, i have to make a call-in loop to subscribe to market data for security, or when using the rest client as IB API endpoints only allows us to do that. Will it be possible that we can pass a list of accounts, symbols to do such operations, (adding the functionalities of ib_async package here) If yes How?

[Voyz]

@dikshantctin thanks for expanding with a description.

How should i move forward if i have to use your package to build a scalable solution to work with a universe of securities.

This heavily depends on what you're trying to do precisely. If you can ask direct questions regarding the areas you'd need help with, we could see if there's anything that can be shared that would aid you.

I can't understand how i can use Ibind to give me live prices for that

You can query them from the REST API:

https://github.com/Voyz/ibind/wiki/API-Reference-%E2%80%90-IbkrClient#client.ibkr_client_mixins.marketdata_mixin.MarketdataMixin.marketdata_history_by_symbols
https://github.com/Voyz/ibind/blob/master/examples/rest_05_marketdata_history.py

Or get a stream from the WS API:

https://github.com/Voyz/ibind/blob/master/examples/ws_03_market_history.py

Will it be possible that we can pass a list of accounts, symbols to do such operations,

Accounts no, you can only work with one account at a time. This is a limitation imposed by the fact that only one session can be authenticated at a time. If you have a gateway with an authenticated session deployed for each account and you can wrangle where the requests are going out to, then you could do it, but it sounds complex.

Symbols yes, IBind supports querying multiple symbols at a time.

adding the functionalities of ib_async package here

Which functionalities in particular? Please be specific and outline them in detail.