[verified] Add Siamchart SET50 fundamental scraper (financial table + stock-info ratios/income)
- fetch_financial(group): parse Siamchart JS store_real_data array -> EPS/Rev/NP (5yr) + PE for every symbol - parse_stock_info_html: head1/body key ratios (PE/P/BV/D/E/DPS/EPS/ROAA/ROAE/NPM/Yield) + income statement (QoQ/YoY) - collect_siamchart.py CLI: --group/--stock-info/--with-info, timestamped JSON snapshot - build_url validates group against ^[A-Z0-9]+(?:-[A-Z0-9]+)*$ (blocker closed) - 10 unit tests; full backend suite 135 OK; independent review deleg_ad2b8dc7 passed=true [verified] tags from requesting-code-review pipeline.
This commit is contained in:
92
backend/scripts/collect_siamchart.py
Normal file
92
backend/scripts/collect_siamchart.py
Normal file
@@ -0,0 +1,92 @@
|
||||
"""Collect Siamchart SET50 fundamental data into a local timestamped snapshot.
|
||||
|
||||
Two modes:
|
||||
|
||||
--group SET50 parse the whole-group financial table (EPS/Rev/NP/PE
|
||||
for every symbol in the group) into `rows`.
|
||||
--stock-info SYMBOL parse one symbol's key ratios + income statement into
|
||||
`info`.
|
||||
|
||||
With --with-info, after fetching the group table we also fetch each symbol's
|
||||
stock-info page (ratios + income) and attach them under `details[SYMBOL]`. This
|
||||
is the full "所有 in one" fundamental snapshot. The output is written to the
|
||||
`--out` path with a UTC timestamp and a data/parser provenance header.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from app import siamchart
|
||||
|
||||
|
||||
def _now_utc() -> str:
|
||||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--group", default=None,
|
||||
help="Siamchart financial group (e.g. SET50).")
|
||||
parser.add_argument("--stock-info", dest="stock_info", metavar="SYMBOL", default=None,
|
||||
help="Fetch one symbol's stock-info instead of a group table.")
|
||||
parser.add_argument("--with-info", action="store_true",
|
||||
help="After the group table, also fetch each symbol's stock-info.")
|
||||
parser.add_argument("--out", type=Path, default=None,
|
||||
help="Write the JSON snapshot to this file (default: stdout).")
|
||||
parser.add_argument("--timeout", type=float, default=30.0)
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.stock_info and args.group:
|
||||
parser.error("choose either --group or --stock-info, not both")
|
||||
if not args.stock_info and not args.group:
|
||||
parser.error("provide --group or --stock-info")
|
||||
|
||||
payload: dict = {
|
||||
"source": "siamchart",
|
||||
"retrieved_at": _now_utc(),
|
||||
}
|
||||
|
||||
if args.stock_info:
|
||||
info = siamchart.fetch_stock_info(args.stock_info, timeout=args.timeout)
|
||||
payload.update({
|
||||
"mode": "stock-info",
|
||||
"symbol": info.symbol,
|
||||
"url": siamchart._build_stock_info_url(info.symbol),
|
||||
"info": info.to_dict(),
|
||||
})
|
||||
else:
|
||||
rows = siamchart.fetch_financial(args.group, timeout=args.timeout)
|
||||
payload.update({
|
||||
"mode": "group",
|
||||
"group": args.group,
|
||||
"url": siamchart.build_url(args.group),
|
||||
"count": len(rows),
|
||||
"rows": [row.to_dict() for row in rows],
|
||||
})
|
||||
if args.with_info:
|
||||
details = {}
|
||||
for row in rows:
|
||||
try:
|
||||
info = siamchart.fetch_stock_info(row.symbol, timeout=args.timeout)
|
||||
details[row.symbol] = info.to_dict()
|
||||
except siamchart.SiamchartError as exc:
|
||||
details[row.symbol] = {"error": str(exc)}
|
||||
payload["details"] = details
|
||||
payload["details_count"] = len(details)
|
||||
|
||||
text = json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2)
|
||||
if args.out:
|
||||
args.out.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.out.write_text(text, encoding="utf-8")
|
||||
print(f"wrote {payload.get('count', payload.get('symbol', ''))} -> {args.out}")
|
||||
else:
|
||||
print(text)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user