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>
216 lines
7.3 KiB
Python
216 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
用 Browserless 抓 Yahoo 主力券商進出頁，輸出精簡摘要
|
||
|
||
用法：
|
||
python3 skills/stocks-query/scripts/yahoo-broker-trading-browserless.py 2330
|
||
python3 skills/stocks-query/scripts/yahoo-broker-trading-browserless.py 3481
|
||
|
||
環境變數（可覆蓋）：
|
||
BROWSERLESS_ENDPOINT=http://192.168.42.124:13000
|
||
BROWSERLESS_TOKEN=6R0W53R135510
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
|
||
def load_dotenv() -> None:
|
||
"""載入 .env（優先目前目錄，其次 workspace 根目錄）。"""
|
||
candidates = [
|
||
Path.cwd() / ".env",
|
||
Path(__file__).resolve().parents[3] / ".env",
|
||
]
|
||
for env_path in candidates:
|
||
if not env_path.exists():
|
||
continue
|
||
for raw in env_path.read_text(encoding="utf-8").splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
k, v = line.split("=", 1)
|
||
k = k.strip()
|
||
v = v.strip().strip('"').strip("'")
|
||
if k and k not in os.environ:
|
||
os.environ[k] = v
|
||
break
|
||
|
||
|
||
load_dotenv()
|
||
|
||
ENDPOINT = os.getenv("BROWSERLESS_ENDPOINT")
|
||
TOKEN = os.getenv("BROWSERLESS_TOKEN")
|
||
if not ENDPOINT or not TOKEN:
|
||
raise RuntimeError("缺少必要環境變數：BROWSERLESS_ENDPOINT / BROWSERLESS_TOKEN")
|
||
STOCK_NO = sys.argv[1] if len(sys.argv) > 1 else "2330"
|
||
|
||
|
||
def fetch_html(stock_no: str) -> str:
|
||
url = f"https://tw.stock.yahoo.com/quote/{stock_no}.TW/broker-trading"
|
||
api = f"{ENDPOINT}/content?token={TOKEN}"
|
||
payload = {
|
||
"url": url,
|
||
"gotoOptions": {"waitUntil": "domcontentloaded", "timeout": 60000},
|
||
}
|
||
|
||
req = urllib.request.Request(
|
||
api,
|
||
data=json.dumps(payload).encode("utf-8"),
|
||
headers={"Content-Type": "application/json"},
|
||
method="POST",
|
||
)
|
||
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=90) as resp:
|
||
text = resp.read().decode("utf-8", errors="replace")
|
||
status = resp.getcode()
|
||
except urllib.error.HTTPError as e:
|
||
body = e.read().decode("utf-8", errors="replace") if hasattr(e, "read") else str(e)
|
||
raise RuntimeError(f"Browserless HTTP {e.code}: {body[:200]}") from e
|
||
|
||
if status != 200:
|
||
raise RuntimeError(f"Browserless HTTP {status}: {text[:200]}")
|
||
if "TimeoutError" in text:
|
||
raise RuntimeError("Browserless 導航逾時（TimeoutError）")
|
||
return text
|
||
|
||
|
||
def pick(html: str, pattern: str, default: str = "-") -> str:
|
||
m = re.search(pattern, html, flags=re.S)
|
||
return m.group(1) if m else default
|
||
|
||
|
||
def parse_brokers(html: str, stock_no: str) -> dict:
|
||
name = pick(html, r'property="og:title" content="([^"(]+)\([0-9]+\.TW\)', "").strip()
|
||
if not name:
|
||
name = pick(html, r"<title>([^<(]+)\([0-9]+\.TW\)", "-").strip()
|
||
|
||
# 嘗試從嵌入 JSON 中抓券商進出資料
|
||
buy_brokers = []
|
||
sell_brokers = []
|
||
|
||
# 方法1：從 brokerBuySellData / topBrokers JSON 結構抓取
|
||
json_match = re.search(r'"brokerBuySellData"\s*:\s*(\{[\s\S]*?\})\s*[,}]', html)
|
||
if json_match:
|
||
try:
|
||
raw = json.loads(json_match.group(1))
|
||
for item in raw.get("buy", raw.get("buyBrokers", []))[:10]:
|
||
buy_brokers.append({
|
||
"name": item.get("brokerName", item.get("name", "-")),
|
||
"buy": str(item.get("buy", item.get("buyVolume", 0))),
|
||
"sell": str(item.get("sell", item.get("sellVolume", 0))),
|
||
"diff": str(item.get("diff", item.get("net", 0))),
|
||
})
|
||
for item in raw.get("sell", raw.get("sellBrokers", []))[:10]:
|
||
sell_brokers.append({
|
||
"name": item.get("brokerName", item.get("name", "-")),
|
||
"buy": str(item.get("buy", item.get("buyVolume", 0))),
|
||
"sell": str(item.get("sell", item.get("sellVolume", 0))),
|
||
"diff": str(item.get("diff", item.get("net", 0))),
|
||
})
|
||
except (json.JSONDecodeError, KeyError):
|
||
pass
|
||
|
||
# 方法2：從 HTML table 結構抓取
|
||
if not buy_brokers and not sell_brokers:
|
||
# 買超券商區塊
|
||
buy_section = ""
|
||
sell_section = ""
|
||
buy_start = html.find("買超券商")
|
||
sell_start = html.find("賣超券商")
|
||
|
||
if buy_start != -1 and sell_start != -1:
|
||
buy_section = html[buy_start:sell_start]
|
||
sell_section = html[sell_start:sell_start + 50000]
|
||
elif buy_start != -1:
|
||
buy_section = html[buy_start:buy_start + 50000]
|
||
|
||
broker_row_pat = re.compile(
|
||
r'<li[^>]*>.*?'
|
||
r'(?:券商|broker)[^<]*</[^>]+>.*?'
|
||
r'([^<]{2,20})</(?:span|div|td)>'
|
||
r'.*?([\d,]+).*?([\d,]+).*?([+-]?[\d,]+)',
|
||
re.S,
|
||
)
|
||
|
||
# 更通用的抓法：連續出現的券商名+數字 pattern
|
||
simple_pat = re.compile(
|
||
r'>([^\d<>]{2,12}(?:證券|期貨|投顧|銀行|金控)?[^<>]{0,6})</(?:span|div|td|a)>'
|
||
r'[^<]*<[^>]+>([\d,]+)</[^>]+>'
|
||
r'[^<]*<[^>]+>([\d,]+)</[^>]+>'
|
||
r'[^<]*<[^>]+>([+-]?[\d,]+)</[^>]+>',
|
||
re.S,
|
||
)
|
||
|
||
for section, target_list in [(buy_section, buy_brokers), (sell_section, sell_brokers)]:
|
||
for m in simple_pat.finditer(section):
|
||
broker_name = m.group(1).strip()
|
||
if len(broker_name) < 2 or broker_name.isdigit():
|
||
continue
|
||
target_list.append({
|
||
"name": broker_name,
|
||
"buy": m.group(2).replace(",", ""),
|
||
"sell": m.group(3).replace(",", ""),
|
||
"diff": m.group(4).replace(",", ""),
|
||
})
|
||
if len(target_list) >= 10:
|
||
break
|
||
|
||
# 日期
|
||
date = pick(html, r"資料日期[：:]\s*([0-9/\-]+)")
|
||
if date == "-":
|
||
date = pick(html, r'"formattedDate"\s*:\s*"([^"]+)"')
|
||
|
||
return {
|
||
"name": name,
|
||
"stock_no": stock_no,
|
||
"date": date,
|
||
"buy_brokers": buy_brokers[:10],
|
||
"sell_brokers": sell_brokers[:10],
|
||
}
|
||
|
||
|
||
def sign(v: str) -> str:
|
||
if not v or v == "-":
|
||
return "-"
|
||
try:
|
||
n = int(v.replace(",", ""))
|
||
return f"+{n:,}" if n >= 0 else f"{n:,}"
|
||
except ValueError:
|
||
return v
|
||
|
||
|
||
def print_summary(d: dict) -> None:
|
||
print(f"{d['name']}（{d['stock_no']}）｜主力券商進出")
|
||
if d["date"] != "-":
|
||
print(f"• 日期：{d['date']}")
|
||
|
||
if d["buy_brokers"]:
|
||
print("• 買超前幾大券商：")
|
||
for b in d["buy_brokers"][:5]:
|
||
print(f" - {b['name']}：買 {b['buy']}／賣 {b['sell']}／差 {sign(b['diff'])} 張")
|
||
else:
|
||
print("• 買超券商：無資料")
|
||
|
||
if d["sell_brokers"]:
|
||
print("• 賣超前幾大券商：")
|
||
for b in d["sell_brokers"][:5]:
|
||
print(f" - {b['name']}：買 {b['buy']}／賣 {b['sell']}／差 {sign(b['diff'])} 張")
|
||
else:
|
||
print("• 賣超券商：無資料")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
html = fetch_html(STOCK_NO)
|
||
data = parse_brokers(html, STOCK_NO)
|
||
print_summary(data)
|
||
except Exception as e:
|
||
print(f"取得失敗：{e}", file=sys.stderr)
|
||
sys.exit(1)
|