summaryrefslogtreecommitdiffstats
path: root/openbb_platform/providers/cboe/openbb_cboe/models/european_indices.py
diff options
context:
space:
mode:
Diffstat (limited to 'openbb_platform/providers/cboe/openbb_cboe/models/european_indices.py')
-rw-r--r--openbb_platform/providers/cboe/openbb_cboe/models/european_indices.py142
1 files changed, 0 insertions, 142 deletions
diff --git a/openbb_platform/providers/cboe/openbb_cboe/models/european_indices.py b/openbb_platform/providers/cboe/openbb_cboe/models/european_indices.py
deleted file mode 100644
index d06a05d0d3e..00000000000
--- a/openbb_platform/providers/cboe/openbb_cboe/models/european_indices.py
+++ /dev/null
@@ -1,142 +0,0 @@
-"""CBOE European Indices Model."""
-
-from datetime import datetime
-from typing import Any, Dict, List, Literal, Optional
-
-import pandas as pd
-from dateutil.relativedelta import relativedelta
-from openbb_cboe.utils.helpers import Europe
-from openbb_core.provider.abstract.fetcher import Fetcher
-from openbb_core.provider.standard_models.european_indices import (
- EuropeanIndicesData,
- EuropeanIndicesQueryParams,
-)
-from openbb_core.provider.utils.helpers import make_request
-from pydantic import Field, field_validator
-
-
-class CboeEuropeanIndicesQueryParams(EuropeanIndicesQueryParams):
- """CBOE European Indices Query.
-
- Source: https://www.cboe.com/europe/indices/
- """
-
- interval: Literal["1d", "1m"] = Field(description="Data granularity.", default="1d")
-
-
-class CboeEuropeanIndicesData(EuropeanIndicesData):
- """CBOE European Indices Data."""
-
- open: Optional[float] = Field(
- default=None,
- description="Opening price for the interval. Only valid when interval is 1m.",
- )
- high: Optional[float] = Field(
- default=None,
- description="High price for the interval. Only valid when interval is 1m.",
- )
- low: Optional[float] = Field(
- default=None,
- description="Low price for the interval. Only valid when interval is 1m.",
- )
- utc_datetime: Optional[datetime] = Field(
- default=None, description="UTC datetime. Only valid when interval is 1m."
- )
-
- @field_validator("date", mode="before", check_fields=False)
- def date_validate(cls, v): # pylint: disable=E0213
- """Return datetime object from string."""
- try:
- return datetime.strptime(v, "%Y-%m-%d")
- except Exception:
- return datetime.strptime(v, "%Y-%m-%dT%H:%M:%S")
-
-
-class CboeEuropeanIndicesFetcher(
- Fetcher[
- CboeEuropeanIndicesQueryParams,
- List[CboeEuropeanIndicesData],
- ]
-):
- """Transform the query, extract and transform the data from the CBOE endpoints."""
-
- @staticmethod
- def transform_query(params: Dict[str, Any]) -> CboeEuropeanIndicesQueryParams:
- """Transform the query."""
- now = datetime.now().date()
- transformed_params = params
- if params.get("start_date") is None:
- transformed_params["start_date"] = now - relativedelta(years=1)
-
- if params.get("end_date") is None:
- transformed_params["end_date"] = now
-
- return CboeEuropeanIndicesQueryParams(**transformed_params)
-
- @staticmethod
- def extract_data(
- query: CboeEuropeanIndicesQueryParams,
- credentials: Optional[Dict[str, str]],
- **kwargs: Any,
- ) -> List[Dict]:
- """Return the raw data from the CBOE endpoint."""
- data = pd.DataFrame()
- query.symbol = query.symbol.upper()
- SYMBOLS = pd.DataFrame(Europe.list_indices())["symbol"].to_list()
-
- if query.symbol not in SYMBOLS:
- raise RuntimeError(
- f"The symbol, {query.symbol},"
- "was not found in the CBOE European Indices directory. "
- "Use `available_indices(europe=True)` to see the full list of indices."
- )
- if query.interval == "1d":
- url = f"https://cdn.cboe.com/api/global/european_indices/index_history/{query.symbol}.json"
-
- r = make_request(url)
- if r.status_code != 200:
- raise RuntimeError(r.status_code)
-
- r_json = r.json()
-
- data = pd.DataFrame.from_records(r_json["data"]).set_index("date")
-
- data.index = pd.to_datetime(data.index, format="%Y-%m-%d")
-
- if query.start_date is not None:
- query.end_date = (
- query.end_date if query.end_date else datetime.now().date()
- )
-
- data = data[
- (data.index >= pd.to_datetime(query.start_date, format="%Y-%m-%d"))
- & (data.index <= pd.to_datetime(query.end_date, format="%Y-%m-%d"))
- ]
- data.index = data.index.astype(str)
- data = data.reset_index()
-
- if query.interval == "1m":
- url = f"https://cdn.cboe.com/api/global/european_indices/intraday_chart_data/{query.symbol}.json"
-
- r = make_request(url)
- if r.status_code != 200:
- raise RuntimeError(r.status_code)
-
- r_json = r.json()["data"]
-
- data = pd.DataFrame.from_records(pd.DataFrame(r_json)["price"])
- data["date"] = pd.DataFrame(r_json)["datetime"]
- data["utc_datetime"] = pd.to_datetime(pd.DataFrame(r_json)["utc_datetime"])
-
- data = data[["utc_datetime", "date", "open", "high", "low", "close"]].round(
- 2
- )
-
- return data.to_dict("records")
-
- @staticmethod
- def transform_data(
- query: CboeEuropeanIndicesQueryParams, data: List[Dict], **kwargs: Any
- ) -> List[CboeEuropeanIndicesData]:
- """Transform the data to the standard format."""
- return [CboeEuropeanIndicesData.model_validate(d) for d in data]