feat(analytics): аллокация портфеля по классу, сектору, стране и валюте

Каждое измерение покрывает ОДИН И ТОТ ЖЕ итог — бумаги плюс кэш. Четыре диаграммы одного портфеля обязаны быть одного размера, иначе экраны противоречат друг другу, поэтому кэш это бакет в каждом разрезе, а не то, что выброшено из тех, куда он неочевидно ложится. Исключение — валютный разрез: деньги в рубле лежат вместе с рублёвыми бумагами, потому что вопрос к этой диаграмме именно такой. Бакет хранится ключом, а не подписью: класс актива как есть, сектор и страна как их пишет источник, плюс два литерала — cash и unknown. Язык живёт в клиенте; зашивать его в данные значит зашивать один язык навсегда. Позиция без цены исключается, а не считается нулём: ноль тихо ужал бы все остальные веса. Короткая позиция сохраняет свою величину, но не уменьшает знаменатель — иначе длинная сторона вылезла бы за 100 %, что на круговой диаграмме не значит ничего. Derived-кэш вынесен в valuation.cash_balances(): одно определение «нашего кэша» для сверки со снапшотом брокера и для аллокации, с одним и тем же исключением покупок с карты, деньги которых баланс счёта никогда не видел.
2026-09-18 14:20:37 +03:00
parent c203ae65fc
commit b58ffb3aac
7 changed files with 426 additions and 3 deletions
@@ -0,0 +1,88 @@
+"""Allocation rules on synthetic holdings — no database."""
+
+from decimal import Decimal
+
+from fintracker.analytics.allocation import CASH, UNKNOWN, Holding, bucket_of, split, weigh
+from fintracker.models import AllocationDimension
+
+D = Decimal
+DIM = AllocationDimension
+
+
+def holding(
+    instrument_id: int,
+    value: str,
+    *,
+    asset_class: str = "share",
+    sector: str | None = "energy",
+    country: str | None = "RU",
+    currency: str = "RUB",
+) -> Holding:
+    return Holding(
+        instrument_id=instrument_id,
+        value_rub=D(value),
+        asset_class=asset_class,
+        sector=sector,
+        country=country,
+        currency=currency,
+    )
+
+
+def test_buckets_are_summed_per_dimension_and_sorted_by_value():
+    rows = split(
+        [
+            holding(1, "100", asset_class="share"),
+            holding(2, "300", asset_class="bond"),
+            holding(3, "50", asset_class="share"),
+        ],
+        {},
+        DIM.asset_class,
+    )
+    assert rows == [("bond", D(300), 1), ("share", D(150), 2)]
+
+
+def test_cash_is_its_own_bucket_everywhere_but_the_currency_chart():
+    holdings = [holding(1, "900", currency="RUB")]
+    for dimension in (DIM.asset_class, DIM.sector, DIM.country):
+        assert (CASH, D(100), 0) in split(holdings, {"RUB": D(100)}, dimension)
+    # the currency question is "how exposed am I to this money", and cash is exposure too
+    assert split(holdings, {"RUB": D(100)}, DIM.currency) == [("RUB", D(1000), 1)]
+
+
+def test_an_unfilled_attribute_becomes_its_own_bucket_not_a_guess():
+    rows = split([holding(1, "100", sector=None, country=None)], {}, DIM.sector)
+    assert rows == [(UNKNOWN, D(100), 1)]
+    assert bucket_of(holding(1, "1", country=None), DIM.country) == UNKNOWN
+
+
+def test_every_dimension_covers_the_same_total():
+    holdings = [
+        holding(1, "600", asset_class="bond", sector="gov", country="RU", currency="RUB"),
+        holding(2, "400", asset_class="etf", sector=None, country="US", currency="USD"),
+    ]
+    cash = {"RUB": D(200)}
+    totals = {
+        dimension: sum(value for _, value, _ in split(holdings, cash, dimension))
+        for dimension in AllocationDimension
+    }
+    assert set(totals.values()) == {D(1200)}
+
+
+def test_weights_add_up_to_one():
+    weighted = weigh(
+        split([holding(1, "300"), holding(2, "100")], {"RUB": D(100)}, DIM.asset_class)
+    )
+    assert sum(w for _, _, _, w in weighted) == D(1)
+
+
+def test_a_short_keeps_its_value_but_does_not_inflate_the_longs():
+    weighted = weigh([("share", D(100), 1), ("bond", D(-50), 1)])
+    shares = next(row for row in weighted if row[0] == "share")
+    shorts = next(row for row in weighted if row[0] == "bond")
+    assert shares[3] == D(1)
+    assert shorts[1] == D(-50)
+
+
+def test_an_empty_portfolio_has_no_buckets():
+    assert split([], {}, DIM.asset_class) == []
+    assert weigh([]) == []