sync/instruments.py - master - SimpleSpace/AutoInvestDataSync

sync/instruments.py
82 строки3 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 dataclasses import dataclass

from sync.client import TInvestClient
from sync.portfolio import PositionData
from sync.utils import money_to_float

logger = logging.getLogger(__name__)

INSTRUMENT_CACHE: dict[str, dict] = {}


def enrich_positions(tinvest: TInvestClient, positions: list[PositionData]) -> None:
    for pos in positions:
        if pos.figi and pos.figi not in INSTRUMENT_CACHE:
            try:
                _fetch_and_cache(tinvest, pos.figi, pos.instrument_type)
                time.sleep(0.3)
            except Exception as e:
                logger.warning("Failed to fetch instrument %s: %s", pos.figi, e)

    for pos in positions:
        if not pos.figi:
            continue
        cached = INSTRUMENT_CACHE.get(pos.figi)
        if not cached:
            continue
        pos.name = cached.get("name", "")
        pos.isin = cached.get("isin", "")
        pos.currency = cached.get("currency", pos.currency or "")
        pos.lot = cached.get("lot", 1)
        pos.sector = cached.get("sector", "")
        pos.country_of_risk = cached.get("country_of_risk", "")
        pos.country_of_risk_name = cached.get("country_of_risk_name", "")
        if not pos.ticker:
            pos.ticker = cached.get("ticker", "")


def _fetch_and_cache(tinvest: TInvestClient, figi: str, instrument_type: str) -> dict:
    data: dict = {}
    try:
        base = tinvest.get_instrument(figi)
        data["name"] = base.instrument.name
        data["ticker"] = base.instrument.ticker
        data["isin"] = base.instrument.isin
        data["currency"] = base.instrument.currency
        data["lot"] = base.instrument.lot
        data["country_of_risk"] = base.instrument.country_of_risk
        data["country_of_risk_name"] = base.instrument.country_of_risk_name
    except Exception:
        pass

    itype = instrument_type.upper() if instrument_type else ""
    try:
        if "SHARE" in itype:
            resp = tinvest.get_share(figi)
            if resp and resp.instrument:
                data["sector"] = resp.instrument.sector
                data["name"] = resp.instrument.name or data.get("name", "")
        elif "BOND" in itype:
            resp = tinvest.get_bond(figi)
            if resp and resp.instrument:
                data["sector"] = resp.instrument.sector
                data["name"] = resp.instrument.name or data.get("name", "")
                data["nominal"] = money_to_float(resp.instrument.nominal)
                data["coupon_quantity_per_year"] = resp.instrument.coupon_quantity_per_year
                data["floating_coupon_flag"] = resp.instrument.floating_coupon_flag
                data["perpetual_flag"] = resp.instrument.perpetual_flag
                if resp.instrument.maturity_date:
                    data["maturity_date"] = resp.instrument.maturity_date.strftime("%Y-%m-%d")
        elif "ETF" in itype:
            resp = tinvest.get_etf(figi)
            if resp and resp.instrument:
                data["name"] = resp.instrument.name or data.get("name", "")
    except Exception as e:
        logger.debug("Detailed fetch failed for %s (%s): %s", figi, itype, e)

    INSTRUMENT_CACHE[figi] = data
    return data