restore and rebuild all stock query scripts

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>
This commit is contained in:
2026-04-10 10:07:31 +08:00
commit 7d72aed984
16 changed files with 2153 additions and 0 deletions

131
scripts/twse-day-trading.py Normal file
View File

@@ -0,0 +1,131 @@
#!/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)

View File

@@ -0,0 +1,128 @@
#!/usr/bin/env python3
"""
查詢 TWSE 外資持股比例統計
用法:
python3 skills/stocks-query/scripts/twse-foreign-holdings.py 2330
python3 skills/stocks-query/scripts/twse-foreign-holdings.py 3481
資料來源:證交所 TWSE 公開 API外資及陸資持股統計
"""
from __future__ import annotations
import json
import sys
import urllib.request
from datetime import datetime, timedelta
def fetch_foreign_holdings(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/fund/MI_QFIIS"
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:
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:
target_table = tbl
break
if not target_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}")
# 外資持股
shares = d.get("外資及陸資持股股數", d.get("外陸資持股股數", "-"))
pct = d.get("外資及陸資持股比率", d.get("外陸資持股比率", d.get("持股比率", "-")))
available = d.get("外資及陸資尚可投資股數", d.get("尚可投資股數", "-"))
upper_limit = d.get("外資及陸資共用法令投資上限比率", d.get("投資上限比率", "-"))
total_issued = d.get("發行股數", "-")
# 外資買賣超相關
foreign_buy = d.get("外資及陸資買進股數", d.get("買進股數", "-"))
foreign_sell = d.get("外資及陸資賣出股數", d.get("賣出股數", "-"))
print(f"• 外資持股股數:{shares}")
print(f"• 外資持股比率:{pct}%") if pct != "-" and "%" not in str(pct) else print(f"• 外資持股比率:{pct}")
if total_issued != "-":
print(f"• 發行股數:{total_issued}")
if available != "-":
print(f"• 尚可投資股數:{available}")
if upper_limit != "-":
print(f"• 投資上限比率:{upper_limit}")
if foreign_buy != "-":
print(f"• 外資買進股數:{foreign_buy}")
if foreign_sell != "-":
print(f"• 外資賣出股數:{foreign_sell}")
if __name__ == "__main__":
stock_no = sys.argv[1] if len(sys.argv) > 1 else "2330"
try:
data = fetch_foreign_holdings(stock_no)
print_summary(stock_no, data)
except Exception as e:
print(f"取得失敗:{e}", file=sys.stderr)
sys.exit(1)

View File

