commit 7d72aed984eac16a9a3d0996f6e8349abf2f6f9f Author: Timmy Date: Fri Apr 10 10:07:31 2026 +0800 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) diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..fa0555f --- /dev/null +++ b/.gitignore @@ -0,0 +1,19 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +*.egg-info/ +dist/ +build/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo + +# OS +.DS_Store +Thumbs.db diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..7036306 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,107 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Overview + +This is a stock query skill for Taiwan stock market data. Python scripts in `scripts/` fetch data from Yahoo Finance Taiwan using a Browserless headless browser service, then output concise Chinese summaries. + +## Environment Variables + +All `*-browserless.py` scripts require these environment variables (loaded from `.env`): +- `BROWSERLESS_ENDPOINT` - Browserless service URL (e.g., `http://192.168.42.124:13000`) +- `BROWSERLESS_TOKEN` - Authentication token for Browserless + +The `.env` file is searched first in current directory, then at workspace root (3 levels up from scripts). + +## Architecture + +Scripts fall into two categories: + +### 1. Browserless Scripts (Dynamic Content Scraping) +Use Browserless to fetch Yahoo pages that require JavaScript rendering: + +| Script | Purpose | +|--------|---------| +| `yahoo-quote-browserless.py` | Individual stock quote (price, OHLC, volume, bid/ask) | +| `yahoo-institutional-browserless.py` | Three major institutional investors (foreign, trust, dealer) | +| `yahoo-technical-browserless.py` | Technical indicators (RSI14, ATR14, KD, Bollinger Bands, 5-day K-line) | +| `yahoo-market-browserless.py` | TAIEX weighted index (^TWII) market summary | +| `yahoo-dividend-browserless.py` | Dividend history (cash/stock dividend, ex-date, yield) | +| `yahoo-broker-trading-browserless.py` | Top broker buy/sell rankings | + +### 2. Direct API Scripts (No Browserless Required) +Use Yahoo Finance Chart API directly: + +| Script | Purpose | +|--------|---------| +| `yahoo-etf-holdings.py` | ETF holdings (top 10, industry distribution, asset allocation) | +| `yahoo-ma-signal.py` | Moving averages (MA5/10/20/60) and 20-day bias | +| `yahoo-macd-signal.py` | MACD indicator (DIF, DEA, histogram) | + +### 3. TWSE Public API Scripts (No Browserless Required) +Use TWSE open data API directly: + +| Script | Purpose | +|--------|---------| +| `twse-margin-trading.py` | Margin trading balance (融資融券餘額) | +| `twse-day-trading.py` | Day trading statistics (當沖交易統計) | +| `twse-foreign-holdings.py` | Foreign investor holdings ratio (外資持股比例) | + +## Common Commands + +### Query individual stock (full analysis) +```bash +# Three scripts are typically run together for complete picture +python3 skills/stocks-query/scripts/yahoo-institutional-browserless.py 2330 +python3 skills/stocks-query/scripts/yahoo-quote-browserless.py 2330 +python3 skills/stocks-query/scripts/yahoo-technical-browserless.py 2330 +``` + +### Query ETF (with holdings) +```bash +python3 skills/stocks-query/scripts/yahoo-quote-browserless.py 0050 +python3 skills/stocks-query/scripts/yahoo-etf-holdings.py 0050 +``` + +### Query market index +```bash +python3 skills/stocks-query/scripts/yahoo-market-browserless.py +``` + +### MA/MACD signals (direct API, no Browserless) +```bash +python3 skills/stocks-query/scripts/yahoo-ma-signal.py 2330 +python3 skills/stocks-query/scripts/yahoo-macd-signal.py 2330 +``` + +### Dividend / Broker trading (Browserless) +```bash +python3 skills/stocks-query/scripts/yahoo-dividend-browserless.py 2330 +python3 skills/stocks-query/scripts/yahoo-broker-trading-browserless.py 2330 +``` + +### TWSE data (margin, day-trading, foreign holdings) +```bash +python3 skills/stocks-query/scripts/twse-margin-trading.py 2330 +python3 skills/stocks-query/scripts/twse-day-trading.py 2330 +python3 skills/stocks-query/scripts/twse-foreign-holdings.py 2330 +``` + +## Script Patterns + +- All scripts accept stock symbol as first command-line argument (default provided) +- Browserless scripts use `urllib` or `subprocess` with `curl` to POST to Browserless `/content` endpoint +- HTML parsing uses regex patterns (not BeautifulSoup) to minimize dependencies +- Output format is Chinese text summaries with bullet points +- Error handling exits with code 1 and prints error message to stderr + +## Stock Number Format + +- Taiwan stocks: 4-digit number (e.g., `2330`, `3481`) +- ETFs: Typically start with `00` (e.g., `0050`, `00922`) +- Yahoo suffix: `.TW` is appended automatically in URLs + +## Skill Integration + +This directory is registered as a skill (see `SKILL.md`). When users input a stock number or ask about "大盤" (market), the scripts should be executed in sequence and results synthesized into a concise Chinese summary with plain-language interpretation of technical indicators. diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..aa0f276 --- /dev/null +++ b/SKILL.md @@ -0,0 +1,84 @@ +--- +name: stocks-query +description: 查詢台股個股與大盤（Yahoo 資料）並整理重點摘要。當使用者輸入股票代號（例如 2330、0050、3481）或詢問「大盤」時使用。流程：個股依序執行法人、報價、技術分析三支 Python 腳本；大盤執行大盤腳本；最後輸出精簡中文摘要，並檢查是否有本地持股庫存紀錄。 +--- + +# stocks-query + +使用 `scripts/stocks/` 內的腳本查詢台股資料，並回覆精簡摘要。 + +## 執行流程 + +### A) 使用者只輸入股票代號（例：2330、0050、3481） + +依序執行（盡量平行呼叫以加快速度）： + +```bash +python3 skills/stocks-query/scripts/yahoo-institutional-browserless.py <代號> +python3 skills/stocks-query/scripts/yahoo-quote-browserless.py <代號> +python3 skills/stocks-query/scripts/yahoo-technical-browserless.py <代號> +python3 skills/stocks-query/scripts/twse-margin-trading.py <代號> +python3 skills/stocks-query/scripts/yahoo-dividend-browserless.py <代號> +python3 skills/stocks-query/scripts/yahoo-market-brief.py +python3 skills/stocks-query/scripts/yahoo-broker-trading-browserless.py <代號> +python3 skills/stocks-query/scripts/twse-day-trading.py <代號> +python3 skills/stocks-query/scripts/twse-foreign-holdings.py <代號> +``` + +若標的**不是** ETF（非 `00` 開頭），額外執行： + +```bash +python3 skills/stocks-query/scripts/twse-monthly-revenue.py <代號> +python3 skills/stocks-query/scripts/twse-pe-pbr.py <代號> +``` + +若標的是 ETF（`00` 開頭），改為額外執行： + +```bash +python3 skills/stocks-query/scripts/yahoo-etf-holdings.py <代號> +``` + +注意：月營收、本益比 對 ETF 無資料，不要對 ETF 執行。 + +再檢查本地是否有該代號的持股／庫存紀錄（例如 `stocks/`、`purchases/` 相關檔）。 + +回覆重點： +- 現價、漲跌、開高低、昨收、成交量、均價、最佳買賣 +- 三大法人（外資／投信／自營商／合計） +- 近 5 日 K 線（開高低收） +- RSI14 +- ATR14 +- KD +- 布林通道 +- 看均線、量能、乖離、MACD（若腳本有提供資料則納入摘要） +- 融資融券：融資餘額（增減）、融券餘額（增減）、券資比、融資使用率 +- 股利殖利率：歷年現金股利、殖利率、最近除息日與填息天數 +- 大盤同步：加權指數現值與漲跌（一行簡要） +- 主力券商：買超／賣超前 15 大券商、主力買賣超合計張數 +- 當沖統計：當沖成交量、佔個股成交比例、大盤當沖佔比 +- 外資持股：外資持有股數與持股比例 +- 非 ETF 額外補：月營收（當月、MoM、YoY、累計）、本益比、股價淨值比 +- ETF 額外補：前十大持股、行業比重、資產分佈 +- 是否查到本地持股紀錄 +- 以上技術面數值後，必須補一段「白話判斷」，用台灣繁中解釋目前偏強／偏弱、是否過熱、波動大不大、短線偏多或偏震盪，避免只丟數字。 + +### B) 使用者詢問「大盤」 + +執行： + +```bash +python3 skills/stocks-query/scripts/yahoo-market-browserless.py +``` + +回覆重點： +- 加權指數 +- 漲跌（點數與百分比） +- 開高低、昨收 +- 成交金額 +- 更新時間 + +## 回覆風格 + +- 使用繁體中文（台灣） +- 簡潔、重點式條列 +- 若某欄位缺值，明確標註「無資料」 diff --git a/scripts/twse-day-trading.py b/scripts/twse-day-trading.py new file mode 100644 index 0000000..947cc3f --- /dev/null +++ b/scripts/twse-day-trading.py @@ -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) diff --git a/scripts/twse-foreign-holdings.py b/scripts/twse-foreign-holdings.py new file mode 100644 index 0000000..97dc667 --- /dev/null +++ b/scripts/twse-foreign-holdings.py @@ -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) diff --git a/scripts/twse-margin-trading.py b/scripts/twse-margin-trading.py new file mode 100644 index 0000000..c64fe9c --- /dev/null +++ b/scripts/twse-margin-trading.py @@ -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) diff --git a/scripts/yahoo-broker-trading-browserless.py b/scripts/yahoo-broker-trading-browserless.py new file mode 100644 index 0000000..ffbcd6c --- /dev/null +++ b/scripts/yahoo-broker-trading-browserless.py @@ -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"([^<(]+)\([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) diff --git a/scripts/yahoo-dividend-browserless.py b/scripts/yahoo-dividend-browserless.py new file mode 100644 index 0000000..ba93f98 --- /dev/null +++ b/scripts/yahoo-dividend-browserless.py @@ -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) diff --git a/scripts/yahoo-etf-holdings.py b/scripts/yahoo-etf-holdings.py new file mode 100644 index 0000000..37dbad8 --- /dev/null +++ b/scripts/yahoo-etf-holdings.py @@ -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() diff --git a/scripts/yahoo-institutional-browserless.py b/scripts/yahoo-institutional-browserless.py new file mode 100644 index 0000000..790b119 --- /dev/null +++ b/scripts/yahoo-institutional-browserless.py @@ -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) diff --git a/scripts/yahoo-ma-signal.py b/scripts/yahoo-ma-signal.py new file mode 100644 index 0000000..fb91f67 --- /dev/null +++ b/scripts/yahoo-ma-signal.py @@ -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() diff --git a/scripts/yahoo-macd-signal.py b/scripts/yahoo-macd-signal.py new file mode 100644 index 0000000..5447ed7 --- /dev/null +++ b/scripts/yahoo-macd-signal.py @@ -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() diff --git a/scripts/yahoo-market-brief.py b/scripts/yahoo-market-brief.py new file mode 120000 index 0000000..1a1cbaa --- /dev/null +++ b/scripts/yahoo-market-brief.py @@ -0,0 +1 @@ +yahoo-market-browserless.py \ No newline at end of file diff --git a/scripts/yahoo-market-browserless.py b/scripts/yahoo-market-browserless.py new file mode 100644 index 0000000..24d13b6 --- /dev/null +++ b/scripts/yahoo-market-browserless.py @@ -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) diff --git a/scripts/yahoo-quote-browserless.py b/scripts/yahoo-quote-browserless.py new file mode 100644 index 0000000..6b64b5f --- /dev/null +++ b/scripts/yahoo-quote-browserless.py @@ -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) diff --git a/scripts/yahoo-technical-browserless.py b/scripts/yahoo-technical-browserless.py new file mode 100644 index 0000000..105dcf9 --- /dev/null +++ b/scripts/yahoo-technical-browserless.py @@ -0,0 +1,303 @@ +#!/usr/bin/env python3 +""" +用 Browserless 透過 Yahoo Chart API 抓技術面摘要（代號可變） + +輸出內容： +- 近 5 個交易日 K 線（開高低收） +- RSI14 +- ATR14 +- KD（9,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)