"""Market movers tools for Schwab MCP Server."""
from typing import Any
from ..client import SchwabClient
GET_MOVERS_SCHEMA = {
"type": "object",
"properties": {
"index": {
"type": "string",
"enum": [
"$DJI",
"$COMPX",
"$SPX",
"NYSE",
"NASDAQ",
"OTCBB",
"INDEX_ALL",
"EQUITY_ALL",
"OPTION_ALL",
"OPTION_PUT",
"OPTION_CALL",
],
"description": (
"Market index: $DJI (Dow Jones), $COMPX (NASDAQ Composite), "
"$SPX (S&P 500), NYSE, NASDAQ, OTCBB, INDEX_ALL, EQUITY_ALL, "
"OPTION_ALL, OPTION_PUT, OPTION_CALL"
),
"default": "$DJI",
},
"direction": {
"type": "string",
"enum": ["up", "down"],
"description": "Mover direction: up (gainers) or down (losers)",
"default": "up",
},
"change": {
"type": "string",
"enum": ["percent", "value"],
"description": "Metric used to rank movers (percent or value change)",
"default": "percent",
},
},
"required": [],
}
def _parse_mover(mover: dict) -> dict[str, Any]:
"""Parse a single mover from API response."""
return {
"symbol": mover.get("symbol"),
"description": mover.get("description"),
# API returns 'last' not 'lastPrice'
"last_price": mover.get("last")
or mover.get("lastPrice")
or mover.get("lastPriceInDouble"),
# API returns 'change' as the change value
"change": mover.get("change")
or mover.get("netChange")
or mover.get("changeInDouble"),
"change_percent": mover.get("netPercentChange")
or mover.get("percentChange")
or mover.get("percentChangeInDouble"),
"direction": mover.get("direction"),
"volume": mover.get("totalVolume") or mover.get("volume"),
"market_share": mover.get("marketShare"),
"trades": mover.get("trades"),
}
async def get_movers(client: SchwabClient, args: dict) -> dict[str, Any]:
"""
Get market movers for an index.
Args:
client: SchwabClient instance
args: Tool arguments with optional 'index', 'direction', and 'change'
Returns:
Market movers data
"""
index = args.get("index", "$DJI")
direction = str(args.get("direction", "up")).lower()
change = str(args.get("change", "percent")).lower()
response = await client.get_movers(index, direction=direction, change=change)
raw_movers: list[dict] = []
if isinstance(response, list):
raw_movers = response
elif isinstance(response, dict):
if isinstance(response.get("screeners"), list):
raw_movers = response["screeners"]
elif isinstance(response.get("movers"), list):
raw_movers = response["movers"]
movers = [_parse_mover(m) for m in raw_movers]
return {
"index": index,
"direction": direction,
"change": change,
"count": len(movers),
"movers": movers,
}
