CoCalc -- services.py

GitHub Repository: wiseplat/python-code
Path: blob/master/ invest-robot-contest_TinkoffBotTwitch-main/venv/lib/python3.8/site-packages/tinkoff/invest/services.py
⁷⁸¹⁹ views
1
# pylint:disable=redefined-builtin,too-many-lines
2
import abc
3
import logging
4
from datetime import datetime
5
from typing import Dict, Generator, Iterable, List, Optional, Tuple
6

7
import grpc
8

9
from tinkoff.invest.market_data_stream.market_data_stream_manager import (
10
    MarketDataStreamManager,
11
)
12

13
from . import _grpc_helpers
14
from ._errors import handle_request_error, handle_request_error_gen
15
from .caching.cache_settings import MarketDataCacheSettings
16
from .caching.instrument_date_range_market_data import InstrumentDateRangeData
17
from .caching.instrument_market_data_storage import InstrumentMarketDataStorage
18
from .grpc import (
19
    instruments_pb2,
20
    instruments_pb2_grpc,
21
    marketdata_pb2,
22
    marketdata_pb2_grpc,
23
    operations_pb2,
24
    operations_pb2_grpc,
25
    orders_pb2,
26
    orders_pb2_grpc,
27
    sandbox_pb2,
28
    sandbox_pb2_grpc,
29
    stoporders_pb2,
30
    stoporders_pb2_grpc,
31
    users_pb2,
32
    users_pb2_grpc,
33
)
34
from .logging import get_tracking_id_from_call, log_request
35
from .metadata import get_metadata
36
from .schemas import (
37
    AssetRequest,
38
    AssetResponse,
39
    AssetsRequest,
40
    AssetsResponse,
41
    BondResponse,
42
    BondsResponse,
43
    BrokerReportRequest,
44
    BrokerReportResponse,
45
    CancelOrderRequest,
46
    CancelOrderResponse,
47
    CancelStopOrderRequest,
48
    CancelStopOrderResponse,
49
    CandleInterval,
50
    CloseSandboxAccountRequest,
51
    CloseSandboxAccountResponse,
52
    CurrenciesResponse,
53
    CurrencyResponse,
54
    EditFavoritesActionType,
55
    EditFavoritesRequest,
56
    EditFavoritesRequestInstrument,
57
    EditFavoritesResponse,
58
    EtfResponse,
59
    EtfsResponse,
60
    FutureResponse,
61
    FuturesResponse,
62
    GenerateBrokerReportRequest,
63
    GenerateDividendsForeignIssuerReportRequest,
64
    GetAccountsRequest,
65
    GetAccountsResponse,
66
    GetAccruedInterestsRequest,
67
    GetAccruedInterestsResponse,
68
    GetBondCouponsRequest,
69
    GetBondCouponsResponse,
70
    GetBrokerReportRequest,
71
    GetCandlesRequest,
72
    GetCandlesResponse,
73
    GetDividendsForeignIssuerReportRequest,
74
    GetDividendsForeignIssuerRequest,
75
    GetDividendsForeignIssuerResponse,
76
    GetDividendsRequest,
77
    GetDividendsResponse,
78
    GetFavoritesRequest,
79
    GetFavoritesResponse,
80
    GetFuturesMarginRequest,
81
    GetFuturesMarginResponse,
82
    GetInfoRequest,
83
    GetInfoResponse,
84
    GetLastPricesRequest,
85
    GetLastPricesResponse,
86
    GetLastTradesRequest,
87
    GetLastTradesResponse,
88
    GetMarginAttributesRequest,
89
    GetMarginAttributesResponse,
90
    GetOrderBookRequest,
91
    GetOrderBookResponse,
92
    GetOrdersRequest,
93
    GetOrdersResponse,
94
    GetOrderStateRequest,
95
    GetStopOrdersRequest,
96
    GetStopOrdersResponse,
97
    GetTradingStatusRequest,
98
    GetTradingStatusResponse,
99
    GetUserTariffRequest,
100
    GetUserTariffResponse,
101
    HistoricCandle,
102
    InstrumentIdType,
103
    InstrumentRequest,
104
    InstrumentResponse,
105
    InstrumentsRequest,
106
    InstrumentStatus,
107
    MarketDataRequest,
108
    MarketDataResponse,
109
    MoneyValue,
110
    OpenSandboxAccountRequest,
111
    OpenSandboxAccountResponse,
112
    OperationsRequest,
113
    OperationsResponse,
114
    OperationState,
115
    OrderDirection,
116
    OrderState,
117
    OrderType,
118
    PortfolioRequest,
119
    PortfolioResponse,
120
    PositionsRequest,
121
    PositionsResponse,
122
    PostOrderRequest,
123
    PostOrderResponse,
124
    PostStopOrderRequest,
125
    PostStopOrderResponse,
126
    Quotation,
127
    SandboxPayInRequest,
128
    SandboxPayInResponse,
129
    ShareResponse,
130
    SharesResponse,
131
    StopOrderDirection,
132
    StopOrderExpirationType,
133
    StopOrderType,
134
    TradesStreamRequest,
135
    TradesStreamResponse,
136
    TradingSchedulesRequest,
137
    TradingSchedulesResponse,
138
    WithdrawLimitsRequest,
139
    WithdrawLimitsResponse,
140
)
141
from .typedefs import AccountId
142
from .utils import (
143
    candle_interval_to_timedelta,
144
    datetime_range_floor,
145
    get_intervals,
146
    now,
147
)
148

149
__all__ = (
150
    "Services",
151
    "InstrumentsService",
152
    "MarketDataService",
153
    "MarketDataStreamService",
154
    "OperationsService",
155
    "OrdersStreamService",
156
    "OrdersService",
157
    "UsersService",
158
    "SandboxService",
159
    "StopOrdersService",
160
)
161
logger = logging.getLogger(__name__)
162

163

164
class ICandleGetter(abc.ABC):
165
    @abc.abstractmethod
166
    def get_all_candles(
167
        self,
168
        *,
169
        from_: datetime,
170
        to: Optional[datetime],
171
        interval: CandleInterval,
172
        figi: str,
173
    ) -> Generator[HistoricCandle, None, None]:
174
        pass
175

176

177
class MarketDataCache(ICandleGetter):
178
    def __init__(self, settings: MarketDataCacheSettings, services: "Services"):
179
        self._settings = settings
180
        self._settings.base_cache_dir.mkdir(parents=True, exist_ok=True)
181
        self._services = services
182
        self._figi_cache_storages: Dict[
183
            Tuple[str, CandleInterval], InstrumentMarketDataStorage
184
        ] = {}
185

186
    def _get_candles_from_net(
187
        self, figi: str, interval: CandleInterval, from_: datetime, to: datetime
188
    ) -> Iterable[HistoricCandle]:
189
        yield from self._services.get_all_candles(
190
            figi=figi,
191
            interval=interval,
192
            from_=from_,
193
            to=to,
194
        )
195

196
    def _with_saving_into_cache(
197
        self,
198
        storage: InstrumentMarketDataStorage,
199
        from_net: Iterable[HistoricCandle],
200
        net_range: Tuple[datetime, datetime],
201
    ) -> Iterable[HistoricCandle]:
