stocks-query/scripts/yahoo-etf-holdings.py

#!/usr/bin/env python3
"""抓 Yahoo 台股 ETF 持股分析頁的前十大持股 / 行業比重 / 資產分佈。"""

from __future__ import annotations

import argparse
import csv
import json
import re
import sys
import urllib.request
from html import unescape

USER_AGENT = "Mozilla/5.0 (OpenClaw; yahoo-etf-holdings)"
BASE_URL = "https://tw.stock.yahoo.com/quote/{symbol}.TW/holding"


def fetch(url: str) -> str:
    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    with urllib.request.urlopen(req, timeout=30) as response:
        return response.read().decode("utf-8", "ignore")


def strip_tags(text: str) -> str:
    return unescape(re.sub(r"<[^>]+>", "", text)).strip()


def extract_date(block: str) -> str:
    m = re.search(r'<time[^>]*datatime="([0-9]{4}/[0-9]{2}/[0-9]{2})"', block)
    return m.group(1) if m else ""


def extract_block(html: str, start_marker: str, end_marker: str | None = None) -> str:
    start = html.find(start_marker)
    if start == -1:
        return ""
    sub = html[start:]
    if end_marker:
        end = sub.find(end_marker)
        if end != -1:
            sub = sub[:end]
    return sub


def parse_rank_list(block: str, value_class: str, limit: int | None = None) -> list[dict[str, str]]:
    pattern = re.compile(
        rf'">(\d+)\.</div>([^<]+)</div><div class="{re.escape(value_class)}">([0-9]+\.[0-9]+%)</div>',
        re.S,
    )
    items = []
    for rank, name, pct in pattern.findall(block):
        items.append({"排名": rank, "名稱": strip_tags(name), "占比": pct})
        if limit and len(items) >= limit:
            break
    return items


def parse_industries(block: str) -> list[dict[str, str]]:
    pattern = re.compile(
        r'<div class="D\(f\) Ai\(c\)">(?:<div[^>]*></div>)?([^<]+)</div><div class="Fx\(n\) Fw\(b\) Pstart\(12px\)">([0-9]+\.[0-9]+%)</div>',
        re.S,
    )
    items = []
    for idx, (name, pct) in enumerate(pattern.findall(block), start=1):
        items.append({"排名": str(idx), "名稱": strip_tags(name), "占比": pct})
    return items


def parse_holding_page(symbol: str) -> dict:
    url = BASE_URL.format(symbol=symbol)
    html = fetch(url)

    asset_block = extract_block(html, '資產分佈</h2>', '前十大持股</h2>')
    top_block = extract_block(html, '前十大持股</h2>', '網友也在看')
    industry_block = extract_block(html, '行業比重</h2>', '持股明細</h2>')

    assets = parse_rank_list(asset_block, 'Fx(n) Pstart(16px)')
    top10 = parse_rank_list(top_block, 'Fx(n) Pstart(16px)', limit=10)
    industries = parse_industries(industry_block)

    return {
        "symbol": symbol,
        "url": url,
        "asset_date": extract_date(asset_block),
        "assets": assets,
        "top10_date": extract_date(top_block),
        "top10": top10,
        "industry_date": extract_date(industry_block),
        "industries": industries,
    }


def print_section(title: str, date: str, items: list[dict[str, str]]) -> None:
    print(f"{title}｜資料時間：{date or '無資料'}")
    for item in items:
        print(f"{item['排名']}. {item['名稱']}：{item['占比']}")
    print()


def write_json(data: dict, out: str | None) -> None:
    text = json.dumps(data, ensure_ascii=False, indent=2)
    if out:
        with open(out, 'w', encoding='utf-8') as f:
            f.write(text)
    else:
        print(text)


def write_csv(data: dict, out: str | None, section: str) -> None:
    mapping = {
        'top': ('top10', 'top10_date'),
        'industry': ('industries', 'industry_date'),
        'asset': ('assets', 'asset_date'),
    }
    rows = []
    if section == 'all':
        for sec in ['top', 'industry', 'asset']:
            key, date_key = mapping[sec]
            rows.extend({"section": sec, "data_date": data[date_key], **item} for item in data[key])
    else:
        key, date_key = mapping[section]
        rows.extend({"section": section, "data_date": data[date_key], **item} for item in data[key])

    fieldnames = ['section', 'data_date', '排名', '名稱', '占比']
    if out:
        f = open(out, 'w', encoding='utf-8-sig', newline='')
    else:
        f = sys.stdout
    try:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)
    finally:
        if out:
            f.close()


def main() -> None:
    parser = argparse.ArgumentParser(description='抓 Yahoo ETF 持股分析頁資料')
    parser.add_argument('symbol', help='ETF 代號，例如 00922、0050')
    parser.add_argument('--format', choices=['text', 'json', 'csv'], default='text')
    parser.add_argument('--section', choices=['all', 'top', 'industry', 'asset'], default='all')
    parser.add_argument('--out', help='輸出檔案路徑')
    args = parser.parse_args()

    data = parse_holding_page(args.symbol)

    if args.format == 'json':
        write_json(data, args.out)
        return
    if args.format == 'csv':
        write_csv(data, args.out, args.section)
        return

    if args.section in ('all', 'top'):
        print_section(f"{args.symbol} 前十大持股", data['top10_date'], data['top10'])
    if args.section in ('all', 'industry'):
        print_section(f"{args.symbol} 行業比重", data['industry_date'], data['industries'])
    if args.section in ('all', 'asset'):
        print_section(f"{args.symbol} 資產分佈", data['asset_date'], data['assets'])


if __name__ == '__main__':
    main()