@@ -0,0 +1,120 @@
#!/usr/bin/env python3
"""
查詢 TWSE 個股融資融券餘額(信用交易統計)
用法:
python3 skills/stocks-query/scripts/twse-margin-trading.py 2330
python3 skills/stocks-query/scripts/twse-margin-trading.py 3481
資料來源:證交所 TWSE 公開 API
"""
from __future__ import annotations
import json
import sys
import urllib.request
from datetime import datetime, timedelta
def fetch_margin(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/marginTrading/MI_MARGN"
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:
continue
# 找到個股那張表title 含「融資融券」或欄位含「股票代號」)
target_table = None
for tbl in tables:
fields = tbl.get("fields", [])
if "股票代號" in fields:
target_table = tbl
break
if not target_table:
continue
fields = target_table["fields"]
for row in target_table.get("data", []):
code = row[fields.index("股票代號")].strip()
if code == stock_no:
result = {"date": d}
for i, field in enumerate(fields):
result[field.strip()] = row[i].strip().replace(",", "")
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("股票名稱", stock_no)
print(f"{name}{stock_no})|融資融券摘要")
print(f"• 日期:{date_fmt}")
# 融資
margin_buy = d.get("融資買進", "-")
margin_sell = d.get("融資賣出", "-")
margin_cash = d.get("融資現償", "-")
margin_balance = d.get("融資餘額", d.get("融資今日餘額", "-"))
margin_limit = d.get("融資限額", "-")
print(f"• 融資買進:{margin_buy}")
print(f"• 融資賣出:{margin_sell}")
print(f"• 融資現償:{margin_cash}")
print(f"• 融資餘額:{margin_balance}")
if margin_limit != "-":
print(f"• 融資限額:{margin_limit}")
# 融券
short_sell = d.get("融券賣出", "-")
short_buy = d.get("融券買進", "-")
short_cash = d.get("融券現償", "-")
short_balance = d.get("融券餘額", d.get("融券今日餘額", "-"))
short_limit = d.get("融券限額", "-")
print(f"• 融券賣出:{short_sell}")
print(f"• 融券買進:{short_buy}")
print(f"• 融券現償:{short_cash}")
print(f"• 融券餘額:{short_balance}")
if short_limit != "-":
print(f"• 融券限額:{short_limit}")
# 券資比
try:
mb = float(margin_balance) if margin_balance != "-" else 0
sb = float(short_balance) if short_balance != "-" else 0
if mb > 0:
ratio = (sb / mb) * 100
print(f"• 券資比:{ratio:.2f}%")
except (ValueError, ZeroDivisionError):
pass
if __name__ == "__main__":
stock_no = sys.argv[1] if len(sys.argv) > 1 else "2330"
try:
data = fetch_margin(stock_no)
print_summary(stock_no, data)
except Exception as e:
print(f"取得失敗:{e}", file=sys.stderr)
sys.exit(1)

View File

@@ -0,0 +1,215 @@
#!/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)

View File

@@ -0,0 +1,190 @@
#!/usr/bin/env python3
"""
用 Browserless 抓 Yahoo 股利政策頁,輸出精簡摘要
用法:
python3 skills/stocks-query/scripts/yahoo-dividend-browserless.py 2330
python3 skills/stocks-query/scripts/yahoo-dividend-browserless.py 0050
環境變數(可覆蓋):
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/dividend"
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_dividend(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()
# 殖利率
dividend_yield = pick(html, r">現金殖利率</span><span[^>]*>([\d.]+%?)")
# 年度股利統計(從 JSON-like 嵌入或 table 抓取)
# 嘗試從頁面的 script data 抓股利歷史
rows = []
# 方法1抓 dividendData JSON
json_match = re.search(r'"dividendData"\s*:\s*(\[[\s\S]*?\])\s*[,}]', html)
if json_match:
try:
raw_data = json.loads(json_match.group(1))
for item in raw_data[:5]: # 最近 5 年
year = item.get("year", item.get("formattedYear", "-"))
cash = item.get("cashDividend", item.get("cash", 0))
stock = item.get("stockDividend", item.get("stock", 0))
total = item.get("totalDividend", 0)
if total == 0:
try:
total = float(cash) + float(stock)
except (ValueError, TypeError):
total = "-"
rows.append({
"year": str(year),
"cash": str(cash),
"stock": str(stock),
"total": str(total),
})
except (json.JSONDecodeError, KeyError):
pass
# 方法2用 regex 抓 table rows
if not rows:
# 常見結構:年度、現金股利、股票股利、合計
row_pattern = re.compile(
r'<tr[^>]*>.*?<td[^>]*>(1\d{2,3})</td>'
r'.*?<td[^>]*>([\d.]+)</td>'
r'.*?<td[^>]*>([\d.]+)</td>'
r'.*?<td[^>]*>([\d.]+)</td>',
re.S,
)
for m in row_pattern.finditer(html):
rows.append({
"year": m.group(1),
"cash": m.group(2),
"stock": m.group(3),
"total": m.group(4),
})
if len(rows) >= 5:
break
# 方法3抓除息日相關
ex_date = pick(html, r">除息日</span><span[^>]*>([^<]+)")
pay_date = pick(html, r">發放日</span><span[^>]*>([^<]+)")
cash_div = pick(html, r">現金股利</span><span[^>]*>([\d.]+)")
stock_div = pick(html, r">股票股利</span><span[^>]*>([\d.]+)")
return {
"name": name,
"stock_no": stock_no,
"dividend_yield": dividend_yield,
"cash_div": cash_div,
"stock_div": stock_div,
"ex_date": ex_date,
"pay_date": pay_date,
"history": rows[:5],
}
def print_summary(d: dict) -> None:
print(f"{d['name']}{d['stock_no']})|股利摘要")
if d["dividend_yield"] != "-":
print(f"• 現金殖利率:{d['dividend_yield']}")
if d["cash_div"] != "-":
print(f"• 現金股利:{d['cash_div']}")
if d["stock_div"] != "-":
print(f"• 股票股利:{d['stock_div']}")
if d["ex_date"] != "-":
print(f"• 除息日:{d['ex_date']}")
if d["pay_date"] != "-":
print(f"• 發放日:{d['pay_date']}")
if d["history"]:
print("• 近年股利:")
for r in d["history"]:
print(f" - {r['year']}年:現金 {r['cash']}/股票 {r['stock']}/合計 {r['total']}")
else:
print("• 近年股利:無資料")
if __name__ == "__main__":
try:
html = fetch_html(STOCK_NO)
data = parse_dividend(html, STOCK_NO)
print_summary(data)
except Exception as e:
print(f"取得失敗:{e}", file=sys.stderr)
sys.exit(1)

View File

@@ -0,0 +1,164 @@
#!/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()

View File

@@ -0,0 +1,158 @@
#!/usr/bin/env python3
"""
用 Browserless 抓 Yahoo 三大法人頁,輸出精簡摘要(代號可變)
用法:
python3 stocks/yahoo-institutional-browserless.py 3481
python3 stocks/yahoo-institutional-browserless.py 2330
"""
import json
import os
from pathlib import Path
import re
import sys
import subprocess
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 "3481"
def fetch_html(stock_no: str) -> str:
url = f"https://tw.stock.yahoo.com/quote/{stock_no}.TW/institutional-trading"
api = f"{ENDPOINT}/content?token={TOKEN}"
payload = json.dumps(
{
"url": url,
"gotoOptions": {"waitUntil": "domcontentloaded", "timeout": 60000},
},
ensure_ascii=False,
)
cmd = [
"curl",
"-sS",
"--max-time",
"90",
"-X",
"POST",
api,
"-H",
"Content-Type: application/json",
"-d",
payload,
]
try:
text = subprocess.check_output(cmd, text=True, timeout=95)
except subprocess.CalledProcessError as e:
raise RuntimeError(f"Browserless 呼叫失敗: {e}") from e
except subprocess.TimeoutExpired as e:
raise RuntimeError("Browserless 呼叫逾時") from e
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_inst(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()
# 抓 summary data避免大範圍 regex 回溯,先縮小搜尋範圍)
anchor = html.find('institutionBuySellSummaryData')
chunk = html[anchor: anchor + 250000] if anchor != -1 else html
rows = []
obj_pat = re.compile(r'\{[^{}]*"formattedDate":"[^"]+"[^{}]*\}')
for om in obj_pat.finditer(chunk):
s = om.group(0)
foreign_m = re.search(r'"foreignDiffVolK":(-?\d+)', s)
trust_m = re.search(r'"investmentTrustDiffVolK":(-?\d+)', s)
dealer_m = re.search(r'"dealerDiffVolK":(-?\d+)', s)
total_m = re.search(r'"totalDiffVolK":(-?\d+)', s)
date_m = re.search(r'"formattedDate":"([^"]+)"', s)
if foreign_m and trust_m and dealer_m and total_m and date_m:
rows.append((foreign_m.group(1), trust_m.group(1), dealer_m.group(1), total_m.group(1), date_m.group(1)))
if not rows:
raise RuntimeError("抓不到三大法人資料(頁面結構可能變更)")
def normalize_date(ds: str) -> str:
return ds.replace("\\u002F", "/")
def date_key(ds: str):
s = normalize_date(ds)
try:
y, m, d = s.split("/")
return int(y), int(m), int(d)
except Exception:
return (0, 0, 0)
foreign, trust, dealer, total, date = max(rows, key=lambda r: date_key(r[4]))
date = normalize_date(date)
return {
"name": name,
"stock_no": stock_no,
"date": date,
"foreign": int(foreign),
"trust": int(trust),
"dealer": int(dealer),
"total": int(total),
}
def sign(v: int) -> str:
return f"+{v:,}" if v >= 0 else f"{v:,}"
def print_summary(d: dict) -> None:
print(f"{d['name']}{d['stock_no']})|法人買賣超摘要")
print(f"• 日期:{d['date']}")
print(f"• 外資:{sign(d['foreign'])}")
print(f"• 投信:{sign(d['trust'])}")
print(f"• 自營商:{sign(d['dealer'])}")
print(f"• 三大法人合計:{sign(d['total'])}")
if __name__ == "__main__":
try:
html = fetch_html(STOCK_NO)
data = parse_inst(html, STOCK_NO)
print_summary(data)
except Exception as e:
print(f"取得失敗:{e}", file=sys.stderr)
sys.exit(1)

View File

@@ -0,0 +1,82 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import json
import sys
import urllib.parse
import urllib.request
def fetch_closes(symbol: str, range_: str = "6mo", interval: str = "1d") -> list[float]:
base = f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}"
params = {"range": range_, "interval": interval}
url = f"{base}?{urllib.parse.urlencode(params)}"
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=30) as resp:
data = json.loads(resp.read().decode("utf-8", errors="replace"))
result = data.get("chart", {}).get("result", [])
if not result:
return []
quote = result[0].get("indicators", {}).get("quote", [])
if not quote:
return []
closes = quote[0].get("close", [])
return [float(x) for x in closes if x is not None]
def sma(values: list[float], window: int) -> float:
return sum(values[-window:]) / window
def main() -> None:
symbol = sys.argv[1] if len(sys.argv) > 1 else "00922.TW"
closes = fetch_closes(symbol)
if len(closes) < 60:
print(f"{symbol} 資料不足:至少需要 60 筆日收盤,當前 {len(closes)}")
return
price = closes[-1]
ma5 = sma(closes, 5)
ma10 = sma(closes, 10)
ma20 = sma(closes, 20)
ma60 = sma(closes, 60)
bias20 = ((price - ma20) / ma20) * 100
print(f"{symbol} 均線摘要")
print(f"- 收盤:{price:.2f}")
print(f"- MA5 {ma5:.2f}")
print(f"- MA10{ma10:.2f}")
print(f"- MA20{ma20:.2f}")
print(f"- MA60{ma60:.2f}")
print(f"- 20日乖離{bias20:.2f}%")
# 簡單判讀
trend = []
if price > ma20:
trend.append("站上月線")
else:
trend.append("跌破月線")
if ma20 > ma60:
trend.append("中期偏多")
else:
trend.append("中期偏弱")
if ma5 > ma20:
trend.append("短線強於月線")
else:
trend.append("短線弱於月線")
print(f"- 判讀:{''.join(trend)}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,91 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import json
import sys
import urllib.parse
import urllib.request
def fetch_closes(symbol: str, range_: str = "6mo", interval: str = "1d") -> list[float]:
base = f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}"
params = {"range": range_, "interval": interval}
url = f"{base}?{urllib.parse.urlencode(params)}"
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=30) as resp:
data = json.loads(resp.read().decode("utf-8", errors="replace"))
result = data.get("chart", {}).get("result", [])
if not result:
return []
quote = result[0].get("indicators", {}).get("quote", [])
if not quote:
return []
closes = quote[0].get("close", [])
return [float(x) for x in closes if x is not None]
def ema(values: list[float], period: int) -> list[float]:
if not values:
return []
k = 2 / (period + 1)
out = [values[0]]
for v in values[1:]:
out.append(v * k + out[-1] * (1 - k))
return out
def macd(closes: list[float]) -> tuple[list[float], list[float], list[float]]:
ema12 = ema(closes, 12)
ema26 = ema(closes, 26)
dif = [a - b for a, b in zip(ema12, ema26)]
dea = ema(dif, 9)
hist = [d - s for d, s in zip(dif, dea)]
return dif, dea, hist
def main() -> None:
symbol = sys.argv[1] if len(sys.argv) > 1 else "00922.TW"
closes = fetch_closes(symbol)
if len(closes) < 35:
print(f"{symbol} 資料不足,無法判斷 MACD至少 35 筆,當前 {len(closes)}")
return
dif, dea, hist = macd(closes)
d0, d1 = dif[-1], dif[-2]
s0, s1 = dea[-1], dea[-2]
h0, h1 = hist[-1], hist[-2]
golden_cross = d1 <= s1 and d0 > s0
death_cross = d1 >= s1 and d0 < s0
low_zone = d0 < 0 and s0 < 0
if golden_cross and low_zone:
signal = "✅ 低位金叉:偏多訊號"
elif golden_cross:
signal = "✅ 金叉:短線轉強"
elif death_cross:
signal = "⚠️ 死叉:短線轉弱"
elif h0 > h1 and h0 > 0:
signal = "📈 多頭延續(柱體擴大)"
elif h0 < h1 and h0 < 0:
signal = "📉 空頭延續(柱體擴大)"
else:
signal = "👀 MACD 震盪中,先觀察"
print(f"{symbol} MACD 摘要")
print(f"- DIF{d0:.4f}")
print(f"- DEA{s0:.4f}")
print(f"- HIST{h0:.4f}")
print(f"- 訊號:{signal}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1 @@
yahoo-market-browserless.py

View File

@@ -0,0 +1,168 @@
#!/usr/bin/env python3
"""
用 Browserless 抓 Yahoo 台股大盤(加權指數)即時行情。
用法:
python3 stocks/yahoo-market-browserless.py
python3 stocks/yahoo-market-browserless.py ^TWII
環境變數(可覆蓋):
BROWSERLESS_ENDPOINT=http://192.168.42.124:13000
BROWSERLESS_TOKEN=6R0W53R135510
"""
import os
import re
import sys
import json
import urllib.request
import urllib.error
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")
SYMBOL = sys.argv[1] if len(sys.argv) > 1 else "^TWII"
def fetch_html(symbol: str) -> str:
url = f"https://tw.stock.yahoo.com/quote/{symbol}"
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 to_float(num: str) -> float | None:
try:
if not num or num == "-":
return None
return float(num.replace(",", ""))
except Exception:
return None
def parse_market(html: str, symbol: str) -> dict:
# 名稱:優先取 title加權指數(^TWII) 走勢圖 - Yahoo股市
name = pick(html, r"<title>([^<(]+)\([^)]*\)\s*走勢圖")
if name == "-":
name = pick(html, r'property="og:title" content="([^<(]+)\([^)]*\)')
price = pick(html, r"Fz\(32px\)[^>]*>([\d,]+\.?\d*)<")
# 漲跌值常出現在百分比前一個 span且前面有紅綠三角形 span
updown = pick(
html,
r'border-width:[^"]+\"></span>([+\-]?[\d,]+\.?\d*)</span><span[^>]*>\([+\-]?[0-9]+\.[0-9]+%\)',
)
pct = pick(html, r"\(([+\-]?[0-9]+\.[0-9]+%)\)</span>")
open_price = pick(html, r">開盤</span><span[^>]*>([\d,]+\.?\d*)")
high = pick(html, r">最高</span><span[^>]*>([\d,]+\.?\d*)")
low = pick(html, r">最低</span><span[^>]*>([\d,]+\.?\d*)")
prev = pick(html, r">昨收</span><span[^>]*>([\d,]+\.?\d*)")
amount_100m = pick(html, r">成交金額\(億\)</span><span[^>]*>([\d,]+\.?\d*)")
time_raw = pick(html, r"(?:開盤|收盤) \| ([0-9/ :]+) 更新")
time_fmt = "-" if time_raw == "-" else f"{time_raw.replace('/', '-')}(台北)"
# 以指數與昨收重算漲跌,避免頁面符號誤判
p = to_float(price)
pv = to_float(prev)
if p is not None and pv is not None:
diff = p - pv
updown = f"{diff:+.2f}"
if pv != 0:
pct = f"{(diff / pv) * 100:+.2f}%"
return {
"name": name,
"symbol": symbol,
"price": price,
"updown": updown,
"pct": pct,
"open": open_price,
"high": high,
"low": low,
"prev": prev,
"amount_100m": amount_100m,
"time": time_fmt,
}
def sign_num(v: str) -> str:
if v in ("-", ""):
return "-"
return v if v.startswith("-") or v.startswith("+") else f"+{v}"
def print_summary(q: dict) -> None:
print(f"{q['name']}{q['symbol']}")
print(f"• 指數:{q['price']}")
print(f"• 漲跌:{sign_num(q['updown'])}{sign_num(q['pct'])}")
print(f"• 開盤/最高/最低:{q['open']}{q['high']}{q['low']}")
print(f"• 昨收:{q['prev']}")
print(f"• 成交金額:{q['amount_100m']}")
print(f"• 資料時間:{q['time']}")
if __name__ == "__main__":
try:
html = fetch_html(SYMBOL)
market = parse_market(html, SYMBOL)
print_summary(market)
except Exception as e:
print(f"取得失敗:{e}", file=sys.stderr)
sys.exit(1)

View File

@@ -0,0 +1,192 @@
#!/usr/bin/env python3
"""
用 Browserless 抓 Yahoo 股市頁面,輸出精簡報價摘要(省 token 版)
用法:
python3 stocks/yahoo-quote-browserless.py 3481
python3 stocks/yahoo-quote-browserless.py 2330
環境變數(可覆蓋):
BROWSERLESS_ENDPOINT=http://192.168.42.124:13000
BROWSERLESS_TOKEN=6R0W53R135510
"""
import os
import re
import sys
import json
import urllib.request
import urllib.error
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 "3481"
def fetch_html(stock_no: str) -> str:
url = f"https://tw.stock.yahoo.com/quote/{stock_no}.TW"
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 to_float(num: str) -> float | None:
try:
if not num or num == "-":
return None
return float(num.replace(",", ""))
except Exception:
return None
def parse_quote(html: str, stock_no: str) -> dict:
# 名稱容錯:有些股票 h1 結構會不一樣
name = pick(html, r"<h1[^>]*>([^<(]+)", "-").strip()
if not name or name in ("-", "Yahoo股市"):
name = pick(html, r'property="og:title" content="([^"(]+)\([0-9]+\.TW\)', "").strip()
if not name or name in ("-", "Yahoo股市"):
name = pick(html, r"<title>([^<(]+)\([0-9]+\.TW\)", "-").strip()
# 價格容錯:支援千分位與不同 class 變化
num_pat = r"([\d,]+\.?\d*)"
price = pick(html, rf"Fz\(32px\)[^>]*>{num_pat}<")
# 漲跌先抓頁面值,後續再用(現價-昨收)覆寫,避免符號抓錯
updown = pick(html, rf"Fz\(20px\)[^>]*>([\d\.\+\-]+)<", "-")
if updown == "-":
updown = pick(html, rf"Fz\(20px\)[^>]*>[\s\S]*?</span>([\d\.\+\-]+)<", "0.00")
pct = pick(html, r"\(([+-]?[0-9]+\.[0-9]+%)\)</span>", "0.00%")
open_price = pick(html, rf">開盤</span><span[^>]*>{num_pat}")
high = pick(html, rf">最高</span><span[^>]*>{num_pat}")
low = pick(html, rf">最低</span><span[^>]*>{num_pat}")
prev = pick(html, rf">昨收</span><span[^>]*>{num_pat}")
avg = pick(html, rf">均價</span><span[^>]*>{num_pat}")
volume_lots = pick(
html,
r'Mb\(4px\)">([0-9,]+)</span><span class="Fz\(12px\) C\(\$c-icon\)">成交量',
"0",
)
volume_shares = "{:,.0f}".format(float(volume_lots.replace(",", "")) * 1000)
bids = re.findall(r"Fw\(n\)[^>]*>([\d,]+\.?\d*)</span>", html)
best_bid = bids[0] if len(bids) >= 1 else "-"
best_ask = bids[5] if len(bids) >= 6 else "-"
time_raw = pick(html, r"開盤 \| ([0-9/ :]+) 更新")
time_fmt = "-" if time_raw == "-" else f"{time_raw.replace('/', '-')}(台北)"
market = "未知" if time_raw == "-" else "開盤中"
# 以現價與昨收重算漲跌,避免頁面箭頭造成正負號誤判
p = to_float(price)
pv = to_float(prev)
if p is not None and pv is not None:
diff = p - pv
updown = f"{diff:+.2f}"
if pv != 0:
pct = f"{(diff / pv) * 100:+.2f}%"
return {
"name": name,
"stock_no": stock_no,
"price": price,
"updown": updown,
"pct": pct,
"open": open_price,
"high": high,
"low": low,
"prev": prev,
"volume_lots": volume_lots,
"volume_shares": volume_shares,
"avg": avg,
"best_bid": best_bid,
"best_ask": best_ask,
"market": market,
"time": time_fmt,
}
def sign_num(v: str) -> str:
if v in ("-", ""):
return "-"
return v if v.startswith("-") or v.startswith("+") else f"+{v}"
def print_summary(q: dict) -> None:
print(f"{q['name']}{q['stock_no']}")
print(f"• 現價:{q['price']}")
print(f"• 漲跌:{sign_num(q['updown'])}{sign_num(q['pct'])}")
print(f"• 開盤/最高/最低:{q['open']}{q['high']}{q['low']}")
print(f"• 昨收:{q['prev']}")
print(f"• 成交量:{q['volume_shares']}Yahoo 顯示 {q['volume_lots']} 張,換算股數)")
print(f"• 均價:{q['avg']}")
print(f"• 最佳買賣:{q['best_bid']}{q['best_ask']}")
print(f"• 市場狀態:{q['market']}")
print(f"• 資料時間:{q['time']}")
if __name__ == "__main__":
try:
html = fetch_html(STOCK_NO)
quote = parse_quote(html, STOCK_NO)
print_summary(quote)
except Exception as e:
print(f"取得失敗:{e}", file=sys.stderr)
sys.exit(1)

View File

@@ -0,0 +1,303 @@
#!/usr/bin/env python3
"""
用 Browserless 透過 Yahoo Chart API 抓技術面摘要(代號可變)
輸出內容:
- 近 5 個交易日 K 線(開高低收)
- RSI14
- ATR14
- KD9,3,3
- 布林通道20,2
用法:
python3 skills/stocks-query/scripts/yahoo-technical-browserless.py 3481
python3 skills/stocks-query/scripts/yahoo-technical-browserless.py 2330
"""
import datetime as dt
import html as html_lib
import json
import math
import os
from pathlib import Path
import re
import subprocess
import sys
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 "3481"
def fetch_chart_json(stock_no: str) -> dict:
api_url = f"https://query1.finance.yahoo.com/v8/finance/chart/{stock_no}.TW?range=3mo&interval=1d"
browserless_api = f"{ENDPOINT}/content?token={TOKEN}"
payload = json.dumps(
{
"url": api_url,
"gotoOptions": {"waitUntil": "domcontentloaded", "timeout": 60000},
},
ensure_ascii=False,
)
cmd = [
"curl",
"-sS",
"--max-time",
"90",
"-X",
"POST",
browserless_api,
"-H",
"Content-Type: application/json",
"-d",
payload,
]
try:
raw = subprocess.check_output(cmd, text=True)
except subprocess.CalledProcessError as e:
raise RuntimeError(f"Browserless 呼叫失敗: {e}") from e
except subprocess.TimeoutExpired as e:
raise RuntimeError("Browserless 呼叫逾時") from e
m = re.search(r"<pre>([\s\S]+)</pre>", raw)
json_text = html_lib.unescape(m.group(1) if m else raw).strip()
data = json.loads(json_text)
if data.get("chart", {}).get("error"):
raise RuntimeError(f"Yahoo API 錯誤: {data['chart']['error']}")
return data
def compute_rsi14(closes: list[float]) -> float | None:
if len(closes) < 15:
return None
gains = []
losses = []
for prev_close, close in zip(closes[:-1], closes[1:]):
change = close - prev_close
gains.append(max(change, 0.0))
losses.append(max(-change, 0.0))
period = 14
avg_gain = sum(gains[:period]) / period
avg_loss = sum(losses[:period]) / period
for gain, loss in zip(gains[period:], losses[period:]):
avg_gain = ((avg_gain * (period - 1)) + gain) / period
avg_loss = ((avg_loss * (period - 1)) + loss) / period
if avg_loss == 0:
return 100.0
rs = avg_gain / avg_loss
return 100 - (100 / (1 + rs))
def compute_atr14(highs: list[float], lows: list[float], closes: list[float]) -> float | None:
if len(highs) < 15 or len(lows) < 15 or len(closes) < 15:
return None
true_ranges = []
prev_close = closes[0]
for high, low, close in zip(highs[1:], lows[1:], closes[1:]):
tr = max(
high - low,
abs(high - prev_close),
abs(low - prev_close),
)
true_ranges.append(tr)
prev_close = close
period = 14
if len(true_ranges) < period:
return None
atr = sum(true_ranges[:period]) / period
for tr in true_ranges[period:]:
atr = ((atr * (period - 1)) + tr) / period
return atr
def compute_kd(highs: list[float], lows: list[float], closes: list[float]) -> tuple[float | None, float | None]:
period = 9
if len(highs) < period or len(lows) < period or len(closes) < period:
return None, None
k_value = 50.0
d_value = 50.0
rsv_values = []
for idx in range(period - 1, len(closes)):
window_high = max(highs[idx - period + 1 : idx + 1])
window_low = min(lows[idx - period + 1 : idx + 1])
close = closes[idx]
if window_high == window_low:
rsv = 50.0
else:
rsv = ((close - window_low) / (window_high - window_low)) * 100
rsv_values.append(rsv)
for rsv in rsv_values:
k_value = ((2 / 3) * k_value) + ((1 / 3) * rsv)
d_value = ((2 / 3) * d_value) + ((1 / 3) * k_value)
return k_value, d_value
def compute_bollinger_bands(closes: list[float]) -> tuple[float | None, float | None, float | None]:
period = 20
if len(closes) < period:
return None, None, None
window = closes[-period:]
middle = sum(window) / period
variance = sum((price - middle) ** 2 for price in window) / period
std_dev = math.sqrt(variance)
upper = middle + (2 * std_dev)
lower = middle - (2 * std_dev)
return upper, middle, lower
def parse_technical_data(data: dict):
result = data["chart"]["result"][0]
symbol = result.get("meta", {}).get("symbol", f"{STOCK_NO}.TW")
timestamps = result.get("timestamp", [])
quote = result.get("indicators", {}).get("quote", [{}])[0]
opens = quote.get("open", [])
closes = quote.get("close", [])
highs = quote.get("high", [])
lows = quote.get("low", [])
kline_rows = []
close_values = []
high_values = []
low_values = []
for ts, open_price, close, high, low in zip(timestamps, opens, closes, highs, lows):
if None in (open_price, close, high, low):
continue
d = dt.datetime.fromtimestamp(ts, tz=dt.timezone(dt.timedelta(hours=8)))
open_float = float(open_price)
close_float = float(close)
high_float = float(high)
low_float = float(low)
kline_rows.append(
(d.strftime("%m/%d"), open_float, high_float, low_float, close_float)
)
close_values.append(close_float)
high_values.append(high_float)
low_values.append(low_float)
if len(kline_rows) < 5:
raise RuntimeError("可用 K 線資料不足 5 筆")
rsi14 = compute_rsi14(close_values)
atr14 = compute_atr14(high_values, low_values, close_values)
k_value, d_value = compute_kd(high_values, low_values, close_values)
boll_upper, boll_middle, boll_lower = compute_bollinger_bands(close_values)
return (
symbol,
kline_rows[-5:],
rsi14,
atr14,
k_value,
d_value,
boll_upper,
boll_middle,
boll_lower,
)
def print_summary(
stock_no: str,
symbol: str,
kline_rows,
rsi14: float | None,
atr14: float | None,
k_value: float | None,
d_value: float | None,
boll_upper: float | None,
boll_middle: float | None,
boll_lower: float | None,
):
print(f"{stock_no}{symbol})|技術面摘要")
print("• 近 5 個交易日 K 線(開/高/低/收)")
for d, o, h, l, c in kline_rows:
print(f" - {d}{o:.2f}{h:.2f}{l:.2f}{c:.2f}")
if rsi14 is None:
print("• RSI14無資料")
else:
print(f"• RSI14{rsi14:.2f}")
if atr14 is None:
print("• ATR14無資料")
else:
print(f"• ATR14{atr14:.2f}")
if k_value is None or d_value is None:
print("• KD無資料")
else:
print(f"• K{k_value:.2f}")
print(f"• D{d_value:.2f}")
if boll_upper is None or boll_middle is None or boll_lower is None:
print("• 布林通道:無資料")
else:
print(f"• 布林上軌:{boll_upper:.2f}")
print(f"• 布林中軌:{boll_middle:.2f}")
print(f"• 布林下軌:{boll_lower:.2f}")
if __name__ == "__main__":
try:
(
symbol,
kline_rows,
rsi14,
atr14,
k_value,
d_value,
boll_upper,
boll_middle,
boll_lower,
) = parse_technical_data(fetch_chart_json(STOCK_NO))
print_summary(
STOCK_NO,
symbol,
kline_rows,
rsi14,
atr14,
k_value,
d_value,
boll_upper,
boll_middle,
boll_lower,
)
except Exception as e:
print(f"取得失敗:{e}", file=sys.stderr)
sys.exit(1)