I have a similar query with streaming multiple

    // Create a Historical Pricing stream - specifying the desired 'interday' interval

    var stream = Summaries.Definition(symbols.Take(10).ToList())

        .Fields("BID", "ASK", "HIGH_1", "LOW_1", "OPEN_PRC", "TRDPRC_1", "ACVOL_UNS")

                                              .Interval(Summaries.Interval.P1D)

                                              .Count(5)

                                              .GetStream();



    // Specify the TSI lambda expressions to capture 'Insert' and 'Update' events

    stream.OnInsert((data, stream) => Common.DisplayTable($"INSERT: {DateTime.Now}", data.Table)

          .OnUpdate((data, stream) => Common.DisplayTable($"UPDATE: {DateTime.Now}", data.Table))

          .OnStatus((status, stream) => Console.WriteLine($"Status: {status}"))

          .OnError((error, stream) => Console.WriteLine($"Error: {error}"))

          .Open();

If I use the above... all of my symbols give me the first "OnInsert", however all the following OnUpdate are only for the FIRST symbol in the list.

All in all- how do I stream daily & intraday bars for multiple symbols at once?

Hello @arek.majka

To stream intraday summary data for multiple instruments, you will need to:

- open a desktop session

- create and open a summary stream for each instrument

- retain streams in a list and close them when no longer required

I provided an example below:

using (ISession session = Sessions.GetSession("desktop"))
{
    session.Open();

    string[] symbols = { //your symbols }

    var summaryStreams = new List<ISummariesStream>();

    foreach (string symbol in symbols)
    {
        // Set definition details and create stream object
        var summaryStream = Summaries.Definition(symbol)
            .Fields("DATE", "OPEN_PRC", "HIGH_1", "LOW_1", "TRDPRC_1") //some specific fields
            .Interval(Summaries.Interval.PT1M) //Intraday, 1 minute bars
            .Sessions(HistoricalPricing.Sessions.normal) //Normal market hours session
            .Count(10) //Initial historical data row count
            .GetStream();

        // Set call back methods and open the stream to receive data
        summaryStream
            .OnInsert((data, stream) => {})
            .OnUpdate((data, stream) => {})
            .OnStatus((status, stream) => {})
            .OnError((error, stream) => {})
            .Open();

        summaryStreams.Add(summaryStream);
    }

    // When not required anymore, streams should be closed. 
    summaryStreams.ForEach(stream => stream.Close());
}

About your second question - "An additional issue, is when I run this on September 11, it only gives me data until September 10. Does the historical data request not include the most bar data as well?"

Yes, historical data does not contain the current bar, it only contains data about intervals that have finished.

For the current bar you will receive stream updates because it is ongoing data and not really "history".

Best regards,

Baciu Wahl Cristian

Hello,

Thank you for the response.

Is there no way to bulk request multiple instruments historically though? I was told by one of your colleagues that we should be using bulk requests instead of going one by one?

The requests are relatively slow; just in the fact that even if it takes a second to get the results on a single query, if you have a few hundred markets that gets to be a very long time.

The same goes with streaming. If we are using Summaries.Definition().GetStream() the startup is often times multiple seconds.

Are there other types of requests I should be using? Please note that we are indeed trying to stream bar data and not bid/ask/trade data, as we don't want to be recreating our own bars. We are upgrading from Eikon so trying to keep the same structure in tact.

Thank you.

Hello @arek.majka

When requesting for historical pricing data you should use a Summaries Definition. There are 2 different endpoints depending on the interval.

For intraday intervals, there is a GET endpoint that only accepts one instrument at a time. You can also specify the number of rows you wish to retrieve.

For interday intervals, there is a POST endpoint, that accepts multiple instruments, but it can only retrieve summary containing 1 row for each instrument from the backend. Retrieving multiple rows for an instrument is done similarly to intraday intervals, via a GET point.

I have provided an example bellow:

internal static class DataRetrievalMultipleCases
{
    public static async Task Run()
    {
        Log.Level = LogLevel.Warn;
        Common.Prompt = false;

        using (ISession session = Sessions.GetSession("desktop"))
        {
            session.Open();

            string[] symbols = { "AAPL.O", "TSLA.O", "MSFT.O", "NVDA.O", "AMZN.O", "AMD.O", };

            var tasks = new List<Task>();

            // Case 1 : Intraday 
            System.Diagnostics.Stopwatch watch = System.Diagnostics.Stopwatch.StartNew();
            foreach (var symbol in symbols)
            {
                tasks.Add(GetIntradaySummaryData(symbol));
            }
            await Task.WhenAll(tasks);
            watch.Stop();
            Console.WriteLine($"Time to complete INTRADAY requests: {watch.Elapsed}");

            // Case 2 : Interday 
            System.Diagnostics.Stopwatch watch2 = System.Diagnostics.Stopwatch.StartNew();
            await GetInterdaySummaryData(symbols);
            watch2.Stop();
            Console.WriteLine($"Time to complete INTERDAY requests: {watch2.Elapsed}");

            Console.ReadLine();
        }
    }

    private static async Task GetIntradaySummaryData(string symbol)
    {
        // Intraday - GET - 1 instrument per request - multiple rows can be retrieved
        var data = await Summaries.Definition(symbol)
                                  .Fields("HIGH_1", "LOW_1")
                                  .Interval(Summaries.Interval.PT5M)
                                  .Count(5)
                                  .GetDataAsync();

        Common.DisplayTable(data);
    }

    private static async Task<IDataSetResponse> GetInterdaySummaryData(string[] symbols)
    {
        // Interday - POST - multiple instruments per request - 1 row per instrument retrieved
        var data = await Summaries.Definition(symbols)
                              .Fields("HIGH_1", "LOW_1")
                              .Interval(Summaries.Interval.P1D)
                              .Count(5) //ignored for multiple instruments
                              .GetDataAsync();

        Common.DisplayTable(data);
        return data;
    }
}

Best regards,

Cristian

Hi Cristian,

Thank you. I am trying to utilize the stream and historical to do the hist+stream at the same time like this:

  var stream = Summaries.Definition(symbol).Fields("OPEN_PRC", "HIGH_1", "LOW_1", "TRDPRC_1", "ACVOL_UNS")
                                  .Interval(Summaries.Interval.P1D)
                                  //.Start(new DateTime(2024, 09, 01))
                                  //.End(new DateTime(2024, 10, 03))
                                  //.GetData()
                                  .Count(10)
                                  .GetStream()
                                  .OnInsert((data, stream) => Common.DisplayTable($"INSERT: {DateTime.Now}", data.Table))
                                  .OnUpdate((data, stream) => Common.DisplayTable($"UPDATE: {DateTime.Now}", data.Table))
                                  .OnStatus((status, stream) => Console.WriteLine($"Status: {status}"))
                                  .OnError((error, stream) => Console.WriteLine($"Error: {error}"))
                                  .Open();

I am having 3 issues.

(1)

The first is options aren't working? When I call this for a future like 1CH25 it works fine, but options give me this

" REQ: [reqIndividual()] : 1C430L24

Error: Failed to load meta data. Failed to load Instrument Metadata for item: 1C430L24"

(2) I'm using the .Count(10) to go 10 days back, for example... because if I try .start() and .end() it tells me "Historical data not available". Is this because if I used .GetData() it would work, but is opening the stream assuming I'm giving a count backwards?

(3)

Aside from the options not working, I can do this for a list of futures. However it only seems to work properly in a synchronous manner using .Open(). If I change it to .OpenAsync() then the first one will return me data, but nothing else works. I have tried converting it to a new function with a list of tasks and then doing a Task.WhenAll(taskList) sort of logic.. but I'm getting the same behavior, the others just do nothing and only the first one works.

Please advise on if it's possible to fire off multiple of these historical/stream requests at the same time.

The general concept is let's say I have 100 markets.. when I start up, I want to recover 60min/1day bars since I last stopped and then start streaming.

Thank you.

Hello @arek.majka

Regarding your 3 issues.

1 - When requesting stream data for a given instrument, the library has to request metadata related to that instrument to be able to process incoming stream updates. In the case of "1C430L24", the metadata provided by the backend contains no summary rules. Without summary rules, summary stream data cannot be processed. In the case of "1CH25", the instrument contains summary rules so it is working OK. I have submitted a ticket for the backend teams about this issue. Awaiting their reply.

2 - The Count method is used to specify in the definition how many records of historical data we want to retrieve. A count of 10, means that you want the latest 10 records of historical data. When opening a stream, the first part of processing is retrieving the initial historical data based on definition specification, followed by stream updates via a web socket connection, for the current bar. The retrieval of historical data works in the same way for both cases.

3 - I have provided below an example how you can open multiple summary streams asynchronously using a list of tasks.

static async Task Main(string[] args)
{
    Log.Level = NLog.LogLevel.Trace;

    using (ISession session = Sessions.GetSession("desktop"))
    {
        session.Open();
        
        var symbols = new List<string> { "AAPL.O", "TSLA.O", "MSFT.O", "NVDA.O" };

        var openStreamTasks = new List<Task<Stream.State>>();

        foreach (var symbol in symbols)
        {
            var streamTask = Summaries
               .Definition(symbol)
               .Fields("OPEN_PRC", "HIGH_1", "LOW_1", "TRDPRC_1", "ACVOL_UNS")
               .Interval(Summaries.Interval.P1D)
               .Count(10)
               .GetStream()
               .OnInsert((data, s) => Common.DisplayTable($"INSERT: {DateTime.Now}", data.Table))
               .OnUpdate((data, s) => Common.DisplayTable($"UPDATE: {DateTime.Now}", data.Table))
               .OnStatus((status, s) => Console.WriteLine($"Status: {status}"))
               .OnError((error, s) => Console.WriteLine($"Error: {error}"))
               .OpenAsync();

            openStreamTasks.Add(streamTask);
        }

        await Task.WhenAll(openStreamTasks);

        // Hit any key to exit program
        Console.ReadLine();
    }
}

Best regards,

Baciu Wahl Cristian