diff options
Diffstat (limited to 'openbb_platform/providers')
11 files changed, 1444 insertions, 366 deletions
diff --git a/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/__init__.py b/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/__init__.py index e4022980de8..38b5757c7b1 100644 --- a/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/__init__.py +++ b/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/__init__.py @@ -1,8 +1,12 @@ """Seeking Alpha Provider module.""" from openbb_core.provider.abstract.provider import Provider -from openbb_seeking_alpha.models.upcoming_release_days import ( - SAUpcomingReleaseDaysFetcher, +from openbb_seeking_alpha.models.calendar_earnings import SACalendarEarningsFetcher +from openbb_seeking_alpha.models.forward_eps_estimates import ( + SAForwardEpsEstimatesFetcher, +) +from openbb_seeking_alpha.models.forward_sales_estimates import ( + SAForwardSalesEstimatesFetcher, ) seeking_alpha_provider = Provider( @@ -11,7 +15,9 @@ seeking_alpha_provider = Provider( description="""Seeking Alpha is a data provider with access to news, analysis, and real-time alerts on stocks.""", fetcher_dict={ - "UpcomingReleaseDays": SAUpcomingReleaseDaysFetcher, + "CalendarEarnings": SACalendarEarningsFetcher, + "ForwardEpsEstimates": SAForwardEpsEstimatesFetcher, + "ForwardSalesEstimates": SAForwardSalesEstimatesFetcher, }, repr_name="Seeking Alpha", ) diff --git a/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/calendar_earnings.py b/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/calendar_earnings.py new file mode 100644 index 00000000000..67b6ee13339 --- /dev/null +++ b/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/calendar_earnings.py @@ -0,0 +1,144 @@ +"""Seeking Alpha Calendar Earnings Model.""" + +# pylint: disable=unused-argument + +import asyncio +import json +from datetime import datetime, timedelta +from typing import Any, Dict, List, Literal, Optional +from warnings import warn + +from openbb_core.provider.abstract.fetcher import Fetcher +from openbb_core.provider.standard_models.calendar_earnings import ( + CalendarEarningsData, + CalendarEarningsQueryParams, +) +from openbb_core.provider.utils.helpers import amake_request +from openbb_seeking_alpha.utils.helpers import HEADERS, date_range +from pydantic import Field, field_validator + + +class SACalendarEarningsQueryParams(CalendarEarningsQueryParams): + """Seeking Alpha Calendar Earnings Query. + + Source: https://seekingalpha.com/earnings/earnings-calendar + """ + + country: Literal["us", "ca"] = Field( + default="us", + description="The country to get calendar data for.", + json_schema_extra={"choices": ["us", "ca"]}, + ) + + +class SACalendarEarningsData(CalendarEarningsData): + """Seeking Alpha Calendar Earnings Data.""" + + market_cap: Optional[float] = Field( + default=None, + description="Market cap of the entity.", + ) + reporting_time: Optional[str] = Field( + default=None, + description="The reporting time - e.g. after market close.", + ) + exchange: Optional[str] = Field( + default=None, + description="The primary trading exchange.", + ) + sector_id: Optional[int] = Field( + default=None, + description="The Seeking Alpha Sector ID.", + ) + + @field_validator("report_date", mode="before", check_fields=False) + @classmethod + def validate_release_date(cls, v): + """Validate the release date.""" + v = v.split("T")[0] + return datetime.strptime(v, "%Y-%m-%d").date() + + +class SACalendarEarningsFetcher( + Fetcher[ + SACalendarEarningsQueryParams, + List[SACalendarEarningsData], + ] +): + """Seeking Alpha Calendar Earnings Fetcher.""" + + @staticmethod + def transform_query(params: Dict[str, Any]) -> SACalendarEarningsQueryParams: + """Transform the query.""" + now = datetime.today().date() + transformed_params = params + if not params.get("start_date"): + transformed_params["start_date"] = now + if not params.get("end_date"): + transformed_params["end_date"] = now + timedelta(days=3) + return SACalendarEarningsQueryParams(**transformed_params) + + @staticmethod + async def aextract_data( + query: SACalendarEarningsQueryParams, + credentials: Optional[Dict[str, str]], + **kwargs: Any, + ) -> List[Dict]: + """Return the raw data from the Seeking Alpha endpoint.""" + results: List[Dict] = [] + dates = [ + date.strftime("%Y-%m-%d") + for date in date_range(query.start_date, query.end_date) + ] + currency = "USD" if query.country == "us" else "CAD" + messages: List = [] + + async def get_date(date, currency): + """Get date for one date.""" + url = ( + f"https://seekingalpha.com/api/v3/earnings_calendar/tickers?" + f"filter%5Bselected_date%5D={date}" + f"&filter%5Bwith_rating%5D=false&filter%5Bcurrency%5D={currency}" + ) + response = await amake_request(url=url, headers=HEADERS) + # Try again if the response is blocked. + if "blockScript" in response: + response = await amake_request(url=url, headers=HEADERS) + if "blockScript" in response: + message = json.dumps(response) + messages.append(message) + warn(message) + if "data" in response: + results.extend(response.get("data")) + + await asyncio.gather(*[get_date(date, currency) for date in dates]) + + if not results: + raise RuntimeError(f"Error with the Seeking Alpha request -> {messages}") + + return results + + @staticmethod + def transform_data( + query: SACalendarEarningsQueryParams, + data: List[Dict], + **kwargs: Any, + ) -> List[SACalendarEarningsData]: + """Transform the data to the standard format.""" + transformed_data: List[SACalendarEarningsData] = [] + for row in sorted(data, key=lambda x: x["attributes"]["release_date"]): + attributes = row.get("attributes", {}) + transformed_data.append( + SACalendarEarningsData.model_validate( + { + "report_date": attributes.get("release_date"), + "reporting_time": attributes.get("release_time"), + "symbol": attributes.get("slug"), + "name": attributes.get("name"), + "market_cap": attributes.get("marketcap"), + "exchange": attributes.get("exchange"), + "sector_id": attributes.get("sector_id"), + } + ) + ) + return transformed_data diff --git a/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/forward_eps_estimates.py b/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/forward_eps_estimates.py new file mode 100644 index 00000000000..fb0c4458eca --- /dev/null +++ b/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/forward_eps_estimates.py @@ -0,0 +1,207 @@ +"""Seeking Alpha Forward EPS Estimates Model.""" + +# pylint: disable=unused-argument + +from typing import Any, Dict, List, Literal, Optional +from warnings import warn + +from openbb_core.provider.abstract.fetcher import Fetcher +from openbb_core.provider.standard_models.forward_eps_estimates import ( + ForwardEpsEstimatesData, + ForwardEpsEstimatesQueryParams, +) +from openbb_core.provider.utils.helpers import amake_request +from openbb_seeking_alpha.utils.helpers import HEADERS, get_seekingalpha_id +from pydantic import Field, field_validator + + +class SAForwardEpsEstimatesQueryParams(ForwardEpsEstimatesQueryParams): + """Seeking Alpha Forward EPS Estimates Query. + + Source: https://seekingalpha.com/earnings/earnings-calendar + """ + + __json_schema_extra__ = {"symbol": {"multiple_items_allowed": True}} + + period: Literal["annual", "quarter"] = Field( + default="quarter", + description="The reporting period.", + json_schema_extra={"choices": ["annual", "quarter"]}, + ) + + @field_validator("symbol", mode="before", check_fields=False) + @classmethod + def check_symbol(cls, value): + """Check the symbol.""" + if not value: + raise RuntimeError("Error: Symbol is a required field for Seeking Alpha.") + return value + + +class SAForwardEpsEstimatesData(ForwardEpsEstimatesData): + """Seeking Alpha Forward EPS Estimates Data.""" + + normalized_actual: Optional[float] = Field( + default=None, + description="Actual normalized EPS.", + ) + period_growth: Optional[float] = Field( + default=None, + description="Estimated (or actual if reported) EPS growth for the period.", + json_schema_extra={"x-unit_measurement": "percent", "x-frontend_multiply": 100}, + ) + low_estimate_gaap: Optional[float] = Field( + default=None, + description="Estimated GAAP EPS low for the period.", + ) + high_estimate_gaap: Optional[float] = Field( + default=None, + description="Estimated GAAP EPS high for the period.", + ) + mean_gaap: Optional[float] = Field( + default=None, + description="Estimated GAAP EPS mean for the period.", + ) + gaap_actual: Optional[float] = Field( + default=None, + description="Actual GAAP EPS.", + ) + + +class SAForwardEpsEstimatesFetcher( + Fetcher[ + SAForwardEpsEstimatesQueryParams, + List[SAForwardEpsEstimatesData], + ] +): + """Seeking Alpha Forward EPS Estimates Fetcher.""" + + @staticmethod + def transform_query(params: Dict[str, Any]) -> SAForwardEpsEstimatesQueryParams: + """Transform the query.""" + return SAForwardEpsEstimatesQueryParams(**params) + + @staticmethod + async def aextract_data( + query: SAForwardEpsEstimatesQueryParams, + credentials: Optional[Dict[str, str]], + **kwargs: Any, + ) -> Dict: + """Return the raw data from the Seeking Alpha endpoint.""" + tickers = query.symbol.split(",") + fp = query.period if query.period == "annual" else "quarterly" + url = "https://seekingalpha.com/api/v3/symbol_data/estimates" + querystring: Dict = { + "estimates_data_items": "eps_normalized_actual,eps_normalized_consensus_low,eps_normalized_consensus_mean," + "eps_normalized_consensus_high,eps_normalized_num_of_estimates," + "eps_gaap_actual,eps_gaap_consensus_low,eps_gaap_consensus_mean,eps_gaap_consensus_high,", + "period_type": fp, + "relative_periods": "-4,-3,-2,-1,0,1,2,3,4,5,6,7,8,9,10,11,12", + } + ids: Dict = {ticker: await get_seekingalpha_id(ticker) for ticker in tickers} + querystring["ticker_ids"] = (",").join(list(ids.values())) + payload: str = "" + response = await amake_request( + url, data=payload, headers=HEADERS, params=querystring + ) + estimates: Dict = response.get("estimates", {}) # type: ignore + if not estimates: + raise RuntimeError(f"No estimates data was returned for: {query.symbol}") + + output: Dict = {"ids": ids, "estimates": estimates} + + return output + + @staticmethod + def transform_data( + query: SAForwardEpsEstimatesQueryParams, + data: Dict, + **kwargs: Any, + ) -> List[SAForwardEpsEstimatesData]: + """Transform the data to the standard format.""" + tickers = query.symbol.split(",") + ids = data.get("ids") + estimates = data.get("estimates") + results: List[SAForwardEpsEstimatesData] = [] + for ticker in tickers: + sa_id = str(ids.get(ticker, "")) + if sa_id == "" or sa_id not in estimates: + warn(f"Symbol Error: No data found for, {ticker}") + seek_object = estimates.get(sa_id) + items = len(seek_object["eps_normalized_num_of_estimates"]) + for i in range(0, items - 4): + eps_estimates: Dict = {} + eps_estimates["symbol"] = ticker + num_estimates = seek_object["eps_normalized_num_of_estimates"].get(str(i)) + if not num_estimates: + continue + period = num_estimates[0].get("period", {}) + if period: + period_type = period.get("periodtypeid") + eps_estimates["calendar_year"] = period.get("calendaryear") + eps_estimates["calendar_period"] = ( + "Q" + str(period.get("calendarquarter", "")) + if period_type == "quarterly" + else "FY" + ) + eps_estimates["date"] = period.get("periodenddate").split("T")[0] + eps_estimates["fiscal_year"] = period.get("fiscalyear") + eps_estimates["fiscal_period"] = ( + "Q" + str(period.get("fiscalquarter", "")) + if period_type == "quarterly" + else "FY" + ) + eps_estimates["number_of_analysts"] = num_estimates[0].get( + "dataitemvalue" + ) + actual = seek_object["eps_normalized_actual"].get(str(i)) + if actual: + eps_estimates["normalized_actual"] = actual[0].get("dataitemvalue") + gaap_actual = seek_object["eps_gaap_actual"].get(str(i)) + if gaap_actual: + eps_estimates["gaap_actual"] = gaap_actual[0].get("dataitemvalue") + low = seek_object["eps_normalized_consensus_low"].get(str(i)) + if low: + eps_estimates["low_estimate"] = low[0].get("dataitemvalue") + gaap_low = seek_object["eps_gaap_consensus_low"].get(str(i)) + if gaap_low: + eps_estimates["low_estimate_gaap"] = gaap_low[0].get( + "dataitemvalue" + ) + high = seek_object["eps_normalized_consensus_high"].get(str(i)) + if high: + eps_estimates["high_estimate"] = high[0].get("dataitemvalue") + gaap_high = seek_object["eps_gaap_consensus_high"].get(str(i)) + if gaap_high: + eps_estimates["high_estimate_gaap"] = gaap_high[0].get( + "dataitemvalue" + ) + mean = seek_object["eps_normalized_consensus_mean"].get(str(i)) + if mean: + mean = mean[0].get("dataitemvalue") + eps_estimates["mean"] = mean + gaap_mean = seek_object["eps_gaap_consensus_mean"].get(str(i)) + if gaap_mean: + eps_estimates["mean_gaap"] = gaap_mean[0].get("dataitemvalue") + # Calculate the estimated growth percent. + this = float(mean) if mean else None + prev = None + percent = None + try: + prev = float( + seek_object["eps_normalized_actual"][str(i - 1)][0].get( + "dataitemvalue" + ) + ) + except KeyError: + prev = float( + seek_object["eps_normalized_consensus_mean"][str(i - 1)][0].get( + "dataitemvalue" + ) + ) + if this and prev: + percent = (this - prev) / prev + eps_estimates["period_growth"] = percent + results.append(SAForwardEpsEstimatesData.model_validate(eps_estimates)) + + return results diff --git a/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/forward_sales_estimates.py b/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/forward_sales_estimates.py new file mode 100644 index 00000000000..c20e371bac0 --- /dev/null +++ b/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/forward_sales_estimates.py @@ -0,0 +1,179 @@ +"""Seeking Alpha Forward Sales Estimates Model.""" + +# pylint: disable=unused-argument + +from typing import Any, Dict, List, Literal, Optional +from warnings import warn + +from openbb_core.provider.abstract.fetcher import Fetcher +from openbb_core.provider.standard_models.forward_sales_estimates import ( + ForwardSalesEstimatesData, + ForwardSalesEstimatesQueryParams, +) +from openbb_core.provider.utils.helpers import amake_request +from openbb_seeking_alpha.utils.helpers import HEADERS, get_seekingalpha_id +from pydantic import Field, field_validator + + +class SAForwardSalesEstimatesQueryParams(ForwardSalesEstimatesQueryParams): + """Seeking Alpha Forward Sales Estimates Query. + + Source: https://seekingalpha.com/earnings/earnings-calendar + """ + + __json_schema_extra__ = {"symbol": {"multiple_items_allowed": True}} + + period: Literal["annual", "quarter"] = Field( + default="quarter", + description="The reporting period.", + json_schema_extra={"choices": ["annual", "quarter"]}, + ) + + @field_validator("symbol", mode="before", check_fields=False) + @classmethod + def check_symbol(cls, value): + """Check the symbol.""" + if not value: + raise RuntimeError("Error: Symbol is a required field for Seeking Alpha.") + return value + + +class SAForwardSalesEstimatesData(ForwardSalesEstimatesData): + """Seeking Alpha Forward Sales Estimates Data.""" + + actual: Optional[int] = Field( + default=None, + description="Actual sales (revenue) for the period.", + ) + period_growth: Optional[float] = Field( + default=None, + description="Estimated (or actual if reported) EPS growth for the period.", + json_schema_extra={"x-unit_measurement": "percent", "x-frontend_multiply": 100}, + ) + + +class SAForwardSalesEstimatesFetcher( + Fetcher[ + SAForwardSalesEstimatesQueryParams, + List[SAForwardSalesEstimatesData], + ] +): + """Seeking Alpha Forward Sales Estimates Fetcher.""" + + @staticmethod + def transform_query(params: Dict[str, Any]) -> SAForwardSalesEstimatesQueryParams: + """Transform the query.""" + return SAForwardSalesEstimatesQueryParams(**params) + + @staticmethod + async def aextract_data( + query: SAForwardSalesEstimatesQueryParams, + credentials: Optional[Dict[str, str]], + **kwargs: Any, + ) -> Dict: + """Return the raw data from the Seeking Alpha endpoint.""" + tickers = query.symbol.split(",") + fp = query.period if query.period == "annual" else "quarterly" + url = "https://seekingalpha.com/api/v3/symbol_data/estimates" + payload = "" + querystring = { + "estimates_data_items": "revenue_actual,revenue_consensus_low,revenue_consensus_mean," + "revenue_consensus_high,revenue_num_of_estimates", + "period_type": fp, + "relative_periods": "-4,-3,-2,-1,0,1,2,3,4,5,6,7,8,9,10,11,12", + } + ids = {ticker: await get_seekingalpha_id(ticker) for ticker in tickers} + querystring["ticker_ids"] = (",").join(list(ids.values())) + response = await amake_request( + url, data=payload, headers=HEADERS, params=querystring + ) + estimates = response.get("estimates", {}) # type: ignore + if not estimates: + raise RuntimeError(f"No estimates data was returned for: {query.symbol}") + output: Dict = {"ids": ids, "estimates": estimates} + + return output + + @staticmethod + def transform_data( + query: SAForwardSalesEstimatesQueryParams, + data: Dict, + **kwargs: Any, + ) -> List[SAForwardSalesEstimatesData]: + """Transform the data to the standard format.""" + tickers = query.symbol.split(",") + ids = data.get("ids") + estimates = data.get("estimates") + results: List[SAForwardSalesEstimatesData] = [] + for ticker in tickers: + sa_id = str(ids.get(ticker, "")) + if sa_id == "" or sa_id not in estimates: + warn(f"Symbol Error: No data found for, {ticker}") + seek_object = estimates.get(sa_id) + items = len(seek_object["revenue_num_of_estimates"]) + for i in range(0, items - 4): + rev_estimates: Dict = {} + rev_estimates["symbol"] = ticker + num_estimates = seek_object["revenue_num_of_estimates"].get(str(i)) + if not num_estimates: + continue + period = num_estimates[0].get("period", {}) + if period: + period_type = period.get("periodtypeid") + rev_estimates["calendar_year"] = period.get("calendaryear") + rev_estimates["calendar_period"] = ( + "Q" + str(period.get("calendarquarter", "")) + if period_type == "quarterly" + else "FY" + ) + rev_estimates["date"] = period.get("periodenddate").split("T")[0] + rev_estimates["fiscal_year"] = period.get("fiscalyear") + rev_estimates["fiscal_period"] = ( + "Q" + str(period.get("fiscalquarter", "")) + if period_type == "quarterly" + else "FY" + ) + rev_estimates["number_of_analysts"] = num_estimates[0].get( + "dataitemvalue" + ) + mean = seek_object["revenue_consensus_mean"].get(str(i)) + if mean: + mean = mean[0].get("dataitemvalue") + rev_estimates["mean"] = int(float(mean)) + actual = ( + seek_object["revenue_actual"][str(i)][0].get("dataitemvalue") + if i < 1 + else None + ) + if actual: + rev_estimates["actual"] = int(float(actual)) + low = seek_object["revenue_consensus_low"].get(str(i)) + if low: + low = low[0].get("dataitemvalue") + rev_estimates["low_estimate"] = int(float(low)) + high = seek_object["revenue_consensus_high"].get(str(i)) + if high: + high = high[0].get("dataitemvalue") + rev_estimates["high_estimate"] = int(float(high)) + # Calculate the estimated growth percent. + this = float(mean) if mean else None + prev = None + percent = None + try: + prev = float( + seek_object["revenue"][str(i - 1)][0].get("dataitemvalue") + ) + except KeyError: + prev = float( + seek_object["revenue_consensus_mean"][str(i - 1)][0].get( + "dataitemvalue" + ) + ) + if this and prev: + percent = (this - prev) / prev + rev_estimates["period_growth"] = percent + results.append( + SAForwardSalesEstimatesData.model_validate(rev_estimates) + ) + + return results diff --git a/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/upcoming_release_days.py b/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/upcoming_release_days.py deleted file mode 100644 index 3dda9734307..00000000000 --- a/openbb_platform/providers/seeking_alpha/openbb_seeking_alpha/models/upcoming_release_days.py +++ /dev/null @@ -1,108 +0,0 @@ -"""Seeking Alpha Upcoming Release Days Model.""" - -from datetime import datetime -from typing import Any, Dict, List, Optional - -import requests -from openbb_core.provider.abstract.fetcher import Fetcher -from openbb_core.provider.standard_models.upcoming_release_days import ( - UpcomingReleaseDaysData, - UpcomingReleaseDaysQueryParams, -) -from openbb_core.provider.utils.descriptions import QUERY_DESCRIPTIONS -from pydantic import Field, field_validator - - -class SAUpcomingReleaseDaysQueryParams(UpcomingReleaseDaysQueryParams): - """Seeking Alpha Upcoming Release Days Query. - - Source: https://seekingalpha.com/api/v3/earnings_calendar/tickers - """ - - limit: int = Field( - description=QUERY_DESCRIPTIONS.get("limit", "") - + "In this case, the number of lookahead days.", - default=10, - ) - - -class SAUpcomingReleaseDaysData(UpcomingReleaseDaysData): - """Seeking Alpha Upcoming Release Days Data.""" - - __alias_dict__ = {"symbol": "slug", "release_time_type": "release_time"} - - sector_id: Optional[int] = Field( - description="The sector ID of the asset.", - ) - - @field_validator("release_date", mode="before", check_fields=False) - def validate_release_date(cls, v: Any) -> Any: # pylint: disable=E0213 - """Validate the release date.""" - v = v.split("T")[0] - return datetime.strptime(v, "%Y-%m-%d").date() - - -class SAUpcomingReleaseDaysFetcher( - Fetcher[ - SAUpcomingReleaseDaysQueryParams, - List[SAUpcomingReleaseDaysData], - ] -): - """Transform the query, extract and transform the data from the Seeking Alpha endpoints.""" - - @staticmethod - def transform_query(params: Dict[str, Any]) -> SAUpcomingReleaseDaysQueryParams: - """Transform the query.""" - return SAUpcomingReleaseDaysQueryParams(**params) - - @staticmethod - def extract_data( - query: SAUpcomingReleaseDaysQueryParams, # pylint: disable=W0613 - credentials: Optional[Dict[str, str]], - **kwargs: Any, - ) -> Dict: - """Return the raw data from the Seeking Alpha endpoint.""" - url = ( - f"https://seekingalpha.com/api/v3/earnings_calendar/tickers?" - f"filter%5Bselected_date%5D={str(datetime.now().date())}" # cspell:disable-line - f"&filter%5Bwith_rating%5D=false&filter%5Bcurrency%5D=USD" # cspell:disable-line - ) - response = requests.get( - url=url, - timeout=5, - ) - if response.status_code != 200: |