Skip to main content
Glama

Taiwan Stock Agent

by clsung
stock_tools.py8.91 kB
import os from datetime import datetime from typing import Any from tw_stock_agent.services.stock_service import StockService from tw_stock_agent.utils.validation import ( StockCodeValidator, StockParameterValidator, validate_stock_request ) from tw_stock_agent.utils.mcp_error_handler import ( mcp_error_handler, MCPResponseFormatter ) from tw_stock_agent.exceptions import ( ParameterValidationError, StockNotFoundError, TwStockAgentError ) TWSE_EQUITIES_URL = "https://isin.twse.com.tw/isin/C_public.jsp?strMode=2" TPEX_EQUITIES_URL = "https://isin.twse.com.tw/isin/C_public.jsp?strMode=4" def update_stock_codes(): """Update stock codes from TWSE and TPEx websites.""" def get_directory(): return os.path.dirname(os.path.abspath(__file__)) try: # Note: to_csv function is not defined in this file # This function would need to be implemented or imported # to_csv(TWSE_EQUITIES_URL, os.path.join(get_directory(), "twse_equities.csv")) # to_csv(TPEX_EQUITIES_URL, os.path.join(get_directory(), "tpex_equities.csv")) pass except Exception as e: import logging logger = logging.getLogger("tw-stock-agent.stock_tools") logger.error(f"Failed to update stock codes: {e}") # 初始化服務 stock_service = StockService() # Helper function to validate multiple stock codes async def validate_and_fetch_multiple_stocks(stock_codes: list[str]) -> dict[str, Any]: """ Helper function to validate and fetch multiple stock data. Args: stock_codes: List of stock codes to validate and fetch Returns: Dictionary of stock data with validation results Raises: ParameterValidationError: If stock codes list is invalid """ if not stock_codes or not isinstance(stock_codes, list): raise ParameterValidationError( parameter_name="stock_codes", parameter_value=stock_codes, expected_format="Non-empty list of stock codes" ) # Validate all stock codes first try: validated_codes = StockCodeValidator.validate_multiple_codes(stock_codes) except Exception as e: raise ParameterValidationError( parameter_name="stock_codes", parameter_value=stock_codes, message=f"Stock code validation failed: {str(e)}" ) # Fetch data for all validated codes results = await stock_service.fetch_multiple_stocks_data(validated_codes) return { "requested_codes": stock_codes, "validated_codes": validated_codes, "results": results, "success_count": sum(1 for r in results.values() if "error" not in r), "error_count": sum(1 for r in results.values() if "error" in r) } @mcp_error_handler("get_stock_data") async def get_stock_data(stock_code: str) -> dict[str, Any]: """ 取得股票基本資料 Args: stock_code: 股票代號,例如:2330 Returns: 股票資料,包含公司概況、產業別、市值等資訊 Raises: InvalidStockCodeError: 股票代號格式錯誤 StockNotFoundError: 找不到股票 StockDataUnavailableError: 股票資料無法取得 """ # Validate request parameters validated_params = validate_stock_request(stock_code=stock_code) # Fetch stock data using the validated stock code result = await stock_service.fetch_stock_data(validated_params["stock_code"]) # Format response for MCP return MCPResponseFormatter.format_stock_data_response(result) @mcp_error_handler("get_price_history") async def get_price_history(stock_code: str, period: str = "1mo") -> dict[str, Any]: """ 取得股票的歷史價格資料 Args: stock_code: 股票代號,例如:2330 period: 時間區間,可選值:'1d', '5d', '1mo', '3mo', '6mo', '1y', 'ytd', 'max' Returns: 歷史價格資料 Raises: InvalidStockCodeError: 股票代號格式錯誤 ParameterValidationError: period參數錯誤 StockDataUnavailableError: 價格資料無法取得 """ # Validate request parameters validated_params = validate_stock_request( stock_code=stock_code, period=period ) # 處理period參數,轉換為天數 days_map = { "1d": 1, "5d": 5, "1mo": 30, "3mo": 90, "6mo": 180, "1y": 365, "ytd": max(1, (datetime.now() - datetime(datetime.now().year, 1, 1)).days), "max": 3650 # 約10年 } validated_period = validated_params["period"] days = days_map.get(validated_period, 30) if days <= 0: raise ParameterValidationError( parameter_name="period", parameter_value=period, message="Invalid period resulting in zero or negative days" ) # Fetch price data price_data = await stock_service.fetch_price_data( validated_params["stock_code"], days ) result = { "stock_code": validated_params["stock_code"], "period": validated_period, "data": price_data, "requested_days": days, "actual_records": len(price_data) if isinstance(price_data, list) else 0 } # Format response for MCP with enhanced structure return MCPResponseFormatter.format_price_data_response(result) @mcp_error_handler("get_best_four_points") async def get_best_four_points(stock_code: str) -> dict[str, Any]: """ 取得四大買賣點分析 Args: stock_code: 股票代號,例如:2330 Returns: 四大買賣點分析結果 Raises: InvalidStockCodeError: 股票代號格式錯誤 StockNotFoundError: 找不到股票 StockDataUnavailableError: 分析資料無法取得 """ # Validate request parameters validated_params = validate_stock_request(stock_code=stock_code) # Fetch analysis data result = await stock_service.get_best_four_points(validated_params["stock_code"]) # Format response for MCP with enhanced technical analysis structure return MCPResponseFormatter.format_technical_analysis_response(result) @mcp_error_handler("get_realtime_data") async def get_realtime_data(stock_code: str) -> dict[str, Any]: """ 取得即時股票資訊 Args: stock_code: 股票代號,例如:2330 Returns: 即時股票資訊 Raises: InvalidStockCodeError: 股票代號格式錯誤 StockNotFoundError: 找不到股票 StockDataUnavailableError: 即時資料無法取得 StockMarketClosedError: 股市休市 """ # Validate request parameters validated_params = validate_stock_request(stock_code=stock_code) # Fetch real-time data result = await stock_service.get_realtime_data(validated_params["stock_code"]) # Format response for MCP with enhanced realtime structure return MCPResponseFormatter.format_realtime_data_response(result) @mcp_error_handler("get_market_overview") async def get_market_overview() -> dict[str, Any]: """ 取得市場概況 Returns: 大盤指數、成交量、漲跌家數等資訊 Raises: StockDataUnavailableError: 市場資料無法取得 """ try: # 使用台積電(2330)作為市場指標 realtime_data = await stock_service.get_realtime_data("2330") # Check if we got valid data if "error" in realtime_data: from tw_stock_agent.exceptions import StockDataUnavailableError raise StockDataUnavailableError( stock_code="market", data_type="market overview", message="Unable to fetch market overview data" ) result = { "date": datetime.now().isoformat(), "taiex": realtime_data.get("current_price"), "volume": realtime_data.get("volume"), "updated_at": realtime_data.get("updated_at"), "market_status": realtime_data.get("market_status", "unknown"), "reference_stock": "2330" # TSMC as market reference } # Format response for MCP with enhanced market overview structure return MCPResponseFormatter.format_market_overview_response(result) except TwStockAgentError: # Re-raise our custom errors raise except Exception as e: from tw_stock_agent.exceptions import StockDataUnavailableError raise StockDataUnavailableError( stock_code="market", data_type="market overview", message=f"Failed to fetch market overview: {str(e)}" )

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/clsung/tw-stock-agent'

If you have feedback or need assistance with the MCP directory API, please join our Discord server