202
        candles = list(from_net)
203
        storage.update(
204
            [InstrumentDateRangeData(date_range=net_range, historic_candles=candles)]
205
        )
206
        logger.debug("From net [\n%s\n%s\n]", str(net_range[0]), str(net_range[1]))
207
        logger.debug(
208
            "From net real [\n%s\n%s\n]",
209
            str(min(list(map(lambda x: x.time, candles)))),
210
            str(max(list(map(lambda x: x.time, candles)))),
211
        )
212

213
        yield from candles
214

215
    def get_all_candles(
216
        self,
217
        *,
218
        from_: datetime,
219
        to: Optional[datetime] = None,
220
        interval: CandleInterval = CandleInterval(0),
221
        figi: str = "",
222
    ) -> Generator[HistoricCandle, None, None]:
223
        to = to or now()
224
        from_, to = datetime_range_floor((from_, to))
225
        logger.debug("Request [\n%s\n%s\n]", str(from_), str(to))
226

227
        processed_time = from_
228
        figi_cache_storage = self._get_figi_cache_storage(figi=figi, interval=interval)
229
        for cached in figi_cache_storage.get(request_range=(from_, to)):
230
            cached_start, cached_end = cached.date_range
231
            cached_candles = list(cached.historic_candles)
232
            assert cached_start >= processed_time
233
            if cached_start > processed_time:
234
                yield from self._with_saving_into_cache(
235
                    storage=figi_cache_storage,
236
                    from_net=self._get_candles_from_net(
237
                        figi, interval, processed_time, cached_start
238
                    ),
239
                    net_range=(processed_time, cached_start),
240
                )
241
            logger.debug(
242
                "Returning from cache [\n%s\n%s\n]", str(cached_start), str(cached_end)
243
            )
244

245
            yield from cached_candles
246
            processed_time = cached_end
247
        if processed_time + candle_interval_to_timedelta(interval) <= to:
248
            yield from self._with_saving_into_cache(
249
                storage=figi_cache_storage,
250
                from_net=self._get_candles_from_net(figi, interval, processed_time, to),
251
                net_range=(processed_time, to),
252
            )
253

254
    def _get_figi_cache_storage(
255
        self, figi: str, interval: CandleInterval
256
    ) -> InstrumentMarketDataStorage:
257
        figi_tuple = (figi, interval)
258
        storage = self._figi_cache_storages.get(figi_tuple)
259
        if storage is None:
260
            storage = InstrumentMarketDataStorage(
261
                figi=figi, interval=interval, settings=self._settings
262
            )
263
            self._figi_cache_storages[figi_tuple] = storage
264
        return storage  # noqa: R504
265

266

267
class Services(ICandleGetter):
268
    def __init__(
269
        self,
270
        channel: grpc.Channel,
271
        token: str,
272
        sandbox_token: Optional[str] = None,
273
        app_name: Optional[str] = None,
274
    ) -> None:
275
        metadata = get_metadata(token, app_name)
276
        sandbox_metadata = get_metadata(sandbox_token or token, app_name)
277
        self.instruments = InstrumentsService(channel, metadata)
278
        self.market_data = MarketDataService(channel, metadata)
279
        self.market_data_stream = MarketDataStreamService(channel, metadata)
280
        self.operations = OperationsService(channel, metadata)
281
        self.orders_stream = OrdersStreamService(channel, metadata)
282
        self.orders = OrdersService(channel, metadata)
283
        self.users = UsersService(channel, metadata)
284
        self.sandbox = SandboxService(channel, sandbox_metadata)
285
        self.stop_orders = StopOrdersService(channel, metadata)
286

287
    def create_market_data_stream(self) -> MarketDataStreamManager:
288
        return MarketDataStreamManager(
289
            market_data_stream_service=self.market_data_stream
290
        )
291

292
    def cancel_all_orders(self, account_id: AccountId) -> None:
293
        orders_service: OrdersService = self.orders
294
        stop_orders_service: StopOrdersService = self.stop_orders
295

296
        orders_response = orders_service.get_orders(account_id=account_id)
297
        for order in orders_response.orders:
298
            orders_service.cancel_order(account_id=account_id, order_id=order.order_id)
299

300
        stop_orders_response = stop_orders_service.get_stop_orders(
301
            account_id=account_id
302
        )
303
        for stop_order in stop_orders_response.stop_orders:
304
            stop_orders_service.cancel_stop_order(
305
                account_id=account_id, stop_order_id=stop_order.stop_order_id
306
            )
307

308
    def get_all_candles(
309
        self,
310
        *,
311
        from_: datetime,
312
        to: Optional[datetime] = None,
313
        interval: CandleInterval = CandleInterval(0),
314
        figi: str = "",
315
    ) -> Generator[HistoricCandle, None, None]:
316
        to = to or now()
317

318
        for local_from_, local_to in get_intervals(interval, from_, to):
319
            candles_response = self.market_data.get_candles(
320
                figi=figi,
321
                interval=interval,
322
                from_=local_from_,
323
                to=local_to,
324
            )
325
            yield from candles_response.candles
326

327

328
class InstrumentsService(_grpc_helpers.Service):
329
    _stub_factory = instruments_pb2_grpc.InstrumentsServiceStub
330

331
    @handle_request_error("TradingSchedules")
332
    def trading_schedules(
333
        self,
334
        *,
335
        exchange: str = "",
336
        from_: Optional[datetime] = None,
337
        to: Optional[datetime] = None,
338
    ) -> TradingSchedulesResponse:
339
        request = TradingSchedulesRequest()
340
        request.exchange = exchange
341
        if from_ is not None:
342
            request.from_ = from_
343
        if to is not None:
344
            request.to = to
345
        response, call = self.stub.TradingSchedules.with_call(
346
            request=_grpc_helpers.dataclass_to_protobuff(
347
                request, instruments_pb2.TradingSchedulesRequest()
348
            ),
349
            metadata=self.metadata,
350
        )
351
        log_request(get_tracking_id_from_call(call), "TradingSchedules")
352
        return _grpc_helpers.protobuf_to_dataclass(response, TradingSchedulesResponse)
353

354
    @handle_request_error("BondBy")
355
    def bond_by(
356
        self,
357
        *,
358
        id_type: InstrumentIdType = InstrumentIdType(0),
359
        class_code: str = "",
360
        id: str = "",
361
    ) -> BondResponse:
362
        request = InstrumentRequest()
363
        request.id_type = id_type
364
        request.class_code = class_code
365
        request.id = id
366
        response, call = self.stub.BondBy.with_call(
367
            request=_grpc_helpers.dataclass_to_protobuff(
368
                request, instruments_pb2.InstrumentRequest()
369
            ),
370
            metadata=self.metadata,
371
        )
372
        log_request(get_tracking_id_from_call(call), "BondBy")
373
        return _grpc_helpers.protobuf_to_dataclass(response, BondResponse)
374

375
    @handle_request_error("Bonds")
376
    def bonds(
377
        self, *, instrument_status: InstrumentStatus = InstrumentStatus(0)
378
    ) -> BondsResponse:
379
        request = InstrumentsRequest()
