# -*- coding: utf-8 -*- """ FinMind 技術指標資料模組 用於獲取台灣股市技術指標和籌碼面數據 API 文檔： - 技術面：https://finmind.github.io/tutor/TaiwanMarket/Technical/ - 籌碼面：https://finmind.github.io/tutor/TaiwanMarket/Chip/ 可用的資料集： - TaiwanStockPER: 個股 PER、PBR 資料 - TaiwanStockMarginPurchaseShortSale: 個股融資融劵表 - TaiwanStockInstitutionalInvestorsBuySell: 法人買賣表 - TaiwanStockShareholding: 外資持股表 技術指標計算： - 本模組會從 FinMind 獲取股價數據，自行計算技術指標（SMA、RSI、MACD 等） - 不需要依賴 Alpha Vantage 或其他外部服務 注意：本模組不使用需要 backer/sponsor 會員資格的功能 """ import json from datetime import datetime from dateutil.relativedelta import relativedelta from typing import Optional import pandas as pd import numpy as np from .finmind_common import ( _make_api_request, format_date, get_default_start_date, normalize_stock_id, FinMindError, FinMindDataNotFoundError, format_output, ) # 指標描述 INDICATOR_DESCRIPTIONS = { "per": "本益比（PER）：股價與每股盈餘之比，用於評估股票價值。較低的 PER 可能表示股票被低估。", "pbr": "股價淨值比（PBR）：股價與每股淨值之比，用於評估公司淨資產價值。PBR 低於 1 可能表示股票被低估。", "dividend_yield": "殖利率：股利與股價之比，衡量股票收益率。較高的殖利率可能吸引存股族。", "margin_purchase": "融資餘額：投資人向券商借錢買股票的未還金額。融資增加可能表示看多情緒。", "short_sale": "融券餘額：投資人借股票賣出的未補回數量。融券增加可能表示看空情緒。", "institutional": "三大法人買賣超：外資、投信、自營商的買賣超情況，是重要的市場動向指標。", "foreign_holding": "外資持股比例：外資持有該股票的比例。外資持續買超可能推升股價。", # 技術指標描述 "sma": "簡單移動平均線（SMA）：過去 N 日收盤價的平均值，用於判斷趨勢方向。", "ema": "指數移動平均線（EMA）：加權移動平均，對近期價格給予較高權重。", "rsi": "相對強弱指標（RSI）：衡量價格動能，RSI>70 為超買，RSI<30 為超賣。", "macd": "MACD：趨勢動能指標，由快線、慢線和柱狀圖組成，用於判斷買賣時機。", "bbands": "布林通道（Bollinger Bands）：由中軌（SMA）和上下軌組成，用於判斷價格波動範圍。", } # 支援的技術指標列表 CALCULATED_INDICATORS = [ "sma", "ema", "rsi", "macd", "bbands", "close_5_sma", "close_10_sma", "close_20_sma", "close_50_sma", "close_100_sma", "close_200_sma", "close_5_ema", "close_10_ema", "close_20_ema", "close_50_ema", ] def _get_stock_price_data(symbol: str, start_date: str, end_date: str) -> pd.DataFrame: """獲取股價數據並轉換為 DataFrame""" response = _make_api_request( dataset="TaiwanStockPrice", data_id=symbol, start_date=start_date, end_date=end_date ) if "data" not in response or not response["data"]: raise FinMindDataNotFoundError(f"找不到 {symbol} 的股價數據") df = pd.DataFrame(response["data"]) df["date"] = pd.to_datetime(df["date"]) df = df.sort_values("date").reset_index(drop=True) # 重命名欄位 df = df.rename(columns={ "max": "high", "min": "low", "Trading_Volume": "volume" }) return df def _calculate_sma(df: pd.DataFrame, period: int) -> pd.Series: """計算簡單移動平均線""" return df["close"].rolling(window=period).mean() def _calculate_ema(df: pd.DataFrame, period: int) -> pd.Series: """計算指數移動平均線""" return df["close"].ewm(span=period, adjust=False).mean() def _calculate_rsi(df: pd.DataFrame, period: int = 14) -> pd.Series: """計算 RSI 指標""" delta = df["close"].diff() gain = delta.where(delta > 0, 0) loss = (-delta).where(delta < 0, 0) avg_gain = gain.rolling(window=period).mean() avg_loss = loss.rolling(window=period).mean() rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) return rsi def _calculate_macd(df: pd.DataFrame, fast: int = 12, slow: int = 26, signal: int = 9) -> dict: """計算 MACD 指標""" ema_fast = df["close"].ewm(span=fast, adjust=False).mean() ema_slow = df["close"].ewm(span=slow, adjust=False).mean() macd_line = ema_fast - ema_slow signal_line = macd_line.ewm(span=signal, adjust=False).mean() histogram = macd_line - signal_line return { "macd": macd_line, "signal": signal_line, "histogram": histogram } def _calculate_bbands(df: pd.DataFrame, period: int = 20, std_dev: float = 2.0) -> dict: """計算布林通道""" sma = df["close"].rolling(window=period).mean() std = df["close"].rolling(window=period).std() return { "upper": sma + (std * std_dev), "middle": sma, "lower": sma - (std * std_dev) } def get_indicator( symbol: str, indicator: str, curr_date: str, look_back_days: int, interval: str = "daily", time_period: int = 14, series_type: str = "close" ) -> str: """ 返回 FinMind 在一個時間窗口內的技術指標/籌碼面數據。 支援的籌碼面指標： - per: 本益比 - pbr: 股價淨值比 - dividend_yield: 殖利率 - margin_purchase: 融資餘額 - short_sale: 融券餘額 - institutional: 三大法人買賣超 - foreign_holding: 外資持股比例 支援的技術指標（從股價計算）： - sma, close_N_sma: 簡單移動平均線 - ema, close_N_ema: 指數移動平均線 - rsi: 相對強弱指標 - macd: MACD 指標 - bbands: 布林通道 Args: symbol: 股票代碼（例如 "2330"） indicator: 要獲取的指標類型 curr_date: 當前交易日期，格式為 YYYY-mm-dd look_back_days: 回溯天數 interval: 時間間隔（保留參數，FinMind 只支援日線） time_period: 用於計算的天數 series_type: 價格類型（保留參數） Returns: str: 包含指標值和描述的字串 """ symbol = normalize_stock_id(symbol) indicator = indicator.lower() # 籌碼面指標 chip_indicators = [ "per", "pbr", "dividend_yield", "margin_purchase", "short_sale", "institutional", "foreign_holding" ] curr_date_dt = datetime.strptime(curr_date, "%Y-%m-%d") before = curr_date_dt - relativedelta(days=look_back_days) start_date = format_date(before) end_date = format_date(curr_date_dt) try: # 籌碼面指標 if indicator in chip_indicators: if indicator in ["per", "pbr", "dividend_yield"]: return _get_per_pbr_indicator(symbol, indicator, start_date, end_date) elif indicator in ["margin_purchase", "short_sale"]: return _get_margin_indicator(symbol, indicator, start_date, end_date) elif indicator == "institutional": return _get_institutional_indicator(symbol, start_date, end_date) elif indicator == "foreign_holding": return _get_foreign_holding_indicator(symbol, start_date, end_date) # 技術指標（從股價計算） else: return _calculate_technical_indicator( symbol=symbol, indicator=indicator, start_date=start_date, end_date=end_date, time_period=time_period, look_back_days=look_back_days ) except Exception as e: print(f"獲取 {indicator} 的 FinMind 指標數據時出錯：{e}") return f"檢索 {indicator} 數據時出錯：{str(e)}" def _calculate_technical_indicator( symbol: str, indicator: str, start_date: str, end_date: str, time_period: int, look_back_days: int ) -> str: """計算技術指標""" indicator = indicator.lower() # 解析指標名稱（例如 close_50_sma -> sma, period=50） period = time_period indicator_type = indicator # 解析 close_N_sma 或 close_N_ema 格式 if "_sma" in indicator: parts = indicator.split("_") for i, p in enumerate(parts): if p.isdigit(): period = int(p) indicator_type = "sma" elif "_ema" in indicator: parts = indicator.split("_") for i, p in enumerate(parts): if p.isdigit(): period = int(p) indicator_type = "ema" # 需要更多歷史數據來計算指標 start_date_dt = datetime.strptime(start_date, "%Y-%m-%d") extended_start = start_date_dt - relativedelta(days=max(period + 50, 300)) extended_start_str = format_date(extended_start) try: # 獲取股價數據 df = _get_stock_price_data(symbol, extended_start_str, end_date) if df.empty: return f"找不到 {symbol} 的股價數據來計算 {indicator}。" # 計算指標 if indicator_type == "sma": df["indicator"] = _calculate_sma(df, period) desc = f"{period} 日簡單移動平均線" elif indicator_type == "ema": df["indicator"] = _calculate_ema(df, period) desc = f"{period} 日指數移動平均線" elif indicator_type == "rsi": df["indicator"] = _calculate_rsi(df, period) desc = f"{period} 日 RSI 相對強弱指標" elif indicator_type == "macd": macd_data = _calculate_macd(df) df["macd"] = macd_data["macd"] df["signal"] = macd_data["signal"] df["histogram"] = macd_data["histogram"] desc = "MACD 指標（12, 26, 9）" elif indicator_type == "bbands": bbands_data = _calculate_bbands(df, period) df["bb_upper"] = bbands_data["upper"] df["bb_middle"] = bbands_data["middle"] df["bb_lower"] = bbands_data["lower"] desc = f"{period} 日布林通道" else: return f"不支援的技術指標 {indicator}。支援的技術指標：sma, ema, rsi, macd, bbands, close_N_sma, close_N_ema" # 過濾日期範圍 start_date_dt = datetime.strptime(start_date, "%Y-%m-%d") df = df[df["date"] >= pd.Timestamp(start_date_dt)] # 只顯示最近的數據 df = df.tail(min(look_back_days, 30)) # 格式化輸出 ind_string = "" if indicator_type == "macd": for _, row in df.iterrows(): date = row["date"].strftime("%Y-%m-%d") macd_val = row["macd"] signal_val = row["signal"] hist_val = row["histogram"] if pd.notna(macd_val): ind_string += f"{date}: MACD={macd_val:.4f}, Signal={signal_val:.4f}, Histogram={hist_val:.4f}\n" elif indicator_type == "bbands": for _, row in df.iterrows(): date = row["date"].strftime("%Y-%m-%d") upper = row["bb_upper"] middle = row["bb_middle"] lower = row["bb_lower"] close = row["close"] if pd.notna(upper): ind_string += f"{date}: Upper={upper:.2f}, Middle={middle:.2f}, Lower={lower:.2f}, Close={close:.2f}\n" else: for _, row in df.iterrows(): date = row["date"].strftime("%Y-%m-%d") value = row["indicator"] close = row["close"] if pd.notna(value): ind_string += f"{date}: {indicator.upper()}={value:.4f}, Close={close:.2f}\n" if not ind_string: ind_string = "指定日期範圍內無足夠數據計算指標。\n" result_str = ( f"## 從 {start_date} 到 {end_date} 的 {desc} ({symbol})：\n\n" + ind_string + "\n\n" + INDICATOR_DESCRIPTIONS.get(indicator_type, "技術指標計算自 FinMind 股價數據。") ) return result_str except FinMindError as e: return f"獲取 {indicator} 數據時出錯：{str(e)}" except Exception as e: return f"計算 {indicator} 時出錯：{str(e)}" def _get_per_pbr_indicator( symbol: str, indicator: str, start_date: str, end_date: str ) -> str: """獲取 PER、PBR、殖利率指標""" try: response = _make_api_request( dataset="TaiwanStockPER", data_id=symbol, start_date=start_date, end_date=end_date ) if "data" in response and response["data"]: data = response["data"] # 欄位映射 field_map = { "per": "PER", "pbr": "PBR", "dividend_yield": "dividend_yield" } field_name = field_map.get(indicator, indicator) # 格式化輸出 ind_string = "" for row in sorted(data, key=lambda x: x.get("date", "")): date = row.get("date", "") value = row.get(field_name, "N/A") ind_string += f"{date}: {value}\n" if not ind_string: ind_string = "指定日期範圍內無可用數據。\n" result_str = ( f"## 從 {start_date} 到 {end_date} 的 {indicator.upper()} 值：\n\n" + ind_string + "\n\n" + INDICATOR_DESCRIPTIONS.get(indicator, "無可用描述。") ) return result_str else: return f"找不到 {symbol} 的 {indicator} 數據。" except FinMindError as e: return f"獲取 {indicator} 數據時出錯：{str(e)}" def _get_margin_indicator( symbol: str, indicator: str, start_date: str, end_date: str ) -> str: """獲取融資融券指標""" try: response = _make_api_request( dataset="TaiwanStockMarginPurchaseShortSale", data_id=symbol, start_date=start_date, end_date=end_date ) if "data" in response and response["data"]: data = response["data"] # 欄位映射 if indicator == "margin_purchase": field_name = "MarginPurchaseTodayBalance" display_name = "融資餘額" else: # short_sale field_name = "ShortSaleTodayBalance" display_name = "融券餘額" # 格式化輸出 ind_string = "" for row in sorted(data, key=lambda x: x.get("date", "")): date = row.get("date", "") value = row.get(field_name, "N/A") ind_string += f"{date}: {value:,}\n" if isinstance(value, (int, float)) else f"{date}: {value}\n" if not ind_string: ind_string = "指定日期範圍內無可用數據。\n" result_str = ( f"## 從 {start_date} 到 {end_date} 的 {display_name} ({symbol})：\n\n" + ind_string + "\n\n" + INDICATOR_DESCRIPTIONS.get(indicator, "無可用描述。") ) return result_str else: return f"找不到 {symbol} 的融資融券數據。" except FinMindError as e: return f"獲取融資融券數據時出錯：{str(e)}" def _get_institutional_indicator( symbol: str, start_date: str, end_date: str ) -> str: """獲取三大法人買賣超指標""" try: response = _make_api_request( dataset="TaiwanStockInstitutionalInvestorsBuySell", data_id=symbol, start_date=start_date, end_date=end_date ) if "data" in response and response["data"]: data = response["data"] df = pd.DataFrame(data) # 按日期分組計算各法人買賣超 ind_string = "" # 獲取唯一日期 dates = sorted(df["date"].unique()) for date in dates: day_data = df[df["date"] == date] # 彙總各法人買賣超 foreign = day_data[day_data["name"].str.contains("外資", na=False)]["buy"].sum() - \ day_data[day_data["name"].str.contains("外資", na=False)]["sell"].sum() investment_trust = day_data[day_data["name"].str.contains("投信", na=False)]["buy"].sum() - \ day_data[day_data["name"].str.contains("投信", na=False)]["sell"].sum() dealer = day_data[day_data["name"].str.contains("自營商", na=False)]["buy"].sum() - \ day_data[day_data["name"].str.contains("自營商", na=False)]["sell"].sum() total = foreign + investment_trust + dealer ind_string += f"{date}: 外資 {foreign:+,} / 投信 {investment_trust:+,} / 自營 {dealer:+,} / 合計 {total:+,}\n" if not ind_string: ind_string = "指定日期範圍內無可用數據。\n" result_str = ( f"## 從 {start_date} 到 {end_date} 的三大法人買賣超 ({symbol})：\n\n" + ind_string + "\n\n" + INDICATOR_DESCRIPTIONS.get("institutional", "無可用描述。") ) return result_str else: return f"找不到 {symbol} 的三大法人買賣超數據。" except FinMindError as e: return f"獲取三大法人數據時出錯：{str(e)}" def _get_foreign_holding_indicator( symbol: str, start_date: str, end_date: str ) -> str: """獲取外資持股比例指標""" try: response = _make_api_request( dataset="TaiwanStockShareholding", data_id=symbol, start_date=start_date, end_date=end_date ) if "data" in response and response["data"]: data = response["data"] # 格式化輸出 ind_string = "" for row in sorted(data, key=lambda x: x.get("date", "")): date = row.get("date", "") holding_percent = row.get("ForeignInvestmentSharesRatio", "N/A") holding_shares = row.get("ForeignInvestmentShares", "N/A") if isinstance(holding_percent, (int, float)): ind_string += f"{date}: {holding_percent:.2f}% ({holding_shares:,} 股)\n" else: ind_string += f"{date}: {holding_percent}\n" if not ind_string: ind_string = "指定日期範圍內無可用數據。\n" result_str = ( f"## 從 {start_date} 到 {end_date} 的外資持股比例 ({symbol})：\n\n" + ind_string + "\n\n" + INDICATOR_DESCRIPTIONS.get("foreign_holding", "無可用描述。") ) return result_str else: return f"找不到 {symbol} 的外資持股數據。" except FinMindError as e: return f"獲取外資持股數據時出錯：{str(e)}" def get_margin_data( symbol: str, start_date: str, end_date: str ) -> str: """ 獲取個股融資融劵表完整數據。 資料區間：2001-01-01 ~ now 返回欄位： - date: 日期 - stock_id: 股票代碼 - MarginPurchaseBuy: 融資買進 - MarginPurchaseSell: 融資賣出 - MarginPurchaseTodayBalance: 融資今日餘額 - ShortSaleBuy: 融券買進 - ShortSaleSell: 融券賣出 - ShortSaleTodayBalance: 融券今日餘額 Returns: str: JSON 格式的融資融券數據 """ symbol = normalize_stock_id(symbol) try: response = _make_api_request( dataset="TaiwanStockMarginPurchaseShortSale", data_id=symbol, start_date=start_date, end_date=end_date ) if "data" in response and response["data"]: result = { "stock_id": symbol, "data_type": "margin_trading", "data": response["data"] } return format_output(result) else: return format_output({ "stock_id": symbol, "data": [], "message": "查無資料" }) except FinMindError as e: return format_output({ "error": str(e), "stock_id": symbol }) def get_institutional_data( symbol: str, start_date: str, end_date: str ) -> str: """ 獲取法人買賣表完整數據。 資料區間：2005-01-01 ~ now 返回欄位： - date: 日期 - stock_id: 股票代碼 - name: 法人名稱（外資、投信、自營商等） - buy: 買進金額 - sell: 賣出金額 Returns: str: JSON 格式的法人買賣數據 """ symbol = normalize_stock_id(symbol) try: response = _make_api_request( dataset="TaiwanStockInstitutionalInvestorsBuySell", data_id=symbol, start_date=start_date, end_date=end_date ) if "data" in response and response["data"]: result = { "stock_id": symbol, "data_type": "institutional_investors", "data": response["data"] } return format_output(result) else: return format_output({ "stock_id": symbol, "data": [], "message": "查無資料" }) except FinMindError as e: return format_output({ "error": str(e), "stock_id": symbol })