From eb4eae5bebca7bca31a6dc94f10d7641d638328b Mon Sep 17 00:00:00 2001 From: Dmitry Date: Fri, 18 Sep 2026 15:06:06 +0300 Subject: [PATCH] =?UTF-8?q?feat(ledger):=20=D0=BA=D0=BE=D1=80=D0=BF=D0=BE?= =?UTF-8?q?=D1=80=D0=B0=D1=82=D0=B8=D0=B2=D0=BD=D1=8B=D0=B5=20=D0=B4=D0=B5?= =?UTF-8?q?=D0=B9=D1=81=D1=82=D0=B2=D0=B8=D1=8F=20=D0=B8=D0=B7=20=D0=BE?= =?UTF-8?q?=D0=BF=D0=B5=D1=80=D0=B0=D1=86=D0=B8=D0=B9=20=D0=B1=D1=80=D0=BE?= =?UTF-8?q?=D0=BA=D0=B5=D1=80=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit derive() выводит сплиты, амортизации и погашения из самого леджера и складывает их в corporate_action; rewrite_splits схлопывает пару безденежных transfer_out/transfer_in в синтетический split. Без этого перерегистрация бумаги реализовалась бы как фиктивный round-trip: срок владения обнулился бы, а с ним и право на ЛДВ. amount_per_unit считается пулом по инструменту и дате: одно действие эмитента приходит отдельной операцией на каждый счёт, и делить надо на суммарную позицию. Позиция для деления реплеится внутри модуля, потому что шаг обязан идти ДО лотов — именно лоты потребляют то, что он пишет. Нет позиции — NULL и находка corporate_action_unpriced, не подставленное число. Модуль пишет и удаляет только split, amortization и repayment. Дивиденды и купоны он не трогает: их календарь придёт из фида эмитента, и чистка снесла бы объявленные. Живых сплитов и погашений в данных нет — эти ветки покрыты только синтетикой. Из реального есть девять BOND_REPAYMENT, которые пулятся в семь амортизаций. --- .../fintracker/ledger/corporate_actions.py | 449 ++++++++++++++++++ backend/src/fintracker/ledger/lots.py | 7 +- backend/src/fintracker/ledger/rebuild.py | 10 +- .../tests/ledger/test_corporate_actions.py | 212 +++++++++ 4 files changed, 676 insertions(+), 2 deletions(-) create mode 100644 backend/src/fintracker/ledger/corporate_actions.py create mode 100644 backend/tests/ledger/test_corporate_actions.py diff --git a/backend/src/fintracker/ledger/corporate_actions.py b/backend/src/fintracker/ledger/corporate_actions.py new file mode 100644 index 0000000..8db3539 --- /dev/null +++ b/backend/src/fintracker/ledger/corporate_actions.py @@ -0,0 +1,449 @@ +"""Corporate actions read back out of the ledger (plan §1.5, §1.6). + +Brokers do not hand over a corporate-action feed: what arrives is the *consequence* — a +bond repayment credited to the account, a quantity that suddenly multiplies. This module +reconstructs the action from those consequences and writes it to `corporate_action`, which +is what `ledger/lots.py` then reads to reshape the lots (a split needs a ratio the trade +feed never states). + +`derive()` is pure — a sequence of `LedgerEvent` in, a list of `DerivedAction` out — so +every rule below is testable without a database. `rebuild_corporate_actions()` is the thin +session wrapper, and it upserts on the table's own unique key rather than rewriting the +table, because a later phase will also fill it from an announcement feed. + +The rules, and why they are what they are: + +* **A split is recognised two ways.** Either the source states the ratio (a `stock_split` + event, or a `split_ratio` left in `meta` by a report parser), or the position changes + with no money moving: a securities transfer out of N units and back in of M on the same + day for the same paper is a re-registration, ratio M/N. Treating that pair as a real + round-trip would realise a phantom P&L and reset the holding period, which would in turn + lose the long-term holding exemption on a paper that was never actually sold. +* **Amortisation is priced per unit from the money and the position.** The payment says + nothing about the nominal; the only way to per-unit figures is what was held when it + landed, so the derivation replays the position alongside the events. +* **A repayment is a fact, not a forecast.** It is recorded with the par actually paid per + unit, which is what the lot engine closes the remaining lots at. +* **Only the three ledger-derivable kinds are managed here.** Dividends and coupons also + arrive as events, but their calendar (ex-date, record date, an announced but unpaid + amount) needs an issuer feed, and half-filling the table from payments would leave rows + that look announced and are not. They stay out until that feed exists. + +Everything derived is `status = paid`: it is money that has already moved, or a quantity +that has already changed. +""" + +from __future__ import annotations + +import logging +from collections import defaultdict +from collections.abc import Sequence +from dataclasses import dataclass, field +from datetime import date +from decimal import Decimal + +from sqlalchemy import delete, select, tuple_ +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.analytics import FINDINGS +from fintracker.ledger.lots import LedgerEvent +from fintracker.models import Event, EventStatus, Instrument +from fintracker.models.ledger import POSITION_KINDS, EventKind +from fintracker.models.pricing import CorporateAction, CorporateActionKind, CorporateActionStatus + +log = logging.getLogger(__name__) + +ZERO = Decimal(0) + +#: Kinds this module owns end to end: it both writes and prunes them. +DERIVED_KINDS = frozenset( + { + CorporateActionKind.stock_split, + CorporateActionKind.amortization, + CorporateActionKind.repayment, + } +) + +#: Kinds the derivation reads — position moves plus the payouts that reshape a bond. +RELEVANT_KINDS = POSITION_KINDS | {EventKind.amortization} + +#: `meta` keys a source may leave a split ratio under, most explicit first. +RATIO_KEYS = ("split_ratio", "ratio") + +_IN_KINDS = frozenset({EventKind.buy, EventKind.transfer_in}) +_OUT_KINDS = frozenset({EventKind.sell, EventKind.transfer_out, EventKind.repayment}) + + +@dataclass(frozen=True) +class DerivedAction: + """One corporate action as the ledger shows it — instrument-wide, not per account. + + A split or a repayment happens to the paper, not to an account: two accounts holding the + same bond see one amortisation. The per-unit figure is therefore pooled across accounts, + which is also what makes `source_id` stable enough to upsert on. + """ + + instrument_id: int + kind: CorporateActionKind + action_date: date + source: str + amount_per_unit: Decimal | None = None + currency: str | None = None + ratio: Decimal | None = None + leg_event_ids: tuple[int, ...] = () + """Events whose position effect IS the action — the transfer pair behind a split. + + They must not be replayed as trades; `rewrite_splits()` swaps them for a split event. + """ + + @property + def source_id(self) -> str: + """Stable key within (instrument, kind, source): one action per paper per day.""" + return self.action_date.isoformat() + + +@dataclass +class DeriveResult: + actions: list[DerivedAction] = field(default_factory=list) + unpriced: list[DerivedAction] = field(default_factory=list) + """Actions whose per-unit amount stayed unknown — no position was open to divide by.""" + + +def derive(events: Sequence[LedgerEvent]) -> DeriveResult: + """Reconstruct the corporate actions a sequence of ledger events implies.""" + ordered = sorted(events, key=_order) + splits = _splits(ordered) + result = DeriveResult(actions=list(splits)) + _payouts(rewrite_splits(ordered, splits), result) + result.actions.sort(key=lambda a: (a.action_date, a.instrument_id, a.kind)) + return result + + +def rewrite_splits( + events: Sequence[LedgerEvent], actions: Sequence[DerivedAction] +) -> list[LedgerEvent]: + """Replace the transfer legs of a re-registration with a single `stock_split` event. + + The ledger can then multiply the open lots instead of closing and reopening them, which + is the whole point of noticing the pair in the first place. + """ + legs = {eid: action for action in actions for eid in action.leg_event_ids} + if not legs: + return list(events) + + rewritten: list[LedgerEvent] = [] + emitted: set[str] = set() + for event in events: + action = legs.get(event.id) + if action is None: + rewritten.append(event) + continue + if action.source_id in emitted: + continue + emitted.add(action.source_id) + rewritten.append( + LedgerEvent( + id=event.id, + account_id=event.account_id, + instrument_id=event.instrument_id, + kind=EventKind.stock_split, + trade_date=action.action_date, + quantity=None, + price=None, + price_currency=None, + amount=ZERO, + currency=event.currency, + fee=None, + ratio=action.ratio, + exchange_traded=event.exchange_traded, + source=event.source, + ) + ) + return rewritten + + +def _splits(events: Sequence[LedgerEvent]) -> list[DerivedAction]: + """Stated ratios first, then the transfer pairs that imply one.""" + actions: list[DerivedAction] = [ + DerivedAction( + instrument_id=event.instrument_id, + kind=CorporateActionKind.stock_split, + action_date=event.trade_date, + source=event.source, + ratio=ratio, + ) + for event in events + if event.kind is EventKind.stock_split and (ratio := _stated_ratio(event)) is not None + ] + stated = {(a.instrument_id, a.action_date) for a in actions} + + for (_, instrument_id, day), legs in _pairs(events).items(): + if (instrument_id, day) in stated: + continue + out_qty = sum((abs(e.quantity or ZERO) for e in legs if e.kind in _OUT_KINDS), start=ZERO) + in_qty = sum((abs(e.quantity or ZERO) for e in legs if e.kind in _IN_KINDS), start=ZERO) + if out_qty <= ZERO or in_qty <= ZERO or in_qty == out_qty: + continue + actions.append( + DerivedAction( + instrument_id=instrument_id, + kind=CorporateActionKind.stock_split, + action_date=day, + source=legs[0].source, + ratio=in_qty / out_qty, + leg_event_ids=tuple(sorted(e.id for e in legs)), + ) + ) + return actions + + +def _pairs(events: Sequence[LedgerEvent]) -> dict[tuple[int, int, date], list[LedgerEvent]]: + """Same-day cashless transfers of one paper on one account, both directions present.""" + buckets: dict[tuple[int, int, date], list[LedgerEvent]] = defaultdict(list) + for event in events: + if event.kind not in {EventKind.transfer_in, EventKind.transfer_out}: + continue + if event.amount != ZERO: + continue # money moved, so it was a trade dressed as a transfer, not a swap + buckets[(event.account_id, event.instrument_id, event.trade_date)].append(event) + return { + key: legs + for key, legs in buckets.items() + if any(e.kind is EventKind.transfer_in for e in legs) + and any(e.kind is EventKind.transfer_out for e in legs) + } + + +def _stated_ratio(event: LedgerEvent) -> Decimal | None: + """The ratio the source itself gave, from the event column or from `meta`.""" + if event.ratio and event.ratio > ZERO: + return event.ratio + for key in RATIO_KEYS: + raw = (event.meta or {}).get(key) + if raw is None: + continue + try: + ratio = Decimal(str(raw)) + except (ArithmeticError, ValueError): + continue + if ratio > ZERO: + return ratio + return None + + +def _payouts(events: Sequence[LedgerEvent], result: DeriveResult) -> None: + """Amortisations and repayments, priced per unit against the position of the day. + + The position has to be replayed here rather than read from `lot`: this module runs + *before* the lot engine, which needs the split ratios it produces. + """ + held: dict[tuple[int, int], Decimal] = defaultdict(lambda: ZERO) + pools: dict[tuple[int, date, CorporateActionKind], _Pool] = {} + + for event in events: + key = (event.account_id, event.instrument_id) + if event.kind is EventKind.stock_split: + held[key] *= event.ratio or Decimal(1) + continue + if event.kind in _IN_KINDS or event.kind in _OUT_KINDS: + if event.kind is EventKind.repayment: + # the repayment itself carries the units redeemed; the held position is the + # fallback for a feed that reports only the money + units = abs(event.quantity) if event.quantity else held[key] + _pool(pools, event, CorporateActionKind.repayment).add(event, units) + held[key] += event.quantity or ZERO + continue + if event.kind is EventKind.amortization: + _pool(pools, event, CorporateActionKind.amortization).add(event, held[key]) + + for pool in pools.values(): + action = pool.finish() + result.actions.append(action) + if action.amount_per_unit is None: + result.unpriced.append(action) + + +@dataclass +class _Pool: + """Accumulator for one (instrument, date, kind) across accounts.""" + + instrument_id: int + day: date + kind: CorporateActionKind + source: str + paid: Decimal = ZERO + qty: Decimal = ZERO + currencies: set[str] = field(default_factory=set) + + def add(self, event: LedgerEvent, qty: Decimal) -> None: + self.paid += abs(event.amount) + self.qty += abs(qty) + self.currencies.add(event.currency.upper()) + + def finish(self) -> DerivedAction: + # no position to divide by means the history starts mid-way: report the action, but + # never invent a per-unit number for it + per_unit = self.paid / self.qty if self.qty > ZERO and self.paid > ZERO else None + return DerivedAction( + instrument_id=self.instrument_id, + kind=self.kind, + action_date=self.day, + source=self.source, + amount_per_unit=per_unit, + currency=self.currencies.pop() if len(self.currencies) == 1 else None, + ratio=None, + ) + + +def _pool( + pools: dict[tuple[int, date, CorporateActionKind], _Pool], + event: LedgerEvent, + kind: CorporateActionKind, +) -> _Pool: + key = (event.instrument_id, event.trade_date, kind) + pool = pools.get(key) + if pool is None: + pool = pools[key] = _Pool( + instrument_id=event.instrument_id, + day=event.trade_date, + kind=kind, + source=event.source, + ) + return pool + + +def _order(event: LedgerEvent) -> tuple[date, int, int]: + """Chronological, opens before closes — the same order the lot engine replays in.""" + return (event.trade_date, 0 if event.kind in _IN_KINDS else 1, event.id) + + +async def rebuild_corporate_actions(session: AsyncSession) -> None: + """Derive corporate actions from the confirmed ledger and upsert them idempotently.""" + events = await load_events(session) + result = derive(events) + + rows = [ + { + "instrument_id": a.instrument_id, + "kind": a.kind, + "status": CorporateActionStatus.paid, + "record_date": None, + "ex_date": a.action_date, + "pay_date": a.action_date, + "amount_per_unit": a.amount_per_unit, + "currency": a.currency, + "ratio": a.ratio, + "source": a.source, + "source_id": a.source_id, + } + for a in result.actions + ] + if rows: + stmt = pg_insert(CorporateAction).values(rows) + await session.execute( + stmt.on_conflict_do_update( + index_elements=["instrument_id", "kind", "source", "source_id"], + set_={ + "status": stmt.excluded.status, + "ex_date": stmt.excluded.ex_date, + "pay_date": stmt.excluded.pay_date, + "amount_per_unit": stmt.excluded.amount_per_unit, + "currency": stmt.excluded.currency, + "ratio": stmt.excluded.ratio, + }, + ) + ) + + await _prune(session, result.actions) + _report(result) + log.info("corporate actions: %s derived", len(result.actions)) + + +async def _prune(session: AsyncSession, actions: Sequence[DerivedAction]) -> None: + """Drop derived rows the ledger no longer implies — an operation was cancelled upstream. + + Only `DERIVED_KINDS` are touched: dividends and coupons in this table come from an + announcement feed, and deleting them here would empty the calendar on every refresh. + """ + stmt = delete(CorporateAction).where(CorporateAction.kind.in_(DERIVED_KINDS)) + if actions: + keys = [(a.instrument_id, a.kind, a.source, a.source_id) for a in actions] + stmt = stmt.where( + tuple_( + CorporateAction.instrument_id, + CorporateAction.kind, + CorporateAction.source, + CorporateAction.source_id, + ).notin_(keys) + ) + await session.execute(stmt) + + +async def load_events(session: AsyncSession) -> list[LedgerEvent]: + """Confirmed, instrument-bearing events the derivation reads.""" + rows = ( + await session.execute( + select( + Event.id, + Event.account_id, + Event.instrument_id, + Event.kind, + Event.trade_date, + Event.quantity, + Event.amount, + Event.currency, + Event.source, + Event.meta, + ) + .join(Instrument, Instrument.id == Event.instrument_id) + .where( + Event.status == EventStatus.confirmed, + Event.kind.in_(RELEVANT_KINDS), + Event.instrument_id.is_not(None), + ) + ) + ).all() + return [ + LedgerEvent( + id=r.id, + account_id=r.account_id, + instrument_id=r.instrument_id, + kind=r.kind, + trade_date=r.trade_date, + quantity=r.quantity, + price=None, + price_currency=None, + amount=r.amount or ZERO, + currency=r.currency, + fee=None, + source=r.source, + meta=r.meta, + ) + for r in rows + ] + + +def _report(result: DeriveResult) -> None: + """Surface the actions that stayed unpriced — a gap in history, not a parsing error.""" + if not result.unpriced: + return + FINDINGS.add( + "corporate_action_unpriced", + "warn", + f"Не удалось вывести сумму на бумагу по {len(result.unpriced)} корпоративным " + "действиям — на дату выплаты позиция по инструменту не числится", + count=len(result.unpriced), + ref={"instruments": sorted({a.instrument_id for a in result.unpriced})}, + ) + + +__all__ = [ + "DERIVED_KINDS", + "DeriveResult", + "DerivedAction", + "derive", + "load_events", + "rebuild_corporate_actions", + "rewrite_splits", +] diff --git a/backend/src/fintracker/ledger/lots.py b/backend/src/fintracker/ledger/lots.py index 452187a..f20b5d7 100644 --- a/backend/src/fintracker/ledger/lots.py +++ b/backend/src/fintracker/ledger/lots.py @@ -36,10 +36,11 @@ reported as a `Shortfall` for `metric_data_quality` to surface rather than guess from __future__ import annotations from collections import defaultdict -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from datetime import date from decimal import Decimal +from typing import Any from fintracker.models.ledger import EventKind @@ -69,6 +70,10 @@ class LedgerEvent: """Split ratio; only read for `stock_split` events.""" exchange_traded: bool = True """False for OTC papers, which do not qualify for the LDV exemption.""" + source: str = "tinvest" + """Who reported it — carried so a derived corporate action can name its origin.""" + meta: Mapping[str, Any] | None = None + """Source payload leftovers; `ledger/corporate_actions.py` reads a split ratio from it.""" @dataclass diff --git a/backend/src/fintracker/ledger/rebuild.py b/backend/src/fintracker/ledger/rebuild.py index c356378..ab3b747 100644 --- a/backend/src/fintracker/ledger/rebuild.py +++ b/backend/src/fintracker/ledger/rebuild.py @@ -19,6 +19,7 @@ from sqlalchemy import delete, insert, select from sqlalchemy.ext.asyncio import AsyncSession from fintracker.analytics import FINDINGS +from fintracker.ledger.corporate_actions import derive, rewrite_splits from fintracker.ledger.lots import LedgerEvent, LotResult, apply from fintracker.models import Event, EventStatus, Instrument, Lot, LotDisposal from fintracker.models.ledger import POSITION_KINDS, EventKind @@ -122,6 +123,8 @@ async def _load_events(session: AsyncSession) -> list[LedgerEvent]: Event.currency, Event.fee, Event.accrued_interest, + Event.source, + Event.meta, Instrument.asset_class, Instrument.board, ) @@ -135,7 +138,7 @@ async def _load_events(session: AsyncSession) -> list[LedgerEvent]: ).all() ratios = await _split_ratios(session) - return [ + events = [ LedgerEvent( id=r.id, account_id=r.account_id, @@ -151,9 +154,14 @@ async def _load_events(session: AsyncSession) -> list[LedgerEvent]: accrued_interest=r.accrued_interest, ratio=ratios.get((r.instrument_id, r.trade_date)), exchange_traded=str(r.asset_class) in EXCHANGE_CLASSES and bool(r.board), + source=r.source, + meta=r.meta, ) for r in rows ] + # a re-registration arrives as a pair of cashless transfers; replaying it as a + # round-trip would realise a phantom P&L, so it becomes the split it actually is + return rewrite_splits(events, derive(events).actions) async def _split_ratios(session: AsyncSession) -> dict[tuple[int, object], Decimal]: diff --git a/backend/tests/ledger/test_corporate_actions.py b/backend/tests/ledger/test_corporate_actions.py new file mode 100644 index 0000000..07d3486 --- /dev/null +++ b/backend/tests/ledger/test_corporate_actions.py @@ -0,0 +1,212 @@ +"""Corporate actions derived from the ledger, and what they do to the lots. + +The pure half runs on synthetic events; the database half checks that a second refresh +writes the same rows instead of a second copy of them. +""" + +from datetime import date +from decimal import Decimal + +from sqlalchemy import select + +from factories import make_account, make_event, make_instrument +from fintracker.db import get_sessionmaker +from fintracker.ledger.corporate_actions import derive, rebuild_corporate_actions, rewrite_splits +from fintracker.ledger.lots import LedgerEvent, apply +from fintracker.models import AssetClass, CorporateAction, CorporateActionKind, EventKind + +D = Decimal +ACC, INS = 1, 10 +_seq = iter(range(1, 10_000)) + + +def ev(kind: EventKind, day: str, *, qty=None, amount="0", **over) -> LedgerEvent: + return LedgerEvent( + id=over.pop("id", next(_seq)), + account_id=over.pop("account_id", ACC), + instrument_id=over.pop("instrument_id", INS), + kind=kind, + trade_date=date.fromisoformat(day), + quantity=D(qty) if qty is not None else None, + price=None, + price_currency=None, + amount=D(amount), + currency="RUB", + fee=None, + ratio=D(over.pop("ratio")) if "ratio" in over else None, + meta=over.pop("meta", None), + ) + + +def buy(day, qty, amount, **kw): + return ev(EventKind.buy, day, qty=qty, amount=amount, **kw) + + +# --- the pure derivation ---------------------------------------------------------------- + + +def test_a_cashless_transfer_pair_is_read_as_a_split(): + actions = derive( + [ + buy("2025-01-10", "10", "-1000"), + ev(EventKind.transfer_out, "2025-06-02", qty="-10"), + ev(EventKind.transfer_in, "2025-06-02", qty="100"), + ] + ).actions + + assert [a.kind for a in actions] == [CorporateActionKind.stock_split] + assert actions[0].ratio == D(10) + assert actions[0].action_date == date(2025, 6, 2) + assert actions[0].source_id == "2025-06-02" + + +def test_a_transfer_pair_that_moved_money_is_a_trade_not_a_split(): + actions = derive( + [ + buy("2025-01-10", "10", "-1000"), + ev(EventKind.transfer_out, "2025-06-02", qty="-10", amount="1500"), + ev(EventKind.transfer_in, "2025-06-02", qty="100", amount="-1500"), + ] + ).actions + + assert actions == [] + + +def test_a_split_ratio_stated_in_meta_is_used(): + actions = derive( + [ + buy("2025-01-10", "10", "-1000"), + ev(EventKind.stock_split, "2025-06-02", meta={"split_ratio": "10"}), + ] + ).actions + + assert [a.ratio for a in actions] == [D(10)] + assert actions[0].leg_event_ids == () # nothing to suppress: no position event to drop + + +def test_a_split_of_ten_multiplies_quantity_and_keeps_the_position_cost(): + events = [ + buy("2025-01-10", "10", "-1000"), + ev(EventKind.transfer_out, "2025-06-02", qty="-10"), + ev(EventKind.transfer_in, "2025-06-02", qty="100"), + ] + result = apply(rewrite_splits(events, derive(events).actions)) + + assert len(result.lots) == 1 # the pair became a split, not a sale and a purchase + lot = result.lots[0] + assert lot.qty_remaining == D(100) + assert lot.cost_per_unit == D(10) + assert lot.cost_per_unit * lot.qty_remaining == D(1000) + assert result.disposals == [] # and so nothing was realised + + +def test_partial_amortisation_is_priced_per_unit_and_lowers_the_cost(): + events = [ + buy("2025-01-10", "4", "-4000"), + ev(EventKind.amortization, "2025-07-01", amount="1000"), + ] + actions = derive(events).actions + + assert [a.kind for a in actions] == [CorporateActionKind.amortization] + assert actions[0].amount_per_unit == D(250) + assert actions[0].currency == "RUB" + + lot = apply(events).lots[0] + assert lot.qty_remaining == D(4) # amortisation returns principal, not paper + assert lot.cost_per_unit == D(750) + + +def test_amortisation_pools_the_same_bond_across_accounts(): + actions = derive( + [ + buy("2025-01-10", "4", "-4000"), + buy("2025-01-10", "6", "-6000", account_id=2), + ev(EventKind.amortization, "2025-07-01", amount="1000"), + ev(EventKind.amortization, "2025-07-01", amount="1500", account_id=2), + ] + ).actions + + assert len(actions) == 1 # one action on the paper, not one per account + assert actions[0].amount_per_unit == D(250) + + +def test_a_repayment_closes_the_lots_at_par_and_is_recorded_as_one(): + events = [ + buy("2025-01-10", "4", "-4000"), + ev(EventKind.amortization, "2025-07-01", amount="1000"), + ev(EventKind.repayment, "2025-12-01", qty="-4", amount="3000"), + ] + actions = [a for a in derive(events).actions if a.kind is CorporateActionKind.repayment] + + assert [a.amount_per_unit for a in actions] == [D(750)] + assert actions[0].action_date == date(2025, 12, 1) + + result = apply(events) + assert result.lots[0].qty_remaining == D(0) + assert result.lots[0].closed_at == date(2025, 12, 1) + # cost fell to 750/unit with the amortisation, par came back at 750 — no phantom loss + assert sum(d.realized_pnl_native for d in result.disposals) == D(0) + + +def test_a_payout_with_no_position_stays_unpriced_instead_of_guessing(): + result = derive([ev(EventKind.amortization, "2025-07-01", amount="1000")]) + + assert [a.amount_per_unit for a in result.actions] == [None] + assert result.unpriced == result.actions + + +# --- the session wrapper ---------------------------------------------------------------- + + +async def _bond_with_amortisation() -> None: + account_id = await make_account(name="Брокерский", source="tinvest") + instrument_id = await make_instrument(ticker="RU000A", asset_class=AssetClass.bond) + await make_event( + date(2025, 1, 10), + account_id=account_id, + instrument_id=instrument_id, + kind=EventKind.buy, + quantity=4, + amount="-4000", + source_id="buy-1", + ) + await make_event( + date(2025, 7, 1), + account_id=account_id, + instrument_id=instrument_id, + kind=EventKind.amortization, + amount="1000", + source_id="amort-1", + ) + + +async def _rebuild() -> list[CorporateAction]: + async with get_sessionmaker()() as session: + await rebuild_corporate_actions(session) + await session.commit() + async with get_sessionmaker()() as session: + return list((await session.execute(select(CorporateAction))).scalars()) + + +async def test_rebuild_writes_the_derived_action(app): + await _bond_with_amortisation() + + rows = await _rebuild() + + assert len(rows) == 1 + assert rows[0].kind is CorporateActionKind.amortization + assert rows[0].amount_per_unit == D(250) + assert rows[0].ex_date == date(2025, 7, 1) + assert rows[0].pay_date == date(2025, 7, 1) + assert rows[0].source == "tinvest" + + +async def test_a_second_rebuild_updates_in_place_instead_of_duplicating(app): + await _bond_with_amortisation() + + first = await _rebuild() + second = await _rebuild() + + assert len(second) == 1 + assert [r.id for r in second] == [r.id for r in first] + assert second[0].amount_per_unit == D(250)