From c203ae65fce922468101ba1c218e29b61567fe49 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Fri, 18 Sep 2026 13:44:50 +0300 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20=D0=BE=D1=86=D0=B5=D0=BD?= =?UTF-8?q?=D0=BA=D0=B0=20=D0=BF=D0=BE=D0=B7=D0=B8=D1=86=D0=B8=D0=B9,=20XI?= =?UTF-8?q?RR=20=D0=B8=20TWR?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pricing/prices.py — цена на дату: последний close тянется вперёд, после 10 дней считается протухшей, но всё ещё используется. Назад не тянется никогда — цена из будущего это выдумка, а не оценка. valuation берёт два разных источника намеренно. Дневная серия — реплей event (только он отвечает, сколько стоило в марте), текущие холдинги — из lot, где есть себестоимость и учтены сплиты. Расхождение между ними на последний день становится находкой, а не поводом выбрать одно из двух. Отчётная единица — scope: all, account:, portfolio:. returns читает только metric_portfolio_value_daily. XIRR — по внешним потокам и терминальной стоимости; TWR — цепочкой V_t / (V_{t-1} + F_t). План пишет формулу как (V_t - F_t) / V_{t-1}, то есть с потоком в конце дня; поток в начале даёт то же число при нулевом потоке, не требует особого случая на первый день и относит движение рынка к деньгам, которые в этот день уже работали. Покупка бумаги без цены трактуется как вывод из оцениваемого портфеля (unvalued_flow_rub): иначе деньги уходят из оценки, а бумага в неё не попадает, и день читается как обвал — именно это фонд денежного рынка без фида MOEX устроил серии 2024 года. Пропускается только день, в который меняется ЧИСЛО неоценённых позиций, и их счётчик уходит в metric_data_quality. Нет цены или курса — NULL и замечание, не ноль: SIBN6P4 в холдингах именно так и выглядит. Валютные «позиции» из сверки исключены, их двойник — cash_snapshot, а не лот; после этого расхождений с брокером ровно пять известных. Проверка из плана закрыта тестами: взнос 100 и 110 через год дают XIRR 10,0 % и TWR 10,0 %, второй взнос двигает XIRR и не трогает TWR подпериодов. --- ...2_метрики_оценки_холдингов_и_доходности.py | 124 +++ backend/src/fintracker/analytics/returns.py | 402 ++++++++ backend/src/fintracker/analytics/valuation.py | 880 ++++++++++++++++++ backend/src/fintracker/pricing/prices.py | 135 +++ backend/tests/analytics/test_prices.py | 68 ++ backend/tests/analytics/test_returns.py | 121 +++ backend/tests/analytics/test_valuation.py | 244 +++++ backend/tests/analytics/test_valuation_db.py | 270 ++++++ 8 files changed, 2244 insertions(+) create mode 100644 backend/alembic/versions/b7424afbb5e2_метрики_оценки_холдингов_и_доходности.py create mode 100644 backend/src/fintracker/analytics/returns.py create mode 100644 backend/src/fintracker/analytics/valuation.py create mode 100644 backend/src/fintracker/pricing/prices.py create mode 100644 backend/tests/analytics/test_prices.py create mode 100644 backend/tests/analytics/test_returns.py create mode 100644 backend/tests/analytics/test_valuation.py create mode 100644 backend/tests/analytics/test_valuation_db.py diff --git a/backend/alembic/versions/b7424afbb5e2_метрики_оценки_холдингов_и_доходности.py b/backend/alembic/versions/b7424afbb5e2_метрики_оценки_холдингов_и_доходности.py new file mode 100644 index 0000000..9af40c1 --- /dev/null +++ b/backend/alembic/versions/b7424afbb5e2_метрики_оценки_холдингов_и_доходности.py @@ -0,0 +1,124 @@ +"""метрики оценки, холдингов и доходности + +Revision ID: b7424afbb5e2 +Revises: 3b1431df06af +Create Date: 2026-09-18 13:28:43.383174 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "b7424afbb5e2" +down_revision: str | None = "3b1431df06af" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "metric_portfolio_value_daily", + sa.Column("scope", sa.String(length=32), nullable=False), + sa.Column("d", sa.Date(), nullable=False), + sa.Column("market_value_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("accrued_interest_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("cash_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("total_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("external_flow_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("unvalued_flow_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("invested_net_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("pnl_total_rub", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("stale_price_count", sa.Integer(), nullable=False), + sa.Column("missing_price_count", sa.Integer(), nullable=False), + sa.Column("missing_fx_count", sa.Integer(), nullable=False), + sa.Column( + "computed_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("scope", "d", name=op.f("pk_metric_portfolio_value_daily")), + ) + op.create_table( + "metric_returns", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("scope", sa.String(length=32), nullable=False), + sa.Column("period", sa.String(length=8), nullable=False), + sa.Column("date_from", sa.Date(), nullable=False), + sa.Column("date_to", sa.Date(), nullable=False), + sa.Column("value_start_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("value_end_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("external_flow_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("abs_pnl_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("xirr", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("twr", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("twr_annualized", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("twr_days_skipped", sa.Integer(), nullable=False), + sa.Column( + "computed_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_metric_returns")), + sa.UniqueConstraint("scope", "period", name=op.f("uq_metric_returns_scope_period")), + ) + op.create_index(op.f("ix_metric_returns_scope"), "metric_returns", ["scope"], unique=False) + op.create_table( + "metric_holding", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("scope", sa.String(length=32), nullable=False), + sa.Column("instrument_id", sa.Integer(), nullable=False), + sa.Column("qty", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("avg_cost", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("cost_currency", sa.String(length=3), nullable=True), + sa.Column("cost_total_rub", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("market_price", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("price_currency", sa.String(length=3), nullable=True), + sa.Column("price_date", sa.Date(), nullable=True), + sa.Column("price_status", sa.String(length=8), nullable=False), + sa.Column("value_native", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("value_rub", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("accrued_interest_rub", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("unrealized_pnl_native", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("unrealized_pnl_rub", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("realized_pnl_rub", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("income_rub", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("weight", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("xirr", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("first_buy_date", sa.Date(), nullable=True), + sa.Column("days_held", sa.Integer(), nullable=True), + sa.Column("ldv_eligible_qty", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column( + "computed_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.ForeignKeyConstraint( + ["instrument_id"], + ["instrument.id"], + name=op.f("fk_metric_holding_instrument_id_instrument"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_metric_holding")), + sa.UniqueConstraint( + "scope", "instrument_id", name=op.f("uq_metric_holding_scope_instrument_id") + ), + ) + op.create_index(op.f("ix_metric_holding_scope"), "metric_holding", ["scope"], unique=False) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f("ix_metric_holding_scope"), table_name="metric_holding") + op.drop_table("metric_holding") + op.drop_index(op.f("ix_metric_returns_scope"), table_name="metric_returns") + op.drop_table("metric_returns") + op.drop_table("metric_portfolio_value_daily") + # ### end Alembic commands ### diff --git a/backend/src/fintracker/analytics/returns.py b/backend/src/fintracker/analytics/returns.py new file mode 100644 index 0000000..a1d2aa2 --- /dev/null +++ b/backend/src/fintracker/analytics/returns.py @@ -0,0 +1,402 @@ +"""XIRR and TWR — the two return numbers, and why there are two (plan §3). + +**XIRR** (money-weighted) answers "what did *I* earn": the internal rate of return of the +actual cash the investor moved across the portfolio boundary, plus the terminal value. It +depends on *when* money was added, so a well-timed deposit flatters it and a badly-timed one +does not. + +**TWR** (time-weighted) answers "how did the portfolio do", stripping the timing of deposits +out entirely by chaining daily sub-period returns. It is what a fund's published return is, +and what a benchmark can be compared against. + +The relationship is the acceptance check in the plan: with a single contribution the two +agree (nothing to time), and a second contribution moves XIRR while leaving every +sub-period of TWR untouched. + + r_t = V_t / (V_{t-1} + F_t) + +The plan writes this as `(V_t − F_t) / V_{t−1}`, which places the flow at the end of the day. +Placing it at the start instead — the form above — is the same number whenever F_t is zero, +handles the very first day (V_{t-1} = 0) without a special case, and credits a deposit with +the market move of the day it arrived, which is what actually happened to it. + +Everything is read from `metric_portfolio_value_daily`, which `analytics/valuation.py` has +already built: the daily total and the day's net external flow, both in RUB. That table is +therefore the single definition of "a flow" — deposits, withdrawals, securities transferred +in or out, and purchases funded straight from a linked card. + +Rates are fractions (0.1 = 10 %) and are the one place floats are allowed: they are +coefficients, not money (conventions.md). +""" + +from __future__ import annotations + +import logging +from collections import defaultdict +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import date +from decimal import Decimal, InvalidOperation + +import pyxirr +from sqlalchemy import delete, insert, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.analytics import FINDINGS, today_local +from fintracker.models import ( + Event, + EventStatus, + MetricHolding, + MetricPortfolioValueDaily, + MetricReturns, +) +from fintracker.pricing.fx import FxTable + +log = logging.getLogger(__name__) + +ZERO = Decimal(0) +ONE = Decimal(1) +DAYS_IN_YEAR = Decimal(365) +RATE_PLACES = Decimal("0.000001") + +PERIODS: tuple[str, ...] = ("1m", "3m", "6m", "ytd", "1y", "3y", "all") + +#: Periods reported even when the data is younger than their nominal length. +ELASTIC_PERIODS = frozenset({"ytd", "all"}) + +MIN_XIRR_DAYS = 30 +"""Below this, annualising a per-instrument return turns a day of noise into a percent a +year. A paper bought yesterday gets no XIRR rather than a triple-digit one.""" + + +@dataclass(frozen=True) +class Point: + """One day of a scope's series: what it was worth, and what crossed the boundary.""" + + d: date + value: Decimal + flow: Decimal + """Net contribution on this day, + into the portfolio.""" + unvalued_flow: Decimal = ZERO + """Cash that moved between the valued portfolio and a position with no price.""" + missing: int = 0 + """How many positions could not be valued on this day — see `twr` for why it matters.""" + + +@dataclass(frozen=True) +class TwrResult: + """A time-weighted return and how much of the period it actually covers.""" + + value: Decimal | None + days_used: int + days_skipped: int + + +# -------------------------------------------------------------------------------------- +# pure core +# -------------------------------------------------------------------------------------- + + +def months_back(d: date, months: int) -> date: + """`d` shifted back whole months, clamped to the end of a shorter month.""" + total = d.year * 12 + (d.month - 1) - months + year, month = divmod(total, 12) + month += 1 + day = min(d.day, _days_in_month(year, month)) + return date(year, month, day) + + +def _days_in_month(year: int, month: int) -> int: + if month == 12: + return 31 + return (date(year, month + 1, 1) - date(year, month, 1)).days + + +def period_start(period: str, end: date) -> date | None: + """First day of `period`, i.e. the day whose closing value is the opening value.""" + match period: + case "1m": + return months_back(end, 1) + case "3m": + return months_back(end, 3) + case "6m": + return months_back(end, 6) + case "1y": + return months_back(end, 12) + case "3y": + return months_back(end, 36) + case "ytd": + return date(end.year, 1, 1) + case "all": + return None + case _: + raise ValueError(f"unknown period: {period}") + + +def xirr(dates: Sequence[date], amounts: Sequence[Decimal]) -> Decimal | None: + """Annualised money-weighted return, or None when the flows admit no solution. + + Sign convention: negative is money the investor committed, positive is money that came + back — so a portfolio's flows are the NEGATIVE of its contributions, and the terminal + value is a positive flow on the last day. + + None is the honest answer for flows that are all one sign (nothing has come back yet, or + nothing went in) and for the pathological cases where the solver does not converge. + """ + if len(dates) < 2: + return None + floats = [float(a) for a in amounts] + if not (any(a > 0 for a in floats) and any(a < 0 for a in floats)): + return None + try: + rate = pyxirr.xirr(list(dates), floats) + except (pyxirr.InvalidPaymentsError, ValueError, ZeroDivisionError): + return None + if rate is None: + return None + return _as_rate(rate) + + +def twr(points: Sequence[Point], *, opening_value: Decimal, opening_missing: int = 0) -> TwrResult: + """Chain daily sub-period returns into the cumulative time-weighted return. + + `points` are the days INSIDE the period; `opening_value` is the close of the day before + it. A day whose capital base is zero or negative contributes a factor of 1: there was + nothing invested to earn a return on, and a negative base would flip the sign of every + later factor. + + A position nobody quotes is treated as being outside the portfolio: buying it is then a + withdrawal and selling it a deposit, which is what `Point.unvalued_flow` carries. Without + that, the cash spent on it would leave the value while the paper never enters it, and the + day would read as a crash — which is what a money-market fund with no MOEX feed did to + the 2024 series. On a quiet day the unquoted position cancels out of both ends of the + ratio and costs nothing. + + One case is still skipped rather than adjusted: a day where the NUMBER of unvalued + positions changed. The value then jumps by a whole position that neither gained nor lost + anything, and no flow describes it. `TwrResult.days_skipped` reports how many such days + a period contains, because a TWR with holes has to say so. + """ + factor = ONE + previous = opening_value + previous_missing = opening_missing + used = skipped = 0 + for point in points: + base = previous + point.flow + point.unvalued_flow + if point.missing != previous_missing: + skipped += 1 + elif base > ZERO: + factor *= point.value / base + used += 1 + previous = point.value + previous_missing = point.missing + return TwrResult(_quantize(factor - ONE) if used else None, used, skipped) + + +def annualize(cumulative: Decimal | None, days: int) -> Decimal | None: + """Scale a cumulative return to a year; None below a year, where it would be noise.""" + if cumulative is None or days < 365 or cumulative <= -ONE: + return None + try: + return _as_rate((1.0 + float(cumulative)) ** (365.0 / days) - 1.0) + except (OverflowError, ValueError): + return None + + +def _as_rate(value: float) -> Decimal | None: + try: + return _quantize(Decimal(repr(value))) + except (InvalidOperation, ValueError): + return None + + +def _quantize(value: Decimal) -> Decimal: + return value.quantize(RATE_PLACES) + + +# -------------------------------------------------------------------------------------- +# I/O +# -------------------------------------------------------------------------------------- + + +async def rebuild_returns(session: AsyncSession) -> None: + """Replace `metric_returns` for every scope, then fill `metric_holding.xirr`.""" + as_of = today_local() + series = await _load_series(session) + + await session.execute(delete(MetricReturns)) + if not series: + return + + rows: list[dict[str, object]] = [] + for scope, points in series.items(): + rows += _rows_for_scope(scope, points, as_of) + if rows: + await session.execute(insert(MetricReturns), rows) + _report_partial_twr(rows) + + await _fill_holding_xirr(session, as_of) + log.info("returns: %s scopes, %s rows", len(series), len(rows)) + + +async def _load_series(session: AsyncSession) -> dict[str, list[Point]]: + """The daily value series per scope, exactly as valuation wrote it.""" + rows = ( + await session.execute( + select( + MetricPortfolioValueDaily.scope, + MetricPortfolioValueDaily.d, + MetricPortfolioValueDaily.total_rub, + MetricPortfolioValueDaily.external_flow_rub, + MetricPortfolioValueDaily.unvalued_flow_rub, + MetricPortfolioValueDaily.missing_price_count, + MetricPortfolioValueDaily.missing_fx_count, + ).order_by(MetricPortfolioValueDaily.scope, MetricPortfolioValueDaily.d) + ) + ).all() + out: dict[str, list[Point]] = defaultdict(list) + for r in rows: + out[r.scope].append( + Point( + d=r.d, + value=Decimal(r.total_rub), + flow=Decimal(r.external_flow_rub), + unvalued_flow=Decimal(r.unvalued_flow_rub), + missing=r.missing_price_count + r.missing_fx_count, + ) + ) + return dict(out) + + +def _rows_for_scope(scope: str, points: Sequence[Point], as_of: date) -> list[dict[str, object]]: + """One row per period the series is long enough to support.""" + if not points: + return [] + first, last = points[0], points[-1] + index = {p.d: i for i, p in enumerate(points)} + + rows: list[dict[str, object]] = [] + for period in PERIODS: + start = period_start(period, as_of) or first.d + if start < first.d: + if period not in ELASTIC_PERIODS: + continue + start = first.d + if start >= last.d: + continue + # the spine is gap-free, so a date inside it is always present + opening_index = index.get(start) + if opening_index is None: + continue + + opening = points[opening_index] + inside = points[opening_index + 1 :] + net_flow = sum((p.flow for p in inside), start=ZERO) + + dates = [opening.d] + amounts = [-opening.value] + for p in inside: + if p.flow: + dates.append(p.d) + amounts.append(-p.flow) + dates.append(last.d) + amounts.append(last.value) + + chain = twr(inside, opening_value=opening.value, opening_missing=opening.missing) + rows.append( + { + "scope": scope, + "period": period, + "date_from": opening.d, + "date_to": last.d, + "value_start_rub": opening.value, + "value_end_rub": last.value, + "external_flow_rub": net_flow, + "abs_pnl_rub": last.value - opening.value - net_flow, + "xirr": xirr(dates, amounts), + "twr": chain.value, + "twr_annualized": annualize(chain.value, chain.days_used), + "twr_days_skipped": chain.days_skipped, + } + ) + return rows + + +async def _fill_holding_xirr(session: AsyncSession, as_of: date) -> None: + """Per-instrument money-weighted return: its own cash effects plus what it is worth now. + + An instrument's `event.amount` is already signed the way XIRR wants it — a purchase is + money out, a dividend is money in — so the flows need no inversion here, unlike the + portfolio-level series where a deposit is a contribution. + """ + from fintracker.analytics.valuation import account_scopes + + fx = await FxTable.load(session) + rows = ( + await session.execute( + select( + Event.account_id, + Event.instrument_id, + Event.trade_date, + Event.amount, + Event.currency, + ).where(Event.status == EventStatus.confirmed, Event.instrument_id.is_not(None)) + ) + ).all() + flows: dict[tuple[int, int], list[tuple[date, Decimal]]] = defaultdict(list) + for r in rows: + rub = fx.to_rub(Decimal(r.amount or 0), r.currency, r.trade_date) + if rub: + flows[(r.account_id, r.instrument_id)].append((r.trade_date, rub)) + + ledger_accounts = {account_id for account_id, _ in flows} + scopes = await account_scopes(session, ledger_accounts) + + holdings = ( + await session.execute( + select( + MetricHolding.id, + MetricHolding.scope, + MetricHolding.instrument_id, + MetricHolding.value_rub, + ).where(MetricHolding.value_rub.is_not(None)) + ) + ).all() + for h in holdings: + account_ids = scopes.get(h.scope) + if not account_ids: + continue + legs: list[tuple[date, Decimal]] = [] + for account_id in account_ids: + legs += flows.get((account_id, h.instrument_id), []) + if not legs: + continue + legs.sort() + if (as_of - legs[0][0]).days < MIN_XIRR_DAYS: + continue + dates = [d for d, _ in legs] + [as_of] + amounts = [a for _, a in legs] + [Decimal(h.value_rub)] + rate = xirr(dates, amounts) + if rate is not None: + await session.execute( + update(MetricHolding).where(MetricHolding.id == h.id).values(xirr=rate) + ) + + +def _report_partial_twr(rows: Sequence[Mapping[str, object]]) -> None: + """Say out loud when the whole-portfolio TWR was built from part of the days.""" + skipped = next( + (r["twr_days_skipped"] for r in rows if r["scope"] == "all" and r["period"] == "all"), + 0, + ) + if not isinstance(skipped, int): + return + if not skipped: + return + FINDINGS.add( + "twr_partial", + "warn", + f"TWR посчитан не по всей истории: {skipped} дн. пропущено, " + "в эти дни часть позиции была без цены", + count=skipped, + ) diff --git a/backend/src/fintracker/analytics/valuation.py b/backend/src/fintracker/analytics/valuation.py new file mode 100644 index 0000000..ce57a8a --- /dev/null +++ b/backend/src/fintracker/analytics/valuation.py @@ -0,0 +1,880 @@ +"""What the portfolio is worth: the daily value series and the current holdings (plan §3). + +Two numbers come out of here, and they are computed from different sources on purpose. + +* **The daily series** replays the ledger: the position on day *d* is the cumulative signed + `event.quantity` up to *d*, the cash is the cumulative `event.amount` per currency. Only a + replay can answer "what was it worth in March", and it is the input TWR needs. +* **The current holdings** come from `lot.qty_remaining`, which is the split- and + amortisation-aware truth and carries the cost basis the replay does not have. + +They must agree on the last day; when they do not, that is a finding, not a number to pick +between (`position_replay_mismatch`). + +Sign conventions follow the ledger: `quantity` is + into the position, `amount` is + into the +account's cash, and `qty_remaining` is stored signed so a short position is negative all the +way through — its market value, cost and unrealised P&L are negative too, which is what makes +a scope's total come out right by plain addition. + +Currency: nothing is converted on the way in. Every RUB figure is the native amount times the +CBR rate **of its own day**, and a day with no rate contributes nothing and increments +`missing_fx_count` instead. Likewise a price: an instrument nobody quotes is left out of the +sums and reported, never valued at zero — the two papers with no feed (SIBN6P4, +NDM_TBNK-PP-FIXPRCNT-08.25) must show up as NULL holdings with a remark. + +External flows (deposits, withdrawals, securities transferred in or out, and purchases funded +straight from a linked card) are extracted here rather than in `returns.py`, because +`invested_net_rub` is what makes the series' P&L column meaningful. `returns.py` then needs +nothing but this table. +""" + +from __future__ import annotations + +import logging +from collections import defaultdict +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass +from datetime import date, timedelta +from decimal import Decimal +from typing import Protocol + +from sqlalchemy import delete, func, insert, select +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.analytics import FINDINGS, today_local +from fintracker.models import ( + Account, + AssetClass, + CashSnapshot, + Event, + EventStatus, + Instrument, + Lot, + LotDisposal, + MetricHolding, + MetricPortfolioValueDaily, + PortfolioAccount, + PositionSnapshot, +) +from fintracker.models.ledger import EXTERNAL_FLOW_KINDS, POSITION_KINDS, EventKind +from fintracker.pricing.fx import FxTable +from fintracker.pricing.prices import STALE_AFTER_DAYS, PriceTable, Quote + +log = logging.getLogger(__name__) + +ZERO = Decimal(0) +RUB = "RUB" +LDV_DAYS = 3 * 365 + +#: Kinds that pay money into (or out of) a position without changing its quantity. +INCOME_KINDS = frozenset( + { + EventKind.dividend, + EventKind.coupon, + EventKind.interest, + EventKind.amortization, + EventKind.tax, + EventKind.tax_refund, + } +) + +#: Asset classes that can qualify for the long-term holding exemption (art. 219.1 NK). +EXCHANGE_CLASSES = frozenset({"share", "bond", "etf", "fund"}) + +#: Positions and cash may differ from the broker's snapshot by this much before it is a finding. +QTY_TOLERANCE = Decimal("0.000001") +CASH_TOLERANCE = Decimal(1) + + +class PriceLookup(Protocol): + def at(self, instrument_id: int, d: date) -> Quote | None: ... + + +class FxLookup(Protocol): + def rate(self, d: date, ccy: str | None) -> Decimal | None: ... + + +# -------------------------------------------------------------------------------------- +# pure core: no session, no I/O — everything below takes plain deltas and lookups +# -------------------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class Deltas: + """Per-day changes the replay walks, keyed so a scope is just a subset of accounts. + + `positions` and `cash` are the raw ledger effects; `flows` is the part of them that + crossed the portfolio boundary, already separated because only the ledger knows which + events those were. + """ + + positions: Mapping[tuple[int, int], Mapping[date, Decimal]] + """(account, instrument) -> day -> signed quantity change.""" + cash: Mapping[tuple[int, str], Mapping[date, Decimal]] + """(account, currency) -> day -> signed cash change.""" + flows: Mapping[tuple[int, str], Mapping[date, Decimal]] + """(account, currency) -> day -> net contribution, + into the portfolio.""" + instrument_cash: Mapping[tuple[int, int, str], Mapping[date, Decimal]] + """(account, instrument, currency) -> day -> cash the instrument took or paid. + + Only used to find the money that crossed into a position nobody quotes: that cash leaves + the valued portfolio without reappearing as market value, and TWR has to know.""" + + +@dataclass(frozen=True) +class DayValue: + """One account on one day. Scope totals are these added up (see `combine`).""" + + d: date + market_value_rub: Decimal + accrued_interest_rub: Decimal + cash_rub: Decimal + external_flow_rub: Decimal + unvalued_flow_rub: Decimal = ZERO + """Cash that moved between the valued portfolio and a position with no price.""" + stale_price_count: int = 0 + missing_price_count: int = 0 + missing_fx_count: int = 0 + + @property + def total_rub(self) -> Decimal: + return self.market_value_rub + self.cash_rub + + @property + def complete(self) -> bool: + """True when nothing was left out of the sums — the P&L column depends on it.""" + return not (self.missing_price_count or self.missing_fx_count) + + +def days_between(start: date, end: date) -> list[date]: + out: list[date] = [] + d = start + while d <= end: + out.append(d) + d += timedelta(days=1) + return out + + +def value_series( + *, + spine: Sequence[date], + deltas: Deltas, + prices: PriceLookup, + fx: FxLookup, +) -> dict[int, list[DayValue]]: + """Replay the ledger over `spine`, returning one `DayValue` per account per day. + + The walk is forward and stateful: running quantities and running cash carry across days, + which is what makes a day with no trades still move with the price and the rate. + """ + accounts = {acc for acc, _ in deltas.positions} | {acc for acc, _ in deltas.cash} + accounts |= {acc for acc, _ in deltas.flows} + accounts |= {acc for acc, _, _ in deltas.instrument_cash} + + qty: dict[tuple[int, int], Decimal] = defaultdict(lambda: ZERO) + cash: dict[tuple[int, str], Decimal] = defaultdict(lambda: ZERO) + out: dict[int, list[DayValue]] = {acc: [] for acc in accounts} + + for d in spine: + for key, by_day in deltas.positions.items(): + delta = by_day.get(d) + if delta: + qty[key] += delta + for key, by_day in deltas.cash.items(): + delta = by_day.get(d) + if delta: + cash[key] += delta + + market: dict[int, Decimal] = defaultdict(lambda: ZERO) + accrued: dict[int, Decimal] = defaultdict(lambda: ZERO) + stale: dict[int, int] = defaultdict(int) + no_price: dict[int, int] = defaultdict(int) + no_fx: dict[int, int] = defaultdict(int) + + for (account_id, instrument_id), held in qty.items(): + if held == ZERO: + continue + quote = prices.at(instrument_id, d) + if quote is None: + no_price[account_id] += 1 + continue + if quote.is_stale_on(d): + stale[account_id] += 1 + rate = fx.rate(d, quote.currency) + if rate is None: + no_fx[account_id] += 1 + continue + market[account_id] += held * quote.total * rate + if quote.accrued_interest: + accrued[account_id] += held * quote.accrued_interest * rate + + cash_rub: dict[int, Decimal] = defaultdict(lambda: ZERO) + for (account_id, ccy), balance in cash.items(): + if balance == ZERO: + continue + rate = fx.rate(d, ccy) + if rate is None: + no_fx[account_id] += 1 + continue + cash_rub[account_id] += balance * rate + + flow_rub: dict[int, Decimal] = defaultdict(lambda: ZERO) + for (account_id, ccy), by_day in deltas.flows.items(): + native = by_day.get(d) + if not native: + continue + rate = fx.rate(d, ccy) + if rate is None: + no_fx[account_id] += 1 + continue + flow_rub[account_id] += native * rate + + # cash that went into (or came out of) a position nobody quotes: it disappears from + # the valued portfolio without turning into market value, so for a time-weighted + # return it behaves exactly like a withdrawal + unvalued_rub: dict[int, Decimal] = defaultdict(lambda: ZERO) + for (account_id, instrument_id, ccy), by_day in deltas.instrument_cash.items(): + native = by_day.get(d) + if not native or prices.at(instrument_id, d) is not None: + continue + rate = fx.rate(d, ccy) + if rate is not None: + unvalued_rub[account_id] += native * rate + + for account_id in accounts: + out[account_id].append( + DayValue( + d=d, + market_value_rub=market[account_id], + accrued_interest_rub=accrued[account_id], + cash_rub=cash_rub[account_id], + external_flow_rub=flow_rub[account_id], + unvalued_flow_rub=unvalued_rub[account_id], + stale_price_count=stale[account_id], + missing_price_count=no_price[account_id], + missing_fx_count=no_fx[account_id], + ) + ) + return out + + +def combine(series: Mapping[int, Sequence[DayValue]], account_ids: Iterable[int]) -> list[DayValue]: + """Add up several accounts' series into one scope's series, day by day.""" + parts = [series[a] for a in account_ids if a in series] + if not parts: + return [] + length = len(parts[0]) + return [ + DayValue( + d=parts[0][i].d, + market_value_rub=sum((p[i].market_value_rub for p in parts), start=ZERO), + accrued_interest_rub=sum((p[i].accrued_interest_rub for p in parts), start=ZERO), + cash_rub=sum((p[i].cash_rub for p in parts), start=ZERO), + external_flow_rub=sum((p[i].external_flow_rub for p in parts), start=ZERO), + unvalued_flow_rub=sum((p[i].unvalued_flow_rub for p in parts), start=ZERO), + stale_price_count=sum(p[i].stale_price_count for p in parts), + missing_price_count=sum(p[i].missing_price_count for p in parts), + missing_fx_count=sum(p[i].missing_fx_count for p in parts), + ) + for i in range(length) + ] + + +@dataclass(frozen=True) +class OpenPosition: + """The open lots of one instrument in one scope, already folded together.""" + + instrument_id: int + qty: Decimal + """Signed: negative while a short is open.""" + cost_native: Decimal + """Signed cost of what is still held, in `cost_currency`.""" + cost_currency: str | None + cost_rub: Decimal | None + """Cost at the CBR rate of each lot's own open date; None when any lot lacked one.""" + first_open: date | None + ldv_qty: Decimal = ZERO + + +@dataclass(frozen=True) +class HoldingValue: + """What one open position is worth now, and how much of that is knowable.""" + + price: Decimal | None + price_currency: str | None + price_date: date | None + status: str + """ok | stale | missing""" + value_native: Decimal | None + value_rub: Decimal | None + accrued_rub: Decimal | None + unrealized_native: Decimal | None + unrealized_rub: Decimal | None + + +def value_holding( + position: OpenPosition, quote: Quote | None, rate: Decimal | None, *, as_of: date +) -> HoldingValue: + """Price one position. Missing price or rate yields NULLs and a status, never a zero. + + Unrealised P&L in RUB compares two different days on purpose: today's value at today's + rate against the cost at the rate of each purchase. That is the number a Russian investor + actually has — the currency revaluation is part of the gain, and the tax code agrees. + """ + if quote is None: + return HoldingValue(None, None, None, "missing", None, None, None, None, None) + + status = "stale" if quote.is_stale_on(as_of) else "ok" + value_native = position.qty * quote.total + accrued_native = position.qty * (quote.accrued_interest or ZERO) + unrealized_native = ( + value_native - position.cost_native if position.cost_currency == quote.currency else None + ) + if rate is None: + return HoldingValue( + quote.price, + quote.currency, + quote.as_of, + status, + value_native, + None, + None, + unrealized_native, + None, + ) + + value_rub = value_native * rate + return HoldingValue( + price=quote.price, + price_currency=quote.currency, + price_date=quote.as_of, + status=status, + value_native=value_native, + value_rub=value_rub, + accrued_rub=accrued_native * rate, + unrealized_native=unrealized_native, + unrealized_rub=(value_rub - position.cost_rub if position.cost_rub is not None else None), + ) + + +def weights(values: Mapping[int, Decimal | None]) -> dict[int, Decimal | None]: + """Share of the scope each holding makes up, over the part that could be valued. + + Shorts are excluded from the base: a negative leg would make the denominator smaller than + the longs it divides, and weights above 100 % mean nothing on an allocation chart. + """ + base = sum((v for v in values.values() if v is not None and v > ZERO), start=ZERO) + if base == ZERO: + return dict.fromkeys(values, None) + return {k: (v / base if v is not None else None) for k, v in values.items()} + + +# -------------------------------------------------------------------------------------- +# I/O: load the ledger, run the pure core, replace the metric tables +# -------------------------------------------------------------------------------------- + + +def _card_funded(meta: object) -> bool: + return bool(meta.get("card_funded")) if isinstance(meta, dict) else False + + +async def _load_deltas(session: AsyncSession, prices: PriceTable) -> tuple[Deltas, date | None]: + """Turn confirmed events into the per-day deltas the replay walks. + + Three separations happen here and nowhere else: + + * a **card-funded trade** leaves no cash delta at all — the money came from a linked card + and went straight into the paper, so the account balance never saw it. Counting the + payment would show a phantom overdraft against the broker's own cash snapshot; + * a **securities transfer** has no cash effect, so its contribution is valued at the + market price of its trade date; + * everything else contributes its `amount` to cash, and only `EXTERNAL_FLOW_KINDS` + contribute to the flow series. + """ + rows = ( + await session.execute( + select( + Event.account_id, + Event.instrument_id, + Event.kind, + Event.trade_date, + Event.quantity, + Event.amount, + Event.currency, + Event.meta, + ) + .join(Account, Account.id == Event.account_id) + .where(Event.status == EventStatus.confirmed) + .order_by(Event.trade_date) + ) + ).all() + if not rows: + return Deltas({}, {}, {}, {}), None + + positions: dict[tuple[int, int], dict[date, Decimal]] = defaultdict( + lambda: defaultdict(Decimal) + ) + cash: dict[tuple[int, str], dict[date, Decimal]] = defaultdict(lambda: defaultdict(Decimal)) + flows: dict[tuple[int, str], dict[date, Decimal]] = defaultdict(lambda: defaultdict(Decimal)) + instrument_cash: dict[tuple[int, int, str], dict[date, Decimal]] = defaultdict( + lambda: defaultdict(Decimal) + ) + unpriced_transfers = 0 + + for r in rows: + ccy = (r.currency or RUB).upper() + amount = Decimal(r.amount or 0) + card = _card_funded(r.meta) + + if r.instrument_id is not None and r.kind in POSITION_KINDS and r.quantity: + positions[(r.account_id, r.instrument_id)][r.trade_date] += Decimal(r.quantity) + + if amount and not card: + cash[(r.account_id, ccy)][r.trade_date] += amount + if amount and r.instrument_id is not None: + instrument_cash[(r.account_id, r.instrument_id, ccy)][r.trade_date] += amount + + if card: + # the card supplied (or absorbed) the cash: a contribution of the opposite sign + flows[(r.account_id, ccy)][r.trade_date] += -amount + elif r.kind in EXTERNAL_FLOW_KINDS: + if amount: + flows[(r.account_id, ccy)][r.trade_date] += amount + elif r.instrument_id is not None and r.quantity: + quote = prices.at(r.instrument_id, r.trade_date) + if quote is None: + unpriced_transfers += 1 + continue + flows[(r.account_id, quote.currency)][r.trade_date] += ( + Decimal(r.quantity) * quote.total + ) + + if unpriced_transfers: + FINDINGS.add( + "unpriced_transfer", + "warn", + f"Переводов бумагами без цены на дату: {unpriced_transfers} — " + "они не вошли во внешние потоки, XIRR занижен или завышен", + count=unpriced_transfers, + ) + + return Deltas(positions, cash, flows, instrument_cash), rows[0].trade_date + + +async def account_scopes(session: AsyncSession, ledger_accounts: set[int]) -> dict[str, set[int]]: + """Every set of accounts the metrics are reported for: all, each one, each portfolio.""" + scopes: dict[str, set[int]] = {"all": set(ledger_accounts)} + for account_id in sorted(ledger_accounts): + scopes[f"account:{account_id}"] = {account_id} + + rows = ( + await session.execute(select(PortfolioAccount.portfolio_id, PortfolioAccount.account_id)) + ).all() + per_portfolio: dict[int, set[int]] = defaultdict(set) + for r in rows: + if r.account_id in ledger_accounts: + per_portfolio[r.portfolio_id].add(r.account_id) + for portfolio_id, ids in per_portfolio.items(): + scopes[f"portfolio:{portfolio_id}"] = ids + return scopes + + +async def _load_open_positions(session: AsyncSession, as_of: date) -> dict[int, list[OpenPosition]]: + """Open lots folded into one position per (account, instrument). + + `lot.cost_total_rub` is the cost of the whole lot at its open date, so the part still + held is prorated by `qty_remaining / qty_open` — the rate of that day, not of today, which + is what makes the unrealised figure a real currency-revaluation number. + """ + rows = ( + await session.execute( + select( + Lot.account_id, + Lot.instrument_id, + Lot.open_date, + Lot.qty_open, + Lot.qty_remaining, + Lot.cost_per_unit, + Lot.cost_currency, + Lot.cost_total_rub, + Instrument.asset_class, + Instrument.board, + ) + .join(Instrument, Instrument.id == Lot.instrument_id) + .where(Lot.qty_remaining != 0) + ) + ).all() + + @dataclass + class _Acc: + qty: Decimal = ZERO + cost_native: Decimal = ZERO + cost_rub: Decimal | None = ZERO + ccy: str | None = None + mixed_ccy: bool = False + first_open: date | None = None + ldv_qty: Decimal = ZERO + + buckets: dict[tuple[int, int], _Acc] = defaultdict(_Acc) + for r in rows: + acc = buckets[(r.account_id, r.instrument_id)] + remaining = Decimal(r.qty_remaining) + acc.qty += remaining + acc.cost_native += Decimal(r.cost_per_unit) * remaining + ccy = r.cost_currency.upper() + if acc.ccy is None: + acc.ccy = ccy + elif acc.ccy != ccy: + acc.mixed_ccy = True + if r.cost_total_rub is None or not r.qty_open: + acc.cost_rub = None + elif acc.cost_rub is not None: + acc.cost_rub += Decimal(r.cost_total_rub) * remaining / Decimal(r.qty_open) + if acc.first_open is None or r.open_date < acc.first_open: + acc.first_open = r.open_date + exchange_traded = str(r.asset_class) in EXCHANGE_CLASSES and bool(r.board) + if exchange_traded and remaining > ZERO and (as_of - r.open_date).days >= LDV_DAYS: + acc.ldv_qty += remaining + + out: dict[int, list[OpenPosition]] = defaultdict(list) + for (account_id, instrument_id), acc in buckets.items(): + if acc.qty == ZERO: + continue + out[account_id].append( + OpenPosition( + instrument_id=instrument_id, + qty=acc.qty, + cost_native=acc.cost_native, + # a position bought in two currencies has no single native cost to compare + cost_currency=None if acc.mixed_ccy else acc.ccy, + cost_rub=acc.cost_rub, + first_open=acc.first_open, + ldv_qty=acc.ldv_qty, + ) + ) + return out + + +async def _realized_by_instrument(session: AsyncSession) -> dict[tuple[int, int], Decimal]: + """Cumulative realised P&L in RUB per (account, instrument), from closed lot legs.""" + rows = ( + await session.execute( + select( + Lot.account_id, + Lot.instrument_id, + func.sum(LotDisposal.realized_pnl_rub), + ) + .join(Lot, Lot.id == LotDisposal.lot_id) + .where(LotDisposal.realized_pnl_rub.is_not(None)) + .group_by(Lot.account_id, Lot.instrument_id) + ) + ).all() + return {(r[0], r[1]): Decimal(r[2]) for r in rows if r[2] is not None} + + +async def _income_by_instrument( + session: AsyncSession, fx: FxTable +) -> tuple[dict[tuple[int, int], Decimal], int]: + """Dividends, coupons, amortisation and the tax on them, in RUB at each payment's date.""" + rows = ( + await session.execute( + select( + Event.account_id, + Event.instrument_id, + Event.trade_date, + Event.amount, + Event.currency, + ).where( + Event.status == EventStatus.confirmed, + Event.kind.in_(INCOME_KINDS), + Event.instrument_id.is_not(None), + ) + ) + ).all() + out: dict[tuple[int, int], Decimal] = defaultdict(lambda: ZERO) + missing = 0 + for r in rows: + rub = fx.to_rub(Decimal(r.amount or 0), r.currency, r.trade_date) + if rub is None: + missing += 1 + continue + out[(r.account_id, r.instrument_id)] += rub + return out, missing + + +def merge_positions(parts: Sequence[OpenPosition]) -> OpenPosition: + """Fold one instrument's positions from several accounts into the scope's position.""" + first = parts[0] + if len(parts) == 1: + return first + currencies = {p.cost_currency for p in parts} + cost_rub: Decimal | None = ZERO + for p in parts: + if p.cost_rub is None: + cost_rub = None + break + assert cost_rub is not None + cost_rub += p.cost_rub + opens = [p.first_open for p in parts if p.first_open is not None] + return OpenPosition( + instrument_id=first.instrument_id, + qty=sum((p.qty for p in parts), start=ZERO), + cost_native=sum((p.cost_native for p in parts), start=ZERO), + cost_currency=currencies.pop() if len(currencies) == 1 else None, + cost_rub=cost_rub, + first_open=min(opens) if opens else None, + ldv_qty=sum((p.ldv_qty for p in parts), start=ZERO), + ) + + +async def _reconcile( + session: AsyncSession, cash_derived: Mapping[tuple[int, str], Decimal] +) -> None: + """Compare derived positions and cash with the broker's own snapshot (plan §1.6). + + This is the detector for operation-type mapping bugs: if a kind moves the position the + wrong way, the broker's number and ours part company. Known real mismatches (redomiciled + tickers, a delisted OTC paper) also land here — the table is a list of things to look at, + not a list of errors. + """ + latest_pos = ( + select(PositionSnapshot.account_id, func.max(PositionSnapshot.as_of).label("as_of")) + .group_by(PositionSnapshot.account_id) + .subquery() + ) + # currency "positions" are how the broker reports cash: their counterpart is the cash + # snapshot below, not a lot, so comparing them against lots would flag every account + snapshots = ( + await session.execute( + select( + PositionSnapshot.account_id, PositionSnapshot.instrument_id, PositionSnapshot.qty + ) + .join( + latest_pos, + (latest_pos.c.account_id == PositionSnapshot.account_id) + & (latest_pos.c.as_of == PositionSnapshot.as_of), + ) + .join(Instrument, Instrument.id == PositionSnapshot.instrument_id) + .where(Instrument.asset_class != AssetClass.currency) + ) + ).all() + derived_pos = { + (r.account_id, r.instrument_id): Decimal(r.qty) + for r in ( + await session.execute( + select(Lot.account_id, Lot.instrument_id, func.sum(Lot.qty_remaining).label("qty")) + .join(Instrument, Instrument.id == Lot.instrument_id) + .where(Instrument.asset_class != AssetClass.currency) + .group_by(Lot.account_id, Lot.instrument_id) + ) + ).all() + } + + mismatched: list[int] = [] + seen: set[tuple[int, int]] = set() + for r in snapshots: + key = (r.account_id, r.instrument_id) + seen.add(key) + if abs(derived_pos.get(key, ZERO) - Decimal(r.qty)) > QTY_TOLERANCE: + mismatched.append(r.instrument_id) + for key, qty in derived_pos.items(): + if key not in seen and abs(qty) > QTY_TOLERANCE: + mismatched.append(key[1]) + + if mismatched: + FINDINGS.add( + "position_vs_snapshot", + "warn", + f"Расхождение позиции с брокерским снапшотом по {len(mismatched)} инструментам", + count=len(mismatched), + ref={"instruments": sorted(set(mismatched))}, + ) + + latest_cash = ( + select(CashSnapshot.account_id, func.max(CashSnapshot.as_of).label("as_of")) + .group_by(CashSnapshot.account_id) + .subquery() + ) + cash_rows = ( + await session.execute( + select(CashSnapshot.account_id, CashSnapshot.currency, CashSnapshot.balance).join( + latest_cash, + (latest_cash.c.account_id == CashSnapshot.account_id) + & (latest_cash.c.as_of == CashSnapshot.as_of), + ) + ) + ).all() + off: list[str] = [] + for r in cash_rows: + key = (r.account_id, r.currency.upper()) + diff = cash_derived.get(key, ZERO) - Decimal(r.balance) + if abs(diff) > CASH_TOLERANCE: + off.append(f"счёт {r.account_id} {r.currency.upper()}: {diff:+.2f}") + if off: + FINDINGS.add( + "cash_vs_snapshot", + "warn", + "Расхождение кэша с брокерским снапшотом — " + "; ".join(sorted(off)), + count=len(off), + ) + + +async def rebuild_valuation(session: AsyncSession) -> None: + """Replace `metric_portfolio_value_daily` and `metric_holding` for every scope.""" + as_of = today_local() + prices = await PriceTable.load(session) + fx = await FxTable.load(session) + deltas, first_day = await _load_deltas(session, prices) + + await session.execute(delete(MetricHolding)) + await session.execute(delete(MetricPortfolioValueDaily)) + if first_day is None: + return + + spine = days_between(min(first_day, as_of), as_of) + per_account = value_series(spine=spine, deltas=deltas, prices=prices, fx=fx) + ledger_accounts = set(per_account) + scopes = await account_scopes(session, ledger_accounts) + + value_rows: list[dict[str, object]] = [] + for scope, account_ids in scopes.items(): + invested = ZERO + for point in combine(per_account, account_ids): + invested += point.external_flow_rub + total = point.total_rub + value_rows.append( + { + "scope": scope, + "d": point.d, + "market_value_rub": point.market_value_rub, + "accrued_interest_rub": point.accrued_interest_rub, + "cash_rub": point.cash_rub, + "total_rub": total, + "external_flow_rub": point.external_flow_rub, + "unvalued_flow_rub": point.unvalued_flow_rub, + "invested_net_rub": invested, + "pnl_total_rub": total - invested if point.complete else None, + "stale_price_count": point.stale_price_count, + "missing_price_count": point.missing_price_count, + "missing_fx_count": point.missing_fx_count, + } + ) + if value_rows: + await session.execute(insert(MetricPortfolioValueDaily), value_rows) + + positions = await _load_open_positions(session, as_of) + realized = await _realized_by_instrument(session) + income, income_missing_fx = await _income_by_instrument(session, fx) + await _write_holdings(session, scopes, positions, realized, income, prices, fx, as_of) + + cash_derived = {key: sum(by_day.values(), start=ZERO) for key, by_day in deltas.cash.items()} + await _reconcile(session, cash_derived) + if income_missing_fx: + FINDINGS.add( + "income_missing_fx", + "warn", + f"Выплат без курса на дату: {income_missing_fx} — они не вошли в доход по позиции", + count=income_missing_fx, + ) + log.info("valuation: %s scopes, %s days", len(scopes), len(spine)) + + +async def _write_holdings( + session: AsyncSession, + scopes: Mapping[str, set[int]], + positions: Mapping[int, list[OpenPosition]], + realized: Mapping[tuple[int, int], Decimal], + income: Mapping[tuple[int, int], Decimal], + prices: PriceTable, + fx: FxTable, + as_of: date, +) -> None: + unpriced: set[int] = set() + stale: set[int] = set() + rows: list[dict[str, object]] = [] + + for scope, account_ids in scopes.items(): + by_instrument: dict[int, list[OpenPosition]] = defaultdict(list) + for account_id in account_ids: + for position in positions.get(account_id, []): + by_instrument[position.instrument_id].append(position) + if not by_instrument: + continue + + merged = {iid: merge_positions(parts) for iid, parts in by_instrument.items()} + valued: dict[int, HoldingValue] = {} + for iid, position in merged.items(): + quote = prices.latest(iid, as_of) + rate = fx.rate(as_of, quote.currency) if quote else None + valued[iid] = value_holding(position, quote, rate, as_of=as_of) + if quote is None: + unpriced.add(iid) + elif valued[iid].status == "stale": + stale.add(iid) + + share = weights({iid: v.value_rub for iid, v in valued.items()}) + for iid, position in merged.items(): + value = valued[iid] + rows.append( + { + "scope": scope, + "instrument_id": iid, + "qty": position.qty, + "avg_cost": (position.cost_native / position.qty if position.qty else None), + "cost_currency": position.cost_currency, + "cost_total_rub": position.cost_rub, + "market_price": value.price, + "price_currency": value.price_currency, + "price_date": value.price_date, + "price_status": value.status, + "value_native": value.value_native, + "value_rub": value.value_rub, + "accrued_interest_rub": value.accrued_rub, + "unrealized_pnl_native": value.unrealized_native, + "unrealized_pnl_rub": value.unrealized_rub, + "realized_pnl_rub": sum( + (realized.get((a, iid), ZERO) for a in account_ids), start=ZERO + ), + "income_rub": sum( + (income.get((a, iid), ZERO) for a in account_ids), start=ZERO + ), + "weight": share[iid], + "xirr": None, # filled by analytics/returns.py, which runs next + "first_buy_date": position.first_open, + "days_held": ( + (as_of - position.first_open).days if position.first_open else None + ), + "ldv_eligible_qty": position.ldv_qty, + } + ) + + if rows: + await session.execute(insert(MetricHolding), rows) + if unpriced: + names = await _instrument_names(session, unpriced) + FINDINGS.add( + "holding_without_price", + "warn", + "Нет цены, стоимость позиции неизвестна: " + ", ".join(names), + count=len(unpriced), + ref={"instruments": sorted(unpriced)}, + ) + if stale: + names = await _instrument_names(session, stale) + FINDINGS.add( + "stale_price", + "info", + f"Цена старше {STALE_AFTER_DAYS} дн., оценка по последней известной: " + + ", ".join(names), + count=len(stale), + ref={"instruments": sorted(stale)}, + ) + + +async def _instrument_names(session: AsyncSession, ids: set[int]) -> list[str]: + rows = ( + await session.execute( + select(Instrument.ticker, Instrument.name).where(Instrument.id.in_(ids)) + ) + ).all() + return sorted(r.ticker or r.name for r in rows) diff --git a/backend/src/fintracker/pricing/prices.py b/backend/src/fintracker/pricing/prices.py new file mode 100644 index 0000000..dbdff59 --- /dev/null +++ b/backend/src/fintracker/pricing/prices.py @@ -0,0 +1,135 @@ +"""Dated prices: the counterpart of `pricing/fx.py` for instrument quotes. + +`price_daily` only has rows for days an instrument actually traded, and valuation needs a +number for every calendar day. The rule is the same one the plan sets out: carry the last +close forward, and after `STALE_AFTER_DAYS` keep using it but mark it stale, because a price +three weeks old is a fact about the feed, not about the market. + +There is deliberately no backward fill. Before the first quote an instrument has no price at +all, and a position valued at a price from its future would be an invention. + +Bonds: `price_daily.close` already carries the money value of the percent quote (the sources +resolve `price_pct` against the nominal schedule), and `accrued_interest` rides alongside so +valuation can report НКД separately and still add it into the market value. +""" + +from __future__ import annotations + +from bisect import bisect_right +from dataclasses import dataclass +from datetime import date +from decimal import Decimal + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.models import PriceDaily, PriceLast + +STALE_AFTER_DAYS = 10 +"""Beyond this, a carried-forward close is still used but reported as stale (plan §7.8).""" + + +@dataclass(frozen=True) +class Quote: + """One instrument's price as known on some day, and where it actually came from.""" + + price: Decimal + currency: str + as_of: date + """The day the price was quoted — not the day it is being used for.""" + accrued_interest: Decimal | None = None + + def age_on(self, d: date) -> int: + return (d - self.as_of).days + + def is_stale_on(self, d: date) -> bool: + return self.age_on(d) > STALE_AFTER_DAYS + + @property + def total(self) -> Decimal: + """Price plus accrued interest — what one unit is worth to sell today.""" + return self.price + (self.accrued_interest or Decimal(0)) + + +class PriceTable: + """`price_daily` (+ `price_last`) in memory, answering "what was it worth on day d".""" + + def __init__(self, series: dict[int, list[Quote]], last: dict[int, Quote]) -> None: + # each series is sorted by as_of, which `at` relies on for its bisect + self._series = series + self._days = {iid: [q.as_of for q in quotes] for iid, quotes in series.items()} + self._last = last + + @classmethod + async def load(cls, session: AsyncSession) -> PriceTable: + rows = ( + await session.execute( + select( + PriceDaily.instrument_id, + PriceDaily.d, + PriceDaily.close, + PriceDaily.currency, + PriceDaily.accrued_interest, + ).order_by(PriceDaily.instrument_id, PriceDaily.d) + ) + ).all() + series: dict[int, list[Quote]] = {} + for r in rows: + series.setdefault(r.instrument_id, []).append( + Quote( + price=Decimal(r.close), + currency=r.currency.upper(), + as_of=r.d, + accrued_interest=( + Decimal(r.accrued_interest) if r.accrued_interest is not None else None + ), + ) + ) + + last_rows = ( + await session.execute( + select(PriceLast.instrument_id, PriceLast.ts, PriceLast.price, PriceLast.currency) + ) + ).all() + last = { + r.instrument_id: Quote( + price=Decimal(r.price), currency=r.currency.upper(), as_of=r.ts.date() + ) + for r in last_rows + } + return cls(series, last) + + def at(self, instrument_id: int, d: date) -> Quote | None: + """Last quote on or before `d`; None when the instrument had no price by then.""" + days = self._days.get(instrument_id) + if not days: + return None + pos = bisect_right(days, d) + if pos == 0: + return None + return self._series[instrument_id][pos - 1] + + def latest(self, instrument_id: int, d: date) -> Quote | None: + """The freshest price known for `d`, preferring `price_last` when it is newer. + + `price_last` is refreshed every 15 minutes during the session, so on the current day + it is ahead of `price_daily`; it carries no accrued interest, so a bond keeps the + daily bar's НКД alongside the newer clean price. + """ + daily = self.at(instrument_id, d) + last = self._last.get(instrument_id) + if last is None or last.as_of > d: + return daily + if daily is None or last.as_of >= daily.as_of: + accrued = daily.accrued_interest if daily and daily.as_of == last.as_of else None + return Quote( + price=last.price, + currency=last.currency, + as_of=last.as_of, + accrued_interest=accrued, + ) + return daily + + @property + def instruments(self) -> set[int]: + return set(self._series) | set(self._last) diff --git a/backend/tests/analytics/test_prices.py b/backend/tests/analytics/test_prices.py new file mode 100644 index 0000000..e0a2554 --- /dev/null +++ b/backend/tests/analytics/test_prices.py @@ -0,0 +1,68 @@ +"""The dated price lookup: carry forward, never backward, and know when it is stale.""" + +from datetime import date, timedelta +from decimal import Decimal + +from fintracker.pricing.prices import STALE_AFTER_DAYS, PriceTable, Quote + +D = Decimal +GAZP, TBRU = 10, 11 + + +def table() -> PriceTable: + return PriceTable( + { + GAZP: [ + Quote(D(100), "RUB", date(2025, 3, 3)), + Quote(D(110), "RUB", date(2025, 3, 6), accrued_interest=D(5)), + ] + }, + {GAZP: Quote(D(120), "RUB", date(2025, 3, 10))}, + ) + + +def at(d: date, instrument_id: int = GAZP) -> Quote: + quote = table().at(instrument_id, d) + assert quote is not None + return quote + + +def latest(d: date, instrument_id: int = GAZP) -> Quote: + quote = table().latest(instrument_id, d) + assert quote is not None + return quote + + +def test_a_quiet_day_reuses_the_last_close(): + assert at(date(2025, 3, 5)).price == D(100) + + +def test_there_is_no_price_before_the_first_quote(): + assert table().at(GAZP, date(2025, 3, 2)) is None + assert table().at(TBRU, date(2025, 3, 5)) is None + + +def test_staleness_is_measured_from_the_day_the_price_was_quoted(): + quoted = date(2025, 3, 6) + quote = at(quoted + timedelta(days=30)) + assert quote.as_of == quoted + assert not quote.is_stale_on(quoted + timedelta(days=STALE_AFTER_DAYS)) + assert quote.is_stale_on(quoted + timedelta(days=STALE_AFTER_DAYS + 1)) + + +def test_total_adds_accrued_interest(): + assert at(date(2025, 3, 6)).total == D(115) + assert at(date(2025, 3, 3)).total == D(100) + + +def test_the_intraday_price_wins_once_the_daily_bar_is_behind(): + assert latest(date(2025, 3, 11)).price == D(120) + # …but never a price from the future of the day being valued + assert latest(date(2025, 3, 7)).price == D(110) + + +def test_price_last_alone_is_enough_to_value_an_instrument(): + only_last = PriceTable({}, {TBRU: Quote(D(7), "RUB", date(2025, 3, 10))}) + assert only_last.at(TBRU, date(2025, 3, 10)) is None + quote = only_last.latest(TBRU, date(2025, 3, 10)) + assert quote is not None and quote.price == D(7) diff --git a/backend/tests/analytics/test_returns.py b/backend/tests/analytics/test_returns.py new file mode 100644 index 0000000..0e4fea2 --- /dev/null +++ b/backend/tests/analytics/test_returns.py @@ -0,0 +1,121 @@ +"""XIRR and TWR on synthetic flows — the acceptance check from the plan, §6 phase 2. + +A single contribution: XIRR and TWR must agree, because there is no timing to weigh. +A second contribution: XIRR moves, every sub-period of TWR stays exactly where it was. +""" + +from datetime import date, timedelta +from decimal import Decimal + +import pytest + +from fintracker.analytics.returns import Point, annualize, months_back, period_start, twr, xirr + +D = Decimal +START = date(2025, 1, 1) + + +def day(n: int) -> date: + return START + timedelta(days=n) + + +def flat_series(values: dict[int, str], *, flows: dict[int, str] | None = None) -> list[Point]: + """Days 1..365 of a series whose value only changes on the days listed.""" + flows = flows or {} + points: list[Point] = [] + current = D(values[0]) + for n in range(1, 366): + if n in values: + current = D(values[n]) + points.append(Point(d=day(n), value=current, flow=D(flows.get(n, "0")))) + return points + + +def test_single_contribution_gives_ten_percent_both_ways(): + # 100 in on day 0, worth 110 a year later + assert xirr([day(0), day(365)], [D(-100), D(110)]) == D("0.100000") + + chain = twr(flat_series({0: "100", 365: "110"}), opening_value=D(100)) + assert chain.value == D("0.100000") + assert chain.days_skipped == 0 + + +def test_a_second_contribution_moves_xirr_but_not_twr(): + one = twr(flat_series({0: "100", 365: "110"}), opening_value=D(100)) + two = twr( + flat_series({0: "100", 182: "150", 365: "165"}, flows={182: "50"}), + opening_value=D(100), + ) + # the portfolio still earned exactly 10 % over the year, whoever paid in when + assert two.value == one.value == D("0.100000") + + money_weighted = xirr([day(0), day(182), day(365)], [D(-100), D(-50), D(165)]) + assert money_weighted is not None + assert money_weighted > D("0.100000") + + +def test_a_withdrawal_does_not_read_as_a_loss(): + chain = twr( + flat_series({0: "100", 182: "50", 365: "55"}, flows={182: "-50"}), + opening_value=D(100), + ) + assert chain.value == D("0.100000") + + +def test_an_unvalued_purchase_is_treated_as_leaving_the_portfolio(): + """Cash spent on a paper nobody quotes must not read as a crash (the money-market fund).""" + points = [ + Point(d=day(1), value=D(83000), flow=D(17000), unvalued_flow=D(-17000), missing=1), + Point(d=day(2), value=D(83830), flow=D(0), unvalued_flow=D(0), missing=1), + ] + chain = twr(points, opening_value=D(83000), opening_missing=1) + # day 1: 83000 / (83000 + 17000 - 17000) = 1; day 2: +1 % + assert chain.value == D("0.010000") + assert chain.days_skipped == 0 + + +def test_a_position_becoming_priced_is_skipped_not_counted_as_profit(): + points = [ + Point(d=day(1), value=D(100), flow=D(0), missing=1), + Point(d=day(2), value=D(132), flow=D(0), missing=0), + Point(d=day(3), value=D(133), flow=D(0), missing=0), + ] + chain = twr(points, opening_value=D(100), opening_missing=1) + assert chain.days_skipped == 1 + # only day 3 is chained: the 32 that appeared on day 2 was never a gain + assert chain.value == D("0.007576") + + +def test_a_period_with_no_usable_day_has_no_twr(): + points = [Point(d=day(1), value=D(100), flow=D(0), missing=1)] + assert twr(points, opening_value=D(100), opening_missing=0).value is None + + +def test_xirr_needs_flows_on_both_sides(): + assert xirr([day(0), day(365)], [D(-100), D(-50)]) is None + assert xirr([day(0)], [D(-100)]) is None + + +def test_annualize_only_above_a_year(): + assert annualize(D("0.1"), 180) is None + assert annualize(D("0.21"), 730) == D("0.100000") + assert annualize(None, 730) is None + + +@pytest.mark.parametrize( + ("period", "expected"), + [ + ("1m", date(2026, 2, 28)), + ("3m", date(2025, 12, 31)), + ("1y", date(2025, 3, 31)), + ("ytd", date(2026, 1, 1)), + ("all", None), + ], +) +def test_period_start(period: str, expected: date | None): + assert period_start(period, date(2026, 3, 31)) == expected + + +def test_months_back_clamps_to_a_shorter_month(): + assert months_back(date(2026, 3, 31), 1) == date(2026, 2, 28) + assert months_back(date(2026, 1, 15), 13) == date(2024, 12, 15) diff --git a/backend/tests/analytics/test_valuation.py b/backend/tests/analytics/test_valuation.py new file mode 100644 index 0000000..32b6cbe --- /dev/null +++ b/backend/tests/analytics/test_valuation.py @@ -0,0 +1,244 @@ +"""Valuation rules on synthetic positions — no database, no source. + +The invariants under test are the ones the multi-currency rule turns on: a missing price or +a missing rate produces NULL and a counter, never a zero; a short position is negative all +the way through; and cash spent on an unquoted paper is tracked so returns can see it. +""" + +from datetime import date, timedelta +from decimal import Decimal + +from fintracker.analytics.valuation import ( + DayValue, + Deltas, + OpenPosition, + combine, + days_between, + merge_positions, + value_holding, + value_series, + weights, +) +from fintracker.pricing.prices import STALE_AFTER_DAYS, Quote + +D = Decimal +ACC, OTHER = 1, 2 +GAZP, USD_ETF, SILENT = 10, 11, 12 +DAY = date(2025, 3, 3) + + +class Prices: + def __init__(self, quotes: dict[int, Quote]) -> None: + self._quotes = quotes + + def at(self, instrument_id: int, d: date) -> Quote | None: + return self._quotes.get(instrument_id) + + +class Fx: + def __init__(self, rates: dict[str, Decimal]) -> None: + self._rates = rates + + def rate(self, d: date, ccy: str | None) -> Decimal | None: + return self._rates.get((ccy or "").upper()) + + +def quote(price: str, *, ccy: str = "RUB", nkd: str | None = None, age: int = 0) -> Quote: + return Quote( + price=D(price), + currency=ccy, + as_of=DAY - timedelta(days=age), + accrued_interest=D(nkd) if nkd else None, + ) + + +def deltas(*, positions=None, cash=None, flows=None, instrument_cash=None) -> Deltas: + return Deltas(positions or {}, cash or {}, flows or {}, instrument_cash or {}) + + +def test_a_position_is_priced_and_converted_at_the_rate_of_its_day(): + series = value_series( + spine=[DAY], + deltas=deltas(positions={(ACC, USD_ETF): {DAY: D(10)}}), + prices=Prices({USD_ETF: quote("12", ccy="USD")}), + fx=Fx({"USD": D(80), "RUB": D(1)}), + ) + point = series[ACC][0] + assert point.market_value_rub == D(9600) # 10 * 12 * 80 + assert point.complete + + +def test_a_bond_carries_its_accrued_interest_into_the_value(): + series = value_series( + spine=[DAY], + deltas=deltas(positions={(ACC, GAZP): {DAY: D(7)}}), + prices=Prices({GAZP: quote("1000", nkd="25")}), + fx=Fx({"RUB": D(1)}), + ) + point = series[ACC][0] + assert point.market_value_rub == D(7175) + assert point.accrued_interest_rub == D(175) + + +def test_an_unquoted_position_is_counted_not_valued_at_zero(): + series = value_series( + spine=[DAY], + deltas=deltas(positions={(ACC, SILENT): {DAY: D(7)}}), + prices=Prices({}), + fx=Fx({"RUB": D(1)}), + ) + point = series[ACC][0] + assert point.market_value_rub == D(0) + assert point.missing_price_count == 1 + assert not point.complete + + +def test_a_currency_with_no_rate_drops_out_and_is_reported(): + series = value_series( + spine=[DAY], + deltas=deltas(cash={(ACC, "XBT"): {DAY: D(5)}}), + prices=Prices({}), + fx=Fx({"RUB": D(1)}), + ) + point = series[ACC][0] + assert point.cash_rub == D(0) + assert point.missing_fx_count == 1 + + +def test_a_short_position_is_negative_throughout(): + series = value_series( + spine=[DAY], + deltas=deltas(positions={(ACC, GAZP): {DAY: D(-4)}}), + prices=Prices({GAZP: quote("100")}), + fx=Fx({"RUB": D(1)}), + ) + assert series[ACC][0].market_value_rub == D(-400) + + +def test_quantities_and_cash_carry_across_days(): + spine = days_between(DAY, DAY + timedelta(days=2)) + series = value_series( + spine=spine, + deltas=deltas( + positions={(ACC, GAZP): {DAY: D(2)}}, + cash={(ACC, "RUB"): {DAY: D(500)}}, + ), + prices=Prices({GAZP: quote("100")}), + fx=Fx({"RUB": D(1)}), + ) + assert [p.total_rub for p in series[ACC]] == [D(700), D(700), D(700)] + + +def test_cash_spent_on_an_unquoted_paper_is_recorded_as_an_unvalued_flow(): + series = value_series( + spine=[DAY], + deltas=deltas( + positions={(ACC, SILENT): {DAY: D(140)}}, + cash={(ACC, "RUB"): {DAY: D(-17000)}}, + instrument_cash={(ACC, SILENT, "RUB"): {DAY: D(-17000)}}, + ), + prices=Prices({}), + fx=Fx({"RUB": D(1)}), + ) + assert series[ACC][0].unvalued_flow_rub == D(-17000) + + +def test_combine_adds_accounts_day_by_day(): + series = value_series( + spine=[DAY], + deltas=deltas( + positions={(ACC, GAZP): {DAY: D(1)}, (OTHER, GAZP): {DAY: D(2)}}, + ), + prices=Prices({GAZP: quote("100")}), + fx=Fx({"RUB": D(1)}), + ) + both = combine(series, [ACC, OTHER]) + assert [p.market_value_rub for p in both] == [D(300)] + assert combine(series, []) == [] + + +def position(**over) -> OpenPosition: + base = { + "instrument_id": GAZP, + "qty": D(10), + "cost_native": D(900), + "cost_currency": "RUB", + "cost_rub": D(900), + "first_open": date(2024, 1, 9), + } + return OpenPosition(**{**base, **over}) + + +def test_a_holding_without_a_price_is_null_with_a_status(): + value = value_holding(position(), None, D(1), as_of=DAY) + assert value.status == "missing" + assert value.value_rub is None + assert value.unrealized_rub is None + + +def test_an_old_price_is_used_but_marked_stale(): + value = value_holding(position(), quote("100", age=STALE_AFTER_DAYS + 1), D(1), as_of=DAY) + assert value.status == "stale" + assert value.value_rub == D(1000) + + +def test_unrealized_compares_todays_value_with_the_cost_of_its_own_day(): + value = value_holding( + position(cost_rub=D(72000), cost_currency="USD", cost_native=D(900)), + quote("100", ccy="USD"), + D(80), + as_of=DAY, + ) + assert value.value_rub == D(80000) + assert value.unrealized_native == D(100) # 1000 - 900 USD + assert value.unrealized_rub == D(8000) # includes the currency revaluation + + +def test_a_holding_with_no_rate_keeps_the_native_value_and_drops_the_rub_one(): + value = value_holding(position(), quote("100"), None, as_of=DAY) + assert value.value_native == D(1000) + assert value.value_rub is None + + +def test_a_short_holding_profits_when_the_price_falls(): + short = position(qty=D(-10), cost_native=D(-1000), cost_rub=D(-1000)) + value = value_holding(short, quote("90"), D(1), as_of=DAY) + assert value.value_rub == D(-900) + assert value.unrealized_rub == D(100) + + +def test_weights_ignore_shorts_and_unvalued_holdings(): + share = weights({1: D(300), 2: D(100), 3: None, 4: D(-50)}) + assert share[1] == D("0.75") + assert share[2] == D("0.25") + assert share[3] is None + assert weights({1: None})[1] is None + + +def test_merging_positions_keeps_the_earliest_open_and_drops_a_mixed_currency(): + merged = merge_positions( + [ + position(qty=D(10), cost_rub=D(900), first_open=date(2024, 5, 1)), + position(qty=D(5), cost_currency="USD", cost_rub=D(500), first_open=date(2024, 1, 9)), + ] + ) + assert merged.qty == D(15) + assert merged.cost_rub == D(1400) + assert merged.cost_currency is None + assert merged.first_open == date(2024, 1, 9) + + +def test_merging_loses_the_rub_cost_when_any_lot_lacked_a_rate(): + merged = merge_positions([position(), position(cost_rub=None)]) + assert merged.cost_rub is None + + +def test_a_day_value_totals_market_and_cash(): + point = DayValue( + d=DAY, + market_value_rub=D(100), + accrued_interest_rub=D(0), + cash_rub=D(25), + external_flow_rub=D(0), + ) + assert point.total_rub == D(125) diff --git a/backend/tests/analytics/test_valuation_db.py b/backend/tests/analytics/test_valuation_db.py new file mode 100644 index 0000000..9147185 --- /dev/null +++ b/backend/tests/analytics/test_valuation_db.py @@ -0,0 +1,270 @@ +"""Valuation and returns end to end: ledger + prices + rates -> metric tables.""" + +from datetime import date, timedelta +from decimal import Decimal + +from sqlalchemy import select + +from factories import ( + make_account, + make_cbr_rate, + make_event, + make_instrument, + make_price, + refresh, +) +from fintracker.analytics import today_local +from fintracker.db import get_sessionmaker +from fintracker.models import ( + AccountKind, + AccountRole, + AssetClass, + EventKind, + MetricDataQuality, + MetricHolding, + MetricPortfolioValueDaily, + MetricReturns, +) + +D = Decimal + + +async def broker_account() -> int: + return await make_account( + name="Брокерский", + kind=AccountKind.broker, + role=AccountRole.investment, + balance=None, + include_in_net_worth=False, + source="tinvest", + ) + + +async def rates(days: list[date], ccy: str, value: str) -> None: + for d in days: + await make_cbr_rate(d, ccy, value) + + +async def holdings(scope: str = "all") -> dict[int, MetricHolding]: + async with get_sessionmaker()() as session: + rows = ( + (await session.execute(select(MetricHolding).where(MetricHolding.scope == scope))) + .scalars() + .all() + ) + return {r.instrument_id: r for r in rows} + + +async def value_days(scope: str = "all") -> dict[date, MetricPortfolioValueDaily]: + async with get_sessionmaker()() as session: + rows = ( + ( + await session.execute( + select(MetricPortfolioValueDaily).where( + MetricPortfolioValueDaily.scope == scope + ) + ) + ) + .scalars() + .all() + ) + return {r.d: r for r in rows} + + +async def findings() -> dict[str, MetricDataQuality]: + async with get_sessionmaker()() as session: + rows = (await session.execute(select(MetricDataQuality))).scalars().all() + return {r.check_name: r for r in rows} + + +async def test_a_bought_position_is_valued_and_the_cash_it_cost_is_gone(app): + t = today_local() + bought = t - timedelta(days=3) + account = await broker_account() + gazp = await make_instrument(ticker="GAZP") + + await make_event(bought, account_id=account, kind=EventKind.deposit, amount="10000") + await make_event( + bought, + account_id=account, + kind=EventKind.buy, + instrument_id=gazp, + quantity="10", + price="900", + amount="-9000", + ) + for offset in range(4): + await make_price(t - timedelta(days=offset), instrument_id=gazp, close="950") + await refresh() + + holding = (await holdings())[gazp] + assert holding.qty == D(10) + assert holding.value_rub == D(9500) + assert holding.cost_total_rub == D(9000) + assert holding.unrealized_pnl_rub == D(500) + assert holding.price_status == "ok" + assert holding.weight == D(1) + + today = (await value_days())[t] + assert today.market_value_rub == D(9500) + assert today.cash_rub == D(1000) + assert today.total_rub == D(10500) + assert today.invested_net_rub == D(10000) + assert today.pnl_total_rub == D(500) + + +async def test_a_foreign_position_is_converted_at_the_rate_of_each_day(app): + t = today_local() + bought = t - timedelta(days=2) + account = await broker_account() + etf = await make_instrument(ticker="SPY", currency="USD", board="SPBXM") + + await make_event( + bought, account_id=account, kind=EventKind.deposit, amount="800", currency="USD" + ) + await make_event( + bought, + account_id=account, + kind=EventKind.buy, + instrument_id=etf, + quantity="8", + price="100", + amount="-800", + currency="USD", + ) + for offset in range(3): + d = t - timedelta(days=offset) + await make_price(d, instrument_id=etf, close="110", currency="USD") + await rates([t - timedelta(days=n) for n in range(5)], "USD", "80") + await refresh() + + holding = (await holdings())[etf] + assert holding.value_native == D(880) + assert holding.value_rub == D(70400) # 880 USD * 80 + assert holding.unrealized_pnl_native == D(80) + + +async def test_a_paper_nobody_quotes_is_null_not_zero(app): + t = today_local() + account = await broker_account() + silent = await make_instrument(ticker="SIBN6P4", asset_class=AssetClass.bond, board="SPBRUBND") + + await make_event( + t - timedelta(days=5), account_id=account, kind=EventKind.deposit, amount="7000" + ) + await make_event( + t - timedelta(days=5), + account_id=account, + kind=EventKind.buy, + instrument_id=silent, + quantity="7", + price="1000", + amount="-7000", + ) + await refresh() + + holding = (await holdings())[silent] + assert holding.qty == D(7) + assert holding.price_status == "missing" + assert holding.value_rub is None + assert holding.unrealized_pnl_rub is None + assert holding.weight is None + + today = (await value_days())[t] + assert today.market_value_rub == D(0) + assert today.missing_price_count == 1 + assert today.pnl_total_rub is None # the total is incomplete, so it is not reported + + assert "holding_without_price" in await findings() + + +async def test_a_card_funded_purchase_is_an_external_flow_and_leaves_the_cash_alone(app): + t = today_local() + bought = t - timedelta(days=1) + account = await broker_account() + fund = await make_instrument(ticker="TMOS", asset_class=AssetClass.etf, board="TQTF") + + await make_event( + bought, + account_id=account, + kind=EventKind.buy, + instrument_id=fund, + quantity="100", + price="50", + amount="-5000", + meta={"operation_type": "OPERATION_TYPE_BUY_CARD", "card_funded": True}, + ) + for offset in range(2): + await make_price(t - timedelta(days=offset), instrument_id=fund, close="50") + await refresh() + + today = (await value_days())[t] + assert today.cash_rub == D(0) # the card paid, the account balance never moved + assert today.market_value_rub == D(5000) + assert today.invested_net_rub == D(5000) + assert today.pnl_total_rub == D(0) + + +async def test_returns_are_reported_per_period_and_agree_with_the_flows(app): + t = today_local() + start = t - timedelta(days=370) + account = await broker_account() + gazp = await make_instrument(ticker="GAZP") + + await make_event(start, account_id=account, kind=EventKind.deposit, amount="1000") + await make_event( + start, + account_id=account, + kind=EventKind.buy, + instrument_id=gazp, + quantity="10", + price="100", + amount="-1000", + ) + d = start + while d <= t: + await make_price(d, instrument_id=gazp, close="100" if d < t - timedelta(days=5) else "110") + d += timedelta(days=1) + await refresh() + + async with get_sessionmaker()() as session: + rows = { + r.period: r + for r in ( + (await session.execute(select(MetricReturns).where(MetricReturns.scope == "all"))) + .scalars() + .all() + ) + } + assert set(rows) >= {"1m", "3m", "1y", "all"} + whole = rows["all"] + assert whole.value_end_rub == D(1100) + assert whole.external_flow_rub == D(0) # the deposit IS the opening value + assert whole.abs_pnl_rub == D(100) + assert whole.twr == D("0.100000") + assert whole.xirr is not None and whole.xirr > D(0) + + +async def test_scopes_cover_the_whole_ledger_and_each_account(app): + t = today_local() + first = await broker_account() + second = await broker_account() + gazp = await make_instrument(ticker="GAZP") + + for account in (first, second): + await make_event( + t - timedelta(days=1), + account_id=account, + kind=EventKind.buy, + instrument_id=gazp, + quantity="1", + price="100", + amount="-100", + ) + await make_price(t, instrument_id=gazp, close="100") + await make_price(t - timedelta(days=1), instrument_id=gazp, close="100") + await refresh() + + assert (await holdings())[gazp].qty == D(2) + assert (await holdings(f"account:{first}"))[gazp].qty == D(1) + assert (await holdings(f"account:{second}"))[gazp].qty == D(1)