feat(ledger): корпоративные действия из операций брокера

derive() выводит сплиты, амортизации и погашения из самого леджера и складывает их в
corporate_action; rewrite_splits схлопывает пару безденежных transfer_out/transfer_in
в синтетический split. Без этого перерегистрация бумаги реализовалась бы как фиктивный
round-trip: срок владения обнулился бы, а с ним и право на ЛДВ.

amount_per_unit считается пулом по инструменту и дате: одно действие эмитента приходит
отдельной операцией на каждый счёт, и делить надо на суммарную позицию. Позиция для
деления реплеится внутри модуля, потому что шаг обязан идти ДО лотов — именно лоты
потребляют то, что он пишет. Нет позиции — NULL и находка corporate_action_unpriced,
не подставленное число.

Модуль пишет и удаляет только split, amortization и repayment. Дивиденды и купоны он
не трогает: их календарь придёт из фида эмитента, и чистка снесла бы объявленные.

Живых сплитов и погашений в данных нет — эти ветки покрыты только синтетикой. Из
реального есть девять BOND_REPAYMENT, которые пулятся в семь амортизаций.

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",
+]

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

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]:

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)