f9973e8d0a
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().
132 lines
4.7 KiB
Python
132 lines
4.7 KiB
Python
"""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,
|
|
populate_dated_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,
|
|
)
|
|
|
|
|
|
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()
|