"""Dated dividend cash-flow ledger for the backtest engine (honest). Replaces the previous single scalar ``final_holdings_yield_proxy``, which credited dividends on **final** holdings at the **final** snapshot yield — wrong for any multi-period backtest (a name held mid-window that was sold would never earn its mid-window dividend, and year-to-year yield is flattened to one number). The ledger is a per-symbol, dated dividend schedule: {symbol, ex_date, record_date, pay_date, per_share, source, retrieved_at} A backtest credits ``per_share * qty_held_on_ex_date`` to cash on ``pay_date`` (respects the ex-date cut-off: shares bought on/after ex-date do not receive that payment). Honesty scope: - **Real entries** carry an ``ex_date`` (and ideally record/pay dates) from a collected source (e.g. the Siamchart per-stock dividend-history page). - Until real per-symbol histories exist, a caller may construct a **DPS estimate** row (``source="dps_annual_proxy"``, no ex_date) that spreads the latest ``DPS`` over the holding period. Such a row is labelled ``estimate=True`` and is never presented as a realised cash flow. - no data for a symbol => it earns no dividend in the backtest (fail closed: we do not fabricate a payment). """ from __future__ import annotations import datetime as dt import json import math from pathlib import Path from typing import Any, Iterable, Mapping, Optional _MIN_QTY = 100 # allocation minimum; used only for sanity documentation class DividendLedgerError(ValueError): pass def _parse_date(value: Any) -> dt.date: s = str(value)[:10] try: return dt.date.fromisoformat(s) except (ValueError, TypeError) as exc: raise DividendLedgerError(f"invalid date: {value!r}") from exc class DividendLedger: """In-memory + persisted dated dividend schedule for SET50 symbols.""" def __init__(self, path: Optional[Path] = None) -> None: self.path = Path(path) if path else None # symbol -> sorted list of entries (by ex_date) self._by_symbol: dict[str, list[dict[str, Any]]] = {} if self.path and self.path.is_file(): self._load() # -- persistence ------------------------------------------------------ def _load(self) -> None: try: payload = json.loads(self.path.read_text(encoding="utf-8")) except (OSError, ValueError) as exc: raise DividendLedgerError(f"cannot load dividend ledger: {exc}") from exc entries = payload.get("entries", []) if isinstance(payload, dict) else [] for e in entries: self.add(e.get("symbol", ""), e, persist=False) def save(self) -> None: if not self.path: return self.path.parent.mkdir(parents=True, exist_ok=True) all_entries: list[dict[str, Any]] = [] for sym in sorted(self._by_symbol): all_entries.extend(sorted(self._by_symbol[sym], key=_entry_ex_date)) self.path.write_text( json.dumps({"entries": all_entries}, ensure_ascii=False, indent=2), encoding="utf-8", ) # -- write ------------------------------------------------------------ def add(self, symbol: str, entry: Mapping[str, Any], persist: bool = True) -> dict[str, Any]: """Register a dated dividend entry for a symbol. `entry` requires ``per_share`` (finite, >= 0). A **real** entry must carry ``ex_date`` (and the ledger uses it as the cut-off). An **estimate** row (``source="dps_annual_proxy"``) may omit ex_date and is flagged ``estimate=True``. """ if not symbol: raise DividendLedgerError("dividend entry requires a symbol") per_share = entry.get("per_share") try: per_share = float(per_share) except (TypeError, ValueError) as exc: raise DividendLedgerError("per_share must be numeric") from exc if not math.isfinite(per_share) or per_share < 0: raise DividendLedgerError("per_share must be finite and >= 0") ex_date = entry.get("ex_date") # a DPS annual proxy row is always an estimate regardless of flags estimate = bool(entry.get("estimate", False)) or entry.get("source") == "dps_annual_proxy" if ex_date and not estimate: _parse_date(ex_date) # validate elif not ex_date and not estimate: raise DividendLedgerError("real dividend entry requires ex_date") normalized = dict(entry) normalized["symbol"] = symbol normalized["per_share"] = per_share normalized["estimate"] = estimate self._by_symbol.setdefault(symbol, []).append(normalized) if persist: self.save() return normalized # -- reads ------------------------------------------------------------ def entries(self, symbol: str) -> list[dict[str, Any]]: return sorted(self._by_symbol.get(symbol, []), key=_entry_ex_date) def symbols(self) -> list[str]: return sorted(self._by_symbol.keys()) def _entry_ex_date(e: Any) -> str: return str(e.get("ex_date") or "") # --------------------------------------------------------------------------- # Backtest integration # --------------------------------------------------------------------------- def build_dps_ledger(snapshot: Mapping[str, Any]) -> DividendLedger: """Build a DPS-annual-proxy dividend ledger from a siamchart snapshot. For each symbol in ``snapshot["rows"]`` with a per-symbol ``ratios.DPS``, register an estimate row (``source="dps_annual_proxy"``) so the backtest can credit ``DPS * qty`` per name instead of multiplying a yield percentage by the (price-dependent) market value. These are **estimates**, clearly flagged, not realised dated cash flows — real ex-date history must be collected separately to upgrade a symbol to ``dated_ledger``. """ ledger = DividendLedger() details = snapshot.get("details", {}) or {} for row in snapshot.get("rows", []): symbol = row.get("symbol") if not symbol: continue ratios = (details.get(symbol) or {}).get("ratios", {}) or {} dps = ratios.get("DPS") try: dps = float(dps) except (TypeError, ValueError): dps = None if dps is None or not (dps > 0): continue # no DPS -> no estimate row (fail closed) ledger.add(symbol, {"per_share": dps, "source": "dps_annual_proxy"}) return ledger def populate_dated_dividends(ledger: DividendLedger, symbols: list[str], history_fetcher) -> dict[str, int]: """Fetch real dated dividend history per symbol and register it. ``history_fetcher(symbol) -> list[(ex_date, dps)]`` (e.g. ``siamchart.fetch_dividend_history``). For each symbol, every dated payment becomes a real ``dated_ledger`` row (``source="siamchart_dated"``, ``estimate=False``). Returns {symbol: rows_added}. Note: a symbol with a real dated history is no longer a "dps_annual_proxy" estimate — the dated ledger is the honest, realised cash-flow model. """ counts: dict[str, int] = {} for symbol in symbols: try: history = history_fetcher(symbol) or [] except Exception: # noqa: BLE001 — one symbol must not abort the rest counts[symbol] = 0 continue n = 0 for ex_date, per_share in history: if per_share <= 0: continue ledger.add(symbol, { "ex_date": ex_date, "per_share": per_share, "source": "siamchart_dated", "estimate": False, }) n += 1 counts[symbol] = n return counts def credit_dividends( ledger: DividendLedger, holdings: Mapping[str, float], on_date: dt.date, ) -> float: """Cash dividends payable to `holdings` as of `on_date`. For each symbol, sums ``per_share * qty`` for entries whose: - real entry: ``ex_date <= on_date < pay_date (if pay_date given)``; the share must be held *before* ex_date, which the caller enforces by only passing holdings that were acquired before ex_date (see callers). - estimate entry (no ex_date): credited pro-rata across the holding window — a NOTE, the caller decides the split; here we credit on first sight of the symbol in `holdings` for simplicity and mark it estimate. Returns the total cash to add. Never negative. """ total = 0.0 for symbol, qty in holdings.items(): if qty <= 0: continue for e in ledger.entries(symbol): if e.get("estimate"): # estimate row: credit once per symbol (caller ensures `on_date` # is a single payment date in the window it simulates). total += float(e["per_share"]) * float(qty) continue ex = _parse_date(e.get("ex_date")) pay = _parse_date(e.get("pay_date")) if e.get("pay_date") else None # credited once the payment is due: at/after ex-date (holder # qualifies) AND at/after the pay date when one is given. if on_date >= ex and (pay is None or on_date >= pay): total += float(e["per_share"]) * float(qty) return total