[verified] Real dated dividend history collector (dps_annual_proxy -> dated_ledger)

Close the last deferred PIT milestone by collecting REAL per-stock dated
dividend cash-flow history from Siamchart, upgrading the dividend ledger
from DPS estimates to dated_ledger.

- backend/app/siamchart.py: parse_dividend_history(html) extracts the
  'ประวัติการปันผล' dividend table (ex_date + per-share DPS) from each
  stock-info page; fetch_dividend_history(symbol) fetches it live.
- backend/app/dividend_ledger.py: populate_dated_dividends(ledger,
  symbols, fetcher) registers every dated payment as a real row
  (estimate=False, source=siamchart_dated); one symbol failing never
  aborts the rest.
- backend/app/__init__.py: DividendLedger persisted at
  data/dividends/ledger.json; POST /api/v1/dividends/update fetches all
  symbols and saves it; use_ledger backtests prefer the dated ledger when
  populated (dividend_method=dated_ledger) and fall back to DPS estimates
  otherwise.
- tests: parser (4) + populate (2) — full backend 292 passed.

Live (real network): update fetched 49/49 symbols, 1410 dated payments;
use_ledger backtest then reports dividend_method=dated_ledger.
The 'eval(' static-scan hit is ast.literal_eval (safe literal parse, no
code execution), not eval().

backend/app/__init__.py

@@ -146,6 +146,10 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
     app.extensions["forward_store"] = ForwardTestStore(
         app.config.get("FORWARD_STORE_PATH") or (data_root / "forward" / "runs.json")
     )
+    from .dividend_ledger import DividendLedger
+    app.extensions["dividend_ledger"] = DividendLedger(
+        app.config.get("DIVIDEND_LEDGER_PATH") or (data_root / "dividends" / "ledger.json")
+    )
 
     # shared in-process daily cache + app-internal data scheduler (independent of
     # Hermes — this app runs on its own server).
@@ -749,7 +753,14 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
             ledger = None
             if use_ledger:
                 from .dividend_ledger import build_dps_ledger
-                ledger = build_dps_ledger(_load_siamchart_snapshot())
+                # prefer the persisted dated ledger (real ex-date history, if
+                # populated via /api/v1/dividends/update); otherwise fall back
+                # to DPS estimates built from the current snapshot.
+                stored = app.extensions.get("dividend_ledger")
+                if stored is not None and stored.symbols():
+                    ledger = stored
+                else:
+                    ledger = build_dps_ledger(_load_siamchart_snapshot())
             if use_pit:
                 from pathlib import Path as _Path
                 from .factor_vintages import FactorVintageStore
@@ -882,6 +893,33 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
         except (ForwardError, sim.SimulationError, OSError) as exc:
             return jsonify({"error": str(exc)}), 400
 
+    @app.post("/api/v1/dividends/update")
+    def dividends_update():
+        """Fetch real dated dividend history for every snapshot symbol and
+        persist a dated dividend ledger.
+
+        Uses ``siamchart.fetch_dividend_history`` for each symbol (ex-date +
+        per-share DPS). Populates the on-disk ledger so ``use_ledger`` backtests
+        credit REAL dated cash flows (dividend_method=dated_ledger) instead of
+        DPS estimates. Returns per-symbol row counts.
+        """
+        from .siamchart import fetch_dividend_history
+        from .dividend_ledger import populate_dated_dividends
+        from .siamchart_factors import build_factor_view
+        ledger = app.extensions["dividend_ledger"]
+        fv = build_factor_view()
+        symbols = [f["symbol"] for f in fv.get("factors", [])]
+        counts = populate_dated_dividends(ledger, symbols, fetch_dividend_history)
+        ledger.save()
+        total = sum(counts.values())
+        return jsonify({
+            "updated_symbols": len([s for s, n in counts.items() if n > 0]),
+            "symbols": len(symbols),
+            "total_payments": total,
+            "counts": counts,
+            "note": "dated_ledger now used by use_ledger backtests",
+        })
+
     @app.get("/api/v1/data/last-refresh")
     def last_refresh():
         """Status of the in-app automatic data refresh (independent of Hermes)."""

backend/app/dividend_ledger.py

@@ -158,6 +158,38 @@ def build_dps_ledger(snapshot: Mapping[str, Any]) -> DividendLedger:
     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],

backend/app/siamchart.py

@@ -386,4 +386,52 @@ def fetch_stock_info(symbol: str, timeout: float = 30.0) -> StockInfo:
     """Fetch and parse a single symbol's stock-info page."""
     url = _build_stock_info_url(symbol)
     html_text = _fetch(url, timeout=timeout)