380
        request.instrument_status = instrument_status
381
        response, call = self.stub.Bonds.with_call(
382
            request=_grpc_helpers.dataclass_to_protobuff(
383
                request, instruments_pb2.InstrumentsRequest()
384
            ),
385
            metadata=self.metadata,
386
        )
387
        log_request(get_tracking_id_from_call(call), "Bonds")
388
        return _grpc_helpers.protobuf_to_dataclass(response, BondsResponse)
389

390
    @handle_request_error("CurrencyBy")
391
    def currency_by(
392
        self,
393
        *,
394
        id_type: InstrumentIdType = InstrumentIdType(0),
395
        class_code: str = "",
396
        id: str = "",
397
    ) -> CurrencyResponse:
398
        request = InstrumentRequest()
399
        request.id_type = id_type
400
        request.class_code = class_code
401
        request.id = id
402
        response, call = self.stub.CurrencyBy.with_call(
403
            request=_grpc_helpers.dataclass_to_protobuff(
404
                request, instruments_pb2.InstrumentRequest()
405
            ),
406
            metadata=self.metadata,
407
        )
408
        log_request(get_tracking_id_from_call(call), "CurrencyBy")
409
        return _grpc_helpers.protobuf_to_dataclass(response, CurrencyResponse)
410

411
    @handle_request_error("Currencies")
412
    def currencies(
413
        self, *, instrument_status: InstrumentStatus = InstrumentStatus(0)
414
    ) -> CurrenciesResponse:
415
        request = InstrumentsRequest()
416
        request.instrument_status = instrument_status
417
        response, call = self.stub.Currencies.with_call(
418
            request=_grpc_helpers.dataclass_to_protobuff(
419
                request, instruments_pb2.InstrumentsRequest()
420
            ),
421
            metadata=self.metadata,
422
        )
423
        log_request(get_tracking_id_from_call(call), "Currencies")
424
        return _grpc_helpers.protobuf_to_dataclass(response, CurrenciesResponse)
425

426
    @handle_request_error("EtfBy")
427
    def etf_by(
428
        self,
429
        *,
430
        id_type: InstrumentIdType = InstrumentIdType(0),
431
        class_code: str = "",
432
        id: str = "",
433
    ) -> EtfResponse:
434
        request = InstrumentRequest()
435
        request.id_type = id_type
436
        request.class_code = class_code
437
        request.id = id
438
        response, call = self.stub.EtfBy.with_call(
439
            request=_grpc_helpers.dataclass_to_protobuff(
440
                request, instruments_pb2.InstrumentRequest()
441
            ),
442
            metadata=self.metadata,
443
        )
444
        log_request(get_tracking_id_from_call(call), "EtfBy")
445
        return _grpc_helpers.protobuf_to_dataclass(response, EtfResponse)
446

447
    @handle_request_error("Etfs")
448
    def etfs(
449
        self, *, instrument_status: InstrumentStatus = InstrumentStatus(0)
450
    ) -> EtfsResponse:
451
        request = InstrumentsRequest()
452
        request.instrument_status = instrument_status
453
        response, call = self.stub.Etfs.with_call(
454
            request=_grpc_helpers.dataclass_to_protobuff(
455
                request, instruments_pb2.InstrumentsRequest()
456
            ),
457
            metadata=self.metadata,
458
        )
459
        log_request(get_tracking_id_from_call(call), "Etfs")
460
        return _grpc_helpers.protobuf_to_dataclass(response, EtfsResponse)
461

462
    @handle_request_error("FutureBy")
463
    def future_by(
464
        self,
465
        *,
466
        id_type: InstrumentIdType = InstrumentIdType(0),
467
        class_code: str = "",
468
        id: str = "",
469
    ) -> FutureResponse:
470
        request = InstrumentRequest()
471
        request.id_type = id_type
472
        request.class_code = class_code
473
        request.id = id
474
        response, call = self.stub.FutureBy.with_call(
475
            request=_grpc_helpers.dataclass_to_protobuff(
476
                request, instruments_pb2.InstrumentRequest()
477
            ),
478
            metadata=self.metadata,
479
        )
480
        log_request(get_tracking_id_from_call(call), "FutureBy")
481
        return _grpc_helpers.protobuf_to_dataclass(response, FutureResponse)
482

483
    @handle_request_error("Futures")
484
    def futures(
485
        self, *, instrument_status: InstrumentStatus = InstrumentStatus(0)
486
    ) -> FuturesResponse:
487
        request = InstrumentsRequest()
488
        request.instrument_status = instrument_status
489
        response, call = self.stub.Futures.with_call(
490
            request=_grpc_helpers.dataclass_to_protobuff(
491
                request, instruments_pb2.InstrumentsRequest()
492
            ),
493
            metadata=self.metadata,
494
        )
495
        log_request(get_tracking_id_from_call(call), "Futures")
496
        return _grpc_helpers.protobuf_to_dataclass(response, FuturesResponse)
497

498
    @handle_request_error("ShareBy")
499
    def share_by(
500
        self,
501
        *,
502
        id_type: InstrumentIdType = InstrumentIdType(0),
503
        class_code: str = "",
504
        id: str = "",
505
    ) -> ShareResponse:
506
        request = InstrumentRequest()
507
        request.id_type = id_type
508
        request.class_code = class_code
509
        request.id = id
510
        response, call = self.stub.ShareBy.with_call(
511
            request=_grpc_helpers.dataclass_to_protobuff(
512
                request, instruments_pb2.InstrumentRequest()
513
            ),
514
            metadata=self.metadata,
515
        )
516
        log_request(get_tracking_id_from_call(call), "ShareBy")
517
        return _grpc_helpers.protobuf_to_dataclass(response, ShareResponse)
518

519
    @handle_request_error("Shares")
520
    def shares(
521
        self, *, instrument_status: InstrumentStatus = InstrumentStatus(0)
522
    ) -> SharesResponse:
523
        request = InstrumentsRequest()
524
        request.instrument_status = instrument_status
525
        response, call = self.stub.Shares.with_call(
526
            request=_grpc_helpers.dataclass_to_protobuff(
527
                request, instruments_pb2.InstrumentsRequest()
528
            ),
529
            metadata=self.metadata,
530
        )
531
        log_request(get_tracking_id_from_call(call), "Shares")
532
        return _grpc_helpers.protobuf_to_dataclass(response, SharesResponse)
533

534
    @handle_request_error("GetAccruedInterests")
535
    def get_accrued_interests(
536
        self,
537
        *,
538
        figi: str = "",
539
        from_: Optional[datetime] = None,
540
        to: Optional[datetime] = None,
541
    ) -> GetAccruedInterestsResponse:
542
        request = GetAccruedInterestsRequest()
543
        request.figi = figi
544
        if from_ is not None:
545
            request.from_ = from_
546
        if to is not None:
547
            request.to = to
548
        response, call = self.stub.GetAccruedInterests.with_call(
549
            request=_grpc_helpers.dataclass_to_protobuff(
550
                request, instruments_pb2.GetAccruedInterestsRequest()
551
            ),
552
            metadata=self.metadata,
553
        )
