Spaces:
Running
Running
File size: 2,624 Bytes
178a66c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | from __future__ import annotations
import os
from typing import Any, Optional
import httpx
DEFAULT_API_BASE_URL = "https://mbonea-investingtest.hf.space"
def get_api_base_url() -> Optional[str]:
"""Return remote API base URL, or None when MCP should use the local database."""
if os.getenv("UWEKEZAJI_MCP_USE_LOCAL", "").strip().lower() in {"1", "true", "yes"}:
return None
return os.getenv("UWEKEZAJI_API_BASE_URL", DEFAULT_API_BASE_URL).rstrip("/")
def days_to_time_range(days: int) -> str:
days = max(1, min(days, 3650))
if days <= 7:
return "1w"
if days <= 30:
return "1m"
if days <= 180:
return "6m"
if days <= 365:
return "1y"
if days <= 730:
return "2y"
return "5y"
class UwekezajiApiClient:
def __init__(self, base_url: str):
self.base_url = base_url.rstrip("/")
self._client = httpx.AsyncClient(timeout=45.0, follow_redirects=True)
async def close(self) -> None:
await self._client.aclose()
async def _get(self, path: str, params: Optional[dict[str, Any]] = None) -> dict[str, Any]:
response = await self._client.get(f"{self.base_url}{path}", params=params)
response.raise_for_status()
payload = response.json()
if not payload.get("success"):
raise RuntimeError(payload.get("message") or f"API request failed for {path}")
data = payload.get("data")
return data if isinstance(data, dict) else {}
async def list_stocks(self) -> list[dict[str, Any]]:
data = await self._get("/stocks/list")
stocks = data.get("stocks") or []
return stocks if isinstance(stocks, list) else []
async def get_prices(
self,
symbol: str,
*,
time_range: str = "6m",
limit: int = 500,
) -> list[dict[str, Any]]:
data = await self._get(
f"/stocks/{symbol.upper()}/prices",
params={"time_range": time_range, "limit": limit},
)
prices = data.get("prices") or []
return prices if isinstance(prices, list) else []
async def get_fundamentals(self, symbol: str) -> dict[str, Any]:
data = await self._get(f"/stocks/{symbol.upper()}/fundamentals")
fundamentals = data.get("fundamentals") or {}
return fundamentals if isinstance(fundamentals, dict) else {}
async def get_company(self, symbol: str) -> dict[str, Any]:
data = await self._get(f"/stocks/{symbol.upper()}/company")
company = data.get("company") or {}
return company if isinstance(company, dict) else {}
|