partial fix fmp models

This commit is contained in:
Diogo Sousa
2023-07-29 16:37:23 +01:00
parent d3c398bf7f
commit d28cbda41b
38 changed files with 48 additions and 92 deletions

View File

@@ -1,7 +1,7 @@
"""FMP Analyst Estimates fetcher."""
from typing import Dict, List, Optional
from typing import Any, Dict, List, Optional
from openbb_provider.abstract.fetcher import Fetcher
from openbb_provider.helpers import data_transformer
@@ -57,19 +57,14 @@ class FMPAnalystEstimatesFetcher(
]
):
@staticmethod
def transform_query(
query: AnalystEstimatesQueryParams, extra_params: Optional[Dict] = None
) -> FMPAnalystEstimatesQueryParams:
return FMPAnalystEstimatesQueryParams(
symbol=query.symbol, period=query.period, limit=query.limit
)
def transform_query(params: Dict[str, Any]) -> FMPAnalystEstimatesQueryParams:
return FMPAnalystEstimatesQueryParams(**params)
@staticmethod
def extract_data(
query: FMPAnalystEstimatesQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPAnalystEstimatesData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
query.period = "quarter" if query.period == "quarterly" else "annual"
@@ -81,5 +76,5 @@ class FMPAnalystEstimatesFetcher(
@staticmethod
def transform_data(
data: List[FMPAnalystEstimatesData],
) -> List[AnalystEstimatesData]:
return data_transformer(data, AnalystEstimatesData)
) -> List[FMPAnalystEstimatesData]:
return data

View File

@@ -1,7 +1,7 @@
"""FMP Available Indices fetcher."""
from typing import Dict, List, Optional
from typing import Any, Dict, List, Optional
from openbb_provider.abstract.fetcher import Fetcher
from openbb_provider.helpers import data_transformer
@@ -39,17 +39,14 @@ class FMPAvailableIndicesFetcher(
]
):
@staticmethod
def transform_query(
query: AvailableIndicesQueryParams, extra_params: Optional[Dict] = None
) -> FMPAvailableIndicesQueryParams:
return FMPAvailableIndicesQueryParams()
def transform_query(params: Dict[str, Any]) -> FMPAvailableIndicesQueryParams:
return FMPAvailableIndicesQueryParams(**params)
@staticmethod
def extract_data(
query: FMPAvailableIndicesQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPAvailableIndicesData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
base_url = "https://financialmodelingprep.com/api/v3"
url = f"{base_url}/symbol/available-indexes?apikey={api_key}"
@@ -58,5 +55,5 @@ class FMPAvailableIndicesFetcher(
@staticmethod
def transform_data(
data: List[FMPAvailableIndicesData],
) -> List[AvailableIndicesData]:
return data_transformer(data, AvailableIndicesData)
) -> List[FMPAvailableIndicesData]:
return data

View File

@@ -130,8 +130,7 @@ class FMPBalanceSheetFetcher(
def extract_data(
query: FMPBalanceSheetQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPBalanceSheetData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
3, f"balance-sheet-statement/{query.symbol}", api_key, query, ["symbol"]

View File

@@ -119,8 +119,7 @@ class FMPCashFlowStatementFetcher(
def extract_data(
query: FMPCashFlowStatementQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPCashFlowStatementData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
3, f"cash-flow-statement/{query.symbol}", api_key, query, ["symbol"]

View File

@@ -61,8 +61,7 @@ class FMPCompanyOverviewFetcher(
def extract_data(
query: FMPCompanyOverviewQueryParams, credentials: Optional[Dict[str, str]]
) -> FMPCompanyOverviewData:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
base_url = "https://financialmodelingprep.com/api/v3/"
request_url = f"{base_url}profile/{query.symbol}?apikey={api_key}"

View File

@@ -76,8 +76,7 @@ class FMPCryptoEODFetcher(
def extract_data(
query: FMPCryptoEODQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPCryptoEODData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
base_url = "https://financialmodelingprep.com/api/v3/"
query_str = get_querystring(query.dict(), ["symbol"])

View File

@@ -75,8 +75,7 @@ class FMPDividendCalendarFetcher(
def extract_data(
query: FMPDividendCalendarQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPDividendCalendarData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
base_url = "https://financialmodelingprep.com/api/v3"
query_str = get_querystring(query.dict(), [])

View File

@@ -64,8 +64,7 @@ class FMPEarningsCalendarFetcher(
def extract_data(
query: FMPEarningsCalendarQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPEarningsCalendarData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
3, f"historical/earning_calendar/{query.symbol}", api_key, query, ["symbol"]

View File

@@ -56,8 +56,7 @@ class FMPEarningsCallTranscriptFetcher(
query: FMPEarningsCallTranscriptQueryParams,
credentials: Optional[Dict[str, str]],
) -> List[FMPEarningsCallTranscriptData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
4,

View File

@@ -61,8 +61,7 @@ class FMPESGRiskRatingFetcher(
def extract_data(
query: FMPESGRiskRatingQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPESGRiskRatingData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
4, "esg-environmental-social-governance-data-ratings", api_key, query

View File

@@ -52,8 +52,7 @@ class FMPESGScoreFetcher(
def extract_data(
query: FMPESGScoreQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPESGScoreData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(4, "esg-environmental-social-governance-data", api_key, query)
return get_data_many(url, FMPESGScoreData)

View File

@@ -54,8 +54,7 @@ class FMPESGSectorFetcher(
def extract_data(
query: FMPESGSectorQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPESGSectorData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
4, "esg-environmental-social-governance-sector-benchmark", api_key, query

View File

@@ -65,8 +65,7 @@ class FMPExecutiveCompensationFetcher(
query: FMPExecutiveCompensationQueryParams,
credentials: Optional[Dict[str, str]],
) -> List[FMPExecutiveCompensationData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(4, "governance/executive_compensation", api_key, query)
return get_data_many(url, FMPExecutiveCompensationData)

View File

@@ -78,8 +78,7 @@ class FMPKeyExecutivesFetcher(
def extract_data(
query: FMPKeyExecutivesQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPKeyExecutivesData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
base_url = "https://financialmodelingprep.com/api/v3/"
request_url = f"{base_url}key-executives/{query.symbol}?apikey={api_key}"

View File

@@ -74,8 +74,7 @@ class FMPForexEODFetcher(
def extract_data(
query: FMPForexEODQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPForexEODData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
base_url = "https://financialmodelingprep.com/api/v3"
query_str = get_querystring(query.dict(), ["symbol"])

View File

@@ -51,8 +51,7 @@ class FMPForexPairsFetcher(
def extract_data(
query: FMPForexPairsQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPForexPairsData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
base_url = "https://financialmodelingprep.com/api/v3"
url = f"{base_url}/symbol/available-forex-currency-pairs?apikey={api_key}"

View File

@@ -42,8 +42,7 @@ class FMPGlobalNewsFetcher(
def extract_data(
query: FMPGlobalNewsQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPGlobalNewsData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(4, "general_news", api_key, query)
return get_data_many(url, FMPGlobalNewsData)

View File

@@ -75,8 +75,7 @@ class FMPHistoricalDividendsFetcher(
def extract_data(
query: FMPHistoricalDividendsQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPHistoricalDividendsData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
3, f"historical-price-full/stock_dividend/{query.symbol}", api_key

View File

@@ -74,8 +74,7 @@ class FMPHistoricalEmployeesFetcher(
def extract_data(
query: FMPHistoricalEmployeesQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPHistoricalEmployeesData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(4, "historical/employee_count", api_key, query)
return get_data_many(url, FMPHistoricalEmployeesData)

View File

@@ -48,8 +48,7 @@ class FMPHistoricalStockSplitsFetcher(
query: FMPHistoricalStockSplitsQueryParams,
credentials: Optional[Dict[str, str]],
) -> List[FMPHistoricalStockSplitsData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
3, f"historical-price-full/stock_split/{query.symbol}", api_key

View File

@@ -117,8 +117,7 @@ class FMPIncomeStatementFetcher(
def extract_data(
query: FMPIncomeStatementQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPIncomeStatementData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
3, f"income-statement/{query.symbol}", api_key, query, ["symbol"]

View File

@@ -107,8 +107,7 @@ class FMPInstitutionalOwnershipFetcher(
query: FMPInstitutionalOwnershipQueryParams,
credentials: Optional[Dict[str, str]],
) -> List[FMPInstitutionalOwnershipData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(4, "institutional-ownership/symbol-ownership", api_key, query)
return get_data_many(url, FMPInstitutionalOwnershipData)

View File

@@ -123,8 +123,7 @@ class FMPKeyMetricsFetcher(
def extract_data(
query: FMPKeyMetricsQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPKeyMetricsData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
3, f"key-metrics/{query.symbol}", api_key, query, exclude=["symbol"]

View File

@@ -79,8 +79,7 @@ class FMPMajorIndicesEODFetcher(
def extract_data(
query: FMPMajorIndicesEODQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPMajorIndicesEODData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
base_url = "https://financialmodelingprep.com/api/v3"
query_str = get_querystring(query.dict(), ["symbol"])

View File

@@ -57,8 +57,7 @@ class FMPPriceTargetFetcher(
def extract_data(
query: FMPPriceTargetQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPPriceTargetData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(4, "price-target", api_key, query)
return get_data_many(url, FMPPriceTargetData)

View File

@@ -63,8 +63,7 @@ class FMPPriceTargetConsensusFetcher(
def extract_data(
query: FMPPriceTargetConsensusQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPPriceTargetConsensusData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(4, "price-target-consensus", api_key, query)
return get_data_many(url, FMPPriceTargetConsensusData)

View File

@@ -57,8 +57,7 @@ class FMPRevenueBusinessLineFetcher(
def extract_data(
query: FMPRevenueBusinessLineQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPRevenueBusinessLineData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
# Type has to be ignored below because we are using something different than the literal type
query.period = "quarter" if query.period == "quarterly" else "annual" # type: ignore

View File

@@ -59,8 +59,7 @@ class FMPRevenueGeographicFetcher(
def extract_data(
query: FMPRevenueGeographicQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPRevenueGeographicData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
query.period = "quarter" if query.period == "quarterly" else "annual" # type: ignore
url = create_url(4, "revenue-geographic-segmentation", api_key, query)

View File

@@ -68,8 +68,7 @@ class FMPSECFilingsFetcher(
def extract_data(
query: FMPSECFilingsQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPSECFilingsData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
3, f"sec_filings/{query.symbol}", api_key, query, exclude=["symbol"]

View File

@@ -64,8 +64,7 @@ class FMPShareStatisticsFetcher(
def extract_data(
query: FMPShareStatisticsQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPShareStatisticsData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(4, "shares_float", api_key, query)
return get_data_many(url, FMPShareStatisticsData)

View File

@@ -76,8 +76,7 @@ class FMPStockEODFetcher(
def extract_data(
query: FMPStockEODQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPStockEODData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
base_url = "https://financialmodelingprep.com/api/v3"
query_str = get_querystring(query.dict(), ["symbol"])

View File

@@ -83,8 +83,7 @@ class FMPStockInsiderTradingFetcher(
def extract_data(
query: FMPStockInsiderTradingQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPStockInsiderTradingData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
# This changes the actual type of a pydantic class, but its a quick and clean way to format properly
query.transactionType = ",".join(query.transactionType) # type: ignore

View File

@@ -144,8 +144,7 @@ class FMPStockMultiplesFetcher(
def extract_data(
query: FMPStockMultiplesQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPStockMultiplesData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
3, f"key-metrics-ttm/{query.symbol}", api_key, query, exclude=["symbol"]

View File

@@ -56,8 +56,7 @@ class FMPStockNewsFetcher(
def extract_data(
query: FMPStockNewsQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPStockNewsData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(3, "stock_news", api_key, query)
return get_data_many(url, FMPStockNewsData)

View File

@@ -51,8 +51,7 @@ class FMPStockOwnershipFetcher(
def extract_data(
query: FMPStockOwnershipQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPStockOwnershipData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(
4,

View File

@@ -46,8 +46,7 @@ class FMPStockPeersFetcher(
def extract_data(
query: FMPStockPeersQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPStockPeersData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
url = create_url(4, "stock_peers", api_key, query)
return get_data_many(url, FMPStockPeersData)

View File

@@ -59,8 +59,7 @@ class FMPStockSplitCalendarFetcher(
def extract_data(
query: FMPStockSplitCalendarQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPStockSplitCalendarData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
query_str = f"from={query.start_date}&to={query.end_date}"
url = create_url(3, f"stock_split_calendar?{query_str}", api_key)

View File

@@ -66,8 +66,7 @@ class FMPTreasuryRatesFetcher(
def extract_data(
query: FMPTreasuryRatesQueryParams, credentials: Optional[Dict[str, str]]
) -> List[FMPTreasuryRatesData]:
if credentials:
api_key = credentials.get("fmp_api_key")
api_key = credentials.get("fmp_api_key") if credentials else ""
base_url = "https://financialmodelingprep.com/api/v4/"
query_str = get_querystring(query.dict(), [])