подписка на свечи через gRPC протокол

[vadakoff]

Обратился в чат за помощью на tinkoff.ru/invest/, сказали что у них нет прав консультировать по протоколам gRPC, OpenAPI. Может здесь мне помогут. Такой вопрос: Как правильно конструировать запросы и обрабатывать ответы по потоковому gRPC протоколу? Вобщем код(python3.10.4):

import asyncio
import grpc
import inspect

import contracts.marketdata_pb2 as marketdata_pb2
import contracts.marketdata_pb2_grpc as marketdata_pb2_grpc
import contracts.users_pb2 as users_pb2
import contracts.users_pb2_grpc as users_pb2_grpc

access_token = '***'

async def main() -> None:
    channel_credentials = grpc.ssl_channel_credentials()
    call_credentials = grpc.access_token_call_credentials(access_token)
    credentials = grpc.composite_channel_credentials(channel_credentials, call_credentials)

    async with grpc.aio.secure_channel('invest-public-api.tinkoff.ru:443', credentials=credentials) as channel:
        stub = marketdata_pb2_grpc.MarketDataStreamServiceStub(channel)
        stream = stub.MarketDataStream(marketdata_pb2.SubscribeCandlesRequest(
            subscription_action=marketdata_pb2.SUBSCRIPTION_ACTION_SUBSCRIBE,
            instruments=[
                marketdata_pb2.CandleInstrument(
                    figi='BBG0013HGFT4',  # USD/RUB
                    interval=marketdata_pb2.SUBSCRIPTION_INTERVAL_ONE_MINUTE)  # m1
            ],
            waiting_close=True,
        ))
        message = await stream.read()  # getting one message
        print(message)

if __name__ == '__main__':
    asyncio.run(main(), debug=True)

Ловлю такое исключение:

Traceback (most recent call last):
  File "/home/ransom/dev/tinkoff-trader-bot/main.py", line 38, in main
    message = await stream.read()
  File "/home/ransom/dev/tinkoff-trader-bot/.venv/lib/python3.10/site-packages/grpc/aio/_call.py", line 362, in read
    await self._raise_for_status()
  File "/home/ransom/dev/tinkoff-trader-bot/.venv/lib/python3.10/site-packages/grpc/aio/_call.py", line 233, in _raise_for_status
    raise asyncio.CancelledError()
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/ransom/dev/tinkoff-trader-bot/main.py", line 44, in <module>
    asyncio.run(main(), debug=True)
  File "/usr/local/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/local/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
asyncio.exceptions.CancelledError

[MasterDimmy]

https://github.com/MasterDimmy/go-iapi

    mds, err := s.b.client.MarketDataStreamServiceClient.MarketDataStream(context.Background())
    if err != nil {
        return err
    }

    //receive
    go func() {
        defer zipologger.HandlePanic()
        for {
            resp, err := mds.Recv()  <<<<< приходят события, свечки
            if err != nil {
                s.b.log.Printf("LoopEvents Recv 1 ERROR: %s", err.Error())
                time.Sleep(time.Second)
                continue
            }
            s.b.log.Printf("LoopEvents Recv 1 data: %+v", resp)

            //exit now ?
            if func() bool {
                m.Lock()
                defer m.Unlock()
                if err := solve(resp); err != nil {  <<<< внутри solve - делаешь что нужно с resp свечкой\данными
                    s.b.log.Println("LoopEvents Recv 1 остановлен by solve: error: %s", err.Error())
                    return true
                }
                return false
            }() {
                return
            }
        }
    }()

    //подписка на свечи 
    err = mds.Send(&tgrpcapi.MarketDataRequest{
        Payload: &tgrpcapi.MarketDataRequest_SubscribeCandlesRequest{
            SubscribeCandlesRequest: &tgrpcapi.SubscribeCandlesRequest{
                SubscriptionAction: tgrpcapi.SubscriptionAction_SUBSCRIPTION_ACTION_SUBSCRIBE,
                Instruments: []*tgrpcapi.CandleInstrument{
                    &tgrpcapi.CandleInstrument{
                        Figi:     s.b.activeInstrument.FIGI,
                        Interval: tgrpcapi.SubscriptionInterval_SUBSCRIPTION_INTERVAL_ONE_MINUTE},
                },
            },
        },
    })
    if err != nil {
        return err
    }

[vadakoff]

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

[MasterDimmy]

При постановке стопов получилось разгадать, почему появляется 30053 ?

Вот результат эксперимента при попытке выставить стопы от 100.00 до 100.20 с шагом 0.01 для AAPL:

ERROR: 100.020000 rpc error: code = InvalidArgument desc = 30053
ERROR: 100.050000 rpc error: code = InvalidArgument desc = 30053
ERROR: 100.080000 rpc error: code = InvalidArgument desc = 30053
ERROR: 100.110000 rpc error: code = InvalidArgument desc = 30053
ERROR: 100.130000 rpc error: code = InvalidArgument desc = 30053
ERROR: 100.160000 rpc error: code = InvalidArgument desc = 30053
ERROR: 100.190000 rpc error: code = InvalidArgument desc = 30053

Стопы 100, 100.01, 100.03, 100.04, 100.06, 100.07 и тд - выставились успешно.

[vadakoff]

Незнаю, что то видимо у них на сервере. Мне пока до заключения ордеров далеко, Сейчас на этапе получения котировок

[AlexanderVolkovTCS]

При постановке стопов получилось разгадать, почему появляется 30053 ?

В заголовке ответа сервера приходит полный текст ошибки

ERROR: 100.020000 rpc error: code = InvalidArgument desc = 30053

цена 100.02 для AAPL должна выставляться корректно. Проверьте, что она правильно преобразуется в формат MoneyValue, в котором дробная часть передается в миллиардных долях. 100.02 = {units=100,nano=20000000}

[AlexanderVolkovTCS]

Вопросы по python лучше адресовать в https://github.com/Tinkoff/invest-python

[vadakoff]

Все понятно, надо было отправлять запрос в стрим посредством метода .write():

import asyncio
import grpc
import inspect

import tinkoff.invest.grpc.marketdata_pb2 as marketdata_pb2
import tinkoff.invest.grpc.marketdata_pb2_grpc as marketdata_pb2_grpc

access_token = '***'

async def main() -> None:
    channel_credentials = grpc.ssl_channel_credentials()
    call_credentials = grpc.access_token_call_credentials(access_token)
    credentials = grpc.composite_channel_credentials(channel_credentials, call_credentials)

    async with grpc.aio.secure_channel('invest-public-api.tinkoff.ru:443', credentials=credentials) as channel:
        stub = marketdata_pb2_grpc.MarketDataStreamServiceStub(channel)
        stream = stub.MarketDataStream()
        await stream.write(
            marketdata_pb2.MarketDataRequest(
                subscribe_candles_request=marketdata_pb2.SubscribeCandlesRequest(
                    subscription_action=marketdata_pb2.SUBSCRIPTION_ACTION_SUBSCRIBE,
                    instruments=[
                        marketdata_pb2.CandleInstrument(
                            figi="BBG0013HGFT4",
                            interval=marketdata_pb2.SUBSCRIPTION_INTERVAL_ONE_MINUTE
                        )
                    ],
                    waiting_close=False
                )
            )
        )

        while True:
            i = await stream.read()
            print(i.candle)

if __name__ == '__main__':
    asyncio.run(main(), debug=True)