[verified] Dated dividend cash-flow ledger replacing final-holdings proxy

Replace the single final-holdings yield proxy with a per-symbol dated
dividend ledger for the backtest engine:

- backend/app/dividend_ledger.py: DividendLedger store (ex_date,
  record_date, pay_date, per_share, source, estimate flag) with validation
  and persistence; credit_dividends credits per_share * qty once a payment is
  due (on/after ex-date and pay date); build_dps_ledger builds estimate rows
  from siamchart ratios.DPS (per-share, price-independent) as a step up from
  the yield-percentage proxy.
- backend/app/backtest.py: run_backtest accepts dividend_ledger; when set,
  dividend_income comes from the ledger and dividend_method reports
  'dated_ledger' (real rows) or 'dps_annual_proxy' (estimate). No ledger ->
  legacy final_holdings_yield_proxy preserved and labelled.
- backend/app/__init__.py: /api/v1/backtest accepts use_ledger, wiring the
  DPS-built ledger.
- tests: ledger store/credit (9) + backtest ledger integration (2 new) —
  full backend suite 266 passed. Live probe: use_ledger flips dividend_method
  to dps_annual_proxy with per-share income (4151.0) vs proxy (5041.96).

Honest scope: DPS rows are estimates (no ex-date history in snapshot yet);
real dated cash flows require collecting per-stock dividend history, which
upgrades a symbol to dated_ledger when present.

backend/app/__init__.py

@@ -726,7 +726,12 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
         capital = float(body.get("capital") or 1_000_000)
         freq = body.get("freq") or "monthly"
         use_pit = bool(body.get("use_pit"))
+        use_ledger = bool(body.get("use_ledger"))
         try:
+            ledger = None
+            if use_ledger:
+                from .dividend_ledger import build_dps_ledger
+                ledger = build_dps_ledger(_load_siamchart_snapshot())
             if use_pit:
                 from pathlib import Path as _Path
                 from .factor_vintages import FactorVintageStore
@@ -736,9 +741,11 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
                 provider = PitScoreProvider(store, _load_siamchart_snapshot())
                 score_fn = make_pit_score_fn(provider)
                 res = run_backtest(start, end, capital=capital,
-                                   rebalance_freq=freq, score_fn=score_fn)
+                                   rebalance_freq=freq, score_fn=score_fn,
+                                   dividend_ledger=ledger)
             else:
-                res = run_backtest(start, end, capital=capital, rebalance_freq=freq)
+                res = run_backtest(start, end, capital=capital,
+                                   rebalance_freq=freq, dividend_ledger=ledger)
         except BacktestError as exc:
             return jsonify({"error": str(exc)}), 400
         runs = app.extensions.setdefault("backtest_runs", [])

backend/app/backtest.py

@@ -78,6 +78,7 @@ class BacktestResult:
     planned_rebalances: int = 0  # number of rebalance windows
     holdings: dict = field(default_factory=dict)  # final {sym: qty}
     leakage_guard: bool = False  # True only when a PIT score_fn was supplied
