sync/analytics.py - master - SimpleSpace/AutoInvestDataSync

sync/analytics.py
112 строк4 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.portfolio import PositionData

logger = logging.getLogger(__name__)


@dataclass
class Allocation:
    name: str = ""
    buy_value: float = 0.0
    current_value: float = 0.0
    profit_loss: float = 0.0
    weight_pct: float = 0.0
    count: int = 0


def calculate_analytics(
    positions: list[PositionData],
    first_buy_dates: dict[str, str],
) -> dict:
    from datetime import datetime

    now = datetime.utcnow()
    total_buy = 0.0
    total_current = 0.0

    for pos in positions:
        if pos.figi and pos.figi in first_buy_dates:
            pos.buy_date = first_buy_dates[pos.figi]

        qty = abs(pos.quantity)
        price_fifo = pos.average_position_price_fifo if pos.average_position_price_fifo > 0 else pos.average_position_price
        pos.base_price_rub = price_fifo * qty
        pos.current_value_rub = pos.current_price * qty

        total_buy += pos.base_price_rub
        total_current += pos.current_value_rub

    for pos in positions:
        if total_current > 0:
            pos.weight_in_portfolio_pct = (pos.current_value_rub / total_current) * 100

        if pos.base_price_rub > 0:
            pos.profit_loss_rub = pos.current_value_rub - pos.base_price_rub
            pos.yield_pct = ((pos.current_value_rub / pos.base_price_rub) - 1) * 100

            if pos.buy_date:
                try:
                    buy_dt = datetime.strptime(pos.buy_date, "%Y-%m-%d")
                    days = (now - buy_dt).days
                    if days > 365:
                        ratio = 365.0 / days
                        pos.yield_annual_pct = ((pos.current_value_rub / pos.base_price_rub) ** ratio - 1) * 100
                    elif days > 0:
                        pos.yield_annual_pct = pos.yield_pct * (365.0 / days)
                except (ValueError, ZeroDivisionError):
                    pass

        if pos.next_payout_amount > 0 and pos.current_value_rub > 0:
            pos.expected_dividend_yield_pct = (pos.next_payout_amount / pos.current_value_rub) * 100

        if pos.expected_dividend_yield_pct > 0:
            pos.avg_monthly_income = pos.next_payout_amount / 12.0

    currency_alloc: dict[str, Allocation] = defaultdict(lambda: Allocation())
    broker_alloc: dict[str, Allocation] = defaultdict(lambda: Allocation())
    class_alloc: dict[str, Allocation] = defaultdict(lambda: Allocation())

    for pos in positions:
        cur_key = pos.currency or pos.current_price_currency or "RUB"
        currency_alloc[cur_key].name = cur_key
        currency_alloc[cur_key].buy_value += pos.base_price_rub
        currency_alloc[cur_key].current_value += pos.current_value_rub
        currency_alloc[cur_key].count += 1

        broker_key = pos.account_name or "Неизвестный"
        broker_alloc[broker_key].name = broker_key
        broker_alloc[broker_key].buy_value += pos.base_price_rub
        broker_alloc[broker_key].current_value += pos.current_value_rub
        broker_alloc[broker_key].count += 1

        class_key = pos.instrument_type_display or pos.instrument_type or "Другое"
        class_alloc[class_key].name = class_key
        class_alloc[class_key].buy_value += pos.base_price_rub
        class_alloc[class_key].current_value += pos.current_value_rub
        class_alloc[class_key].count += 1

    for alloc_dict in (currency_alloc, broker_alloc, class_alloc):
        for alloc in alloc_dict.values():
            if total_current > 0:
                alloc.weight_pct = (alloc.current_value / total_current) * 100
            alloc.profit_loss = alloc.current_value - alloc.buy_value

    sorted_by_yield = sorted(positions, key=lambda p: p.yield_annual_pct or 0, reverse=True)
    top_gainers = sorted_by_yield[:5]
    top_losers = sorted_by_yield[-5:][::-1]

    return {
        "total_buy": total_buy,
        "total_current": total_current,
        "total_pnl": total_current - total_buy,
        "currency_allocation": dict(currency_alloc),
        "broker_allocation": dict(broker_alloc),
        "class_allocation": dict(class_alloc),
        "top_gainers": top_gainers,
        "top_losers": top_losers,
    }