1 files changed, 144 insertions, 0 deletions
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