stocks-query/scripts/yahoo-technical-browserless.py

#!/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)