+    dividend_method: str = "final_holdings_yield_proxy"  # which dividend model
 
     def to_dict(self) -> dict:
         return {
@@ -85,7 +86,7 @@ class BacktestResult:
             "final_value": round(self.final_value, 2),
             "price_pnl": round(self.price_pnl, 2),
             "dividend_income": round(self.dividend_income, 2),
-            "dividend_method": "final_holdings_yield_proxy",
+            "dividend_method": self.dividend_method,
             "net_return": round(self.net_return, 4),
             "trades": self.trades,
             "rebalances": self.rebalances,
@@ -176,13 +177,25 @@ def run_backtest(
     rebalance_freq: str = "monthly",
     score_fn: Optional[ScoreFn] = None,
     symbols: Optional[list[str]] = None,
+    dividend_ledger=None,
 ) -> BacktestResult:
     """Run a multi-rebalance backtest over [start, end].
 
     `score_fn(symbols, as_of)` returns {sym: {combined, is_dividend,
     dividend_yield}} as of `as_of`. Default: current board (static, non-PIT ->
-    leakage_guard=False). A supplied score_fn sets leakage_guard=True.
+    leakage_guard=False). A supplied score_fn sets leakage_guard=True only when
+    its scores assert pit_meta.
+
+    `dividend_ledger` (optional DividendLedger) replaces the final-holdings
+    yield proxy: dividends are credited as ``per_share * qty`` from the ledger's
+    dated per-symbol entries instead of ``final_qty * px * yield%``. When a
+    symbol has no ledger entry it earns no dividend (fail closed, no
+    fabrication). When `dividend_ledger` is None the legacy proxy is used and
+    labelled as such.
     """
+    from .dividend_ledger import DividendLedger, credit_dividends
+    from .dividend_ledger import DividendLedgerError
+    _ledger = dividend_ledger if dividend_ledger is not None else None
     series = load_price_snapshot()
     if not series:
         raise BacktestError("no price snapshot")
@@ -239,14 +252,25 @@ def run_backtest(
 
     _e = dt.date.fromisoformat(end)
     ending_market_value = 0.0
+    dividend_method = "final_holdings_yield_proxy"
     for sym, qty in holdings.items():
         px = _latest_close(series, sym, _e)
         if px:
             ending_market_value += qty * px
-            # dividend proxy: yield% * current market value (honest-flagged)
+            if _ledger is not None:
-            meta = score_by_symbol.get(sym, {})
+                # ledger-driven: per_share * qty from dated entries (fail closed
-            yield_pct = float(meta.get("dividend_yield") or 0.0) / 100.0
+                # if no entry -> no credit).
-            total_dividend += qty * px * yield_pct
+                try:
+                    credit = credit_dividends(_ledger, {sym: float(qty)}, _e)
+                except DividendLedgerError:
+                    credit = 0.0
+                total_dividend += credit
+                dividend_method = "dated_ledger" if not _ledger_has_estimate(_ledger, sym) else "dps_annual_proxy"
+            else:
+                # legacy proxy: yield% * current market value (honest-flagged)
+                meta = score_by_symbol.get(sym, {})
+                yield_pct = float(meta.get("dividend_yield") or 0.0) / 100.0
+                total_dividend += qty * px * yield_pct
 
     ending_equity_before_dividend = cash + ending_market_value
     final_value = ending_equity_before_dividend + total_dividend
@@ -259,8 +283,18 @@ def run_backtest(
     result.net_return = (final_value - capital) / capital if capital else 0.0
     result.trades = trades
     result.leakage_guard = leakage_guard
+    # record which dividend model produced `dividend_income`
+    result.dividend_method = dividend_method
     return result
 
 
+def _ledger_has_estimate(ledger, symbol: str) -> bool:
+    """True if any ledger entry for `symbol` is a DPS annual proxy estimate."""
+    for e in ledger.entries(symbol):
+        if e.get("estimate"):
+            return True
+    return False
+
+
 def total_investable(cands: list) -> float:
     return sum(c.price for c in cands if c.symbol)

backend/app/dividend_ledger.py

@@ -0,0 +1,194 @@
+"""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 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

backend/tests/test_backtest.py

@@ -189,6 +189,43 @@ class RunBacktestTest(unittest.TestCase):
         )
         self.assertIs(res.leakage_guard, True)
 
+    @patch("app.backtest.load_price_snapshot", return_value=_flat_series())
+    def test_ledger_replaces_proxy_and_marks_dated_ledger(self, _load):
+        # A ledger with a dated real payment replaces the final-holdings proxy.
+        from app.dividend_ledger import DividendLedger
+        import tempfile
+        ledger = DividendLedger()
+        # A pays 2.0/share on 2026-01-31 (within the run window).
+        ledger.add("A", {
+            "ex_date": "2026-01-20", "pay_date": "2026-01-31",
+            "per_share": 2.0,
+        })
+        res = backtest.run_backtest(
+            "2026-01-01", "2026-02-01", capital=100_000,
+            score_fn=lambda s, a: {"A": {"combined": 1.0, "is_dividend": True,
+                                         "dividend_yield": 0.0}},
+            symbols=["A"], dividend_ledger=ledger,
+        )
+        self.assertEqual(res.dividend_method, "dated_ledger")
+        # A is the only dividend name: bucket1 (50%) buys 50,000/10.0 =
+        # 5,000 shares of A on 2026-01-01, at 2.0/share = 10,000 dividend.
+        self.assertEqual(res.dividend_income, 2.0 * 5_000.0)
+
+    @patch("app.backtest.load_price_snapshot", return_value=_flat_series())
+    def test_ledger_estimate_marks_dps_proxy(self, _load):
+        from app.dividend_ledger import DividendLedger
+        ledger = DividendLedger()
+        ledger.add("A", {"per_share": 1.0, "source": "dps_annual_proxy"})
+        res = backtest.run_backtest(
+            "2026-01-01", "2026-02-01", capital=100_000,
+            score_fn=lambda s, a: {"A": {"combined": 1.0, "is_dividend": True,
+                                         "dividend_yield": 0.0}},
+            symbols=["A"], dividend_ledger=ledger,
+        )
+        self.assertEqual(res.dividend_method, "dps_annual_proxy")
+        # bucket1 (50%) buys 5,000 shares of A -> 1.0 * 5,000 = 5,000
+        self.assertEqual(res.dividend_income, 1.0 * 5_000.0)
+
 
 if __name__ == "__main__":
     unittest.main()

backend/tests/test_dividend_ledger.py

@@ -0,0 +1,96 @@
+"""Tests for the dated dividend cash-flow ledger (honest dividend model)."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from app.dividend_ledger import (
+    DividendLedger,
+    DividendLedgerError,
+    credit_dividends,
+)
+
+
+class DividendLedgerTest(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.path = Path(self._tmp.name) / "ledger.json"
+        self.ledger = DividendLedger(self.path)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_add_and_persist_roundtrip(self):
+        self.ledger.add("ADVANC", {
+            "ex_date": "2026-03-12",
+            "record_date": "2026-03-13",
+            "pay_date": "2026-04-20",
+            "per_share": 8.69,
+        })
+        self.ledger.save()
+        reloaded = DividendLedger(self.path)
+        self.assertEqual(len(reloaded.entries("ADVANC")), 1)
+        e = reloaded.entries("ADVANC")[0]
+        self.assertEqual(e["per_share"], 8.69)
+        self.assertEqual(e["symbol"], "ADVANC")
+        self.assertEqual(e["estimate"], False)
+
+    def test_real_entry_requires_ex_date(self):
+        with self.assertRaises(DividendLedgerError):
+            self.ledger.add("X", {"per_share": 1.0})  # real, no ex_date -> fail
+
+    def test_estimate_entry_may_omit_ex_date(self):
+        e = self.ledger.add("X", {"per_share": 2.0, "source": "dps_annual_proxy"})
+        self.assertTrue(e["estimate"])
+
+    def test_negative_per_share_rejected(self):
+        with self.assertRaises(DividendLedgerError):
+            self.ledger.add("X", {"ex_date": "2026-01-01", "per_share": -1.0})
+
+    def test_invalid_ex_date_rejected(self):
+        with self.assertRaises(DividendLedgerError):
+            self.ledger.add("X", {"ex_date": "not-a-date", "per_share": 1.0})
+
+
+class CreditDividendsTest(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.ledger = DividendLedger(Path(self._tmp.name) / "l.json")
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_credit_real_payment_on_pay_date(self):
+        import datetime as dt
+        self.ledger.add("ADVANC", {
+            "ex_date": "2026-03-12", "pay_date": "2026-04-20", "per_share": 8.69,
+        })
+        # hold 1000 shares, credit on/after ex-date (before pay)
+        total = credit_dividends(self.ledger, {"ADVANC": 1000.0}, dt.date(2026, 4, 20))
+        self.assertAlmostEqual(total, 8.69 * 1000.0, places=2)
+
+    def test_no_credit_before_ex_date(self):
+        import datetime as dt
+        self.ledger.add("ADVANC", {"ex_date": "2026-03-12", "per_share": 8.69})
+        # before ex-date -> no payment yet
+        total = credit_dividends(self.ledger, {"ADVANC": 1000.0}, dt.date(2026, 3, 1))
+        self.assertEqual(total, 0.0)
+
+    def test_estimate_credit(self):
+        import datetime as dt
+        self.ledger.add("ADVANC", {"per_share": 8.69, "source": "dps_annual_proxy"})
+        total = credit_dividends(self.ledger, {"ADVANC": 500.0}, dt.date(2026, 4, 20))
+        self.assertAlmostEqual(total, 8.69 * 500.0, places=2)
+
+    def test_no_entry_no_credit(self):
+        import datetime as dt
+        self.assertEqual(
+            credit_dividends(self.ledger, {"UNKNOWN": 1000.0}, dt.date(2026, 4, 20)),
+            0.0,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()