"""Unified schemas for stock data across all providers."""
from typing import Literal, Optional, Any
from datetime import datetime
from pydantic import BaseModel, Field
# ============================================================================
# Input Schemas
# ============================================================================
class SearchSymbolInput(BaseModel):
"""Input for search_symbol tool."""
query: str = Field(..., description="Search query for stock symbols")
limit: int = Field(10, ge=1, le=50, description="Maximum number of results")
market_hint: Optional[str] = Field(None, description="Market hint (US, HK, CN)")
class GetQuoteInput(BaseModel):
"""Input for get_quote tool."""
symbol: str = Field(..., description="Stock symbol (e.g., AAPL, 0700.HK, 600519.SS)")
prefer: Optional[Literal["yahoo", "finnhub", "futu", "auto"]] = Field(None, description="Preferred data source")
class GetHistoryInput(BaseModel):
"""Input for get_history tool."""
symbol: str = Field(..., description="Stock symbol")
start_date: str = Field(..., description="Start date in YYYY-MM-DD format")
end_date: str = Field(..., description="End date in YYYY-MM-DD format")
interval: Literal["1d", "1wk", "1mo"] = Field("1d", description="Data interval")
adjust: Literal["none", "qfq", "hfq"] = Field("qfq", description="Adjustment type (none, qfq=forward, hfq=backward)")
class GetFundamentalsInput(BaseModel):
"""Input for get_fundamentals tool."""
symbol: str = Field(..., description="Stock symbol")
period: Literal["ttm", "annual"] = Field("ttm", description="Period type")
class GetFinancialStatementsInput(BaseModel):
"""Input for get_financial_statements tool."""
symbol: str = Field(..., description="Stock symbol")
statement: Literal["income", "balance", "cashflow"] = Field(..., description="Statement type")
period: Literal["annual", "quarterly"] = Field("annual", description="Period frequency")
class ResolveSymbolInput(BaseModel):
"""Input for resolve_symbol tool."""
symbol: str = Field(..., description="Stock symbol (Yahoo or Futu format)")
class SubscribeQuoteInput(BaseModel):
"""Input for subscribe_quote tool."""
symbols: list[str] = Field(..., description="List of stock symbols")
sub_types: list[Literal["QUOTE", "KLINE", "ORDER_BOOK", "TICKER"]] = Field(..., description="Subscription types")
enable: bool = Field(True, description="Enable or disable subscription")
class GetOrderBookInput(BaseModel):
"""Input for get_order_book tool."""
symbol: str = Field(..., description="Stock symbol")
num: int = Field(10, description="Number of entries (1-10)")
class GetTickerInput(BaseModel):
"""Input for get_ticker tool."""
symbol: str = Field(..., description="Stock symbol")
num: int = Field(100, description="Number of entries (1-1000)")
# ============================================================================
# Output Schemas
# ============================================================================
class SymbolSearchResult(BaseModel):
"""A single symbol search result."""
symbol: str
name: str
market: Optional[str] = None
exchange: Optional[str] = None
source: str = "finnhub"
yahoo_symbol: Optional[str] = None
futu_code: Optional[str] = None
class Quote(BaseModel):
"""Unified quote data."""
symbol: str
name: Optional[str] = None
market: Optional[str] = None
exchange: Optional[str] = None
currency: Optional[str] = None
price: Optional[float] = None
change: Optional[float] = None
change_pct: Optional[float] = None
open: Optional[float] = None
high: Optional[float] = None
low: Optional[float] = None
prev_close: Optional[float] = None
volume: Optional[int] = None
timestamp: str
source: str
source_detail: Optional[dict] = None
class HistoryRecord(BaseModel):
"""A single historical data point."""
date: str
open: float
high: float
low: float
close: float
volume: int
class Fundamentals(BaseModel):
"""Unified fundamental metrics."""
symbol: str
currency: Optional[str] = None
# Valuation
market_cap: Optional[float] = None
pe: Optional[float] = None
pb: Optional[float] = None
ps: Optional[float] = None
dividend_yield: Optional[float] = None
# Profitability
roe: Optional[float] = None
roa: Optional[float] = None
gross_margin: Optional[float] = None
operating_margin: Optional[float] = None
net_margin: Optional[float] = None
# Financial Health
debt_to_equity: Optional[float] = None
current_ratio: Optional[float] = None
quick_ratio: Optional[float] = None
# Performance
revenue_ttm: Optional[float] = None
net_income_ttm: Optional[float] = None
free_cash_flow_ttm: Optional[float] = None
updated_at: str
source: str = "finnhub"
raw: Optional[dict] = None
class FinancialStatementItem(BaseModel):
"""A single period in a financial statement."""
period_end: str
revenue: Optional[float] = None
gross_profit: Optional[float] = None
operating_income: Optional[float] = None
net_income: Optional[float] = None
eps: Optional[float] = None
# Balance sheet specific
total_assets: Optional[float] = None
total_liabilities: Optional[float] = None
total_equity: Optional[float] = None
cash: Optional[float] = None
# Cashflow specific
operating_cash_flow: Optional[float] = None
investing_cash_flow: Optional[float] = None
financing_cash_flow: Optional[float] = None
free_cash_flow: Optional[float] = None
raw: Optional[dict] = None
class FinancialStatements(BaseModel):
"""Financial statements data."""
symbol: str
statement: Literal["income", "balance", "cashflow"]
period: Literal["annual", "quarterly"]
currency: Optional[str] = None
items: list[FinancialStatementItem]
source: str = "finnhub"
# ============================================================================
# Error Schema
# ============================================================================
class ErrorDetail(BaseModel):
"""Structured error response."""
code: Literal[
"PROVIDER_TIMEOUT",
"SYMBOL_NOT_FOUND",
"INVALID_ARGUMENT",
"PROVIDER_RATE_LIMIT",
"PROVIDER_ERROR",
"MISSING_API_KEY",
"NO_DATA",
"PROVIDER_CONNECT_FAILED",
"HISTORY_NOT_AVAILABLE",
"SUBSCRIPTION_LIMIT",
"INVALID_SYMBOL_FORMAT"
]
message: str
details: Optional[dict] = None
