338 lines
12 KiB
Python
338 lines
12 KiB
Python
"""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,
|
|
}
|