554
        log_request(get_tracking_id_from_call(call), "GetAccruedInterests")
555
        return _grpc_helpers.protobuf_to_dataclass(
556
            response, GetAccruedInterestsResponse
557
        )
558

559
    @handle_request_error("GetFuturesMargin")
560
    def get_futures_margin(self, *, figi: str = "") -> GetFuturesMarginResponse:
561
        request = GetFuturesMarginRequest()
562
        request.figi = figi
563
        response, call = self.stub.GetFuturesMargin.with_call(
564
            request=_grpc_helpers.dataclass_to_protobuff(
565
                request, instruments_pb2.GetFuturesMarginRequest()
566
            ),
567
            metadata=self.metadata,
568
        )
569
        log_request(get_tracking_id_from_call(call), "GetFuturesMargin")
570
        return _grpc_helpers.protobuf_to_dataclass(response, GetFuturesMarginResponse)
571

572
    @handle_request_error("GetInstrumentBy")
573
    def get_instrument_by(
574
        self,
575
        *,
576
        id_type: InstrumentIdType = InstrumentIdType(0),
577
        class_code: str = "",
578
        id: str = "",
579
    ) -> InstrumentResponse:
580
        request = InstrumentRequest()
581
        request.id_type = id_type
582
        request.class_code = class_code
583
        request.id = id
584
        response, call = self.stub.GetInstrumentBy.with_call(
585
            request=_grpc_helpers.dataclass_to_protobuff(
586
                request, instruments_pb2.InstrumentRequest()
587
            ),
588
            metadata=self.metadata,
589
        )
590
        log_request(get_tracking_id_from_call(call), "GetInstrumentBy")
591
        return _grpc_helpers.protobuf_to_dataclass(response, InstrumentResponse)
592

593
    @handle_request_error("GetDividends")
594
    def get_dividends(
595
        self,
596
        *,
597
        figi: str = "",
598
        from_: Optional[datetime] = None,
599
        to: Optional[datetime] = None,
600
    ) -> GetDividendsResponse:
601
        request = GetDividendsRequest()
602
        request.figi = figi
603
        if from_ is not None:
604
            request.from_ = from_
605
        if to is not None:
606
            request.to = to
607
        response, call = self.stub.GetDividends.with_call(
608
            request=_grpc_helpers.dataclass_to_protobuff(
609
                request, instruments_pb2.GetDividendsRequest()
610
            ),
611
            metadata=self.metadata,
612
        )
613
        log_request(get_tracking_id_from_call(call), "GetDividends")
614
        return _grpc_helpers.protobuf_to_dataclass(response, GetDividendsResponse)
615

616
    @handle_request_error("GetBondCoupons")
617
    def get_bond_coupons(
618
        self,
619
        *,
620
        figi: str = "",
621
        from_: Optional[datetime] = None,
622
        to: Optional[datetime] = None,
623
    ) -> GetBondCouponsResponse:
624
        request = GetBondCouponsRequest()
625
        request.figi = figi
626
        if from_ is not None:
627
            request.from_ = from_
628
        if to is not None:
629
            request.to = to
630
        response, call = self.stub.GetBondCoupons.with_call(
631
            request=_grpc_helpers.dataclass_to_protobuff(
632
                request, instruments_pb2.GetBondCouponsRequest()
633
            ),
634
            metadata=self.metadata,
635
        )
636
        log_request(get_tracking_id_from_call(call), "GetBondCoupons")
637
        return _grpc_helpers.protobuf_to_dataclass(response, GetBondCouponsResponse)
638

639
    @handle_request_error("GetAssetBy")
640
    def get_asset_by(
641
        self,
642
        *,
643
        id: str = "",
644
    ) -> AssetResponse:
645
        request = AssetRequest()
646
        request.id = id
647
        response, call = self.stub.GetAssetBy.with_call(
648
            request=_grpc_helpers.dataclass_to_protobuff(
649
                request, instruments_pb2.AssetRequest()
650
            ),
651
            metadata=self.metadata,
652
        )
653
        log_request(get_tracking_id_from_call(call), "GetAssetBy")
654
        return _grpc_helpers.protobuf_to_dataclass(response, AssetResponse)
655

656
    @handle_request_error("GetAssets")
657
    def get_assets(
658
        self,
659
    ) -> AssetsResponse:
660
        request = AssetsRequest()
661
        response, call = self.stub.GetAssets.with_call(
662
            request=_grpc_helpers.dataclass_to_protobuff(
663
                request, instruments_pb2.AssetsRequest()
664
            ),
665
            metadata=self.metadata,
666
        )
667
        log_request(get_tracking_id_from_call(call), "GetAssets")
668
        return _grpc_helpers.protobuf_to_dataclass(response, AssetsResponse)
669

670
    @handle_request_error("GetFavorites")
671
    def get_favorites(
672
        self,
673
    ) -> GetFavoritesResponse:
674
        request = GetFavoritesRequest()
675
        response, call = self.stub.GetFavorites.with_call(
676
            request=_grpc_helpers.dataclass_to_protobuff(
677
                request, instruments_pb2.GetFavoritesRequest()
678
            ),
679
            metadata=self.metadata,
680
        )
681
        log_request(get_tracking_id_from_call(call), "GetFavorites")
682
        return _grpc_helpers.protobuf_to_dataclass(response, GetFavoritesResponse)
683

684
    @handle_request_error("EditFavorites")
685
    def edit_favorites(
686
        self,
687
        *,
688
        instruments: Optional[List[EditFavoritesRequestInstrument]] = None,
689
        action_type: Optional[EditFavoritesActionType] = None,
690
    ) -> EditFavoritesResponse:
691
        request = EditFavoritesRequest()
692
        if action_type is not None:
693
            request.action_type = action_type
694
        if instruments is not None:
695
            request.instruments = instruments
696
        response, call = self.stub.EditFavorites.with_call(
697
            request=_grpc_helpers.dataclass_to_protobuff(
698
                request, instruments_pb2.EditFavoritesRequest()
699
            ),
700
            metadata=self.metadata,
701
        )
702
        log_request(get_tracking_id_from_call(call), "EditFavorites")
703
        return _grpc_helpers.protobuf_to_dataclass(response, EditFavoritesResponse)
704

705

706
class MarketDataService(_grpc_helpers.Service):
707
    _stub_factory = marketdata_pb2_grpc.MarketDataServiceStub
708

709
    @handle_request_error("GetCandles")
710
    def get_candles(
711
        self,
712
        *,
713
        figi: str = "",
714
        from_: Optional[datetime] = None,
715
        to: Optional[datetime] = None,
716
        interval: CandleInterval = CandleInterval(0),
717
    ) -> GetCandlesResponse:
718
        request = GetCandlesRequest()
719
        request.figi = figi
720
        if from_ is not None:
721
            request.from_ = from_
722
        if to is not None:
723
            request.to = to
724
        request.interval = interval
725
        response, call = self.stub.GetCandles.with_call(
726
            request=_grpc_helpers.dataclass_to_protobuff(
727
                request, marketdata_pb2.GetCandlesRequest()
728
            ),
729
            metadata=self.metadata,
730
        )
731
        log_request(get_tracking_id_from_call(call), "GetCandles")
