ada/fin-tracker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
47908632635172f341fe905f694623c34da51bf2
fin-tracker/backend/tests/analytics/test_income.py
T
Dmitry 15f5812ea4 feat(analytics): доходы, ребалансировка, налоги, бенчмарки и цели — фаза 4 
Второй источник выплат: sources/tinvest/sync_events.py (GetDividends,
GetBondCoupons, GetBondEvents) и sources/moex/payouts.py (ISS bondization +
dividends). Приоритет между ними — pricing/payouts.resolve_payouts, решается
на чтении, а не на записи: corporate_action уникален по (instrument_id, kind,
source, source_id), обе версии сосуществуют, и правило можно поменять без
ресинка истории. Амортизация от MOEX идёт в bond_nominal_schedule, а не
в corporate_action — этим типом безраздельно владеет
ledger/corporate_actions.py.

analytics/income.py — metric_income_monthly (факт) и metric_income_calendar
(прошлое и прогноз) с basis paid/announced/history на каждой строке, три
источника числа не смешиваются. analytics/rebalance.py — сделки по
portfolio_target пропорционально внутри бакета, лоты только вниз, покупки не
занимают у ещё не свершившихся продаж. analytics/tax.py — оценка, не замена
справки брокера: дивиденды/купоны gross, реализованный результат из
lot_disposal с переоценкой каждой ноги на свою дату. analytics/benchmarks.py —
TWR индекса на сетке портфеля, kind (price/total_return) не скрывается.
analytics/goals.py — прогресс цели и нужный взнос по trailing XIRR.

Четыре шага зарегистрированы в register_steps: benchmarks после returns
(общая сетка дат), rebalance после allocation (её веса, не пересчитывает),
income и tax после lots (нужен lot_disposal).
2026-09-19 10:42:50 +03:00

702 lines
23 KiB
Python
Raw Blame History

