Получение цены бумаги в реальном времени через websocket/streaming

[victordoshenko]

Добрый день!

Помогите, пожалуйста, мне нужен простой пример на C#, в котором бы организовывалась подписка (subscribe, если правильно называю и применяю данный термин) на событие изменения последней цены по заданной бумаге. Как организовывать такую подписку? Как вешать обработчик?

Смотрю пример кода, но тяжело разобраться(это единственный пример по streaming в данном sdk, не уверен, что он вообще подходит для данной темы):

StreamingResponseTests.cs

using System;
using FluentAssertions;
using Newtonsoft.Json;
using Tinkoff.Trading.OpenApi.Models;
using Tinkoff.Trading.OpenApi.Tests.TestHelpers;
using Xunit;
namespace Tinkoff.Trading.OpenApi.Tests
{
    public class StreamingResponseTests
    {
        [Fact]
        public void DeserializeCandleTest()
        {
            var streamingResponse = JsonConvert.DeserializeObject<StreamingResponse>(JsonFile.Read("streaming-candle-response"));
            var response = streamingResponse as StreamingResponse<CandlePayload>;
            var expectedResponse = new CandleResponse(
                new CandlePayload(
                    64.0575m,
                    64.0578m,
                    64.0579m,
                    64.0573m,
                    156,
                    new DateTime(2019, 08, 07, 15, 35, 00, DateTimeKind.Utc),
                    CandleInterval.FiveMinutes,
                    "BBG0013HGFT4"),
                new DateTime(2019, 08, 07, 15, 35, 01, 029, DateTimeKind.Utc).AddTicks(7213));
            response.Should().BeEquivalentTo(expectedResponse);
        }
    }
}

Я хочу получать данные о цене бумаги в реальном времени, т.е. HTTP протокол мне не очень подходит, т.к. нужно каждую секунду(или чаще) отправлять самому запрос на сервер и получать ответ, т.е. примерно так:

var orderbook = await _context.MarketOrderbookAsync(figi, 0);
this.lbPrice.BeginInvoke((MethodInvoker)(() => this.lbPrice.Text = orderbook.LastPrice.ToString()));

Как преобразовать этот код в Streaming API протокол, чтобы при изменении цены бумаги в реальном времени сервер сам дергал нужную функцию на клиенте? (естественно, очень грубо и некорректно описал смысл, но если расписывать правильно, займет много места. В общем нужен пример организации взаимодействия с сервером через websocket-ы). Спасибо!

[dlinnozmey]

К сожалению, примера под рукой нет, но последовательность следующая. Создаёте обработчик:

private void OnStreamingEventReceived(object s, StreamingEventReceivedEventArgs e){
// Здесь тело обработчика
}

Подписываете его к контексту: Context.StreamingEventReceived += OnStreamingEventReceived;

И отправляете через контекст запрос на подписку к, например, стакану: await Context.SendStreamingRequestAsync(StreamingRequest.SubscribeOrderbook(instrument.Figi, 20));

В теле обработчика через e.Event отбираете события стакана (e.Event == "orderbook"), приводите e.Response к типу OrderbookResponse и пользуетесь данными.

[victordoshenko]

Спасибо, помогло! Только не смог вывести LastPrice, т.к. его нет в OrderbookResponse. Поэтому считаю последнюю цену как среднее арифметическое между Ask и Bid:

        private void OnStreamingEventReceived(object s, StreamingEventReceivedEventArgs e)
        {
            // Здесь тело обработчика
            if (e.Response.Event == "orderbook")
            {
                var ore = (OrderbookResponse)e.Response;
                this.lbPrice.BeginInvoke((MethodInvoker)(() => this.lbPrice.Text = ((ore.Payload.Asks[0][0] + ore.Payload.Bids[0][0])/2).ToString()));
            }
        }

[dlinnozmey]

Спасибо, помогло!

Только не смог вывести LastPrice, т.к. его нет в OrderbookResponse. Поэтому считаю последнюю цену как среднее арифметическое между Ask и Bid:

        private void OnStreamingEventReceived(object s, StreamingEventReceivedEventArgs e)

        {

            // Здесь тело обработчика

            if (e.Response.Event == "orderbook")

            {

                var ore = (OrderbookResponse)e.Response;

                this.lbPrice.BeginInvoke((MethodInvoker)(() => this.lbPrice.Text = ((ore.Payload.Asks[0][0] + ore.Payload.Bids[0][0])/2).ToString()));

            }

        }

Чтобы узнать последнюю цену - подписывайтесь на свечи и берите цену закрытия получаемой свечи)

[rus-art]

Если не сложно - можете дополнить пример и сделать PR, другим тоже пригодится

[victordoshenko]

