feat(analytics): оценка позиций, XIRR и TWR

pricing/prices.py — цена на дату: последний close тянется вперёд, после 10 дней
считается протухшей, но всё ещё используется. Назад не тянется никогда — цена из
будущего это выдумка, а не оценка.

valuation берёт два разных источника намеренно. Дневная серия — реплей event
(только он отвечает, сколько стоило в марте), текущие холдинги — из lot, где
есть себестоимость и учтены сплиты. Расхождение между ними на последний день
становится находкой, а не поводом выбрать одно из двух. Отчётная единица —
scope: all, account:<id>, portfolio:<id>.

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 подпериодов.

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 ###

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,
+    )

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)

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)

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)

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)

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)

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)