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().
438 lines
17 KiB
Python
438 lines
17 KiB
Python
"""Siamchart fundamental-data fetcher/parser for SET50 research factors.
|
|
|
|
Siamchart (siamchart.com) serves its stock financial table as **server-rendered
|
|
HTML** — the HAR capture (`har-derived-api-client`) shows only Cloudflare
|
|
speculation/rum telemetry, no JSON/XHR API. Per that skill's guidance we scrape
|
|
and parse the HTML directly instead of deriving an API client.
|
|
|
|
Primary page (all SET50 symbols in one table):
|
|
|
|
https://siamchart.com/stock-financial/SET50
|
|
├── table.tbl thead: Name, No., EPS21..EPS25 (YoY%), Rev21..Rev25 (YoY%),
|
|
│ NP21..NP25 (YoY%), PE
|
|
└── table.tbl tbody row: [Symbol, No.,
|
|
EPS21, EPS22, EPS23, EPS24, EPS25,
|
|
Rev21, Rev22, Rev23, Rev24, Rev25,
|
|
NP21, NP22, NP23, NP24, NP25,
|
|
PE]
|
|
|
|
Each metric cell is rendered as ``<value> (+YoY%)`` except the base row where a
|
|
``+Infinity`` / ``-Infinity`` / ``NaN`` YoY may appear. PE is a bare number.
|
|
|
|
Per-stock detail pages live at ``/stock-info/<SYMBOL>/`` (PE, P/BV, D/E, DPS,
|
|
EPS, ROAA/ROAE/NPM, Yield%, dividend history, full income statement with
|
|
QoQ/YoY) and ``/stock-chart/<SYMBOL>/`` for prices; both are out of scope for
|
|
this module and are exercised elsewhere.
|
|
|
|
This module is pure: it never touches the network itself. The URL is built from
|
|
a group name, the HTML is fetched by the fetching helper (or injected directly
|
|
for tests), and parsing is a deterministic pure function over the HTML string.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import html
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional
|
|
from urllib.request import Request, urlopen
|
|
|
|
# Browser user-agent captured from the Siamchart HAR so the server treats us as
|
|
# a normal browser rather than a default library client (python-urllib gets 403).
|
|
_USER_AGENT = (
|
|
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
|
|
)
|
|
|
|
_BASE_URL = "https://siamchart.com/stock-financial/"
|
|
|
|
# Column layout of table.tbl for a financial group page.
|
|
_FINANCIAL_COLUMNS = [
|
|
"symbol",
|
|
"no",
|
|
"eps21", "eps22", "eps23", "eps24", "eps25",
|
|
"rev21", "rev22", "rev23", "rev24", "rev25",
|
|
"np21", "np22", "np23", "np24", "np25",
|
|
"pe",
|
|
]
|
|
|
|
|
|
class SiamchartError(Exception):
|
|
"""Raised when a Siamchart HTML page cannot be fetched or parsed."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SiamchartRow:
|
|
"""One parsed row of the Siamchart financial group table."""
|
|
|
|
symbol: str
|
|
no: int
|
|
eps: "dict[int, Optional[float]]" = field(default_factory=dict)
|
|
eps_yoy: "dict[int, Optional[float]]" = field(default_factory=dict)
|
|
revenue: "dict[int, Optional[float]]" = field(default_factory=dict)
|
|
revenue_yoy: "dict[int, Optional[float]]" = field(default_factory=dict)
|
|
net_profit: "dict[int, Optional[float]]" = field(default_factory=dict)
|
|
net_profit_yoy: "dict[int, Optional[float]]" = field(default_factory=dict)
|
|
pe: Optional[float] = None
|
|
raw: list = field(default_factory=list)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"symbol": self.symbol,
|
|
"no": self.no,
|
|
"eps": self.eps,
|
|
"eps_yoy": self.eps_yoy,
|
|
"revenue": self.revenue,
|
|
"revenue_yoy": self.revenue_yoy,
|
|
"net_profit": self.net_profit,
|
|
"net_profit_yoy": self.net_profit_yoy,
|
|
"pe": self.pe,
|
|
}
|
|
|
|
|
|
def build_url(group: str = "SET50") -> str:
|
|
"""Build the Siamchart financial-group URL for a given group name.
|
|
|
|
The group is the string used in the URL path segment exactly as Siamchart
|
|
produces it when the ``fund_type`` dropdown is changed (e.g. ``SET50``).
|
|
It is validated to a strict token (uppercase alphanumerics and hyphen) so it
|
|
can never turn into partial-list/traversal URL (e.g. ``../etc``) and remains
|
|
safe if group ever reaches this builder from request input.
|
|
"""
|
|
if not group:
|
|
raise SiamchartError("group must be a non-empty string")
|
|
group = group.strip()
|
|
if not re.match(r"^[A-Z0-9]+(?:-[A-Z0-9]+)*$", group):
|
|
raise SiamchartError(
|
|
f"invalid Siamchart group: {group!r} "
|
|
"(expected uppercase alphanumerics, optionally hyphenated)"
|
|
)
|
|
return f"{_BASE_URL}{group}"
|
|
|
|
|
|
def _fetch(url: str, timeout: float = 30.0) -> str:
|
|
"""Fetch ``url`` and return its decoded HTML text body."""
|
|
req = Request(url, headers={"User-Agent": _USER_AGENT, "Accept": "text/html"})
|
|
try:
|
|
with urlopen(req, timeout=timeout) as resp:
|
|
raw = resp.read()
|
|
except Exception as exc: # network errors are all non-recoverable here
|
|
raise SiamchartError(f"failed to fetch {url}: {exc}") from exc
|
|
# Siamchart serves UTF-8; fall back to latin-1 only if the header says so.
|
|
for encoding in ("utf-8", "latin-1"):
|
|
try:
|
|
return raw.decode(encoding)
|
|
except UnicodeDecodeError:
|
|
continue
|
|
raise SiamchartError("unable to decode Siamchart response")
|
|
|
|
|
|
def _split_metric(value: str) -> tuple[Optional[float], Optional[float]]:
|
|
"""Parse a ``<value>`` cell into a (value, None) tuple.
|
|
|
|
The PE cell in ``store_real_data`` is a bare number; the YoY% is only
|
|
rendered in the web table and is not present in the raw array. We keep this
|
|
helper so PE parsing is explicit and consistent with the (value, yoy) shape
|
|
used elsewhere. An empty value returns (None, None).
|
|
"""
|
|
text = html.unescape(value).strip()
|
|
if not text:
|
|
return None, None
|
|
numbers = re.findall(r"[-+]?\d+(?:[.,]\d+)?", text.replace(",", ""))
|
|
try:
|
|
val = float(numbers[0]) if numbers else None
|
|
except ValueError:
|
|
val = None
|
|
return val, None
|
|
|
|
|
|
def parse_financial_html(html_text: str) -> list[SiamchartRow]:
|
|
"""Parse a Siamchart financial-group HTML page into rows.
|
|
|
|
Siamchart renders its financial table **client-side** — the raw HTML does
|
|
not contain a static ``<tbody>``; instead the data ships as a JavaScript
|
|
array literal named ``store_real_data`` (assigned from ``real_data``). This
|
|
is a pure ``[symbol_row, ...]`` structure decoder over that literal, so it
|
|
works without executing JS or a browser.
|
|
|
|
Each element is a 22-position list:
|
|
[0] symbol, [1] symbol, [2] full-name, [3] marker,
|
|
[4..8] EPS21..EPS25,
|
|
[9] separator,
|
|
[10..14] Rev21..Rev25,
|
|
[15] separator,
|
|
[16..20] NP21..NP25,
|
|
[21] PE (bare number)
|
|
Strings are unquoted numbers or empty; we coerce sensibly.
|
|
"""
|
|
anchor = re.search(r"var\s+store_real_data\s*=\s*(\[.*?\n\]\s*;)", html_text, re.S)
|
|
if not anchor:
|
|
# Fall back to any bare assignment (some Siamchart variants set
|
|
# `store_real_data = [...]` without the var keyword).
|
|
anchor = re.search(r"store_real_data\s*=\s*(\[.*?\]\s*;)", html_text, re.S)
|
|
if not anchor:
|
|
raise SiamchartError("no store_real_data array found in Siamchart HTML")
|
|
js_literal = anchor.group(1)
|
|
data = _parse_js_array(js_literal)
|
|
|
|
rows: list[SiamchartRow] = []
|
|
for entry in data:
|
|
if not isinstance(entry, list) or len(entry) < 22:
|
|
continue
|
|
symbol = str(entry[0]).strip() if entry[0] else ""
|
|
if not symbol or not re.match(r"^[A-Z0-9]+$", symbol):
|
|
continue
|
|
try:
|
|
no = int(entry[3])
|
|
except (ValueError, TypeError):
|
|
no = 0
|
|
eps_v, eps_y = _years(entry, 4)
|
|
rev_v, rev_y = _years(entry, 10)
|
|
np_v, np_y = _years(entry, 16)
|
|
pe, _ = _split_metric(str(entry[21]))
|
|
rows.append(
|
|
SiamchartRow(
|
|
symbol=symbol,
|
|
no=no,
|
|
eps=eps_v,
|
|
eps_yoy=eps_y,
|
|
revenue=rev_v,
|
|
revenue_yoy=rev_y,
|
|
net_profit=np_v,
|
|
net_profit_yoy=np_y,
|
|
pe=pe,
|
|
raw=list(entry),
|
|
)
|
|
)
|
|
if not rows:
|
|
raise SiamchartError("no symbol rows parsed from Siamchart financial table")
|
|
return rows
|
|
|
|
|
|
def _years(entry: list, start: int, count: int = 5) -> tuple[dict, dict]:
|
|
"""Extract ``count`` numeric values from a JS array row starting at ``start``.
|
|
|
|
Returns (values, yoy) pairs; in ``store_real_data`` each cell is a plain
|
|
number (YoY% is already baked into the adjacent web-column only when the
|
|
table is rendered — here YoY is not present, so yoy is left empty).
|
|
"""
|
|
values: dict[int, Optional[float]] = {}
|
|
yoy: dict[int, Optional[float]] = {}
|
|
for i in range(count):
|
|
idx = start + i
|
|
if idx >= len(entry):
|
|
break
|
|
cell = entry[idx]
|
|
if cell is None or cell == "":
|
|
values[i + 1] = None
|
|
continue
|
|
try:
|
|
values[i + 1] = float(str(cell).replace(",", ""))
|
|
except (ValueError, TypeError):
|
|
values[i + 1] = None
|
|
return values, yoy
|
|
|
|
|
|
def _parse_js_array(js: str) -> list:
|
|
"""Parse a JavaScript array literal into Python (JSON-compatible-safe).
|
|
|
|
Siamchart uses single-quoted strings and trailing commas in places; we
|
|
normalize by converting single quotes to double quotes for string tokens,
|
|
then use ``ast.literal_eval`` for safe parsing (no code execution).
|
|
Strings containing an apostrophe inside a name would break this, but stock
|
|
company names on Siamchart do not contain apostrophes.
|
|
"""
|
|
import ast
|
|
|
|
# Trim the leading '[' and trailing '];'
|
|
js = js.strip()
|
|
if js.endswith(";"):
|
|
js = js[:-1]
|
|
# Convert single-quoted strings to double-quoted: replace 'X' -> "X"
|
|
# (safe here because names have no inner single quotes or doubles).
|
|
converted = re.sub(r"'([^']*)'", lambda m: '"' + m.group(1).replace('"', '\\"') + '"', js)
|
|
try:
|
|
return ast.literal_eval(converted)
|
|
except (ValueError, SyntaxError) as exc:
|
|
raise SiamchartError(f"failed to parse store_real_data array: {exc}") from exc
|
|
|
|
|
|
def fetch_financial(group: str = "SET50", timeout: float = 30.0) -> list[SiamchartRow]:
|
|
"""Fetch and parse the Siamchart financial group table for ``group``."""
|
|
url = build_url(group)
|
|
html_text = _fetch(url, timeout=timeout)
|
|
return parse_financial_html(html_text)
|
|
|
|
|
|
def _build_stock_info_url(symbol: str) -> str:
|
|
symbol = symbol.strip().upper()
|
|
if not re.match(r"^[A-Z0-9]+$", symbol):
|
|
raise SiamchartError("symbol must be an uppercase SET ticker, e.g. PTT")
|
|
return f"https://siamchart.com/stock-info/{symbol}/"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class StockInfo:
|
|
"""Key fundamental ratios and latest income-statement snapshot for a symbol."""
|
|
|
|
symbol: str
|
|
ratios: "dict[str, Optional[float]]" = field(default_factory=dict)
|
|
# Latest income statement: label -> (value, qoq_pct, yoy_pct)
|
|
income: "dict[str, tuple[Optional[float], Optional[float], Optional[float]]]" = (
|
|
field(default_factory=dict)
|
|
)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"symbol": self.symbol,
|
|
"ratios": self.ratios,
|
|
"income": {
|
|
k: {"value": v[0], "qoq_pct": v[1], "yoy_pct": v[2]}
|
|
for k, v in self.income.items()
|
|
},
|
|
}
|
|
|
|
|
|
def parse_stock_info_html(html_text: str, symbol: str) -> StockInfo:
|
|
"""Parse key ratios + latest income statement from a stock-info page.
|
|
|
|
Siamchart renders the summary ratios as a **vertical table**: one row per
|
|
metric, ``<tr><td title=...><b>P/E</b></td><td class="remove_col">value
|
|
<br>(QoQ)<br>(YoY)</td>...`` across five periods. The first ``remove_col``
|
|
cell holds the latest period's value plus QoQ and YoY deltas on separate
|
|
lines.
|
|
|
|
The income statement rows embed the label in a ``<td><b>รวมรายได้</b></td>``
|
|
followed by the same five-period ``remove_col`` cells; we read the first one.
|
|
"""
|
|
ratios: dict[str, Optional[float]] = {}
|
|
income: dict[str, tuple[Optional[float], Optional[float], Optional[float]]] = {}
|
|
|
|
# --- Key ratios: head1 header row followed by a body row ---
|
|
# Located after "ราคาปิดที่ 41", i.e.:
|
|
# <tr><td class="head1">PE</td>...<td class="head1">Yield %</td></tr>
|
|
# <tr><td class="body">9.42</td><td class="body">0.97</td>...</tr>
|
|
hdr = re.search(
|
|
r'<td[^>]*class="head1"[^>]*>PE</td>(.*?)</tr>', html_text, re.S
|
|
)
|
|
if hdr:
|
|
labels = [
|
|
html.unescape(re.sub(r"<[^>]+>", "", x)).strip()
|
|
for x in re.findall(r'<td[^>]*class="head1"[^>]*>(.*?)</td>', hdr.group(0), re.S)
|
|
]
|
|
labels = [x for x in labels if x]
|
|
body_match = re.search(
|
|
r'<td[^>]*class="body"[^>]*>(.*?)</td>\s*</tr>',
|
|
html_text[hdr.end():],
|
|
re.S,
|
|
)
|
|
if body_match:
|
|
body_vals = [
|
|
html.unescape(re.sub(r"<[^>]+>", "", x)).strip()
|
|
for x in re.findall(r'<td[^>]*class="body"[^>]*>(.*?)</td>', body_match.group(0), re.S)
|
|
]
|
|
for label, val in zip(labels, body_vals):
|
|
ratios[label] = _float_or_none(val.split("\n")[0].strip())
|
|
|
|
# --- Income statement headline rows ---
|
|
# Each row is <tr><td><b>LABEL</b></td><td class="remove_col">...</td>.
|
|
# We match the full <b>...</b> label so a prefix like "กำไร" cannot bind to
|
|
# the wrong row, and we only fill a key once (a later lossy variant of the
|
|
# same row, e.g. "กำไร (ขาดทุน) สุทธิ", must not overwrite "กำไรสุทธิ").
|
|
income_labels = [
|
|
("รวมรายได้", "total_revenue"),
|
|
("รวมค่าใช้จ่าย", "total_cost"),
|
|
("กำไรขั้นต้น", "gross_profit"),
|
|
("กำไรสุทธิ", "net_profit"),
|
|
("กำไร (ขาดทุน) สุทธิ", "net_profit"),
|
|
]
|
|
for label, canonical in income_labels:
|
|
if canonical in income:
|
|
continue # already captured by an earlier full match
|
|
m = re.search(r"<b>" + re.escape(label) + r"</b></td>(.*?)</tr>", html_text, re.S)
|
|
if not m:
|
|
continue
|
|
cells = re.findall(r'class="remove_col"[^>]*>(.*?)</td>', m.group(1), re.S)
|
|
if not cells:
|
|
continue
|
|
first = html.unescape(re.sub(r"<[^>]+>", "", cells[0])) or ""
|
|
# "1,568,599.55(+14.48)(+12.26)" — the first full number (thousand-
|
|
# separated) is the value; bracketed tokens are QoQ and YoY percentages.
|
|
first_num = re.search(r"[-+]?\d{1,3}(?:,\d{3})*(?:\.\d+)?", first)
|
|
value = _float_or_none(first_num.group(0)) if first_num else None
|
|
qoq = yoy = None
|
|
for mm in re.finditer(r"\(([-+]?\d+(?:[.,]\d+)?)\)", first):
|
|
pct = _float_or_none(mm.group(1))
|
|
if qoq is None:
|
|
qoq = pct
|
|
elif yoy is None:
|
|
yoy = pct
|
|
income[canonical] = (value, qoq, yoy)
|
|
|
|
return StockInfo(symbol=symbol, ratios=ratios, income=income)
|
|
|
|
|
|
def _float_or_none(text: str) -> Optional[float]:
|
|
text = text.replace(",", "").strip()
|
|
if not text:
|
|
return None
|
|
try:
|
|
return float(text)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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)
|