sync/portfolio.py - master - SimpleSpace/AutoInvestDataSync

sync/portfolio.py
154 строки6 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
from dataclasses import dataclass, field

from sync.client import TInvestClient
from sync.utils import money_to_float, quotation_to_float
from t_tech.invest import Account, PortfolioPosition

logger = logging.getLogger(__name__)

INSTRUMENT_TYPE_NAMES = {
    "INSTRUMENT_TYPE_BOND": "Облигации",
    "INSTRUMENT_TYPE_SHARE": "Акции",
    "INSTRUMENT_TYPE_CURRENCY": "Валюта",
    "INSTRUMENT_TYPE_ETF": "Фонды",
    "INSTRUMENT_TYPE_FUTURES": "Фьючерсы",
    "INSTRUMENT_TYPE_OPTION": "Опционы",
    "INSTRUMENT_TYPE_SP": "Структурные продукты",
}


@dataclass
class PositionData:
    figi: str = ""
    instrument_uid: str = ""
    position_uid: str = ""
    ticker: str = ""
    instrument_type: str = ""
    instrument_type_display: str = ""
    quantity: float = 0.0
    quantity_lots: float = 0.0
    average_position_price: float = 0.0
    average_position_price_currency: str = ""
    average_position_price_fifo: float = 0.0
    current_price: float = 0.0
    current_price_currency: str = ""
    expected_yield: float = 0.0
    expected_yield_fifo: float = 0.0
    current_nkd: float = 0.0
    blocked: bool = False
    account_id: str = ""
    account_name: str = ""
    name: str = ""
    isin: str = ""
    currency: str = ""
    lot: int = 1
    sector: str = ""
    country_of_risk: str = ""
    country_of_risk_name: str = ""
    nominal: float = 0.0
    coupon_quantity_per_year: int = 0
    maturity_date: str | None = None
    floating_coupon_flag: bool = False
    perpetual_flag: bool = False
    next_payout_date: str | None = None
    next_payout_amount: float = 0.0
    next_payout_currency: str = ""
    buy_date: str | None = None
    base_price_rub: float = 0.0
    current_value_rub: float = 0.0
    profit_loss_rub: float = 0.0
    yield_pct: float = 0.0
    yield_annual_pct: float = 0.0
    expected_dividend_yield_pct: float = 0.0
    avg_monthly_income: float = 0.0
    weight_in_portfolio_pct: float = 0.0


@dataclass
class PortfolioSummary:
    total_amount_portfolio: float = 0.0
    total_amount_currencies: float = 0.0
    total_amount_shares: float = 0.0
    total_amount_bonds: float = 0.0
    total_amount_etfs: float = 0.0
    total_amount_futures: float = 0.0
    total_amount_options: float = 0.0
    expected_yield: float = 0.0


@dataclass
class AccountPortfolio:
    account: Account | None = None
    account_name: str = ""
    summary: PortfolioSummary | None = None
    positions: list[PositionData] = field(default_factory=list)


def _parse_portfolio_position(pos: PortfolioPosition, account_id: str, account_name: str) -> PositionData:
    qty = quotation_to_float(pos.quantity)
    pd = PositionData(
        figi=pos.figi,
        instrument_uid=pos.instrument_uid,
        position_uid=pos.position_uid,
        ticker=pos.ticker,
        instrument_type=pos.instrument_type,
        instrument_type_display=INSTRUMENT_TYPE_NAMES.get(pos.instrument_type, pos.instrument_type),
        quantity=qty,
        quantity_lots=quotation_to_float(pos.quantity_lots),
        blocked=pos.blocked,
        account_id=account_id,
        account_name=account_name,
    )
    if pos.average_position_price:
        pd.average_position_price = money_to_float(pos.average_position_price)
        pd.average_position_price_currency = pos.average_position_price.currency
    if pos.average_position_price_fifo:
        pd.average_position_price_fifo = money_to_float(pos.average_position_price_fifo)
    if pos.current_price:
        pd.current_price = money_to_float(pos.current_price)
        pd.current_price_currency = pos.current_price.currency
    if pos.expected_yield:
        pd.expected_yield = quotation_to_float(pos.expected_yield)
    if pos.expected_yield_fifo:
        pd.expected_yield_fifo = quotation_to_float(pos.expected_yield_fifo)
    if pos.current_nkd:
        pd.current_nkd = money_to_float(pos.current_nkd)
    return pd


def fetch_portfolio(tinvest: TInvestClient, accounts: list[Account]) -> list[AccountPortfolio]:
    result = []
    for account in accounts:
        account_name = tinvest.get_account_display_name(account)
        ap = AccountPortfolio(account=account, account_name=account_name)

        portfolio_resp = tinvest.get_portfolio(account.id)

        summary = PortfolioSummary(
            total_amount_portfolio=money_to_float(portfolio_resp.total_amount_portfolio),
            total_amount_currencies=money_to_float(portfolio_resp.total_amount_currencies),
            total_amount_shares=money_to_float(portfolio_resp.total_amount_shares),
            total_amount_bonds=money_to_float(portfolio_resp.total_amount_bonds),
            total_amount_etfs=money_to_float(portfolio_resp.total_amount_etf),
            total_amount_futures=money_to_float(portfolio_resp.total_amount_futures),
            total_amount_options=money_to_float(portfolio_resp.total_amount_options),
            expected_yield=money_to_float(portfolio_resp.expected_yield),
        )
        ap.summary = summary

        for pos in portfolio_resp.positions:
            qty = quotation_to_float(pos.quantity)
            if abs(qty) < 1e-9:
                continue
            pd = _parse_portfolio_position(pos, account.id, account_name)
            ap.positions.append(pd)
            logger.debug("Position: %s qty=%.4f type=%s", pd.figi, pd.quantity, pd.instrument_type)

        result.append(ap)
        logger.info("Account '%s': %d positions, total=%.2f RUB",
                     account_name, len(ap.positions), summary.total_amount_portfolio)

    return result