732
        return _grpc_helpers.protobuf_to_dataclass(response, GetCandlesResponse)
733

734
    @handle_request_error("GetLastPrices")
735
    def get_last_prices(
736
        self, *, figi: Optional[List[str]] = None
737
    ) -> GetLastPricesResponse:
738
        figi = figi or []
739

740
        request = GetLastPricesRequest()
741
        request.figi = figi
742
        response, call = self.stub.GetLastPrices.with_call(
743
            request=_grpc_helpers.dataclass_to_protobuff(
744
                request, marketdata_pb2.GetLastPricesRequest()
745
            ),
746
            metadata=self.metadata,
747
        )
748
        log_request(get_tracking_id_from_call(call), "GetLastPrices")
749
        return _grpc_helpers.protobuf_to_dataclass(response, GetLastPricesResponse)
750

751
    @handle_request_error("GetOrderBook")
752
    def get_order_book(self, *, figi: str = "", depth: int = 0) -> GetOrderBookResponse:
753
        request = GetOrderBookRequest()
754
        request.figi = figi
755
        request.depth = depth
756
        response, call = self.stub.GetOrderBook.with_call(
757
            request=_grpc_helpers.dataclass_to_protobuff(
758
                request, marketdata_pb2.GetOrderBookRequest()
759
            ),
760
            metadata=self.metadata,
761
        )
762
        log_request(get_tracking_id_from_call(call), "GetOrderBook")
763
        return _grpc_helpers.protobuf_to_dataclass(response, GetOrderBookResponse)
764

765
    @handle_request_error("GetTradingStatus")
766
    def get_trading_status(self, *, figi: str = "") -> GetTradingStatusResponse:
767
        request = GetTradingStatusRequest()
768
        request.figi = figi
769
        response, call = self.stub.GetTradingStatus.with_call(
770
            request=_grpc_helpers.dataclass_to_protobuff(
771
                request, marketdata_pb2.GetTradingStatusRequest()
772
            ),
773
            metadata=self.metadata,
774
        )
775
        log_request(get_tracking_id_from_call(call), "GetTradingStatus")
776
        return _grpc_helpers.protobuf_to_dataclass(response, GetTradingStatusResponse)
777

778
    @handle_request_error("GetLastTrades")
779
    def get_last_trades(
780
        self,
781
        *,
782
        figi: str = "",
783
        from_: Optional[datetime] = None,
784
        to: Optional[datetime] = None,
785
    ) -> GetLastTradesResponse:
786
        request = GetLastTradesRequest()
787
        request.figi = figi
788
        if from_ is not None:
789
            request.from_ = from_
790
        if to is not None:
791
            request.to = to
792
        response, call = self.stub.GetLastTrades.with_call(
793
            request=_grpc_helpers.dataclass_to_protobuff(
794
                request, marketdata_pb2.GetLastTradesRequest()
795
            ),
796
            metadata=self.metadata,
797
        )
798
        log_request(get_tracking_id_from_call(call), "GetLastTrades")
799
        return _grpc_helpers.protobuf_to_dataclass(response, GetLastTradesResponse)
800

801

802
class MarketDataStreamService(_grpc_helpers.Service):
803
    _stub_factory = marketdata_pb2_grpc.MarketDataStreamServiceStub
804

805
    @staticmethod
806
    def _convert_market_data_stream_request(
807
        request_iterator: Iterable[MarketDataRequest],
808
    ) -> Iterable[marketdata_pb2.MarketDataRequest]:
809
        for request in request_iterator:
810
            yield _grpc_helpers.dataclass_to_protobuff(
811
                request, marketdata_pb2.MarketDataRequest()
812
            )
813

814
    @handle_request_error_gen("MarketDataStream")
815
    def market_data_stream(
816
        self,
817
        request_iterator: Iterable[MarketDataRequest],
818
    ) -> Iterable[MarketDataResponse]:
819
        for response in self.stub.MarketDataStream(
820
            request_iterator=self._convert_market_data_stream_request(request_iterator),
821
            metadata=self.metadata,
822
        ):
823
            yield _grpc_helpers.protobuf_to_dataclass(response, MarketDataResponse)
824

825

826
class OperationsService(_grpc_helpers.Service):
827
    _stub_factory = operations_pb2_grpc.OperationsServiceStub
828

829
    @handle_request_error("GetOperations")
830
    def get_operations(
831
        self,
832
        *,
833
        account_id: str = "",
834
        from_: Optional[datetime] = None,
835
        to: Optional[datetime] = None,
836
        state: OperationState = OperationState(0),
837
        figi: str = "",
838
    ) -> OperationsResponse:
839
        request = OperationsRequest()
840
        request.account_id = account_id
841
        if from_ is not None:
842
            request.from_ = from_
843
        if to is not None:
844
            request.to = to
845
        request.state = state
846
        request.figi = figi
847
        response, call = self.stub.GetOperations.with_call(
848
            request=_grpc_helpers.dataclass_to_protobuff(
849
                request, operations_pb2.OperationsRequest()
850
            ),
851
            metadata=self.metadata,
852
        )
853
        log_request(get_tracking_id_from_call(call), "GetOperations")
854
        return _grpc_helpers.protobuf_to_dataclass(response, OperationsResponse)
855

856
    @handle_request_error("GetPortfolio")
857
    def get_portfolio(self, *, account_id: str = "") -> PortfolioResponse:
858
        request = PortfolioRequest()
859
        request.account_id = account_id
860
        response, call = self.stub.GetPortfolio.with_call(
861
            request=_grpc_helpers.dataclass_to_protobuff(
862
                request, operations_pb2.PortfolioRequest()
863
            ),
864
            metadata=self.metadata,
865
        )
866
        log_request(get_tracking_id_from_call(call), "GetPortfolio")
867
        return _grpc_helpers.protobuf_to_dataclass(response, PortfolioResponse)
868

869
    @handle_request_error("GetPositions")
870
    def get_positions(self, *, account_id: str = "") -> PositionsResponse:
871
        request = PositionsRequest()
872
        request.account_id = account_id
873
        response, call = self.stub.GetPositions.with_call(
874
            request=_grpc_helpers.dataclass_to_protobuff(
875
                request, operations_pb2.PositionsRequest()
876
            ),
877
            metadata=self.metadata,
878
        )
879
        log_request(get_tracking_id_from_call(call), "GetPositions")
880
        return _grpc_helpers.protobuf_to_dataclass(response, PositionsResponse)
881

882
    @handle_request_error("GetWithdrawLimits")
883
    def get_withdraw_limits(self, *, account_id: str = "") -> WithdrawLimitsResponse:
884
        request = WithdrawLimitsRequest()
885
        request.account_id = account_id
886
        response, call = self.stub.GetWithdrawLimits.with_call(
887
            request=_grpc_helpers.dataclass_to_protobuff(
888
                request, operations_pb2.WithdrawLimitsRequest()
889
            ),
890
            metadata=self.metadata,
891
        )
892
        log_request(get_tracking_id_from_call(call), "GetWithdrawLimits")
893
        return _grpc_helpers.protobuf_to_dataclass(response, WithdrawLimitsResponse)