-    return parse_stock_info_html(html_text, symbol) 
+    return parse_stock_info_html(html_text, symbol)
+
+
+# ---------------------------------------------------------------------------
+# Dividend history (per-stock ex-date + DPS)
+# ---------------------------------------------------------------------------
+def parse_dividend_history(html_text: str) -> list[tuple[str, float]]:
+    """Extract the per-stock dividend history table from a stock-info page.
+
+    Siamchart renders a "ประวัติการปันผล" (dividend history) table on each
+    stock-info page:
+
+        <table>
+          <tr><td class="head1" colspan=4>ประวัติการปันผล</td></tr>
+          <tr><td class="head2">วันที่</td><td class="head2">ปันผล</td>...</tr>
+          <tr><td class="body">2002-04-04</td><td class="body">2.5</td>...</tr>
+          ...
+
+    Returns a chronological list of ``(ex_date, dividend_per_share)`` pairs.
+    The date column is the dividend's dated cut-off (ex-date); the amount is
+    per-share (THB). This is REAL dated cash-flow history — exactly what the
+    ``dividend_ledger`` needs to upgrade a symbol from a DPS estimate to a
+    ``dated_ledger``.
+    """
+    match = re.search(r"ประวัติการปันผล</td></tr>(.*?)</table>", html_text, re.S)
+    if not match:
+        return []
+    body = match.group(1)
+    rows: list[tuple[str, float]] = []
+    for tr in re.findall(r"<tr[^>]*>(.*?)</tr>", body, re.S):
+        cells = re.findall(r'<td[^>]*class="(?:body|remove_col)"[^>]*>(.*?)</td>', tr, re.S)
+        if len(cells) < 2:
+            continue
+        date = re.sub(r"<[^>]+>", "", cells[0]).strip()
+        amount = re.sub(r"<[^>]+>", "", cells[1]).strip()
+        if not re.match(r"^\d{4}-\d{2}-\d{2}$", date):
+            continue
+        try:
+            rows.append((date, float(amount)))
+        except ValueError:
+            continue
+    return sorted(set(rows))  # dedupe + chronological
+
+
+def fetch_dividend_history(symbol: str, timeout: float = 30.0) -> list[tuple[str, float]]:
+    """Fetch a symbol's real dated dividend history from its stock-info page."""
+    url = _build_stock_info_url(symbol)
+    html_text = _fetch(url, timeout=timeout)
+    return parse_dividend_history(html_text)

backend/tests/test_dividend_ledger.py

@@ -10,6 +10,7 @@ from app.dividend_ledger import (
     DividendLedger,
     DividendLedgerError,
     credit_dividends,
+    populate_dated_dividends,
 )
 
 
@@ -92,5 +93,39 @@ class CreditDividendsTest(unittest.TestCase):
         )
 
 
+class PopulateDatedTest(unittest.TestCase):
+    def test_populates_real_dated_rows_and_marks_not_estimate(self):
+        import tempfile
+        from pathlib import Path
+        ledger = DividendLedger(Path(tempfile.mkdtemp()) / "l.json")
+        fake_fetcher = lambda sym: {  # noqa: E731
+            "PTT": [("2024-02-29", 1.2), ("2024-08-28", 0.8), ("2025-03-06", 1.3)],
+            "AOT": [("2025-12-11", 0.81)],
+        }.get(sym, [])
+        counts = populate_dated_dividends(ledger, ["PTT", "AOT", "NONE"], fake_fetcher)
+        self.assertEqual(counts["PTT"], 3)
+        self.assertEqual(counts["AOT"], 1)
+        self.assertEqual(counts["NONE"], 0)
+        entries = ledger.entries("PTT")
+        self.assertEqual(len(entries), 3)
+        self.assertFalse(entries[0]["estimate"])  # real dated row
+        self.assertEqual(entries[0]["ex_date"], "2024-02-29")
+        self.assertEqual(entries[0]["per_share"], 1.2)
+
+    def test_fetcher_error_for_one_symbol_does_not_abort_others(self):
+        import tempfile
+        from pathlib import Path
+        ledger = DividendLedger(Path(tempfile.mkdtemp()) / "l.json")
+
+        def flaky(sym):
+            if sym == "PTT":
+                raise OSError("boom")
+            return [("2025-03-06", 1.3)]
+
+        counts = populate_dated_dividends(ledger, ["PTT", "AOT"], flaky)
+        self.assertEqual(counts["PTT"], 0)
+        self.assertEqual(counts["AOT"], 1)
+
+
 if __name__ == "__main__":
     unittest.main()

backend/tests/test_siamchart_dividend.py

@@ -0,0 +1,46 @@
+"""Tests for the Siamchart dividend-history parser (real ex-date/DPS)."""
+
+from __future__ import annotations
+
+import unittest
+
+from app.siamchart import parse_dividend_history
+
+
+_DIV_TABLE = """<table>
+<tr><td class="head1" colspan=4>ประวัติการปันผล</td></tr>
+<tr><td class="head2" style="width:100px;">วันที่</td><td class="head2" style="width:50px;">ปันผล</td></tr>
+<tr><td class="body">2002-04-04</td><td class="body">2.5</td></tr>
+<tr><td class="body">2024-02-29</td><td class="body">1.2</td></tr>
+<tr><td class="body">2025-03-06</td><td class="body">1.3</td></tr>
+<tr><td class="body">2025-10-01</td><td class="body">0.9</td></tr>
+</table>"""
+
+
+class ParseDividendHistoryTest(unittest.TestCase):
+    def test_extracts_chronological_dated_dps(self):
+        rows = parse_dividend_history(_DIV_TABLE)
+        self.assertEqual(
+            rows,
+            [("2002-04-04", 2.5), ("2024-02-29", 1.2),
+             ("2025-03-06", 1.3), ("2025-10-01", 0.9)],
+        )
+
+    def test_returns_empty_when_no_table(self):
+        self.assertEqual(parse_dividend_history("<html>no dividend table</html>"), [])
+
+    def test_deduplicates_identical_rows(self):
+        html = _DIV_TABLE + '<tr><td class="body">2025-03-06</td><td class="body">1.3</td></tr>'
+        rows = parse_dividend_history(html)
+        self.assertEqual(len([r for r in rows if r == ("2025-03-06", 1.3)]), 1)
+
+    def test_skips_non_date_rows(self):
+        html = """<table><tr><td class="head1">ประวัติการปันผล</td></tr>
+        <tr><td class="body">N/A</td><td class="body">2.0</td></tr>
+        <tr><td class="body">2020-01-01</td><td class="body">1.0</td></tr></table>"""
+        rows = parse_dividend_history(html)
+        self.assertEqual(rows, [("2020-01-01", 1.0)])
+
+
+if __name__ == "__main__":
+    unittest.main()