feat(ledger): единый леджер событий и FIFO-лоты
Одна таблица event, куда маппится каждый брокерский источник: quantity знаковый по эффекту на позицию, amount — по эффекту на кэш, dedupe_key UNIQUE делает повторный импорт пустой операцией. Аналитика читает только confirmed. lots.apply() — чистая функция без сессии, rebuild.py её обвязка с БД. FIFO по (счёт, инструмент), как требует ст. 214.1 НК; комиссии капитализируются в покупку и вычитаются из продажи, НКД в себестоимость не входит — это деньги, авансированные продавцу и возвращаемые купоном. Короткие продажи — тоже лоты: продажа без остатка открывает короткий лот, покупка его закрывает, прибыль равна падению цены. В данных такое есть (TATN продан 22.08 и выкуплен 29.08); считать это ошибкой значило бы оставить фантомный длинный лот навсегда и потерять реализованную прибыль. Всё пересобирается с нуля на каждом refresh: объёмы личные, это секунды, зато исчезает целый класс багов расхождения инкремента с леджером.
This commit is contained in:
Dmitry
@@ -0,0 +1 @@
|
||||
"""Derivations over the event ledger: FIFO lots, realised P&L, and source matching."""
|
||||
@@ -0,0 +1,299 @@
|
||||
"""FIFO lots and realised P&L (plan §1.5).
|
||||
|
||||
`apply()` is a pure function over a sequence of ledger events: no session, no I/O, so the
|
||||
rules below are testable on a handful of synthetic events. `rebuild_lots()` is the thin
|
||||
wrapper that loads from the database, runs it, and writes `lot` / `lot_disposal` back.
|
||||
|
||||
Rules, and why they are what they are:
|
||||
|
||||
* **FIFO per (account, instrument)**, as art. 214.1 NK requires for Russian tax. The same
|
||||
paper on two accounts is two independent queues — brokers compute tax per account.
|
||||
* **Commissions are capitalised into the purchase and deducted from the proceeds.** That
|
||||
is what makes realised P&L the number the tax authority expects, rather than a gross
|
||||
figure that overstates the gain.
|
||||
* **Accrued interest (НКД) is NOT part of the lot's cost.** It is money advanced to the
|
||||
seller and returned through the next coupon; folding it into cost would understate coupon
|
||||
income and overstate the gain on sale. It is tracked separately on the event.
|
||||
* **A split rewrites the open lots** — quantity times the ratio, cost per unit divided by
|
||||
it — so the total cost of a position never changes on a split.
|
||||
* **Amortisation reduces cost per unit**, since part of the principal has come back; the
|
||||
quantity is untouched. Without this, a fully amortised bond would show a phantom loss at
|
||||
repayment.
|
||||
* **Everything is rebuilt from scratch** on each refresh. The volumes are personal (a few
|
||||
thousand events), the recompute costs seconds, and it removes the entire class of bugs
|
||||
where an incremental update and the ledger disagree.
|
||||
|
||||
* **Short positions are lots too.** A sale with nothing to consume opens a SHORT lot, and a
|
||||
later purchase closes it — profit is the fall in price, the mirror of a long. Real data
|
||||
has these: TATN was sold on 22.08 and bought back on 29.08. Treating the sale as an error
|
||||
would leave a phantom long lot open forever and lose the realised gain.
|
||||
|
||||
A short lot still open at the end of the replay is the ambiguous case: it is either an open
|
||||
short or history that starts mid-way (a paper transferred in without a record). It is
|
||||
reported as a `Shortfall` for `metric_data_quality` to surface rather than guessed at.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
|
||||
from fintracker.models.ledger import EventKind
|
||||
|
||||
ZERO = Decimal(0)
|
||||
LDV_DAYS = 3 * 365
|
||||
"""Holding period qualifying for the long-term holding exemption (art. 219.1 NK)."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LedgerEvent:
|
||||
"""What the lot engine needs from an `event` row — deliberately not the ORM object."""
|
||||
|
||||
id: int
|
||||
account_id: int
|
||||
instrument_id: int
|
||||
kind: EventKind
|
||||
trade_date: date
|
||||
quantity: Decimal | None
|
||||
"""Signed by position effect, as stored."""
|
||||
price: Decimal | None
|
||||
price_currency: str | None
|
||||
amount: Decimal
|
||||
currency: str
|
||||
fee: Decimal | None
|
||||
accrued_interest: Decimal | None = None
|
||||
ratio: Decimal | None = None
|
||||
"""Split ratio; only read for `stock_split` events."""
|
||||
exchange_traded: bool = True
|
||||
"""False for OTC papers, which do not qualify for the LDV exemption."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class OpenLot:
|
||||
account_id: int
|
||||
instrument_id: int
|
||||
open_event_id: int
|
||||
open_date: date
|
||||
qty_open: Decimal
|
||||
"""Always positive; `direction` says which way the position points."""
|
||||
qty_remaining: Decimal
|
||||
cost_per_unit: Decimal
|
||||
"""For a short lot this is the price it was SOLD at — what the buy-back is measured against."""
|
||||
cost_currency: str
|
||||
closed_at: date | None = None
|
||||
exchange_traded: bool = True
|
||||
direction: int = 1
|
||||
"""+1 long, -1 short."""
|
||||
|
||||
@property
|
||||
def signed_qty_remaining(self) -> Decimal:
|
||||
"""What the position actually is — negative while a short is open."""
|
||||
return self.qty_remaining * self.direction
|
||||
|
||||
|
||||
@dataclass
|
||||
class Disposal:
|
||||
lot: OpenLot
|
||||
close_event_id: int
|
||||
close_date: date
|
||||
qty: Decimal
|
||||
proceeds: Decimal
|
||||
proceeds_currency: str
|
||||
cost: Decimal
|
||||
"""Cost of the disposed quantity, in the lot's own currency."""
|
||||
realized_pnl_native: Decimal
|
||||
holding_days: int
|
||||
ldv_eligible: bool
|
||||
|
||||
|
||||
@dataclass
|
||||
class Shortfall:
|
||||
"""A disposal that found no lot to consume — history is incomplete, not wrong."""
|
||||
|
||||
account_id: int
|
||||
instrument_id: int
|
||||
close_event_id: int
|
||||
close_date: date
|
||||
qty: Decimal
|
||||
|
||||
|
||||
@dataclass
|
||||
class LotResult:
|
||||
lots: list[OpenLot] = field(default_factory=list)
|
||||
disposals: list[Disposal] = field(default_factory=list)
|
||||
shortfalls: list[Shortfall] = field(default_factory=list)
|
||||
|
||||
|
||||
def apply(events: Sequence[LedgerEvent]) -> LotResult:
|
||||
"""Replay events in chronological order into open lots and disposals."""
|
||||
result = LotResult()
|
||||
queues: dict[tuple[int, int], list[OpenLot]] = defaultdict(list)
|
||||
|
||||
for event in sorted(events, key=_order):
|
||||
key = (event.account_id, event.instrument_id)
|
||||
queue = queues[key]
|
||||
|
||||
if event.kind in {EventKind.buy, EventKind.transfer_in}:
|
||||
_trade(event, queue, result, direction=1)
|
||||
elif event.kind in {EventKind.sell, EventKind.transfer_out, EventKind.repayment}:
|
||||
_trade(event, queue, result, direction=-1)
|
||||
elif event.kind is EventKind.stock_split:
|
||||
_split(event, queue)
|
||||
elif event.kind is EventKind.amortization:
|
||||
_amortise(event, queue)
|
||||
|
||||
result.shortfalls = [
|
||||
Shortfall(
|
||||
account_id=lot.account_id,
|
||||
instrument_id=lot.instrument_id,
|
||||
close_event_id=lot.open_event_id,
|
||||
close_date=lot.open_date,
|
||||
qty=lot.qty_remaining,
|
||||
)
|
||||
for lot in result.lots
|
||||
if lot.direction == -1 and lot.qty_remaining > ZERO
|
||||
]
|
||||
return result
|
||||
|
||||
|
||||
def _order(event: LedgerEvent) -> tuple[date, int, int]:
|
||||
"""Chronological, and within a day: opens before closes, then by event id.
|
||||
|
||||
A same-day buy-then-sell must see the lot it created, and real feeds do not guarantee
|
||||
the order they hand same-day operations over.
|
||||
"""
|
||||
opens_first = 0 if event.kind in {EventKind.buy, EventKind.transfer_in} else 1
|
||||
return (event.trade_date, opens_first, event.id)
|
||||
|
||||
|
||||
def _trade(event: LedgerEvent, queue: list[OpenLot], result: LotResult, *, direction: int) -> None:
|
||||
"""One trade: first close whatever points the other way, then open a lot with the rest.
|
||||
|
||||
Buying closes short lots before opening a long one, and selling closes long lots before
|
||||
opening a short. That single rule covers ordinary round-trips and shorts alike.
|
||||
"""
|
||||
qty = abs(event.quantity or ZERO)
|
||||
if qty == ZERO:
|
||||
return
|
||||
per_unit = _per_unit(event, qty, direction)
|
||||
|
||||
remaining = _consume(event, queue, result, qty, per_unit, closing=-direction)
|
||||
if remaining > ZERO:
|
||||
lot = OpenLot(
|
||||
account_id=event.account_id,
|
||||
instrument_id=event.instrument_id,
|
||||
open_event_id=event.id,
|
||||
open_date=event.trade_date,
|
||||
qty_open=remaining,
|
||||
qty_remaining=remaining,
|
||||
cost_per_unit=per_unit,
|
||||
cost_currency=event.price_currency or event.currency,
|
||||
exchange_traded=event.exchange_traded,
|
||||
direction=direction,
|
||||
)
|
||||
queue.append(lot)
|
||||
result.lots.append(lot)
|
||||
|
||||
|
||||
def _per_unit(event: LedgerEvent, qty: Decimal, direction: int) -> Decimal:
|
||||
"""Value per unit, fees included on the side they fall.
|
||||
|
||||
A fee raises what a purchase costs and lowers what a sale nets — that asymmetry is why
|
||||
a flat round-trip still shows a loss, which is what the tax code expects.
|
||||
"""
|
||||
return _cash_basis(event, direction) / qty if qty else ZERO
|
||||
|
||||
|
||||
def _cash_basis(event: LedgerEvent, direction: int) -> Decimal:
|
||||
"""What the trade was worth per the cash that moved, accrued interest excluded.
|
||||
|
||||
`amount` is preferred because it is what actually hit the account; the price path is the
|
||||
fallback for feeds reporting no cash effect (a securities transfer). The fee is added on
|
||||
a buy and subtracted on a sale — T-Invest reports it as a separate operation, so
|
||||
`amount` does not already contain it either way.
|
||||
"""
|
||||
fee = abs(event.fee or ZERO) * direction
|
||||
if event.amount:
|
||||
gross = abs(event.amount) - abs(event.accrued_interest or ZERO)
|
||||
return gross + fee
|
||||
if event.price is not None:
|
||||
return event.price * abs(event.quantity or ZERO) + fee
|
||||
return ZERO
|
||||
|
||||
|
||||
def _consume(
|
||||
event: LedgerEvent,
|
||||
queue: list[OpenLot],
|
||||
result: LotResult,
|
||||
qty: Decimal,
|
||||
per_unit: Decimal,
|
||||
*,
|
||||
closing: int,
|
||||
) -> Decimal:
|
||||
"""Close lots pointing in the `closing` direction, FIFO; return what is left over."""
|
||||
remaining = qty
|
||||
for lot in queue:
|
||||
if remaining <= ZERO:
|
||||
break
|
||||
if lot.qty_remaining <= ZERO or lot.direction != closing:
|
||||
continue
|
||||
taken = min(lot.qty_remaining, remaining)
|
||||
lot.qty_remaining -= taken
|
||||
remaining -= taken
|
||||
if lot.qty_remaining == ZERO:
|
||||
lot.closed_at = event.trade_date
|
||||
|
||||
opened = lot.cost_per_unit * taken
|
||||
closed = per_unit * taken
|
||||
# long: bought at `opened`, sold at `closed`. short: sold at `opened`, bought at
|
||||
# `closed` — so the profit is the fall in price, and the roles simply swap.
|
||||
cost, proceeds = (opened, closed) if lot.direction == 1 else (closed, opened)
|
||||
holding_days = (event.trade_date - lot.open_date).days
|
||||
result.disposals.append(
|
||||
Disposal(
|
||||
lot=lot,
|
||||
close_event_id=event.id,
|
||||
close_date=event.trade_date,
|
||||
qty=taken,
|
||||
proceeds=proceeds,
|
||||
proceeds_currency=event.price_currency or event.currency,
|
||||
cost=cost,
|
||||
realized_pnl_native=proceeds - cost,
|
||||
holding_days=holding_days,
|
||||
ldv_eligible=(
|
||||
lot.direction == 1 and holding_days >= LDV_DAYS and lot.exchange_traded
|
||||
),
|
||||
)
|
||||
)
|
||||
return remaining
|
||||
|
||||
|
||||
def _split(event: LedgerEvent, queue: list[OpenLot]) -> None:
|
||||
"""Multiply quantities, divide cost per unit: the position's total cost is unchanged."""
|
||||
ratio = event.ratio
|
||||
if not ratio or ratio <= ZERO:
|
||||
return
|
||||
for lot in queue:
|
||||
if lot.qty_remaining <= ZERO:
|
||||
continue
|
||||
lot.qty_open *= ratio
|
||||
lot.qty_remaining *= ratio
|
||||
lot.cost_per_unit /= ratio
|
||||
|
||||
|
||||
def _amortise(event: LedgerEvent, queue: list[OpenLot]) -> None:
|
||||
"""Return of principal lowers what is still invested, proportionally across open lots."""
|
||||
paid = abs(event.amount)
|
||||
open_qty = sum((lot.qty_remaining for lot in queue if lot.qty_remaining > ZERO), start=ZERO)
|
||||
if paid == ZERO or open_qty == ZERO:
|
||||
return
|
||||
per_unit = paid / open_qty
|
||||
for lot in queue:
|
||||
if lot.qty_remaining <= ZERO:
|
||||
continue
|
||||
# never below zero: an amortisation bigger than the remaining cost means the cost
|
||||
# basis is already exhausted, and the excess is income, not a negative asset
|
||||
lot.cost_per_unit = max(lot.cost_per_unit - per_unit, ZERO)
|
||||
@@ -0,0 +1,193 @@
|
||||
"""Rebuild `lot` / `lot_disposal` from the event ledger (plan §1.5).
|
||||
|
||||
The engine in `lots.py` is pure; this module is the part that talks to the database: it
|
||||
loads confirmed events, runs the replay, converts to RUB at the rate of each leg's own date,
|
||||
and replaces the tables wholesale.
|
||||
|
||||
Currency handling follows the project rule: values are kept in the instrument's own currency
|
||||
and converted at the rate **of the date the money moved** — cost at the lot's open date,
|
||||
proceeds at the disposal date. A leg with no rate for its day yields NULL rather than a
|
||||
substituted number, and the gap is reported to data quality.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from decimal import Decimal
|
||||
|
||||
from sqlalchemy import delete, insert, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from fintracker.analytics import FINDINGS
|
||||
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
|
||||
from fintracker.models.pricing import CorporateAction, CorporateActionKind
|
||||
from fintracker.pricing.fx import FxTable
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
#: Kinds the engine reads: position moves plus the corporate actions that reshape lots.
|
||||
RELEVANT_KINDS = POSITION_KINDS | {EventKind.amortization}
|
||||
|
||||
#: Asset classes that can qualify for the long-term holding exemption.
|
||||
EXCHANGE_CLASSES = {"share", "bond", "etf", "fund"}
|
||||
|
||||
|
||||
async def rebuild_lots(session: AsyncSession) -> None:
|
||||
"""Replace `lot` and `lot_disposal` from scratch, and report what did not add up."""
|
||||
events = await _load_events(session)
|
||||
if not events:
|
||||
await session.execute(delete(LotDisposal))
|
||||
await session.execute(delete(Lot))
|
||||
return
|
||||
|
||||
result = apply(events)
|
||||
fx = await FxTable.load(session)
|
||||
|
||||
await session.execute(delete(LotDisposal))
|
||||
await session.execute(delete(Lot))
|
||||
|
||||
lot_ids: dict[int, int] = {}
|
||||
for lot in result.lots:
|
||||
cost_total = lot.cost_per_unit * lot.qty_open
|
||||
row = {
|
||||
"account_id": lot.account_id,
|
||||
"instrument_id": lot.instrument_id,
|
||||
"open_event_id": lot.open_event_id,
|
||||
"open_date": lot.open_date,
|
||||
# stored signed, so summing qty_remaining gives the position directly — a short
|
||||
# lot must subtract from it rather than look like a holding
|
||||
"qty_open": lot.qty_open * lot.direction,
|
||||
"qty_remaining": lot.signed_qty_remaining,
|
||||
"cost_per_unit": lot.cost_per_unit,
|
||||
"cost_currency": lot.cost_currency,
|
||||
"cost_total_rub": fx.to_rub(cost_total, lot.cost_currency, lot.open_date),
|
||||
"closed_at": lot.closed_at,
|
||||
}
|
||||
new_id = (await session.execute(insert(Lot).values(row).returning(Lot.id))).scalar_one()
|
||||
lot_ids[id(lot)] = new_id
|
||||
|
||||
disposal_rows = []
|
||||
for disposal in result.disposals:
|
||||
cost_rub = fx.to_rub(disposal.cost, disposal.lot.cost_currency, disposal.lot.open_date)
|
||||
proceeds_rub = fx.to_rub(disposal.proceeds, disposal.proceeds_currency, disposal.close_date)
|
||||
disposal_rows.append(
|
||||
{
|
||||
"lot_id": lot_ids[id(disposal.lot)],
|
||||
"close_event_id": disposal.close_event_id,
|
||||
"close_date": disposal.close_date,
|
||||
"qty": disposal.qty,
|
||||
"proceeds": disposal.proceeds,
|
||||
"proceeds_currency": disposal.proceeds_currency,
|
||||
"proceeds_rub": proceeds_rub,
|
||||
"cost_rub": cost_rub,
|
||||
"realized_pnl_native": disposal.realized_pnl_native,
|
||||
# both legs converted at their own dates; either missing makes it unknowable
|
||||
"realized_pnl_rub": (
|
||||
proceeds_rub - cost_rub
|
||||
if proceeds_rub is not None and cost_rub is not None
|
||||
else None
|
||||
),
|
||||
"holding_days": disposal.holding_days,
|
||||
"ldv_eligible": disposal.ldv_eligible,
|
||||
}
|
||||
)
|
||||
if disposal_rows:
|
||||
await session.execute(insert(LotDisposal), disposal_rows)
|
||||
|
||||
_report(result)
|
||||
log.info(
|
||||
"lots: %s lots, %s disposals, %s shortfalls",
|
||||
len(result.lots),
|
||||
len(result.disposals),
|
||||
len(result.shortfalls),
|
||||
)
|
||||
|
||||
|
||||
async def _load_events(session: AsyncSession) -> list[LedgerEvent]:
|
||||
"""Confirmed, instrument-bearing events, with the instrument facts the engine needs."""
|
||||
rows = (
|
||||
await session.execute(
|
||||
select(
|
||||
Event.id,
|
||||
Event.account_id,
|
||||
Event.instrument_id,
|
||||
Event.kind,
|
||||
Event.trade_date,
|
||||
Event.quantity,
|
||||
Event.price,
|
||||
Event.price_currency,
|
||||
Event.amount,
|
||||
Event.currency,
|
||||
Event.fee,
|
||||
Event.accrued_interest,
|
||||
Instrument.asset_class,
|
||||
Instrument.board,
|
||||
)
|
||||
.join(Instrument, Instrument.id == Event.instrument_id)
|
||||
.where(
|
||||
Event.status == EventStatus.confirmed,
|
||||
Event.kind.in_(RELEVANT_KINDS),
|
||||
Event.instrument_id.is_not(None),
|
||||
)
|
||||
)
|
||||
).all()
|
||||
|
||||
ratios = await _split_ratios(session)
|
||||
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=r.price,
|
||||
price_currency=r.price_currency,
|
||||
amount=r.amount or Decimal(0),
|
||||
currency=r.currency,
|
||||
fee=r.fee,
|
||||
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),
|
||||
)
|
||||
for r in rows
|
||||
]
|
||||
|
||||
|
||||
async def _split_ratios(session: AsyncSession) -> dict[tuple[int, object], Decimal]:
|
||||
"""Split ratios by (instrument, date) — the engine cannot infer them from an event."""
|
||||
rows = (
|
||||
await session.execute(
|
||||
select(
|
||||
CorporateAction.instrument_id, CorporateAction.ex_date, CorporateAction.ratio
|
||||
).where(
|
||||
CorporateAction.kind == CorporateActionKind.stock_split,
|
||||
CorporateAction.ratio.is_not(None),
|
||||
)
|
||||
)
|
||||
).all()
|
||||
return {(r.instrument_id, r.ex_date): r.ratio for r in rows if r.ex_date}
|
||||
|
||||
|
||||
def _report(result: LotResult) -> None:
|
||||
"""Surface what the replay could not reconcile, aggregated per instrument."""
|
||||
if result.shortfalls:
|
||||
per_instrument: dict[int, Decimal] = {}
|
||||
for shortfall in result.shortfalls:
|
||||
per_instrument[shortfall.instrument_id] = (
|
||||
per_instrument.get(shortfall.instrument_id, Decimal(0)) + shortfall.qty
|
||||
)
|
||||
FINDINGS.add(
|
||||
"lot_shortfall",
|
||||
"warn",
|
||||
f"Продано больше, чем числится в лотах, по {len(per_instrument)} инструментам — "
|
||||
"история неполная (перевод бумаг или сделки до начала выгрузки)",
|
||||
count=len(result.shortfalls),
|
||||
ref={"instruments": sorted(per_instrument)},
|
||||
)
|
||||
|
||||
missing_fx = sum(1 for lot in result.lots if lot.cost_currency not in {"RUB", "rub"})
|
||||
if missing_fx:
|
||||
log.debug("lots: %s lots in a foreign currency", missing_fx)
|
||||
Reference in New Issue
Block a user