894

895
    @handle_request_error("GetBrokerReport")
896
    def get_broker_report(
897
        self,
898
        *,
899
        generate_broker_report_request: Optional[GenerateBrokerReportRequest] = None,
900
        get_broker_report_request: Optional[GetBrokerReportRequest] = None,
901
    ) -> BrokerReportResponse:
902
        request = BrokerReportRequest()
903
        if generate_broker_report_request:
904
            request.generate_broker_report_request = generate_broker_report_request
905
        if get_broker_report_request:
906
            request.get_broker_report_request = get_broker_report_request
907
        response, call = self.stub.GetBrokerReport.with_call(
908
            request=_grpc_helpers.dataclass_to_protobuff(
909
                request, operations_pb2.BrokerReportRequest()
910
            ),
911
            metadata=self.metadata,
912
        )
913
        log_request(get_tracking_id_from_call(call), "GetBrokerReport")
914
        return _grpc_helpers.protobuf_to_dataclass(response, BrokerReportResponse)
915

916
    @handle_request_error("GetDividendsForeignIssuer")
917
    def get_dividends_foreign_issuer(
918
        self,
919
        *,
920
        generate_div_foreign_issuer_report: Optional[
921
            GenerateDividendsForeignIssuerReportRequest
922
        ] = None,
923
        get_div_foreign_issuer_report: Optional[
924
            GetDividendsForeignIssuerReportRequest
925
        ] = None,
926
    ) -> GetDividendsForeignIssuerResponse:
927
        request = GetDividendsForeignIssuerRequest()
928
        if generate_div_foreign_issuer_report is not None:
929
            request.generate_div_foreign_issuer_report = (
930
                generate_div_foreign_issuer_report
931
            )
932
        if get_div_foreign_issuer_report is not None:
933
            request.get_div_foreign_issuer_report = get_div_foreign_issuer_report
934
        response, call = self.stub.GetDividendsForeignIssuer.with_call(
935
            request=_grpc_helpers.dataclass_to_protobuff(
936
                request, operations_pb2.GetDividendsForeignIssuerRequest()
937
            ),
938
            metadata=self.metadata,
939
        )
940
        log_request(get_tracking_id_from_call(call), "GetDividendsForeignIssuer")
941
        return _grpc_helpers.protobuf_to_dataclass(
942
            response, GetDividendsForeignIssuerResponse
943
        )
944

945

946
class OrdersStreamService(_grpc_helpers.Service):
947
    _stub_factory = orders_pb2_grpc.OrdersStreamServiceStub
948

949
    @handle_request_error_gen("TradesStream")
950
    def trades_stream(self, accounts: List[str]) -> Iterable[TradesStreamResponse]:
951
        request = TradesStreamRequest(accounts=accounts)
952
        for response in self.stub.TradesStream(
953
            request=_grpc_helpers.dataclass_to_protobuff(
954
                request, orders_pb2.TradesStreamRequest()
955
            ),
956
            metadata=self.metadata,
957
        ):
958
            yield _grpc_helpers.protobuf_to_dataclass(response, TradesStreamResponse)
959

960

961
class OrdersService(_grpc_helpers.Service):
962
    _stub_factory = orders_pb2_grpc.OrdersServiceStub
963

964
    @handle_request_error("PostOrder")
965
    def post_order(
966
        self,
967
        *,
968
        figi: str = "",
969
        quantity: int = 0,
970
        price: Optional[Quotation] = None,
971
        direction: OrderDirection = OrderDirection(0),
972
        account_id: str = "",
973
        order_type: OrderType = OrderType(0),
974
        order_id: str = "",
975
    ) -> PostOrderResponse:
976
        request = PostOrderRequest()
977
        request.figi = figi
978
        request.quantity = quantity
979
        if price is not None:
980
            request.price = price
981
        request.direction = direction
982
        request.account_id = account_id
983
        request.order_type = order_type
984
        request.order_id = order_id
985
        response, call = self.stub.PostOrder.with_call(
986
            request=_grpc_helpers.dataclass_to_protobuff(
987
                request, orders_pb2.PostOrderRequest()
988
            ),
989
            metadata=self.metadata,
990
        )
991
        log_request(get_tracking_id_from_call(call), "PostOrder")
992
        return _grpc_helpers.protobuf_to_dataclass(response, PostOrderResponse)
993

994
    @handle_request_error("CancelOrder")
995
    def cancel_order(
996
        self, *, account_id: str = "", order_id: str = ""
997
    ) -> CancelOrderResponse:
998
        request = CancelOrderRequest()
999
        request.account_id = account_id
1000
        request.order_id = order_id
1001
        response, call = self.stub.CancelOrder.with_call(
1002
            request=_grpc_helpers.dataclass_to_protobuff(
1003
                request, orders_pb2.CancelOrderRequest()
1004
            ),
1005
            metadata=self.metadata,
1006
        )
1007
        log_request(get_tracking_id_from_call(call), "CancelOrder")
1008
        return _grpc_helpers.protobuf_to_dataclass(response, CancelOrderResponse)
1009

1010
    @handle_request_error("GetOrderState")
1011
    def get_order_state(
1012
        self, *, account_id: str = "", order_id: str = ""
1013
    ) -> OrderState:
1014
        request = GetOrderStateRequest()
1015
        request.account_id = account_id
1016
        request.order_id = order_id
1017
        response, call = self.stub.GetOrderState.with_call(
1018
            request=_grpc_helpers.dataclass_to_protobuff(
1019
                request, orders_pb2.GetOrderStateRequest()
1020
            ),
1021
            metadata=self.metadata,
1022
        )
1023
        log_request(get_tracking_id_from_call(call), "GetOrderState")
1024
        return _grpc_helpers.protobuf_to_dataclass(response, OrderState)
1025

1026
    @handle_request_error("GetOrders")
1027
    def get_orders(self, *, account_id: str = "") -> GetOrdersResponse:
1028
        request = GetOrdersRequest()
1029
        request.account_id = account_id
1030
        response, call = self.stub.GetOrders.with_call(
1031
            request=_grpc_helpers.dataclass_to_protobuff(
1032
                request, orders_pb2.GetOrdersRequest()
1033
            ),
1034
            metadata=self.metadata,
1035
        )
1036
        log_request(get_tracking_id_from_call(call), "GetOrders")
1037
        return _grpc_helpers.protobuf_to_dataclass(response, GetOrdersResponse)
1038

1039

1040
class UsersService(_grpc_helpers.Service):
1041
    _stub_factory = users_pb2_grpc.UsersServiceStub
1042

1043
    @handle_request_error("GetAccounts")
1044
    def get_accounts(self) -> GetAccountsResponse:
1045
        request = GetAccountsRequest()
1046
        response, call = self.stub.GetAccounts.with_call(
1047
            request=_grpc_helpers.dataclass_to_protobuff(
1048
                request, users_pb2.GetAccountsRequest()
1049
            ),
1050
            metadata=self.metadata,
1051
        )
1052
        log_request(get_tracking_id_from_call(call), "GetAccounts")
1053
        return _grpc_helpers.protobuf_to_dataclass(response, GetAccountsResponse)
