refactored RevenueGeographic for stocks.fa.revgeo

This commit is contained in:
Pratyush Shukla
2023-08-02 01:35:51 +05:30
parent c9bdc0fb6e
commit 60f9033909
2 changed files with 43 additions and 76 deletions

View File

@@ -9,32 +9,24 @@ from openbb_provider.models.revenue_geographic import (
RevenueGeographicData,
RevenueGeographicQueryParams,
)
from pydantic import validator
from openbb_fmp.utils.helpers import create_url, get_data
# This part is only provided by FMP and not by the other providers for now.
from pydantic import validator
class FMPRevenueGeographicQueryParams(RevenueGeographicQueryParams):
"""FMP Revenue Geographic QueryParams.
Source: https://financialmodelingprep.com/developer/docs/#Revenue-Geographic-Segmentation
Parameter
---------
symbol : Optional[str]
The symbol of the company if no CIK is provided.
period : Literal["annual", "quarterly"]
The period of the income statement. Default is "annual".
structure : Literal["hierarchical", "flat"]
The structure of the revenue geographic. Default is "flat".
Source: https://site.financialmodelingprep.com/developer/docs/revenue-geographic-by-segments-api/
"""
class FMPRevenueGeographicData(RevenueGeographicData):
"""FMP Revenue Geographic Data."""
@validator("date", pre=True)
def time_validate(cls, v): # pylint: disable=E0213
def date_validate(cls, v): # pylint: disable=E0213
return datetime.strptime(v, "%Y-%m-%d")
@@ -48,11 +40,7 @@ class FMPRevenueGeographicFetcher(
):
@staticmethod
def transform_query(params: Dict[str, Any]) -> FMPRevenueGeographicQueryParams:
transformed_params = params
transformed_params["period"] = (
"annual" if params.get("period", "") == "annually" else "quarter"
)
return FMPRevenueGeographicQueryParams(**transformed_params)
return FMPRevenueGeographicQueryParams(**params)
@staticmethod
def extract_data(
@@ -60,31 +48,29 @@ class FMPRevenueGeographicFetcher(
) -> List[FMPRevenueGeographicData]:
api_key = credentials.get("fmp_api_key") if credentials else ""
query.period = "annual" if query.period == "annually" else "quarter"
url = create_url(4, "revenue-geographic-segmentation", api_key, query)
data = get_data(url)
if isinstance(data, dict):
raise ValueError("Expected list of dicts, got dict")
data = [
{**v, "date": k} for d in data for k, v in d.items() if isinstance(v, dict)
]
data = [
{
"date": d.get("date"),
"americas": d.get("Americas Segment")
or d.get("North America Segment")
or d.get("Americas")
or d.get("North America"),
"europe": d.get("Europe Segment") or d.get("Europe"),
"greater_china": d.get("Greater China Segment")
or d.get("Greater China"),
"japan": d.get("Japan Segment") or d.get("Japan") or d.get("J P"),
"rest_of_asia_pacific": d.get("Rest of Asia Pacific Segment")
or d.get("Rest of Asia Pacific")
or d.get("Asia-Pacific"),
**v,
"date": k,
"americas": v.get("Americas Segment", v.get("North America")),
"europe": v.get("Europe Segment", v.get("Europe")),
"greater_china": v.get("Greater China Segment", v.get("Greater China")),
"japan": v.get("Japan Segment", v.get("Japan", v.get("J P"))),
"rest_of_asia_pacific": v.get(
"Rest of Asia Pacific Segment", v.get("Asia-Pacific")
),
}
for d in data
for k, v in d.items()
if isinstance(v, dict)
]
return [FMPRevenueGeographicData(**d) for d in data]
@@ -93,5 +79,4 @@ class FMPRevenueGeographicFetcher(
def transform_data(
data: List[FMPRevenueGeographicData],
) -> List[FMPRevenueGeographicData]:
# Parse data to RevenueGeographicData
return data

View File

@@ -1,60 +1,42 @@
"""Revenue by geographic segments data model."""
from datetime import date
from datetime import date as dateType
from typing import Literal, Optional
from pydantic import Field
from openbb_provider.abstract.data import Data
from openbb_provider.abstract.query_params import QueryParams
from openbb_provider.descriptions import QUERY_DESCRIPTIONS
from openbb_provider.descriptions import QUERY_DESCRIPTIONS, DATA_DESCRIPTIONS
from openbb_provider.models.base import BaseSymbol
from pydantic import Field
class RevenueGeographicQueryParams(QueryParams, BaseSymbol):
"""Revenue geographic query.
Parameter
---------
symbol : str
The symbol of the company.
period : Literal["quarterly", "annually"]
The period of the income statement.
structure : Literal["hierarchical", "flat"]
The structure of the revenue geographic. Should always be flat.
"""
"""Revenue by Geographic Segments Query."""
period: Literal["quarterly", "annually"] = Field(
default="quarterly", description=QUERY_DESCRIPTIONS.get("period", "")
default="annually", description=QUERY_DESCRIPTIONS.get("period", "")
)
structure: Literal["hierarchical", "flat"] = Field(
default="flat"
) # should always be flat
default="flat", description="The structure of the returned data."
) # should always be flat # should always be flat
class RevenueGeographicData(Data):
"""Revenue by geographic segments data.
"""Revenue by Geographic Segments Data."""
Returns
-------
date : date
The date of the revenue.
americas : Optional[int]
The revenue of the America segment.
europe : Optional[int]
The revenue of the Europe segment.
greater_china : Optional[int]
The revenue of the Greater China segment.
japan : Optional[int]
The revenue of the Japan segment.
rest_of_asia_pacific : Optional[int]
The revenue of the Rest of Asia Pacific segment.
"""
date: date
americas: Optional[int]
europe: Optional[int]
greater_china: Optional[int]
japan: Optional[int]
rest_of_asia_pacific: Optional[int]
date: dateType = Field(description=DATA_DESCRIPTIONS.get("date", ""))
americas: Optional[int] = Field(
description="The revenue from the the American segment."
)
europe: Optional[int] = Field(
description="The revenue from the the European segment."
)
greater_china: Optional[int] = Field(
description="The revenue from the the Greater China segment."
)
japan: Optional[int] = Field(description="The revenue from the the Japan segment.")
rest_of_asia_pacific: Optional[int] = Field(
description="The revenue from the the Rest of Asia Pacific segment."
)