From 068dff22d74e04c2bfb440e071964fe60b740ee3 Mon Sep 17 00:00:00 2001 From: Kunthawat Greethong Date: Thu, 27 Aug 2026 11:46:15 +0700 Subject: [PATCH] [verified] Dated dividend cash-flow ledger replacing final-holdings proxy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 11 +- backend/app/backtest.py | 46 +++++- backend/app/dividend_ledger.py | 194 ++++++++++++++++++++++++++ backend/tests/test_backtest.py | 37 +++++ backend/tests/test_dividend_ledger.py | 96 +++++++++++++ 5 files changed, 376 insertions(+), 8 deletions(-) create mode 100644 backend/app/dividend_ledger.py create mode 100644 backend/tests/test_dividend_ledger.py diff --git a/backend/app/__init__.py b/backend/app/__init__.py index 3c9e6f0..fc87449 100644 --- a/backend/app/__init__.py +++ b/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", []) diff --git a/backend/app/backtest.py b/backend/app/backtest.py index 2e213ca..ecb1b7f 100644 --- a/backend/app/backtest.py +++ b/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) - meta = score_by_symbol.get(sym, {}) - yield_pct = float(meta.get("dividend_yield") or 0.0) / 100.0 - total_dividend += qty * px * yield_pct + if _ledger is not None: + # ledger-driven: per_share * qty from dated entries (fail closed + # if no entry -> no credit). + 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) diff --git a/backend/app/dividend_ledger.py b/backend/app/dividend_ledger.py new file mode 100644 index 0000000..f9b2d74 --- /dev/null +++ b/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 diff --git a/backend/tests/test_backtest.py b/backend/tests/test_backtest.py index 6bb93d9..3ff3321 100644 --- a/backend/tests/test_backtest.py +++ b/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() diff --git a/backend/tests/test_dividend_ledger.py b/backend/tests/test_dividend_ledger.py new file mode 100644 index 0000000..fb6788c --- /dev/null +++ b/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()