summaryrefslogtreecommitdiffstats
path: root/openbb_platform/providers/fmp/openbb_fmp/models/etf_holdings_performance.py
diff options
context:
space:
mode:
Diffstat (limited to 'openbb_platform/providers/fmp/openbb_fmp/models/etf_holdings_performance.py')
-rw-r--r--openbb_platform/providers/fmp/openbb_fmp/models/etf_holdings_performance.py77
1 files changed, 0 insertions, 77 deletions
diff --git a/openbb_platform/providers/fmp/openbb_fmp/models/etf_holdings_performance.py b/openbb_platform/providers/fmp/openbb_fmp/models/etf_holdings_performance.py
deleted file mode 100644
index 0a6f18b8339..00000000000
--- a/openbb_platform/providers/fmp/openbb_fmp/models/etf_holdings_performance.py
+++ /dev/null
@@ -1,77 +0,0 @@
-"""FMP ETF Holdings Performance Model."""
-
-# pylint: disable=unused-argument
-
-from typing import Any, Dict, List, Optional
-
-from openbb_core.provider.abstract.fetcher import Fetcher
-from openbb_fmp.models.etf_holdings import FMPEtfHoldingsFetcher
-from openbb_fmp.models.price_performance import (
- FMPPricePerformanceData,
- FMPPricePerformanceFetcher,
- FMPPricePerformanceQueryParams,
-)
-from pandas import DataFrame
-
-
-class FMPEtfHoldingsPerformanceQueryParams(FMPPricePerformanceQueryParams):
- """FMP ETF Holdings Performance Query."""
-
-
-class FMPEtfHoldingsPerformanceData(FMPPricePerformanceData):
- """FMP ETF Holdings Performance Data."""
-
-
-class FMPEtfHoldingsPerformanceFetcher(
- Fetcher[
- FMPEtfHoldingsPerformanceQueryParams,
- List[FMPEtfHoldingsPerformanceData],
- ]
-):
- """Transform the query, extract and transform the data from the FMP endpoints."""
-
- @staticmethod
- def transform_query(params: Dict[str, Any]) -> FMPEtfHoldingsPerformanceQueryParams:
- """Transform the query."""
- return FMPEtfHoldingsPerformanceQueryParams(**params)
-
- @staticmethod
- async def aextract_data(
- query: FMPEtfHoldingsPerformanceQueryParams,
- credentials: Optional[Dict[str, str]],
- **kwargs: Any,
- ) -> List[Dict]:
- """Return the raw data from the FMP endpoint."""
- # Get the holdings data
- holdings = await FMPEtfHoldingsFetcher().aextract_data(
- FMPEtfHoldingsFetcher.transform_query({"symbol": query.symbol}),
- credentials,
- **kwargs,
- )
- if holdings is None:
- raise RuntimeError(f"No holdings data found for {query.symbol}.")
- holdings_list = DataFrame(holdings).asset.unique().tolist()
- # Split into chunks of 500
- chunks = [holdings_list[i : i + 500] for i in range(0, len(holdings_list), 500)]
- # Get price performance for the holdings
- holdings_performance: List[Dict] = []
- for holding_chunk in chunks:
- holdings_str = (
- ",".join(holding_chunk) if len(holding_chunk) > 1 else holding_chunk[0]
- )
- _performance = await FMPPricePerformanceFetcher().aextract_data(
- FMPPricePerformanceFetcher.transform_query({"symbol": holdings_str}),
- credentials,
- **kwargs,
- )
- holdings_performance.extend(_performance)
- return holdings_performance
-
- @staticmethod
- def transform_data(
- query: FMPEtfHoldingsPerformanceQueryParams,
- data: List[Dict],
- **kwargs: Any,
- ) -> List[FMPEtfHoldingsPerformanceData]:
- """Return the transformed data."""
- return [FMPEtfHoldingsPerformanceData.model_validate(d) for d in data]