Updated fmp_client.py file

Author: dimtics

.gitignore

@@ -1,5 +1,6 @@
 # Python-generated files
 __pycache__/
+.mypy_cache
 *.py[cod]
 *$py.class
 *.so

pyproject.toml

@@ -8,6 +8,7 @@ authors = [
 ]
 requires-python = ">=3.12"
 dependencies = [
+    "asyncio>=3.4.3",
     "dash>=2.18.2",
     "duckdb>=1.1.3",
     "fastapi>=0.115.4",
@@ -33,3 +34,6 @@ dev = [
     "pytest>=8.3.3",
     "ruff>=0.7.3",
 ]
+
+[tool.setuptools]
+package-dir = {"" = "src"}

src/stock_valuation_app/config.yml src/config.ymlsrc/stock_valuation_app/config.yml renamed to src/config.yml

@@ -1,8 +1,8 @@
 api:
   base_url: "https://financialmodelingprep.com/api/v3"
-  annual_ratios: "ratios/"
-  annual_financial_growth: "financial-growth/"
-  rating: "rating/"
+  annual_ratios: "ratios"
+  annual_financial_growth: "financial-growth"
+  rating: "rating"
 
 valuation:
   default_pe_ratio: 15

src/main.py

@@ -0,0 +1,13 @@
+import asyncio
+from stock_valuation_app.data import fmp_client
+
+async def main() -> None:
+    """Main function to fetch data from the Financial Modeling Prep API."""
+    api_client = fmp_client.FMPClient()
+    rating_endpoint = fmp_client.get_endpoint("annual_ratios")
+    data = await api_client.fetch_data(rating_endpoint, "AAPL")
+    print(data)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/stock_valuation_app/__init__.py

@@ -1,9 +0,0 @@
-from fastapi import FastAPI
-from .api.routes import router
-from .ui.dashboard import app as dash_app
-
-app = FastAPI()
-app.include_router(router)
-
-# Mount Dash app
-app.mount("/dashboard", dash_app.server)

src/stock_valuation_app/api/__init__.py

@@ -0,0 +1,2 @@
+from fastapi import APIRouter, Depends
+from stock_valuation_app.data.fmp_client import FMPClient

src/stock_valuation_app/api/routes.py

@@ -1,27 +1,63 @@
+import os
+import asyncio
 from dataclasses import dataclass, field
-from urllib.parse import urljoin
+from typing import Any
 import httpx
-from typing import Any, Optional
-from pydantic import BaseModel
+import utils
+from stock_valuation_app.models.stock import CombinedModel
+
+
+def get_endpoint(url: str) -> str:
+    """Get endpoints from the configuration."""
+    api_config = utils.get_section_config("api")
+    return api_config.get(f"{url}")
+
+
+def get_base_url() -> str:
+    """Get the base URL from the configuration."""
+    return get_endpoint("base_url")
+
+
+def get_api_key() -> str:
+    """Get the API key from the .env file."""
+    return os.getenv("FMP_API_KEY", "")
 
 
 @dataclass
 class FMPClient:
     """A client for interacting with the Financial Modeling Prep API."""
-    base_url: str = field(default_factory=utils.get_base_url)
-    api_key: str = field(default_factory=utils.get_api_key)
+    base_url: str = field(default_factory=get_base_url)
+    api_key: str = field(default_factory=get_api_key)
+    metric_types: list[str] = field(default_factory=lambda: ["profile", "rating", "ratios", "key-metrics", "financial-growth",])
 
+    async def get_data(self, client: httpx.Client, url: str) -> dict[str, Any]:
+        """Call API endpoint asynchronously"""
+        response = await client.get(url)
+        data = response.json()
+        return data
 
-    async def fetch_data(self, endpoint: str, symbol: str, period: Optional[str] = "annual"):  # params: dict[str, Any]
-        """Fetch data from the Financial Modeling Prep API."""
-        base_endpoint = urljoin(self.base_url, endpoint)
-        if period is None:
-            url = f"{base_endpoint}/{symbol}?apikey={self.api_key}"
-        else:
-            url = f"{base_endpoint}/{symbol}?period={period}&apikey={self.api_key}"
+    async def fetch_data(self, ticker: str) -> dict[str, list[dict[str, Any]]]:
+        urls = []
+        for metric in self.metric_types:
+            if metric in ["profile", "rating"]:
+                endpoint = f"{self.base_url}/{metric}/{ticker}?apikey={self.api_key}"
+            else:
+                endpoint = f"{self.base_url}/{metric}/{ticker}?period=annual&apikey={self.api_key}"
+            urls.append(endpoint)
 
         async with httpx.AsyncClient() as client:
-            response = await client.get(url)
-            response.raise_for_status()
-            data = response.json()
-            return data
+            tasks = []
+            for url in urls:
+                tasks.append(asyncio.create_task(self.get_data(client, url)))
+
+            results = await asyncio.gather(*tasks)
+
+            # Rename some metric types to match with fields defined in the CombinedModel
+            replace_metric_types = {"key-metrics": "key_metrics", "financial-growth": "growth",}
+            new_metric_types = [replace_metric_types.get(item, item) for item in self.metric_types]
+
+            # Create combined records dict for validation
+            records = dict(zip(new_metric_types, results))
+
+            # Validate the combined records
+            return CombinedModel(**records).model_dump()