get_price_history
Retrieve historical stock price data for Taiwan-listed companies to analyze market trends and performance over specified time periods.
Instructions
Get historical price data for a specific stock.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| stock_code | Yes | ||
| period | No | 1mo |
Implementation Reference
- mcp_server.py:104-131 (handler)MCP tool handler registered with @mcp.tool(name="get_price_history"). Calls core get_price_history helper, handles exceptions, formats response with PriceHistoryResponse Pydantic model.
@mcp.tool(name="get_price_history", description="Get historical price data for a specific stock.", ) async def get_price_history_tool( stock_code: str, period: str = "1mo" ) -> PriceHistoryResponse: """Get historical price data for a specific stock.""" try: raw_data = await get_price_history(stock_code, period) # Extract clean data without _metadata for Pydantic model from tw_stock_agent.utils.mcp_error_handler import MCPResponseFormatter clean_data = MCPResponseFormatter.extract_metadata_for_model(raw_data) return PriceHistoryResponse(**clean_data) except TwStockAgentError as e: # Return error response in proper format return PriceHistoryResponse( stock_code=stock_code, period=period, data=[], error=e.message ) except Exception as e: return PriceHistoryResponse( stock_code=stock_code, period=period, data=[], error=f"Unexpected error: {str(e)}" ) - Core implementation of get_price_history: validates parameters, maps period to days, fetches historical price data from StockService, constructs and formats the response dictionary.
@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) - Pydantic model PriceHistoryResponse defining the structure, validation, and computed fields for the tool's output response.
class PriceHistoryResponse(BaseStockResponse): """Enhanced response model for price history data.""" period: str = Field( ..., description="Time period (時間區間)", example="1mo" ) data: List[PriceDataPoint] = Field( default_factory=list, description="Price data points (價格資料)" ) requested_days: Optional[int] = Field( None, description="Number of days requested", alias="requestedDays" ) actual_records: Optional[int] = Field( None, description="Actual number of records returned", alias="actualRecords" ) def __init__(self, **data): super().__init__(**data) self.metadata.data_type = "price_history" self.metadata.record_count = len(self.data) @field_validator("period") @classmethod def validate_period(cls, v: str) -> str: """Validate period format.""" valid_periods = {'1d', '5d', '1mo', '3mo', '6mo', '1y', 'ytd', 'max'} if v not in valid_periods: raise ValueError(f"Invalid period. Must be one of: {', '.join(valid_periods)}") return v @computed_field @property def price_summary(self) -> Optional[Dict[str, Any]]: """Calculate price summary statistics.""" if not self.data: return None prices = [] volumes = [] for point in self.data: if point.close_price: prices.append(point.close_price.amount) if point.volume: volumes.append(point.volume) if not prices: return None return { "min_price": float(min(prices)), "max_price": float(max(prices)), "avg_price": float(sum(prices) / len(prices)), "total_volume": sum(volumes) if volumes else None, "avg_volume": sum(volumes) / len(volumes) if volumes else None, "trading_days": len(self.data) } @computed_field @property def overall_change(self) -> Optional[Dict[str, Any]]: """Calculate overall change for the period.""" if len(self.data) < 2: return None first_price = None last_price = None # Find first valid price for point in self.data: if point.close_price: first_price = point.close_price.amount break # Find last valid price for point in reversed(self.data): if point.close_price: last_price = point.close_price.amount break if not first_price or not last_price or first_price == 0: return None change_amount = last_price - first_price change_percentage = float((change_amount / first_price) * 100) return { "period_start_price": float(first_price), "period_end_price": float(last_price), "change_amount": float(change_amount), "change_percentage": change_percentage, "is_positive": change_amount > 0 }