diff --git a/scripts/twse-monthly-revenue.py b/scripts/twse-monthly-revenue.py new file mode 100644 index 0000000..dff75c9 --- /dev/null +++ b/scripts/twse-monthly-revenue.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +""" +查詢個股月營收(Yahoo Finance TW + Browserless) + +用法: + python3 skills/stocks-query/scripts/twse-monthly-revenue.py 2330 + python3 skills/stocks-query/scripts/twse-monthly-revenue.py 3481 + +資料來源:Yahoo 奇摩股市(營收頁),營收單位:千元 +""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +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/revenue" + 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_revenue(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"([^<(]+)\([0-9]+\.TW\)", stock_no).strip() + + # 找 JSON 中的月營收資料(data 陣列內含 revenueMoM 欄位的物件) + obj_pat = re.compile(r'\{[^{}]*"revenueMoM":[^{}]*\}') + matches = obj_pat.findall(html) + + if not matches: + raise RuntimeError("抓不到月營收資料(頁面結構可能變更)") + + rows = [] + for m in matches: + try: + obj = json.loads(m.replace("\\u002F", "/")) + rows.append(obj) + except json.JSONDecodeError: + continue + + if not rows: + raise RuntimeError("無法解析月營收 JSON 資料") + + # 依日期排序(最新在前) + rows.sort(key=lambda r: r.get("date", ""), reverse=True) + + return { + "name": name, + "stock_no": stock_no, + "latest": rows[0], + "history": rows[:6], + } + + +def fmt_pct(val) -> str: + """將小數比例轉為百分比字串。""" + try: + return f"{float(val) * 100:+.2f}%" + except (ValueError, TypeError): + return "-" + + +def fmt_rev(val: str) -> str: + """千元營收加上千分位格式。""" + cleaned = str(val).replace(",", "") + try: + return f"{int(cleaned):,}" + except (ValueError, TypeError): + return val + + +def print_summary(d: dict) -> None: + name = d["name"] + stock_no = d["stock_no"] + latest = d["latest"] + date = latest.get("date", "-") + + print(f"{name}({stock_no})|月營收摘要") + print(f"• 期間:{date}") + print(f"• 當月營收:{fmt_rev(latest.get('revenue', '-'))} 千元") + print(f"• 去年同月營收:{fmt_rev(latest.get('lastYearRevenue', '-'))} 千元") + print(f"• MoM(月增率):{fmt_pct(latest.get('revenueMoM'))}") + print(f"• YoY(年增率):{fmt_pct(latest.get('revenueYoY'))}") + print(f"• 當月累計營收:{fmt_rev(latest.get('revenueAcc', '-'))} 千元") + print(f"• 累計 YoY:{fmt_pct(latest.get('revenueYoYAcc'))}") + + # 近期趨勢 + history = d.get("history", []) + if len(history) > 1: + print(f"\n近 {len(history)} 月趨勢:") + for row in history: + rd = row.get("date", "-") + rv = fmt_rev(row.get("revenue", "-")) + mom = fmt_pct(row.get("revenueMoM")) + yoy = fmt_pct(row.get("revenueYoY")) + print(f" {rd}|{rv} 千元|MoM {mom}|YoY {yoy}") + + +if __name__ == "__main__": + try: + html = fetch_html(STOCK_NO) + data = parse_revenue(html, STOCK_NO) + print_summary(data) + except Exception as e: + print(f"取得失敗:{e}", file=sys.stderr) + sys.exit(1) diff --git a/scripts/twse-pe-pbr.py b/scripts/twse-pe-pbr.py new file mode 100644 index 0000000..7594c9c --- /dev/null +++ b/scripts/twse-pe-pbr.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python3 +""" +查詢 TWSE 個股本益比/股價淨值比/殖利率 + +用法: + python3 skills/stocks-query/scripts/twse-pe-pbr.py 2330 + python3 skills/stocks-query/scripts/twse-pe-pbr.py 3481 + +資料來源:證交所 TWSE 公開 API(個股日本益比、殖利率及股價淨值比) +""" + +from __future__ import annotations + +import json +import sys +import urllib.request +from datetime import datetime, timedelta + + +def fetch_pe_pbr(stock_no: str, date_str: str | None = None) -> dict: + """從 TWSE 抓本益比/股價淨值比,自動往前回溯最多 3 個月找到有資料的月份。""" + if date_str: + dates_to_try = [date_str] + else: + today = datetime.now() + dates_to_try = [] + for i in range(4): + dt = today - timedelta(days=30 * i) + dates_to_try.append(dt.replace(day=1).strftime("%Y%m%d")) + + for d in dates_to_try: + url = ( + f"https://www.twse.com.tw/rwd/zh/afterTrading/BWIBBU" + f"?date={d}&stockNo={stock_no}&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 + + rows = data.get("data", []) + fields = data.get("fields", []) + if not rows or not fields: + continue + + # 取最新一筆 + latest = rows[-1] + result = {} + for i, field in enumerate(fields): + val = str(latest[i]).strip() if i < len(latest) else "-" + result[field.strip()] = val + + # 附上近期歷史 + result["_history"] = [] + for row in rows: + entry = {} + for i, field in enumerate(fields): + val = str(row[i]).strip() if i < len(row) else "-" + entry[field.strip()] = val + result["_history"].append(entry) + + return result + + raise RuntimeError(f"找不到 {stock_no} 的本益比/淨值比資料(已回溯 3 個月)") + + +def print_summary(stock_no: str, d: dict) -> None: + print(f"{stock_no}|本益比/股價淨值比摘要") + + date_val = d.get("日期", "-") + pe = d.get("本益比", "-") + pbr = d.get("股價淨值比", "-") + dividend_yield = d.get("殖利率(%)", "-") + fiscal_year = d.get("股利年度", "-") + fiscal_quarter = d.get("財報年/季", "-") + + print(f"• 最新日期:{date_val}") + print(f"• 本益比:{pe}") + print(f"• 股價淨值比:{pbr}") + if dividend_yield != "-" and "%" not in str(dividend_yield): + print(f"• 殖利率:{dividend_yield}%") + else: + print(f"• 殖利率:{dividend_yield}") + print(f"• 股利年度:{fiscal_year}") + print(f"• 財報年/季:{fiscal_quarter}") + + # 近期趨勢 + history = d.get("_history", []) + if len(history) > 1: + recent = history[-5:] + print(f"\n近期趨勢(最近 {len(recent)} 筆):") + for entry in recent: + h_date = entry.get("日期", "-") + h_pe = entry.get("本益比", "-") + h_pbr = entry.get("股價淨值比", "-") + h_dy = entry.get("殖利率(%)", "-") + print(f" {h_date}|PE {h_pe}|PBR {h_pbr}|殖利率 {h_dy}%") + + +if __name__ == "__main__": + stock_no = sys.argv[1] if len(sys.argv) > 1 else "2330" + try: + data = fetch_pe_pbr(stock_no) + print_summary(stock_no, data) + except Exception as e: + print(f"取得失敗:{e}", file=sys.stderr) + sys.exit(1)