sync/operations.py - master - SimpleSpace/AutoInvestDataSync

sync/operations.py
67 строк2 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 collections import defaultdict
from dataclasses import dataclass

from sync.client import TInvestClient
from sync.utils import money_to_float
from t_tech.invest import OperationType

logger = logging.getLogger(__name__)


@dataclass
class PurchaseRecord:
    month: str = ""
    total_rub: float = 0.0
    count: int = 0


@dataclass
class FirstBuyDate:
    figi: str = ""
    date: str = ""


def fetch_operations(
    tinvest: TInvestClient,
    accounts: list,
    from_dt=None,
    to_dt=None,
) -> tuple[list[PurchaseRecord], dict[str, str]]:
    buy_operations_by_month: dict[str, PurchaseRecord] = {}
    first_buy_dates: dict[str, str] = {}

    for account in accounts:
        logger.info("Fetching operations for account %s", account.name or account.id[:8])
        try:
            items = tinvest.get_operations_by_cursor(
                account_id=account.id,
                from_=from_dt,
                to=to_dt,
            )
        except Exception as e:
            logger.error("Failed to fetch operations for %s: %s", account.id[:8], e)
            continue

        for op in items:
            if not op.figi:
                continue

            if op.date and op.figi:
                date_str = op.date.strftime("%Y-%m-%d")
                if op.figi not in first_buy_dates or date_str < first_buy_dates[op.figi]:
                    if op.type in (OperationType.OPERATION_TYPE_BUY, OperationType.OPERATION_TYPE_BUY_CARD):
                        first_buy_dates[op.figi] = date_str

            if op.type in (OperationType.OPERATION_TYPE_BUY, OperationType.OPERATION_TYPE_BUY_CARD):
                month_key = op.date.strftime("%Y-%m") if op.date else "unknown"
                payment = money_to_float(op.payment) if op.payment else 0.0
                if month_key not in buy_operations_by_month:
                    buy_operations_by_month[month_key] = PurchaseRecord(month=month_key)
                buy_operations_by_month[month_key].total_rub += abs(payment)
                buy_operations_by_month[month_key].count += 1

    purchases = sorted(buy_operations_by_month.values(), key=lambda x: x.month)
    return purchases, first_buy_dates