stocks-query/scripts/twse-day-trading.py

#!/usr/bin/env python3
"""
查詢 TWSE 個股當日沖銷交易統計

用法：
  python3 skills/stocks-query/scripts/twse-day-trading.py 2330
  python3 skills/stocks-query/scripts/twse-day-trading.py 3481

資料來源：證交所 TWSE 公開 API（當日沖銷交易標的及成交量值）
"""

from __future__ import annotations

import json
import sys
import urllib.request
from datetime import datetime, timedelta


def fetch_day_trading(stock_no: str, date_str: str | None = None) -> dict:
    """從 TWSE 抓當沖資料，自動往前回溯最多 10 天找到有資料的日期。"""
    if date_str:
        dates_to_try = [date_str]
    else:
        today = datetime.now()
        dates_to_try = [(today - timedelta(days=i)).strftime("%Y%m%d") for i in range(10)]

    for d in dates_to_try:
        url = (
            f"https://www.twse.com.tw/rwd/zh/dayTrading/dayTrading"
            f"?date={d}&selectType=ALL&response=json"
        )
        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                data = json.loads(resp.read().decode("utf-8", errors="replace"))
        except Exception:
            continue

        if data.get("stat") != "OK":
            continue

        tables = data.get("tables", [])
        if not tables:
            # 有些 API 回傳格式不同，直接看 data 欄位
            fields = data.get("fields", [])
            rows = data.get("data", [])
            if fields and rows:
                tables = [{"fields": fields, "data": rows}]
            else:
                continue

        # 找含「代號」欄位的表
        target_table = None
        for tbl in tables:
            fields = tbl.get("fields", [])
            field_str = "".join(fields)
            if "代號" in field_str or "證券代號" in field_str:
                target_table = tbl
                break

        if not target_table:
            # 嘗試用第一個 table
            if tables:
                target_table = tables[0]
            else:
                continue

        fields = target_table["fields"]

        # 找代號欄位的 index
        code_idx = None
        for i, f in enumerate(fields):
            if "代號" in f:
                code_idx = i
                break

        if code_idx is None:
            continue

        for row in target_table.get("data", []):
            code = row[code_idx].strip()
            if code == stock_no:
                result = {"date": d}
                for i, field in enumerate(fields):
                    result[field.strip()] = row[i].strip().replace(",", "") if i < len(row) else "-"
                return result

    raise RuntimeError(f"找不到 {stock_no} 的當沖資料（已回溯 10 天）")


def print_summary(stock_no: str, d: dict) -> None:
    date_fmt = f"{d['date'][:4]}/{d['date'][4:6]}/{d['date'][6:]}"
    name = d.get("證券名稱", d.get("名稱", stock_no))
    print(f"{name}（{stock_no}）｜當沖交易摘要")
    print(f"• 日期：{date_fmt}")

    # 當沖買賣量
    day_buy = d.get("當日沖銷交易買進成交股數", d.get("買進成交股數", "-"))
    day_sell = d.get("當日沖銷交易賣出成交股數", d.get("賣出成交股數", "-"))
    day_amount = d.get("當日沖銷交易成交金額", d.get("成交金額", "-"))

    print(f"• 當沖買進股數：{day_buy}")
    print(f"• 當沖賣出股數：{day_sell}")

    if day_amount != "-":
        print(f"• 當沖成交金額：{day_amount}")

    # 當沖比例
    ratio = d.get("當日沖銷交易佔該日成交量比率", d.get("佔成交量比率", "-"))
    if ratio == "-":
        # 嘗試自己算
        try:
            total_vol = float(d.get("成交股數", "0"))
            day_vol = float(day_buy) if day_buy != "-" else 0
            if total_vol > 0 and day_vol > 0:
                ratio = f"{(day_vol / total_vol) * 100:.2f}%"
        except (ValueError, ZeroDivisionError):
            pass

    print(f"• 當沖佔比：{ratio}")


if __name__ == "__main__":
    stock_no = sys.argv[1] if len(sys.argv) > 1 else "2330"
    try:
        data = fetch_day_trading(stock_no)
        print_summary(stock_no, data)
    except Exception as e:
        print(f"取得失敗：{e}", file=sys.stderr)
        sys.exit(1)