sync/client.py - master - SimpleSpace/AutoInvestDataSync

sync/client.py
132 строки5 KB
Alex Just
Start commit for mvp 0.0.5 version
04 май 2026, 22:47
04 май 2026, 22:47e12863e
from __future__ import annotations

import logging
import time
from datetime import datetime

from t_tech.invest import (
    Account,
    AccountType,
    Client,
    GetOperationsByCursorRequest,
    InstrumentIdType,
    OperationState,
)

logger = logging.getLogger(__name__)

ACCOUNT_TYPE_NAMES = {
    AccountType.ACCOUNT_TYPE_TINKOFF: "Брокерский счёт",
    AccountType.ACCOUNT_TYPE_TINKOFF_IIS: "ИИС",
    AccountType.ACCOUNT_TYPE_INVEST_BOX: "Инвесткопилка",
    AccountType.ACCOUNT_TYPE_INVEST_FUND: "Инвестфонд",
    AccountType.ACCOUNT_TYPE_DEBIT: "Дебетовый счёт",
    AccountType.ACCOUNT_TYPE_SAVING: "Накопительный счёт",
}


class TInvestClient:
    def __init__(self, token: str):
        self._token = token
        self._client = Client(token)
        self._services = None

    def __enter__(self):
        self._services = self._client.__enter__()
        return self

    def __exit__(self, *args):
        self._client.__exit__(*args)
        self._services = None

    @property
    def svc(self):
        if self._services is None:
            raise RuntimeError("Use 'with TInvestClient(token) as client:'")
        return self._services

    def get_accounts(self) -> list[Account]:
        logger.info("Fetching accounts...")
        resp = self.svc.users.get_accounts()
        accounts = [a for a in resp.accounts if a.status.name == "ACCOUNT_STATUS_OPEN"]
        logger.info("Found %d open account(s)", len(accounts))
        return accounts

    def get_account_display_name(self, account: Account) -> str:
        if account.name:
            return account.name
        return ACCOUNT_TYPE_NAMES.get(account.type, f"Счёт {account.id[:8]}")

    def get_portfolio(self, account_id: str):
        logger.info("Fetching portfolio for account %s", account_id[:8])
        return self.svc.operations.get_portfolio(account_id=account_id)

    def get_positions(self, account_id: str):
        logger.info("Fetching positions for account %s", account_id[:8])
        return self.svc.operations.get_positions(account_id=account_id)

    def get_instrument(self, instrument_id: str, id_type: str = "figi"):
        logger.debug("Fetching instrument %s", instrument_id)
        id_type_enum = {
            "figi": InstrumentIdType.INSTRUMENT_ID_TYPE_FIGI,
            "uid": InstrumentIdType.INSTRUMENT_ID_TYPE_UID,
            "isin": InstrumentIdType.INSTRUMENT_ID_TYPE_ISIN,
            "ticker": InstrumentIdType.INSTRUMENT_ID_TYPE_TICKER,
        }.get(id_type, InstrumentIdType.INSTRUMENT_ID_TYPE_FIGI)
        return self.svc.instruments.get_instrument_by(
            id_type=id_type_enum,
            id=instrument_id,
        )

    def get_share(self, instrument_id: str):
        return self.svc.instruments.share_by(id_type=InstrumentIdType.INSTRUMENT_ID_TYPE_FIGI, id=instrument_id)

    def get_bond(self, instrument_id: str):
        return self.svc.instruments.bond_by(id_type=InstrumentIdType.INSTRUMENT_ID_TYPE_FIGI, id=instrument_id)

    def get_etf(self, instrument_id: str):
        return self.svc.instruments.etf_by(id_type=InstrumentIdType.INSTRUMENT_ID_TYPE_FIGI, id=instrument_id)

    def get_dividends(self, instrument_id: str, from_: datetime | None = None, to: datetime | None = None):
        return self.svc.instruments.get_dividends(
            instrument_id=instrument_id,
            from_=from_,
            to=to,
        )

    def get_bond_coupons(self, instrument_id: str, from_: datetime | None = None, to: datetime | None = None):
        return self.svc.instruments.get_bond_coupons(
            instrument_id=instrument_id,
            from_=from_,
            to=to,
        )

    def get_operations_by_cursor(
        self,
        account_id: str,
        from_: datetime | None = None,
        to: datetime | None = None,
        limit: int = 100,
    ):
        all_items = []
        cursor = ""
        has_next = True
        while has_next:
            req = GetOperationsByCursorRequest(
                account_id=account_id,
                from_=from_,
                to=to,
                cursor=cursor,
                limit=limit,
                state=OperationState.OPERATION_STATE_EXECUTED,
            )
            resp = self.svc.operations.get_operations_by_cursor(request=req)
            all_items.extend(resp.items)
            has_next = resp.has_next
            cursor = resp.next_cursor
            if has_next:
                time.sleep(0.5)
        return all_items

    def get_last_prices(self, figi_list: list[str]):
        return self.svc.market_data.get_last_prices(figi=figi_list)