"""Income: the pure forecast rules first, then the rebuild end to end.
The checks the plan names are here by name: every payout received in the last 12 months has a
`paid` calendar row, and a quarterly payer produces exactly four future entries carrying its
last amount. The third test is the one that catches real money: a coupon after an
amortisation, which must shrink with the nominal instead of staying at par.
"""
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from sqlalchemy import select
from factories import make_account, make_cbr_rate, make_event, make_instrument, refresh
from fintracker.analytics import FINDINGS, today_local
from fintracker.analytics.income import (
BondFacts,
Entry,
Payment,
Payout,
add_months,
bond_entries,
coupon_per_unit,
detect_frequency,
drop_shadowed,
fold_payments,
history_entries,
monthly_rows,
nominal_at,
project_dates,
rebuild_income,
regular,
resolve_actions_fallback,
)
from fintracker.db import get_sessionmaker
from fintracker.models import (
AccountKind,
AccountRole,
AssetClass,
BondNominalSchedule,
CorporateAction,
CorporateActionKind,
CorporateActionStatus,
Event,
EventKind,
IncomeBasis,
Instrument,
MetricIncomeCalendar,
MetricIncomeMonthly,
)
D = Decimal
# --------------------------------------------------------------------------------------
# pure rules
# --------------------------------------------------------------------------------------
def quarterly(n: int, *, end: date = date(2026, 9, 1)) -> list[date]:
return sorted(add_months(end, -3 * k) for k in range(n))
def test_frequency_is_snapped_to_the_three_buckets_the_plan_allows():
assert detect_frequency(quarterly(8)) == 4
assert detect_frequency([date(2024, 6, 1), date(2024, 12, 1), date(2025, 6, 1)]) == 2
assert detect_frequency([date(2024, 6, 1), date(2025, 6, 1), date(2026, 6, 1)]) == 1
def test_a_single_payment_reads_as_annual_and_no_payment_reads_as_nothing():
# one payment says nothing about spacing, but dropping a payout we have actually seen
# would hide it entirely — annual is the commonest Russian dividend
assert detect_frequency([date(2026, 5, 20)]) == 1
assert detect_frequency([]) is None
def test_irregular_spacing_is_flagged_but_still_forecast():
assert regular(quarterly(5))
assert not regular([date(2025, 1, 10), date(2025, 2, 10), date(2026, 5, 10)])
def test_a_quarterly_payer_gives_exactly_four_dates_in_a_year():
last = date(2026, 6, 15)
dates = project_dates(last, 4, start=date(2026, 7, 1), end=date(2027, 6, 30))
assert dates == [date(2026, 9, 15), date(2026, 12, 15), date(2027, 3, 15), date(2027, 6, 15)]
def test_a_coupon_follows_the_nominal_in_force_on_its_own_date():
schedule = [(date(2024, 1, 1), D(1000)), (date(2026, 6, 1), D(500))]
assert nominal_at(schedule, date(2026, 5, 31)) == D(1000)
assert nominal_at(schedule, date(2026, 6, 1)) == D(500)
assert nominal_at(schedule, date(2023, 1, 1)) is None
# a coupon published against par halves once half the principal has been repaid
assert coupon_per_unit(D(40), D(1000), D(500)) == D(20)
assert coupon_per_unit(D(40), D(1000), D(1000)) == D(40)
assert coupon_per_unit(D(40), None, D(500)) == D(40)
def test_bond_entries_cover_coupon_amortisation_and_redemption():
facts = BondFacts(
nominal=D(1000),
nominal_schedule=((date(2024, 1, 1), D(1000)), (date(2026, 11, 1), D(600))),
maturity_date=date(2027, 5, 1),
currency="RUB",
)
coupons = [
Payout(1, "coupon", "announced", None, date(2026, 10, 1), D(40), "RUB"),
Payout(1, "coupon", "announced", None, date(2027, 4, 1), D(40), "RUB"),
]
entries = bond_entries(1, facts, coupons, D(10), start=date(2026, 9, 18), end=date(2027, 9, 18))
by_kind = {(e.kind, e.expected_date): e for e in entries}
assert by_kind[("coupon", date(2026, 10, 1))].amount == D(400)
# after the amortisation the same published coupon is worth 60 % of itself
assert by_kind[("coupon", date(2027, 4, 1))].amount == D(240)
assert by_kind[("amortization", date(2026, 11, 1))].amount == D(4000)
assert by_kind[("repayment", date(2027, 5, 1))].amount == D(6000)
assert all(e.basis is IncomeBasis.schedule for e in entries)
def test_an_announced_payout_displaces_the_projection_of_the_same_payment():
announced = [
Entry(
1,
"dividend",
date(2026, 10, 12),
date(2026, 10, 9),
D(20),
D(5),
D(100),
"RUB",
IncomeBasis.announced,
)
]
projected = [
Entry(
1, "dividend", date(2026, 10, 20), None, D(20), D(4), D(80), "RUB", IncomeBasis.history
),
Entry(
1, "dividend", date(2027, 4, 20), None, D(20), D(4), D(80), "RUB", IncomeBasis.history
),
]
kept = drop_shadowed(announced, projected)
# the declared autumn payment wins; the undeclared spring one survives
assert [e.expected_date for e in kept] == [date(2027, 4, 20)]
def test_the_fallback_resolver_prefers_the_strongest_status():
same_day = date(2026, 10, 12)
payouts = [
Payout(1, "dividend", "forecast", None, same_day, D(3), "RUB"),
Payout(1, "dividend", "announced", None, same_day, D(5), "RUB"),
Payout(1, "dividend", "cancelled", None, date(2026, 11, 1), D(9), "RUB"),
]
resolved = resolve_actions_fallback(payouts)
assert [(p.status, p.amount_per_unit) for p in resolved] == [("announced", D(5))]
def payment(d: date, amount: str, *, held: str = "10", kind: str = "dividend") -> Payment:
return Payment(
account_id=1,
instrument_id=1,
kind=kind,
d=d,
currency="RUB",
amount=D(amount),
tax=D("0"),
held_qty=D(held),
)
def test_history_extrapolates_the_last_amount_per_unit_onto_the_current_position():
payments = [payment(d, "1000") for d in quarterly(8)]
entries, steady = history_entries(
1, payments, D(5), start=date(2026, 9, 18), end=date(2027, 9, 17)
)
assert steady
assert len(entries) == 4
# 1000 ₽ on 10 units, now holding 5 — half the money, not the same money
assert {e.amount for e in entries} == {D(500)}
assert all(e.basis is IncomeBasis.history for e in entries)
def test_payments_fold_per_instrument_kind_and_day_across_accounts():
d = date(2026, 8, 12)
entries = fold_payments([payment(d, "600", held="6"), payment(d, "400", held="4")])
assert len(entries) == 1
assert (entries[0].amount, entries[0].qty, entries[0].per_unit) == (D(1000), D(10), D(100))
assert entries[0].basis is IncomeBasis.paid
def test_monthly_rows_group_by_month_kind_and_currency():
rows = monthly_rows(
[
payment(date(2026, 8, 3), "100"),
payment(date(2026, 8, 20), "200"),
payment(date(2026, 8, 20), "300", kind="coupon"),
payment(date(2026, 9, 1), "400"),
]
)
assert rows[(date(2026, 8, 1), "dividend", "RUB")] == (D(300), D(0), 2)
assert rows[(date(2026, 8, 1), "coupon", "RUB")] == (D(300), D(0), 1)
assert rows[(date(2026, 9, 1), "dividend", "RUB")] == (D(400), D(0), 1)
# --------------------------------------------------------------------------------------
# the rebuild, against the database
# --------------------------------------------------------------------------------------
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 make_bond(
*,
ticker: str = "RU000A0",
nominal: str = "1000",
maturity: date | None = None,
currency: str = "RUB",
) -> int:
async with get_sessionmaker()() as session:
bond = Instrument(
asset_class=AssetClass.bond,
ticker=ticker,
board="TQOB",
name=ticker,
currency=currency,
nominal=D(nominal),
nominal_currency=currency,
maturity_date=maturity,
)
session.add(bond)
await session.commit()
await session.refresh(bond)
return bond.id
async def make_payout(
d: date,
*,
account_id: int,
instrument_id: int,
amount: str,
kind: EventKind = EventKind.dividend,
tax: str | None = None,
currency: str = "RUB",
) -> None:
"""A received payout. `make_event` has no `tax`, and the tax column is the point here."""
async with get_sessionmaker()() as session:
session.add(
Event(
account_id=account_id,
instrument_id=instrument_id,
kind=kind,
ts=datetime.combine(d, datetime.min.time(), tzinfo=UTC),
trade_date=d,
amount=D(amount),
currency=currency,
tax=D(tax) if tax is not None else None,
tax_currency=currency if tax is not None else None,
source="tinvest",
source_id=f"pay-{instrument_id}-{d}-{amount}-{kind}",
dedupe_key=f"tinvest:pay-{instrument_id}-{d}-{amount}-{kind}",
)
)
await session.commit()
async def make_action(
*,
instrument_id: int,
kind: CorporateActionKind,
status: CorporateActionStatus,
pay_date: date | None = None,
record_date: date | None = None,
amount_per_unit: str | None = None,
currency: str = "RUB",
source: str = "moex",
) -> None:
async with get_sessionmaker()() as session:
session.add(
CorporateAction(
instrument_id=instrument_id,
kind=kind,
status=status,
pay_date=pay_date,
record_date=record_date,
amount_per_unit=D(amount_per_unit) if amount_per_unit is not None else None,
currency=currency,
source=source,
source_id=f"{kind}-{pay_date}",
)
)
await session.commit()
async def make_nominal(instrument_id: int, effective: date, nominal: str) -> None:
async with get_sessionmaker()() as session:
session.add(
BondNominalSchedule(
instrument_id=instrument_id,
effective_date=effective,
nominal=D(nominal),
currency="RUB",
source="moex",
)
)
await session.commit()
async def rebuild() -> None:
async with get_sessionmaker()() as session:
await rebuild_income(session)
await session.commit()
async def calendar(scope: str = "all") -> list[MetricIncomeCalendar]:
async with get_sessionmaker()() as session:
return list(
(
await session.execute(
select(MetricIncomeCalendar)
.where(MetricIncomeCalendar.scope == scope)
.order_by(MetricIncomeCalendar.expected_date)
)
)
.scalars()
.all()
)
async def monthly(scope: str = "all") -> list[MetricIncomeMonthly]:
async with get_sessionmaker()() as session:
return list(
(
await session.execute(
select(MetricIncomeMonthly)
.where(MetricIncomeMonthly.scope == scope)
.order_by(MetricIncomeMonthly.month, MetricIncomeMonthly.kind)
)
)
.scalars()
.all()
)
def forecast_within(rows, months: int = 12) -> list[MetricIncomeCalendar]:
"""Future rows inside a half-open window of `months`, the way the API reads them."""
today = today_local()
end = add_months(today, months)
return [r for r in rows if r.basis is not IncomeBasis.paid and today <= r.expected_date < end]
async def test_every_payout_of_the_last_year_has_a_paid_calendar_row(app):
"""Plan check: каждый полученный дивиденд/купон за 12 мес имеет запись календаря."""
today = today_local()
account = await broker_account()
share = await make_instrument(ticker="SBER", name="Сбербанк")
bond = await make_bond()
await make_event(
today - timedelta(days=400),
account_id=account,
kind=EventKind.buy,
instrument_id=share,
quantity="20",
price="250",
amount="-5000",
)
await make_event(
today - timedelta(days=400),
account_id=account,
kind=EventKind.buy,
instrument_id=bond,
quantity="10",
price="1000",
amount="-10000",
)
paid_days = [30, 120, 210, 300]
for offset in paid_days:
await make_payout(
today - timedelta(days=offset),
account_id=account,
instrument_id=share,
amount="400",
)
await make_payout(
today - timedelta(days=60),
account_id=account,
instrument_id=bond,
amount="400",
kind=EventKind.coupon,
)
await refresh()
await rebuild()
rows = await calendar()
paid = {(r.instrument_id, r.kind, r.expected_date) for r in rows if r.basis is IncomeBasis.paid}
for offset in paid_days:
assert (share, "dividend", today - timedelta(days=offset)) in paid
assert (bond, "coupon", today - timedelta(days=60)) in paid
assert len(paid) == len(paid_days) + 1
async def test_a_quarterly_payer_gives_four_future_entries_with_the_last_amount(app):
"""Plan check: квартальный плательщик даёт 4 будущих записи с последней суммой."""
today = today_local()
account = await broker_account()
share = await make_instrument(ticker="LKOH", name="Лукойл")
await make_event(
add_months(today, -30),
account_id=account,
kind=EventKind.buy,
instrument_id=share,
quantity="10",
price="100",
amount="-1000",
)
for k in range(1, 9): # eight payments, one a quarter, the last three months ago
await make_payout(
add_months(today, -3 * k),
account_id=account,
instrument_id=share,
amount="500",
)
await refresh()
await rebuild()
future = forecast_within(await calendar())
assert len(future) == 4
assert {r.basis for r in future} == {IncomeBasis.history}
assert {r.amount for r in future} == {D(500)}
assert {r.per_unit for r in future} == {D(50)}
async def test_a_coupon_shrinks_with_the_nominal_after_an_amortisation(app):
today = today_local()
account = await broker_account()
bond = await make_bond(ticker="RU000AMORT")
await make_event(
today - timedelta(days=30),
account_id=account,
kind=EventKind.buy,
instrument_id=bond,
quantity="10",
price="1000",
amount="-10000",
)
await make_nominal(bond, today - timedelta(days=700), "1000")
await make_nominal(bond, add_months(today, 2), "500")
for months in (1, 4):
await make_action(
instrument_id=bond,
kind=CorporateActionKind.coupon,
status=CorporateActionStatus.announced,
pay_date=add_months(today, months),
amount_per_unit="40",
)
await refresh()
await rebuild()
coupons = {r.expected_date: r for r in await calendar() if r.kind == "coupon"}
before = coupons[add_months(today, 1)]
after = coupons[add_months(today, 4)]
assert before.basis is IncomeBasis.schedule
assert (before.per_unit, before.amount) == (D(40), D(400))
# half the principal has been repaid, so the same published coupon pays half
assert (after.per_unit, after.amount) == (D(20), D(200))
# ...and the amortisation itself is a payment, priced off the step in the schedule
amortisation = next(r for r in await calendar() if r.kind == "amortization")
assert (amortisation.expected_date, amortisation.amount) == (add_months(today, 2), D(5000))
async def test_a_sold_position_leaves_the_forecast_and_a_halved_one_halves_it(app):
today = today_local()
account = await broker_account()
kept = await make_instrument(ticker="GAZP", name="Газпром")
gone = await make_instrument(ticker="MGNT", name="Магнит")
for instrument in (kept, gone):
await make_event(
add_months(today, -18),
account_id=account,
kind=EventKind.buy,
instrument_id=instrument,
quantity="20",
price="100",
amount="-2000",
)
await make_payout(
add_months(today, -12),
account_id=account,
instrument_id=instrument,
amount="2000",
)
await make_event(
add_months(today, -2),
account_id=account,
kind=EventKind.sell,
instrument_id=gone,
quantity="-20",
price="100",
amount="2000",
)
await make_event(
add_months(today, -2),
account_id=account,
kind=EventKind.sell,
instrument_id=kept,
quantity="-10",
price="100",
amount="1000",
)
await refresh()
await rebuild()
future = forecast_within(await calendar())
assert [r.instrument_id for r in future] == [kept]
# 2000 ₽ on 20 units, 10 units left: the forecast follows the position, not the history
assert future[0].amount == D(1000)
async def test_an_announced_dividend_beats_the_history_of_the_same_payment(app):
today = today_local()
account = await broker_account()
share = await make_instrument(ticker="TATN", name="Татнефть")
await make_event(
add_months(today, -18),
account_id=account,
kind=EventKind.buy,
instrument_id=share,
quantity="10",
price="500",
amount="-5000",
)
await make_payout(add_months(today, -12), account_id=account, instrument_id=share, amount="300")
announced_on = today + timedelta(days=10)
await make_action(
instrument_id=share,
kind=CorporateActionKind.dividend,
status=CorporateActionStatus.announced,
record_date=today + timedelta(days=7),
pay_date=announced_on,
amount_per_unit="45",
)
await refresh()
await rebuild()
future = forecast_within(await calendar())
assert len(future) == 1
row = future[0]
assert (row.basis, row.expected_date, row.amount) == (
IncomeBasis.announced,
announced_on,
D(450),
)
assert row.record_date == today + timedelta(days=7)
async def test_history_groups_by_month_kind_and_currency_and_sums_the_tax(app):
today = today_local()
account = await broker_account()
share = await make_instrument(ticker="PHOR", name="ФосАгро")
bond = await make_bond(ticker="RU000TAX")
await make_event(
add_months(today, -12),
account_id=account,
kind=EventKind.buy,
instrument_id=share,
quantity="10",
price="100",
amount="-1000",
)
month = add_months(today.replace(day=5), -3)
await make_payout(month, account_id=account, instrument_id=share, amount="870", tax="130")
await make_payout(
month + timedelta(days=10),
account_id=account,
instrument_id=share,
amount="435",
tax="65",
)
await make_payout(
month + timedelta(days=2),
account_id=account,
instrument_id=bond,
amount="400",
kind=EventKind.coupon,
)
await refresh()
await rebuild()
rows = {(r.month, r.kind): r for r in await monthly()}
dividends = rows[(month.replace(day=1), "dividend")]
assert (dividends.amount, dividends.tax_withheld, dividends.payment_count) == (
D(1305),
D(195),
2,
)
assert dividends.currency == "RUB"
assert rows[(month.replace(day=1), "coupon")].amount == D(400)
async def test_a_payment_without_a_rate_keeps_its_row_and_loses_only_the_rouble_column(app):
today = today_local()
account = await broker_account()
share = await make_instrument(ticker="TCS", name="TCS Group", currency="USD")
await make_event(
add_months(today, -12),
account_id=account,
kind=EventKind.buy,
instrument_id=share,
quantity="10",
price="10",
amount="-100",
currency="USD",
)
paid_on = add_months(today, -2)
await make_payout(paid_on, account_id=account, instrument_id=share, amount="20", currency="USD")
await refresh()
FINDINGS.reset()
await rebuild()
row = next(r for r in await calendar() if r.basis is IncomeBasis.paid)
assert (row.amount, row.currency, row.amount_rub) == (D(20), "USD", None)
assert next(r for r in await monthly()).amount_rub is None
assert any(f.check_name == "income_missing_fx" for f in FINDINGS.items)
async def test_a_rate_on_the_payment_date_fills_the_rouble_column(app):
today = today_local()
account = await broker_account()
share = await make_instrument(ticker="TCS", name="TCS Group", currency="USD")
await make_event(
add_months(today, -12),
account_id=account,
kind=EventKind.buy,
instrument_id=share,
quantity="10",
price="10",
amount="-100",
currency="USD",
)
paid_on = add_months(today, -2)
await make_cbr_rate(paid_on, "USD", "90")
await make_payout(paid_on, account_id=account, instrument_id=share, amount="20", currency="USD")
await refresh()
await rebuild()
row = next(r for r in await calendar() if r.basis is IncomeBasis.paid)
assert row.amount_rub == D(1800)
async def test_an_instrument_with_neither_schedule_nor_history_is_a_warning_not_a_zero(app):
today = today_local()
account = await broker_account()
share = await make_instrument(ticker="SILENT", name="Ничего не платит")
await make_event(
add_months(today, -6),
account_id=account,
kind=EventKind.buy,
instrument_id=share,
quantity="10",
price="100",
amount="-1000",
)
await refresh()
FINDINGS.reset()
await rebuild()
assert forecast_within(await calendar()) == []
finding = next(f for f in FINDINGS.items if f.check_name == "income_without_history")
assert finding.severity == "warn"
assert finding.ref == {"instruments": [share]}
async def test_the_tables_are_rebuilt_from_scratch_on_every_run(app):
today = today_local()
account = await broker_account()
share = await make_instrument(ticker="ROSN", name="Роснефть")
await make_event(
add_months(today, -12),
account_id=account,
kind=EventKind.buy,
instrument_id=share,
quantity="10",
price="100",
amount="-1000",
)
await make_payout(add_months(today, -6), account_id=account, instrument_id=share, amount="500")
await refresh()
await rebuild()
before = len(await calendar())
await rebuild()
assert len(await calendar()) == before
assert before > 0
Reference in New Issue View Git Blame Copy Permalink