1054

1055
    @handle_request_error("GetMarginAttributes")
1056
    def get_margin_attributes(
1057
        self, *, account_id: str = ""
1058
    ) -> GetMarginAttributesResponse:
1059
        request = GetMarginAttributesRequest()
1060
        request.account_id = account_id
1061
        response, call = self.stub.GetMarginAttributes.with_call(
1062
            request=_grpc_helpers.dataclass_to_protobuff(
1063
                request, users_pb2.GetMarginAttributesRequest()
1064
            ),
1065
            metadata=self.metadata,
1066
        )
1067
        log_request(get_tracking_id_from_call(call), "GetMarginAttributes")
1068
        return _grpc_helpers.protobuf_to_dataclass(
1069
            response, GetMarginAttributesResponse
1070
        )
1071

1072
    @handle_request_error("GetUserTariff")
1073
    def get_user_tariff(self) -> GetUserTariffResponse:
1074
        request = GetUserTariffRequest()
1075
        response, call = self.stub.GetUserTariff.with_call(
1076
            request=_grpc_helpers.dataclass_to_protobuff(
1077
                request, users_pb2.GetUserTariffRequest()
1078
            ),
1079
            metadata=self.metadata,
1080
        )
1081
        log_request(get_tracking_id_from_call(call), "GetUserTariff")
1082
        return _grpc_helpers.protobuf_to_dataclass(response, GetUserTariffResponse)
1083

1084
    @handle_request_error("GetInfo")
1085
    def get_info(self) -> GetInfoResponse:
1086
        request = GetInfoRequest()
1087
        response, call = self.stub.GetInfo.with_call(
1088
            request=_grpc_helpers.dataclass_to_protobuff(
1089
                request, users_pb2.GetInfoRequest()
1090
            ),
1091
            metadata=self.metadata,
1092
        )
1093
        log_request(get_tracking_id_from_call(call), "GetInfo")
1094
        return _grpc_helpers.protobuf_to_dataclass(response, GetInfoResponse)
1095

1096

1097
class SandboxService(_grpc_helpers.Service):
1098
    _stub_factory = sandbox_pb2_grpc.SandboxServiceStub
1099

1100
    @handle_request_error("OpenSandboxAccount")
1101
    def open_sandbox_account(self) -> OpenSandboxAccountResponse:
1102
        request = OpenSandboxAccountRequest()
1103
        response, call = self.stub.OpenSandboxAccount.with_call(
1104
            request=_grpc_helpers.dataclass_to_protobuff(
1105
                request, sandbox_pb2.OpenSandboxAccountRequest()
1106
            ),
1107
            metadata=self.metadata,
1108
        )
1109
        log_request(get_tracking_id_from_call(call), "OpenSandboxAccount")
1110
        return _grpc_helpers.protobuf_to_dataclass(response, OpenSandboxAccountResponse)
1111

1112
    @handle_request_error("GetSandboxAccounts")
1113
    def get_sandbox_accounts(self) -> GetAccountsResponse:
1114
        request = GetAccountsRequest()
1115
        response, call = self.stub.GetSandboxAccounts.with_call(
1116
            request=_grpc_helpers.dataclass_to_protobuff(
1117
                request, users_pb2.GetAccountsRequest()
1118
            ),
1119
            metadata=self.metadata,
1120
        )
1121
        log_request(get_tracking_id_from_call(call), "GetSandboxAccounts")
1122
        return _grpc_helpers.protobuf_to_dataclass(response, GetAccountsResponse)
1123

1124
    @handle_request_error("CloseSandboxAccount")
1125
    def close_sandbox_account(
1126
        self, *, account_id: str = ""
1127
    ) -> CloseSandboxAccountResponse:
1128
        request = CloseSandboxAccountRequest()
1129
        request.account_id = account_id
1130
        response, call = self.stub.CloseSandboxAccount.with_call(
1131
            request=_grpc_helpers.dataclass_to_protobuff(
1132
                request, sandbox_pb2.CloseSandboxAccountRequest()
1133
            ),
1134
            metadata=self.metadata,
1135
        )
1136
        log_request(get_tracking_id_from_call(call), "CloseSandboxAccount")
1137
        return _grpc_helpers.protobuf_to_dataclass(
1138
            response, CloseSandboxAccountResponse
1139
        )
1140

1141
    @handle_request_error("PostSandboxOrder")
1142
    def post_sandbox_order(
1143
        self,
1144
        *,
1145
        figi: str = "",
1146
        quantity: int = 0,
1147
        price: Optional[Quotation] = None,
1148
        direction: OrderDirection = OrderDirection(0),
1149
        account_id: str = "",
1150
        order_type: OrderType = OrderType(0),
1151
        order_id: str = "",
1152
    ) -> PostOrderResponse:
1153
        request = PostOrderRequest()
1154
        request.figi = figi
1155
        request.quantity = quantity
1156
        if price is not None:
1157
            request.price = price
1158
        request.direction = direction
1159
        request.account_id = account_id
1160
        request.order_type = order_type
1161
        request.order_id = order_id
1162
        response, call = self.stub.PostSandboxOrder.with_call(
1163
            request=_grpc_helpers.dataclass_to_protobuff(
1164
                request, orders_pb2.PostOrderRequest()
1165
            ),
1166
            metadata=self.metadata,
1167
        )
1168
        log_request(get_tracking_id_from_call(call), "PostSandboxOrder")
1169
        return _grpc_helpers.protobuf_to_dataclass(response, PostOrderResponse)
1170

1171
    @handle_request_error("GetSandboxOrders")
1172
    def get_sandbox_orders(self, *, account_id: str = "") -> GetOrdersResponse:
1173
        request = GetOrdersRequest()
1174
        request.account_id = account_id
1175
        response, call = self.stub.GetSandboxOrders.with_call(
1176
            request=_grpc_helpers.dataclass_to_protobuff(
1177
                request, orders_pb2.GetOrdersRequest()
1178
            ),
1179
            metadata=self.metadata,
1180
        )
1181
        log_request(get_tracking_id_from_call(call), "GetSandboxOrders")
1182
        return _grpc_helpers.protobuf_to_dataclass(response, GetOrdersResponse)
1183

1184
    @handle_request_error("CancelSandboxOrder")
1185
    def cancel_sandbox_order(
1186
        self, *, account_id: str = "", order_id: str = ""
1187
    ) -> CancelOrderResponse:
1188
        request = CancelOrderRequest()
1189
        request.account_id = account_id
1190
        request.order_id = order_id
1191
        response, call = self.stub.CancelSandboxOrder.with_call(
1192
            request=_grpc_helpers.dataclass_to_protobuff(
1193
                request, orders_pb2.CancelOrderRequest()
1194
            ),
1195
            metadata=self.metadata,
1196
        )
1197
        log_request(get_tracking_id_from_call(call), "CancelSandboxOrder")
1198
        return _grpc_helpers.protobuf_to_dataclass(response, CancelOrderResponse)
1199

1200
    @handle_request_error("GetSandboxOrderState")
1201
    def get_sandbox_order_state(
1202
        self, *, account_id: str = "", order_id: str = ""
1203
    ) -> OrderState:
1204
        request = GetOrderStateRequest()
1205
        request.account_id = account_id
1206
        request.order_id = order_id
1207
        response, call = self.stub.GetSandboxOrderState.with_call(
1208
            request=_grpc_helpers.dataclass_to_protobuff(
1209
                request, orders_pb2.GetOrderStateRequest()
1210
            ),
1211
            metadata=self.metadata,
1212
        )
1213
        log_request(get_tracking_id_from_call(call), "GetSandboxOrderState")
1214
        return _grpc_helpers.protobuf_to_dataclass(response, OrderState)
1215

1216
    @handle_request_error("GetSandboxPositions")
1217
    def get_sandbox_positions(self, *, account_id: str = "") -> PositionsResponse:
1218
        request = PositionsRequest()
1219
        request.account_id = account_id
1220
        response, call = self.stub.GetSandboxPositions.with_call(
1221
            request=_grpc_helpers.dataclass_to_protobuff(
1222
                request, operations_pb2.PositionsRequest()
1223
            ),
1224
            metadata=self.metadata,
1225
        )
1226
        log_request(get_tracking_id_from_call(call), "GetSandboxPositions")
1227
        return _grpc_helpers.protobuf_to_dataclass(response, PositionsResponse)
1228

1229
    @handle_request_error("GetSandboxOperations")
1230
    def get_sandbox_operations(
1231
        self,
1232
        *,
1233
        account_id: str = "",
1234
        from_: Optional[datetime] = None,
1235
        to: Optional[datetime] = None,
1236
        state: OperationState = OperationState(0),
1237
        figi: str = "",
1238
    ) -> OperationsResponse:
1239
        request = OperationsRequest()
1240
        request.account_id = account_id
1241
        if from_ is not None:
1242
            request.from_ = from_
1243
        if to is not None:
1244
            request.to = to
1245
        request.state = state
1246
        request.figi = figi
1247
        response, call = self.stub.GetSandboxOperations.with_call(
1248
            request=_grpc_helpers.dataclass_to_protobuff(
1249
                request, operations_pb2.OperationsRequest()
1250
            ),
1251
            metadata=self.metadata,
1252
        )
1253
        log_request(get_tracking_id_from_call(call), "GetSandboxOperations")
1254
        return _grpc_helpers.protobuf_to_dataclass(response, OperationsResponse)
1255

1256
    @handle_request_error("GetSandboxPortfolio")
1257
    def get_sandbox_portfolio(self, *, account_id: str = "") -> PortfolioResponse:
1258
        request = PortfolioRequest()
1259
        request.account_id = account_id
1260
        response, call = self.stub.GetSandboxPortfolio.with_call(
1261
            request=_grpc_helpers.dataclass_to_protobuff(
1262
                request, operations_pb2.PortfolioRequest()
1263
            ),
1264
            metadata=self.metadata,
1265
        )
1266
        log_request(get_tracking_id_from_call(call), "GetSandboxPortfolio")
1267
        return _grpc_helpers.protobuf_to_dataclass(response, PortfolioResponse)
1268

1269
    @handle_request_error("SandboxPayIn")
1270
    def sandbox_pay_in(
1271
        self, *, account_id: str = "", amount: Optional[MoneyValue] = None
1272
    ) -> SandboxPayInResponse:
1273
        request = SandboxPayInRequest()
1274
        request.account_id = account_id
1275
        if amount is not None:
1276
            request.amount = amount
1277
        response, call = self.stub.SandboxPayIn.with_call(
1278
            request=_grpc_helpers.dataclass_to_protobuff(
1279
                request, sandbox_pb2.SandboxPayInRequest()
1280
            ),
1281
            metadata=self.metadata,
1282
        )
1283
        log_request(get_tracking_id_from_call(call), "SandboxPayIn")
1284
        return _grpc_helpers.protobuf_to_dataclass(response, SandboxPayInResponse)
1285

1286

1287
class StopOrdersService(_grpc_helpers.Service):
1288
    _stub_factory = stoporders_pb2_grpc.StopOrdersServiceStub
1289

1290
    @handle_request_error("PostStopOrder")
1291
    def post_stop_order(
1292
        self,
1293
        *,
1294
        figi: str = "",
1295
        quantity: int = 0,
1296
        price: Optional[Quotation] = None,
1297
        stop_price: Optional[Quotation] = None,
1298
        direction: StopOrderDirection = StopOrderDirection(0),
1299
        account_id: str = "",
1300
        expiration_type: StopOrderExpirationType = StopOrderExpirationType(0),
1301
        stop_order_type: StopOrderType = StopOrderType(0),
1302
        expire_date: Optional[datetime] = None,
1303
    ) -> PostStopOrderResponse:
1304
        request = PostStopOrderRequest()
1305
        request.figi = figi
1306
        request.quantity = quantity
1307
        if price is not None:
1308
            request.price = price
1309
        if stop_price is not None:
1310
            request.stop_price = stop_price
1311
        request.direction = direction
1312
        request.account_id = account_id
1313
        request.expiration_type = expiration_type
1314
        request.stop_order_type = stop_order_type
1315
        if expire_date is not None:
1316
            request.expire_date = expire_date
1317
        response, call = self.stub.PostStopOrder.with_call(
1318
            request=_grpc_helpers.dataclass_to_protobuff(
1319
                request, stoporders_pb2.PostStopOrderRequest()
1320
            ),
1321
            metadata=self.metadata,
1322
        )
1323
        log_request(get_tracking_id_from_call(call), "PostStopOrder")
1324
        return _grpc_helpers.protobuf_to_dataclass(response, PostStopOrderResponse)
1325

1326
    @handle_request_error("GetStopOrders")
1327
    def get_stop_orders(self, *, account_id: str = "") -> GetStopOrdersResponse:
1328
        request = GetStopOrdersRequest()
1329
        request.account_id = account_id
1330
        response, call = self.stub.GetStopOrders.with_call(
1331
            request=_grpc_helpers.dataclass_to_protobuff(
1332
                request, stoporders_pb2.GetStopOrdersRequest()
1333
            ),
1334
            metadata=self.metadata,
1335
        )
1336
        log_request(get_tracking_id_from_call(call), "GetStopOrders")
1337
        return _grpc_helpers.protobuf_to_dataclass(response, GetStopOrdersResponse)
1338

1339
    @handle_request_error("CancelStopOrder")
1340
    def cancel_stop_order(
1341
        self, *, account_id: str = "", stop_order_id: str = ""
1342
    ) -> CancelStopOrderResponse:
1343
        request = CancelStopOrderRequest()
1344
        request.account_id = account_id
1345
        request.stop_order_id = stop_order_id
1346
        response, call = self.stub.CancelStopOrder.with_call(
1347
            request=_grpc_helpers.dataclass_to_protobuff(
1348
                request, stoporders_pb2.CancelStopOrderRequest()
1349
            ),
1350
            metadata=self.metadata,
1351
        )
1352
        log_request(get_tracking_id_from_call(call), "CancelStopOrder")
1353
        return _grpc_helpers.protobuf_to_dataclass(response, CancelStopOrderResponse)
1354

1355
Product

Resources

Company
