Timmy
7d72aed984
Recover 7 existing scripts from remote repo and rebuild 5 missing ones: - twse-margin-trading.py (融資融券) - twse-day-trading.py (當沖交易) - twse-foreign-holdings.py (外資持股) - yahoo-dividend-browserless.py (股利資訊) - yahoo-broker-trading-browserless.py (主力券商進出) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
#!/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)
|