From 48eba043003471efdfdd81ecaf93f08c022bc1eb Mon Sep 17 00:00:00 2001 From: Timmy Date: Thu, 26 Mar 2026 18:10:52 +0800 Subject: [PATCH] Initial commit: stocks-query skill Add scripts for querying Taiwan stock market data via Yahoo Finance: - Individual stock quotes (price, OHLC, volume, bid/ask) - Three major institutional investors (foreign, trust, dealer) - Technical indicators (RSI14, ATR14, KD, Bollinger Bands, MACD, MA) - ETF holdings analysis (top 10, industry distribution, asset allocation) - TAIEX weighted index market summary Co-Authored-By: Claude Opus 4.6 (1M context) --- .gitignore | 19 ++ CLAUDE.md | 83 ++++++ SKILL.md | 64 +++++ scripts/yahoo-etf-holdings.py | 164 +++++++++++ scripts/yahoo-institutional-browserless.py | 158 +++++++++++ scripts/yahoo-ma-signal.py | 82 ++++++ scripts/yahoo-macd-signal.py | 91 +++++++ scripts/yahoo-market-browserless.py | 168 ++++++++++++ scripts/yahoo-quote-browserless.py | 192 +++++++++++++ scripts/yahoo-technical-browserless.py | 303 +++++++++++++++++++++ 10 files changed, 1324 insertions(+) create mode 100644 .gitignore create mode 100644 CLAUDE.md create mode 100644 SKILL.md create mode 100644 scripts/yahoo-etf-holdings.py create mode 100644 scripts/yahoo-institutional-browserless.py create mode 100644 scripts/yahoo-ma-signal.py create mode 100644 scripts/yahoo-macd-signal.py create mode 100644 scripts/yahoo-market-browserless.py create mode 100644 scripts/yahoo-quote-browserless.py create mode 100644 scripts/yahoo-technical-browserless.py 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..9964967 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,83 @@ +# 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 | + +### 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) | + +## 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 +``` + +## 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..093a110 --- /dev/null +++ b/SKILL.md @@ -0,0 +1,64 @@ +--- +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 <代號> +``` + +若標的是 ETF（建議以台股 ETF 常見代號型態判斷，如 `00` 開頭），補充執行： + +```bash +python3 skills/stocks-query/scripts/yahoo-etf-holdings.py <代號> +``` + +若不是 ETF，跳過這一步，不要多跑持股分析。 + +再檢查本地是否有該代號的持股／庫存紀錄（例如 `stocks/`、`purchases/` 相關檔）。 + +回覆重點： +- 現價、漲跌、開高低、昨收、成交量、均價、最佳買賣 +- 三大法人（外資／投信／自營商／合計） +- 近 5 日 K 線（開高低收） +- RSI14 +- ATR14 +- KD +- 布林通道 +- 看均線、量能、乖離、MACD（若腳本有提供資料則納入摘要） +- ETF 額外補：前十大持股、行業比重、資產分佈 +- 是否查到本地持股紀錄 +- 以上技術面數值後，必須補一段「白話判斷」，用台灣繁中解釋目前偏強／偏弱、是否過熱、波動大不大、短線偏多或偏震盪，避免只丟數字。 + +### B) 使用者詢問「大盤」 + +執行： + +```bash +python3 skills/stocks-query/scripts/yahoo-market-browserless.py +``` + +回覆重點： +- 加權指數 +- 漲跌（點數與百分比） +- 開高低、昨收 +- 成交金額 +- 更新時間 + +## 回覆風格 + +- 使用繁體中文（台灣） +- 簡潔、重點式條列 +- 若某欄位缺值，明確標註「無資料」 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']*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+)\.([^<]+)
([0-9]+\.[0-9]+%)
', + 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'
(?:]*>
)?([^<]+)
([0-9]+\.[0-9]+%)
', + 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, '資產分佈', '前十大持股') + top_block = extract_block(html, '前十大持股', '網友也在看') + industry_block = extract_block(html, '行業比重', '持股明細') + + 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"([^<(]+)\([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-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)