set50-system/backend/app/dividend_ledger.py

"""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