"""FIFO engine rules, on synthetic events — no database, no source.""" from datetime import date, timedelta from decimal import Decimal from fintracker.ledger.lots import LDV_DAYS, LedgerEvent, apply from fintracker.models.ledger import EventKind D = Decimal ACC, INS = 1, 10 _seq = iter(range(1, 10_000)) def ev(kind: EventKind, day: str, *, qty=None, amount="0", price=None, **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=D(price) if price is not None else None, price_currency="RUB", amount=D(amount), currency="RUB", fee=D(over.pop("fee")) if "fee" in over else None, accrued_interest=D(over.pop("nkd")) if "nkd" in over else None, ratio=D(over.pop("ratio")) if "ratio" in over else None, exchange_traded=over.pop("exchange_traded", True), ) def buy(day, qty, amount, **kw): return ev(EventKind.buy, day, qty=qty, amount=amount, **kw) def sell(day, qty, amount, **kw): return ev(EventKind.sell, day, qty=f"-{qty}", amount=amount, **kw) def test_fifo_consumes_the_oldest_lot_first(): result = apply( [ buy("2025-01-10", "10", "-1000"), buy("2025-02-10", "10", "-1500"), sell("2025-03-10", "10", "1400"), ] ) assert len(result.disposals) == 1 disposal = result.disposals[0] assert disposal.lot.open_date == date(2025, 1, 10) # the January lot, not February assert disposal.cost == D(1000) assert disposal.realized_pnl_native == D(400) # the February lot is untouched assert [lot.qty_remaining for lot in result.lots] == [D(0), D(10)] def test_a_sale_spanning_two_lots_splits_into_two_disposals(): result = apply( [ buy("2025-01-10", "10", "-1000"), buy("2025-02-10", "10", "-2000"), sell("2025-03-10", "15", "2250"), ] ) assert [d.qty for d in result.disposals] == [D(10), D(5)] # 10 @ cost 100 sold at 150 -> +500; 5 @ cost 200 sold at 150 -> -250 assert [d.realized_pnl_native for d in result.disposals] == [D(500), D(-250)] assert sum(d.realized_pnl_native for d in result.disposals) == D(250) def test_fees_are_capitalised_into_cost_and_deducted_from_proceeds(): """The tax-relevant number: a buy costs more than the price, a sale nets less.""" result = apply( [buy("2025-01-10", "10", "-1000", fee="10"), sell("2025-02-10", "10", "1000", fee="10")] ) disposal = result.disposals[0] assert disposal.cost == D(1010) assert disposal.proceeds == D(990) assert disposal.realized_pnl_native == D(-20) # a flat round-trip still loses both fees def test_accrued_interest_is_not_part_of_the_lot_cost(): """НКД is advanced to the seller and returned by the next coupon — not an investment.""" result = apply([buy("2025-01-10", "10", "-1050", nkd="50")]) assert result.lots[0].cost_per_unit == D(100) def test_split_multiplies_quantity_and_divides_cost(): result = apply( [buy("2025-01-10", "10", "-1000"), ev(EventKind.stock_split, "2025-02-01", ratio="10")] ) lot = result.lots[0] assert lot.qty_remaining == D(100) assert lot.cost_per_unit == D(10) assert lot.qty_remaining * lot.cost_per_unit == D(1000) # total cost unchanged def test_amortisation_lowers_cost_per_unit_without_touching_quantity(): result = apply( [buy("2025-01-10", "10", "-10000"), ev(EventKind.amortization, "2025-06-01", amount="2000")] ) lot = result.lots[0] assert lot.qty_remaining == D(10) assert lot.cost_per_unit == D(800) def test_amortised_bond_repaid_at_par_shows_no_phantom_loss(): """Without amortisation lowering the basis, the repayment would look like a big loss.""" result = apply( [ buy("2025-01-10", "10", "-10000"), ev(EventKind.amortization, "2025-06-01", amount="2000"), ev(EventKind.repayment, "2025-12-01", qty="-10", amount="8000"), ] ) assert result.disposals[0].realized_pnl_native == D(0) def test_ldv_marks_only_holdings_of_three_years_or_more(): open_day = date(2021, 1, 10) long_enough = open_day + timedelta(days=LDV_DAYS) result = apply( [buy(open_day.isoformat(), "10", "-1000"), sell(long_enough.isoformat(), "10", "1500")] ) assert result.disposals[0].ldv_eligible is True def test_ldv_does_not_apply_a_day_early(): open_day = date(2021, 1, 10) too_soon = open_day + timedelta(days=LDV_DAYS - 1) result = apply( [buy(open_day.isoformat(), "10", "-1000"), sell(too_soon.isoformat(), "10", "1500")] ) assert result.disposals[0].ldv_eligible is False def test_ldv_does_not_apply_to_otc_papers(): open_day = date(2021, 1, 10) long_enough = open_day + timedelta(days=LDV_DAYS) result = apply( [ buy(open_day.isoformat(), "10", "-1000", exchange_traded=False), sell(long_enough.isoformat(), "10", "1500"), ] ) assert result.disposals[0].ldv_eligible is False def test_selling_more_than_held_reports_a_shortfall_instead_of_inventing_a_lot(): result = apply([buy("2025-01-10", "5", "-500"), sell("2025-02-10", "8", "800")]) assert result.disposals[0].qty == D(5) assert [s.qty for s in result.shortfalls] == [D(3)] def test_a_same_day_buy_is_available_to_a_same_day_sale(): """Feeds do not order same-day operations; the engine must not depend on luck.""" sale = sell("2025-01-10", "10", "1200", id=1) purchase = buy("2025-01-10", "10", "-1000", id=2) result = apply([sale, purchase]) assert not result.shortfalls assert result.disposals[0].realized_pnl_native == D(200) def test_accounts_keep_independent_queues(): """Tax is computed per account, so the same paper elsewhere must not be consumed.""" result = apply( [ buy("2025-01-10", "10", "-1000", account_id=1), buy("2025-01-11", "10", "-2000", account_id=2), sell("2025-02-10", "10", "1500", account_id=2), ] ) disposal = result.disposals[0] assert disposal.lot.account_id == 2 assert disposal.cost == D(2000) # account 2's own lot, not account 1's cheaper one def test_transfer_in_opens_a_lot_and_transfer_out_closes_one(): result = apply( [ ev(EventKind.transfer_in, "2025-01-10", qty="10", amount="0", price="100"), ev(EventKind.transfer_out, "2025-02-10", qty="-4", amount="0", price="120"), ] ) assert result.lots[0].cost_per_unit == D(100) assert result.disposals[0].qty == D(4) def test_a_short_round_trip_realises_the_fall_in_price(): """Real case (TATN): sold on 22.08, bought back on 29.08 — profit is the price drop.""" result = apply( [ sell("2025-08-22", "10", "6751"), buy("2025-08-29", "10", "-6505"), ] ) assert not result.shortfalls assert result.disposals[0].realized_pnl_native == D(246) assert sum(lot.signed_qty_remaining for lot in result.lots) == D(0) def test_an_open_short_is_reported_as_a_shortfall(): """Either an open short or history starting mid-way — both need a human to look.""" result = apply([sell("2025-08-22", "10", "6751")]) assert [s.qty for s in result.shortfalls] == [D(10)] assert result.lots[0].signed_qty_remaining == D(-10) def test_a_short_is_never_ldv_eligible(): result = apply( [ sell("2021-01-10", "10", "1500"), buy((date(2021, 1, 10) + timedelta(days=LDV_DAYS)).isoformat(), "10", "-1000"), ] ) assert result.disposals[0].ldv_eligible is False def test_selling_past_a_long_position_opens_a_short_for_the_excess(): result = apply([buy("2025-01-10", "5", "-500"), sell("2025-02-10", "8", "800")]) assert result.disposals[0].qty == D(5) assert [s.qty for s in result.shortfalls] == [D(3)] assert sum(lot.signed_qty_remaining for lot in result.lots) == D(-3)