From dc057e9aebc868a1ebf64be31d1503cdeec6aeeb Mon Sep 17 00:00:00 2001 From: Kunthawat Greethong Date: Fri, 28 Aug 2026 10:26:12 +0700 Subject: [PATCH] [verified] Task 3: portfolio accounting ledger (fees, avg cost, dated dividends) --- backend/app/portfolio_ledger.py | 337 +++++++++++++++++++++++++ backend/tests/test_portfolio_ledger.py | 176 +++++++++++++ 2 files changed, 513 insertions(+) create mode 100644 backend/app/portfolio_ledger.py create mode 100644 backend/tests/test_portfolio_ledger.py diff --git a/backend/app/portfolio_ledger.py b/backend/app/portfolio_ledger.py new file mode 100644 index 0000000..c5b75ce --- /dev/null +++ b/backend/app/portfolio_ledger.py @@ -0,0 +1,337 @@ +"""Portfolio accounting ledger for the event-driven backtest (Task 3). + +This is the single auditable owner of cash, positions, cost basis, dividend +receivables, transaction costs, and realized/unrealized P&L. It does not decide +*what* to hold (that is the rebalancer, Task 4) — it executes and accounts for +orders and dividend cash flows. + +Decisions (confirmed with the user): + * transaction fee is all-in 0.3% of notional on every buy and every sell; + no VAT/tax is added on top; + * realized P&L is reported GROSS (before fees); all fees (buy + sell) are + tracked together in ``fees`` and subtracted once in reconciliation, so the + net trading gain = realized_pnl - fees attributable to sells; + * dividend cash becomes available exactly 30 calendar days after ex-date + (``ex_date_plus_30d`` assumption, not an observed payment date); + * unpaid dividends are receivables and CANNOT fund purchases. + +Accounting invariant (must reconcile after every event): + ending_equity - initial_capital + = realized_trading_pnl + unrealized_trading_pnl + + dividend_cash_received + accrued_dividend_receivable + - transaction_costs +""" + +from __future__ import annotations + +import datetime as dt +import math +from dataclasses import dataclass, field +from typing import Iterable, Optional + +from .backtest_events import ( + DIVIDEND_PAYMENT_LAG_DAYS, + DividendEntitlementEvent, + DividendPaymentEvent, +) + +# all-in transaction fee: 0.3% of notional on each side +FEE_RATE = 0.003 +# share-lot unit: trades happen in multiples of 100 +LOT_SIZE = 100 + + +class PortfolioLedgerError(ValueError): + pass + + +def _round_money(x: float) -> float: + return round(float(x), 2) + + +@dataclass +class Position: + symbol: str + qty: int = 0 + average_cost: float = 0.0 + + def market_value(self, price: float) -> float: + return self.qty * price + + +@dataclass +class Trade: + date: str + signal_date: str + symbol: str + side: str # "buy" | "sell" + qty: int + price: float + notional: float + fees: float + cost_basis_released: float = 0.0 # avg cost * qty released (sells only) + realized_pnl: float = 0.0 # sells only + cash_after: float = 0.0 + reason: str = "" + + def to_dict(self) -> dict: + return { + "date": self.date, "signal_date": self.signal_date, + "symbol": self.symbol, "side": self.side, "qty": self.qty, + "price": _round_money(self.price), + "notional": _round_money(self.notional), + "fees": _round_money(self.fees), + "cost_basis_released": _round_money(self.cost_basis_released), + "realized_pnl": _round_money(self.realized_pnl), + "cash_after": _round_money(self.cash_after), + "reason": self.reason, + } + + +@dataclass +class DividendLedgerEntry: + symbol: str + ex_date: str + assumed_payment_date: str + qty_entitled: int + per_share: float + amount: float + status: str = "receivable" # receivable -> paid + timing_method: str = "ex_date_plus_30d" + + def to_dict(self) -> dict: + return { + "symbol": self.symbol, "ex_date": self.ex_date, + "assumed_payment_date": self.assumed_payment_date, + "qty_entitled": self.qty_entitled, + "per_share": self.per_share, "amount": _round_money(self.amount), + "status": self.status, "timing_method": self.timing_method, + } + + +@dataclass +class PortfolioState: + initial_capital: float + cash: float = 0.0 + positions: dict[str, Position] = field(default_factory=dict) + receivables: list[DividendLedgerEntry] = field(default_factory=list) + trades: list[Trade] = field(default_factory=list) + realized_pnl: float = 0.0 + dividend_cash_received: float = 0.0 + fees: float = 0.0 + # running log of dividend events for reporting + dividends: list[dict] = field(default_factory=list) + + def __post_init__(self) -> None: + self.cash = float(self.initial_capital) + + +class PortfolioLedger: + """Auditable cash/position/dividend accounting engine.""" + + def __init__( + self, + initial_capital: float, + fee_rate: float = FEE_RATE, + lot_size: int = LOT_SIZE, + ) -> None: + if initial_capital < 0: + raise PortfolioLedgerError("initial capital cannot be negative") + if fee_rate < 0 or fee_rate >= 1: + raise PortfolioLedgerError("fee rate must be in [0, 1)") + self.state = PortfolioState(initial_capital=float(initial_capital)) + self.fee_rate = float(fee_rate) + self.lot_size = int(lot_size) + + # -- positions --------------------------------------------------------- + def qty(self, symbol: str) -> int: + return self.state.positions.get(symbol, Position(symbol)).qty + + def position(self, symbol: str) -> Optional[Position]: + return self.state.positions.get(symbol) + + def positions(self) -> list[Position]: + return [p for p in self.state.positions.values() if p.qty > 0] + + # -- buys/sells -------------------------------------------------------- + def _validate_lot(self, qty: int) -> None: + if qty <= 0: + raise PortfolioLedgerError("order qty must be positive") + if qty % self.lot_size != 0: + raise PortfolioLedgerError( + f"order qty must be a multiple of {self.lot_size} (got {qty})" + ) + + def buy(self, symbol: str, qty: int, price: float, *, + date: str, signal_date: str, reason: str = "") -> Trade: + self._validate_lot(qty) + if price <= 0: + raise PortfolioLedgerError("buy price must be positive") + notional = qty * price + fee = _round_money(notional * self.fee_rate) + total = notional + fee + if total > self.state.cash + 1e-6: + raise PortfolioLedgerError( + f"insufficient cash for buy: need {total:.2f}, have " + f"{self.state.cash:.2f}" + ) + self.state.cash -= total + pos = self.state.positions.setdefault(symbol, Position(symbol=symbol)) + if pos.qty == 0: + pos.average_cost = price + else: + total_cost = pos.qty * pos.average_cost + notional + pos.average_cost = total_cost / (pos.qty + qty) + pos.qty += qty + self.state.fees += fee + trade = Trade( + date=date, signal_date=signal_date, symbol=symbol, side="buy", + qty=qty, price=price, notional=notional, fees=fee, + cash_after=_round_money(self.state.cash), reason=reason, + ) + self.state.trades.append(trade) + return trade + + def sell(self, symbol: str, qty: int, price: float, *, + date: str, signal_date: str, reason: str = "") -> Trade: + self._validate_lot(qty) + if price <= 0: + raise PortfolioLedgerError("sell price must be positive") + pos = self.state.positions.get(symbol) + if pos is None or pos.qty < qty: + raise PortfolioLedgerError( + f"insufficient position to sell {qty} of {symbol}" + ) + notional = qty * price + fee = _round_money(notional * self.fee_rate) + proceeds = notional - fee + cost_basis_released = qty * pos.average_cost + # realized P&L is GROSS (before sell fee). The sell fee is tracked in + # ``state.fees`` and subtracted once in reconciliation, so the reported + # net gain = realized_pnl - fees. This keeps the accounting identity + # balanced (fees are a separate line, not counted twice). + realized = notional - cost_basis_released + self.state.cash += proceeds + pos.qty -= qty + # when fully exited, drop the position entirely + if pos.qty == 0: + del self.state.positions[symbol] + self.state.fees += fee + self.state.realized_pnl += realized + trade = Trade( + date=date, signal_date=signal_date, symbol=symbol, side="sell", + qty=qty, price=price, notional=notional, fees=fee, + cost_basis_released=cost_basis_released, realized_pnl=realized, + cash_after=_round_money(self.state.cash), reason=reason, + ) + self.state.trades.append(trade) + return trade + + # -- dividends --------------------------------------------------------- + def record_dividend_entitlement( + self, symbol: str, ex_date: str, per_share: float + ) -> None: + """Capture entitlement on ex-date using shares held *before* ex-date. + + Only dated (non-estimate) dividends produce receivables. The amount is + ``qty_held * per_share`` and cash becomes available on + ``ex_date + DIVIDEND_PAYMENT_LAG_DAYS``. + """ + qty = self.qty(symbol) + if qty <= 0: + return # no shares held -> no entitlement + if per_share <= 0: + return + ex = dt.date.fromisoformat(ex_date) + pay = ex + dt.timedelta(days=DIVIDEND_PAYMENT_LAG_DAYS) + amount = qty * per_share + self.state.receivables.append(DividendLedgerEntry( + symbol=symbol, ex_date=ex_date, + assumed_payment_date=pay.isoformat(), + qty_entitled=qty, per_share=per_share, amount=amount, + )) + self.state.dividends.append({ + "symbol": symbol, "ex_date": ex_date, + "payment_date": pay.isoformat(), "qty": qty, + "per_share": per_share, "amount": _round_money(amount), + "status": "receivable", + }) + + def pay_due_dividends(self, on_date: str) -> list[DividendLedgerEntry]: + """Credit any receivable whose assumed payment date is <= on_date. + + Paid dividends become spendable cash (available to later buys) and are + removed from receivables. + """ + paid: list[DividendLedgerEntry] = [] + cutoff = dt.date.fromisoformat(on_date) + still_receivable: list[DividendLedgerEntry] = [] + for r in self.state.receivables: + pay_date = dt.date.fromisoformat(r.assumed_payment_date) + if pay_date <= cutoff: + self.state.cash += r.amount + self.state.dividend_cash_received += r.amount + r.status = "paid" + paid.append(r) + # update the dividends log entry status + for d in self.state.dividends: + if (d["symbol"] == r.symbol and d["ex_date"] == r.ex_date + and d["status"] == "receivable"): + d["status"] = "paid" + else: + still_receivable.append(r) + self.state.receivables = still_receivable + return paid + + # -- valuation / reconciliation --------------------------------------- + def total_receivable(self) -> float: + return sum(r.amount for r in self.state.receivables) + + def market_value(self, prices: dict[str, float]) -> float: + total = 0.0 + for sym, pos in self.state.positions.items(): + px = prices.get(sym) + if px is not None: + total += pos.qty * px + return total + + def unrealized_pnl(self, prices: dict[str, float]) -> float: + total = 0.0 + for sym, pos in self.state.positions.items(): + px = prices.get(sym) + if px is not None: + total += pos.qty * (px - pos.average_cost) + return total + + def equity(self, prices: dict[str, float]) -> float: + """Total equity: cash + market value of current holdings.""" + return self.state.cash + self.market_value(prices) + + def reconcile(self, prices: dict[str, float]) -> dict: + """Return the full accounting breakdown for [end] reporting. + + Invariant: equity_before_receivable - initial_capital == + realized + unrealized + dividend_cash - fees. + (Accrued receivable is reported separately and is NOT part of spendable + equity until paid.) + """ + realized = self.state.realized_pnl + unrealized = self.unrealized_pnl(prices) + div_cash = self.state.dividend_cash_received + fees = self.state.fees + equity = self.equity(prices) + lhs = equity - self.state.initial_capital + rhs = realized + unrealized + div_cash - fees + return { + "equity": equity, + "cash": self.state.cash, + "market_value": self.market_value(prices), + "realized_trading_pnl": realized, + "unrealized_trading_pnl": unrealized, + "dividend_cash_received": div_cash, + "dividend_receivable": self.total_receivable(), + "transaction_costs": fees, + "notional_return": lhs, + "reconciled_rhs": rhs, + "balanced": abs(lhs - rhs) < 0.01, + } diff --git a/backend/tests/test_portfolio_ledger.py b/backend/tests/test_portfolio_ledger.py new file mode 100644 index 0000000..0083445 --- /dev/null +++ b/backend/tests/test_portfolio_ledger.py @@ -0,0 +1,176 @@ +"""Tests for the portfolio accounting ledger (Task 3).""" + +from __future__ import annotations + +import unittest + +from app.backtest_events import DIVIDEND_PAYMENT_LAG_DAYS +from app.portfolio_ledger import ( + FEE_RATE, + LOT_SIZE, + PortfolioLedger, + PortfolioLedgerError, +) + + +SHARES = LOT_SIZE # 100 + + +class BuySellTest(unittest.TestCase): + def test_buy_100_lots_and_average_cost(self): + ledger = PortfolioLedger(1_000_000) + ledger.buy("A", SHARES, 100.0, date="2026-01-05", signal_date="2026-01-01") + pos = ledger.position("A") + assert pos is not None + self.assertEqual(pos.qty, SHARES) + self.assertEqual(pos.average_cost, 100.0) + # fee 0.3% of 100*100=10,000 -> 30 + expected_cost = 10_000 * FEE_RATE + self.assertAlmostEqual(ledger.state.fees, expected_cost) + self.assertAlmostEqual( + ledger.state.cash, 1_000_000 - 10_000 - expected_cost, places=2 + ) + + def test_average_cost_after_second_buy_at_higher_price(self): + ledger = PortfolioLedger(1_000_000) + ledger.buy("A", SHARES, 100.0, date="2026-01-05", signal_date="2026-01-01") + ledger.buy("A", SHARES, 200.0, date="2026-02-05", signal_date="2026-02-01") + pos = ledger.position("A") + assert pos is not None + self.assertEqual(pos.qty, 2 * SHARES) + # avg = (100*100 + 200*100) / 200 = 150 + self.assertAlmostEqual(pos.average_cost, 150.0, places=2) + + def test_realized_pnl_uses_average_cost(self): + ledger = PortfolioLedger(1_000_000) + ledger.buy("A", SHARES, 100.0, date="2026-01-05", signal_date="2026-01-01") + # sell at 110 -> avg cost 100; gross realized = 11000 - 10000 = 1000 + trade = ledger.sell("A", SHARES, 110.0, date="2026-03-05", signal_date="2026-03-01") + self.assertAlmostEqual(trade.realized_pnl, 1_000.0, places=2) + self.assertAlmostEqual(ledger.state.realized_pnl, 1_000.0, places=2) + # sell fee = 0.3% * 11000 = 33 tracked separately + self.assertAlmostEqual(trade.fees, 33.0, places=2) + # full exit drops the position + self.assertIsNone(ledger.position("A")) + + def test_buy_requires_lot_multiple(self): + ledger = PortfolioLedger(1_000_000) + with self.assertRaises(PortfolioLedgerError): + ledger.buy("A", 50, 100.0, date="2026-01-05", signal_date="2026-01-01") + + def test_buy_requires_enough_cash(self): + ledger = PortfolioLedger(10_000) + # 100 shares * 500 = 50,000 + fee > 10,000 + with self.assertRaises(PortfolioLedgerError): + ledger.buy("A", SHARES, 500.0, date="2026-01-05", signal_date="2026-01-01") + + def test_sell_more_than_held_rejected(self): + ledger = PortfolioLedger(1_000_000) + ledger.buy("A", SHARES, 100.0, date="2026-01-05", signal_date="2026-01-01") + with self.assertRaises(PortfolioLedgerError): + ledger.sell("A", 2 * SHARES, 100.0, date="2026-03-05", signal_date="2026-03-01") + + +class DividendTest(unittest.TestCase): + def test_entitlement_uses_shares_held_before_ex_date(self): + ledger = PortfolioLedger(1_000_000) + ledger.buy("A", 2 * SHARES, 100.0, date="2026-01-05", signal_date="2026-01-01") + ledger.record_dividend_entitlement("A", "2026-03-10", 1.5) + self.assertEqual(len(ledger.state.receivables), 1) + r = ledger.state.receivables[0] + self.assertEqual(r.qty_entitled, 2 * SHARES) + self.assertAlmostEqual(r.amount, 2 * SHARES * 1.5, places=2) + # payment = ex_date + 30 calendar days + self.assertEqual(r.assumed_payment_date, "2026-04-09") + self.assertEqual(r.timing_method, "ex_date_plus_30d") + # NOT yet cash + self.assertAlmostEqual(ledger.state.dividend_cash_received, 0.0) + + def test_no_entitlement_when_no_shares_held(self): + ledger = PortfolioLedger(1_000_000) + ledger.record_dividend_entitlement("A", "2026-03-10", 1.5) + self.assertEqual(len(ledger.state.receivables), 0) + + def test_payment_credits_cash_at_ex_date_plus_30(self): + ledger = PortfolioLedger(1_000_000) + ledger.buy("A", 2 * SHARES, 100.0, date="2026-01-05", signal_date="2026-01-01") + ledger.record_dividend_entitlement("A", "2026-03-10", 1.5) + cash_before = ledger.state.cash + # before payment date -> no credit + ledger.pay_due_dividends("2026-04-08") + self.assertAlmostEqual(ledger.state.cash, cash_before, places=2) + # on/after payment date -> credited + ledger.pay_due_dividends("2026-04-09") + self.assertAlmostEqual( + ledger.state.cash, cash_before + 2 * SHARES * 1.5, places=2 + ) + self.assertAlmostEqual( + ledger.state.dividend_cash_received, 2 * SHARES * 1.5, places=2 + ) + self.assertEqual(len(ledger.state.receivables), 0) + + def test_receivable_is_not_spendable_before_payment(self): + ledger = PortfolioLedger(50_000) + ledger.buy("A", SHARES, 100.0, date="2026-01-05", signal_date="2026-01-01") + ledger.record_dividend_entitlement("A", "2026-03-10", 1.5) + # dividend receivable = 150, but cash is only 10,030-ish; can't buy 100*200 + with self.assertRaises(PortfolioLedgerError): + ledger.buy("B", SHARES, 400.0, date="2026-03-15", signal_date="2026-03-10") + # after payment, cash grows and can fund the buy + ledger.pay_due_dividends("2026-04-09") + ledger.buy("B", SHARES, 10.0, date="2026-04-10", signal_date="2026-04-09") + self.assertEqual(ledger.qty("B"), SHARES) + + +class ReconcileTest(unittest.TestCase): + def test_equity_matches_market_value_plus_cash(self): + ledger = PortfolioLedger(1_000_000) + ledger.buy("A", SHARES, 100.0, date="2026-01-05", signal_date="2026-01-01") + prices = {"A": 110.0} + self.assertAlmostEqual( + ledger.equity(prices), + ledger.state.cash + 100 * 110.0, + places=2, + ) + + def test_reconciled_balanced_with_sell(self): + ledger = PortfolioLedger(100_000) + ledger.buy("A", SHARES, 50.0, date="2026-01-05", signal_date="2026-01-01") + ledger.sell("A", SHARES, 60.0, date="2026-02-05", signal_date="2026-02-01") + prices = {} + r = ledger.reconcile(prices) + self.assertTrue(r["balanced"]) + # notional return equals realized pnl - fees + self.assertAlmostEqual( + r["notional_return"], r["realized_trading_pnl"] - r["transaction_costs"], + places=1, + ) + + def test_reconciled_balanced_with_holding_and_dividend(self): + ledger = PortfolioLedger(100_000) + ledger.buy("A", SHARES, 50.0, date="2026-01-05", signal_date="2026-01-01") + ledger.record_dividend_entitlement("A", "2026-03-10", 1.0) + ledger.pay_due_dividends("2026-04-09") + prices = {"A": 55.0} + r = ledger.reconcile(prices) + self.assertTrue(r["balanced"]) + # equity = cash(+div) + 100*55 + expected_equity = ledger.state.cash + 100 * 55.0 + self.assertAlmostEqual(r["equity"], expected_equity, places=2) + + +class FeeConfigTest(unittest.TestCase): + def test_custom_fee_rate(self): + ledger = PortfolioLedger(1_000_000, fee_rate=0.0) + ledger.buy("A", SHARES, 100.0, date="2026-01-05", signal_date="2026-01-01") + self.assertEqual(ledger.state.fees, 0.0) + + def test_invalid_fee_rate_rejected(self): + with self.assertRaises(PortfolioLedgerError): + PortfolioLedger(1000, fee_rate=1.0) + with self.assertRaises(PortfolioLedgerError): + PortfolioLedger(1000, fee_rate=-0.1) + + +if __name__ == "__main__": + unittest.main()