summaryrefslogtreecommitdiffstats
path: root/openbb_platform/providers/sec/openbb_sec/models/compare_company_facts.py
diff options
context:
space:
mode:
Diffstat (limited to 'openbb_platform/providers/sec/openbb_sec/models/compare_company_facts.py')
-rw-r--r--openbb_platform/providers/sec/openbb_sec/models/compare_company_facts.py185
1 files changed, 185 insertions, 0 deletions
diff --git a/openbb_platform/providers/sec/openbb_sec/models/compare_company_facts.py b/openbb_platform/providers/sec/openbb_sec/models/compare_company_facts.py
new file mode 100644
index 00000000000..eee4db2ad6a
--- /dev/null
+++ b/openbb_platform/providers/sec/openbb_sec/models/compare_company_facts.py
@@ -0,0 +1,185 @@
+"""SEC Compare Company Facts Model."""
+
+# pylint: disable=unused-argument
+
+from typing import Any, Dict, List, Optional, Union
+from warnings import warn
+
+from openbb_core.provider.abstract.annotated_result import AnnotatedResult
+from openbb_core.provider.abstract.fetcher import Fetcher
+from openbb_core.provider.standard_models.compare_company_facts import (
+ CompareCompanyFactsData,
+ CompareCompanyFactsQueryParams,
+)
+from openbb_core.provider.utils.descriptions import DATA_DESCRIPTIONS
+from openbb_core.provider.utils.errors import EmptyDataError
+from openbb_sec.utils.frames import (
+ FACT_CHOICES,
+ FACTS,
+ FISCAL_PERIODS,
+ get_concept,
+ get_frame,
+)
+from pydantic import Field, field_validator
+
+
+class SecCompareCompanyFactsQueryParams(CompareCompanyFactsQueryParams):
+ """SEC Compare Company Facts Query.
+
+ Source: https://www.sec.gov/edgar/sec-api-documentation
+
+ The xbrl/frames API aggregates one fact for each reporting entity
+ that is last filed that most closely fits the calendrical period requested.
+
+ Because company financial calendars can start and end on any month or day and even change in length from quarter to
+ quarter according to the day of the week, the frame data is assembled by the dates that best align with a calendar
+ quarter or year. Data users should be mindful different reporting start and end dates for facts contained in a frame.
+ """
+
+ __json_schema_extra__ = {
+ "symbol": {"multiple_items_allowed": True},
+ }
+
+ fact: FACT_CHOICES = Field(
+ default="Revenues",
+ description="Fact or concept from the SEC taxonomy, in UpperCamelCase. Defaults to, 'Revenues'."
+ + " AAPL, MSFT, GOOG, BRK-A currently report revenue as, 'RevenueFromContractWithCustomerExcludingAssessedTax'."
+ + " In previous years, they have reported as 'Revenues'.",
+ json_schema_extra={"choices": sorted(FACTS)},
+ )
+ year: Optional[int] = Field(
+ default=None,
+ description="The year to retrieve the data for. If not provided, the current year is used."
+ + " When symbol(s) are provided, excluding the year will return all reported values for the concept.",
+ )
+ fiscal_period: Optional[FISCAL_PERIODS] = Field(
+ default=None,
+ description="The fiscal period to retrieve the data for."
+ + " If not provided, the most recent quarter is used."
+ + " This parameter is ignored when a symbol is supplied.",
+ )
+ instantaneous: bool = Field(
+ default=False,
+ description="Whether to retrieve instantaneous data. See the notes above for more information."
+ + " Defaults to False. Some facts are only available as instantaneous data."
+ + "\nThe function will automatically attempt the inverse of this parameter"
+ + " if the initial fiscal quarter request fails."
+ + " This parameter is ignored when a symbol is supplied.",
+ )
+ use_cache: bool = Field(
+ default=True,
+ description="Whether to use cache for the request. Defaults to True.",
+ )
+
+ @field_validator("fact", mode="before", check_fields=False)
+ @classmethod
+ def validate_fact(cls, v):
+ """Set the default state."""
+ if not v or v == "":
+ return "Revenues"
+ return v
+
+
+class SecCompareCompanyFactsData(CompareCompanyFactsData):
+ """SEC Compare Company Facts Data."""
+
+ __alias_dict__ = {
+ "reported_date": "filed",
+ "period_beginning": "start",
+ "period_ending": "end",
+ "fiscal_year": "fy",
+ "fiscal_period": "fp",
+ "name": "entityName",
+ "accession": "accn",
+ "value": "val",
+ "location": "loc",
+ }
+
+ cik: Union[str, int] = Field(
+ description=DATA_DESCRIPTIONS.get("cik", ""),
+ )
+ location: Optional[str] = Field(
+ default=None,
+ description="Geographic location of the reporting entity.",
+ )
+ form: Optional[str] = Field(
+ default=None,
+ description="The SEC form associated with the fact or concept.",
+ )
+ frame: Optional[str] = Field(
+ default=None,
+ description="The frame ID associated with the fact or concept, if applicable.",
+ )
+ accession: str = Field(
+ description="SEC filing accession number associated with the reported fact or concept.",
+ )
+ fact: str = Field(
+ description="The display name of the fact or concept.",
+ )
+ unit: str = Field(
+ default=None,
+ description="The unit of measurement for the fact or concept.",
+ )
+
+
+class SecCompareCompanyFactsFetcher(
+ Fetcher[SecCompareCompanyFactsQueryParams, List[SecCompareCompanyFactsData]]
+):
+ """SEC Compare Company Facts Fetcher."""
+
+ @staticmethod
+ def transform_query(params: Dict[str, Any]) -> SecCompareCompanyFactsQueryParams:
+ """Transform the query."""
+ return SecCompareCompanyFactsQueryParams(**params)
+
+ @staticmethod
+ async def aextract_data(
+ query: SecCompareCompanyFactsQueryParams,
+ credentials: Optional[Dict[str, str]],
+ **kwargs: Any,
+ ) -> Dict:
+ """Return the raw data from the SEC endpoint."""
+ results: Dict = {}
+ if not query.symbol:
+ results = await get_frame(
+ fact=query.fact,
+ year=query.year,
+ fiscal_period=query.fiscal_period,
+ instantaneous=query.instantaneous,
+ use_cache=query.use_cache,
+ )
+ if query.symbol is not None:
+ if query.instantaneous is True:
+ warn(
+ "The 'instantaneous' parameter is ignored when a symbol is supplied."
+ )
+ if query.fiscal_period is not None:
+ warn(
+ "The 'fiscal_period' parameter is ignored when a symbol is supplied."
+ )
+ results = await get_concept(
+ symbol=query.symbol,
+ fact=query.fact,
+ year=query.year,
+ use_cache=query.use_cache,
+ )
+ if not results:
+ raise EmptyDataError("The request was returned empty.")
+
+ return results
+
+ @staticmethod
+ def transform_data(
+ query: SecCompareCompanyFactsQueryParams,
+ data: Dict,
+ **kwargs: Any,
+ ) -> AnnotatedResult[List[SecCompareCompanyFactsData]]:
+ """Transform the data and validate the model."""
+ if not data:
+ raise EmptyDataError("The request was returned empty.")
+ metadata = data.get("metadata")
+ results_data = data.get("data", [])
+ return AnnotatedResult(
+ result=[SecCompareCompanyFactsData.model_validate(d) for d in results_data], # type: ignore
+ metadata=metadata,
+ )