Если не сложно - можете дополнить пример и сделать PR, другим тоже пригодится

Как делать Pool Request? В какой бранч? Как создавать бранч? Есть инструкция?

[victordoshenko]

Если не сложно - можете дополнить пример и сделать PR, другим тоже пригодится

Как делать Pool Request? В какой бранч? Как создавать бранч? Есть инструкция?

В общем разобраться с вашими ветками не хватило скилла, кидаю код формы как есть:

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Windows.Forms;
using Tinkoff.Trading.OpenApi.Models;
using Tinkoff.Trading.OpenApi.Network;
using System.IO;

namespace TinkoffApiWebSocket
{
    public partial class Form1 : Form
    {
        public Context _context;
        private string _accountId;
        private string figi;
        private List<string> figis;

        public Form1()
        {
            InitializeComponent();
        }

        private async void Form1_Load(object sender, EventArgs e)
        {
            var token = File.ReadAllText("token.txt").Trim();
            var connection = ConnectionFactory.GetConnection(token);
            _context = connection.Context;

            try
            {
                var prodAccount = await _context.AccountsAsync();
                _accountId = prodAccount.First().BrokerAccountId;
            } catch (Exception ex)
            {
                MessageBox.Show("Please check correct token in your token.txt. " + ex.Message);
                return;
            }
            await CheckBalanceAsync();

            figis = new List<string>();
            var instrumentList = (await _context.MarketStocksAsync());
            foreach (var instrument in instrumentList.Instruments.Where(r => r.Ticker == "TSLA" || 
                                                                             r.Ticker == "AAPL" ||
                                                                             r.Ticker == "AMZN" ||
                                                                             r.Ticker == "GOOG" ||
                                                                             r.Ticker == "NVDA" ||
                                                                             r.Ticker == "NFLX" ||
                                                                             r.Ticker == "MSFT" ||
                                                                             r.Ticker == "FB" ))
            {
                cbInstrument.Items.Add(instrument.Ticker);
                figis.Add(instrument.Figi);
            }

            _context.StreamingEventReceived += OnStreamingEventReceived;
            cbInstrument.SelectedIndex = 0;
        }

        private async Task CheckBalanceAsync()
        {
            var portfolio = await _context.PortfolioCurrenciesAsync(_accountId);
            string s = "";
            foreach (var currency in portfolio.Currencies) s += $"{currency.Balance} {currency.Currency}  ";
            lbBalance.Text = s;
        }

        private async void cbInstrument_SelectedIndexChanged(object sender, EventArgs e)
        {
            await _context.SendStreamingRequestAsync(StreamingRequest.UnsubscribeOrderbook(figi, 1));
            figi = figis[cbInstrument.SelectedIndex];
            //updatePrice();
            await _context.SendStreamingRequestAsync(StreamingRequest.SubscribeOrderbook(figi, 1));
        }

/*
        private async void updatePrice()
        {
            if (figi == null) { return; }
            var orderbook = await _context.MarketOrderbookAsync(figi, 0);
            this.lbPrice.BeginInvoke((MethodInvoker)(() => this.lbPrice.Text = orderbook.LastPrice.ToString()));
        }
*/
        private void OnStreamingEventReceived(object s, StreamingEventReceivedEventArgs e)
        {
            if (e.Response.Event == "orderbook")
            {
                var ore = (OrderbookResponse)e.Response;
                this.lbPrice.BeginInvoke((MethodInvoker)(() => this.lbPrice.Text = ((ore.Payload.Asks[0][0] + ore.Payload.Bids[0][0])/2).ToString()));
            }
        }

    }
}

[victordoshenko]

Обнаружился первый глюк представленного решения: спустя несколько минут после первой подписки по какой-то причине не обновляется информация в (OrderbookResponse)e.Response.Payload. Работоспособность восстанавливается при отмене подписки и подписке заново (на скриншоте ситуация - поменял бумагу с TSLA на MSFT и обратно - цена стала обновляться нормально. Спустя несколько минут ситуация повторяется - цена замирает и перестает обновляться). На скриншоте видно, что актуальная цена 453.10, а в OrderbookResponse находится 448.79 UPD: если не переключать бумагу(не переподписываться), то спустя примерно 1 минуту работоспособность восстанавливается, затем снова пропадает, потом снова восстанавливается, и т.д. Примерный интервал пропадания и починки - 1 минута. UPD2: Похоже, эта проблема уже поднималась и уже ведутся работы https://github.com/TinkoffCreditSystems/invest-openapi/issues/323. Очень жду исправления.

[ssolarevIDBS]

Получил: System.InvalidOperationException: 'The WebSocket is not connected.' Либо если не через Await : payload = Figi Not Found.