summaryrefslogtreecommitdiffstats
path: root/openbb_platform/providers/cboe/openbb_cboe/models/equity_info.py
diff options
context:
space:
mode:
Diffstat (limited to 'openbb_platform/providers/cboe/openbb_cboe/models/equity_info.py')
-rw-r--r--openbb_platform/providers/cboe/openbb_cboe/models/equity_info.py140
1 files changed, 0 insertions, 140 deletions
diff --git a/openbb_platform/providers/cboe/openbb_cboe/models/equity_info.py b/openbb_platform/providers/cboe/openbb_cboe/models/equity_info.py
deleted file mode 100644
index 296baab8bd3..00000000000
--- a/openbb_platform/providers/cboe/openbb_cboe/models/equity_info.py
+++ /dev/null
@@ -1,140 +0,0 @@
-"""CBOE Equity Info Model."""
-
-import concurrent.futures
-from datetime import datetime
-from typing import Any, Dict, List, Optional
-
-import pandas as pd
-from openbb_cboe.utils.helpers import (
- get_cboe_directory,
- get_cboe_index_directory,
- get_ticker_info,
- get_ticker_iv,
-)
-from openbb_core.provider.abstract.fetcher import Fetcher
-from openbb_core.provider.standard_models.equity_info import (
- EquityInfoData,
- EquityInfoQueryParams,
-)
-from pydantic import Field
-
-
-class CboeEquityInfoQueryParams(EquityInfoQueryParams):
- """CBOE Equity Info Query.
-
- Source: https://www.cboe.com/
- """
-
-
-class CboeEquityInfoData(EquityInfoData):
- """CBOE Equity Info Data."""
-
- type: Optional[str] = Field(default=None, description="Type of asset.")
- exchange_id: Optional[int] = Field(
- default=None, description="The Exchange ID number."
- )
- tick: Optional[str] = Field(
- default=None, description="Whether the last sale was an up or down tick."
- )
- bid: Optional[float] = Field(default=None, description="Current bid price.")
- bid_size: Optional[float] = Field(default=None, description="Bid lot size.")
- ask: Optional[float] = Field(default=None, description="Current ask price.")
- ask_size: Optional[float] = Field(default=None, description="Ask lot size.")
- volume: Optional[float] = Field(
- default=None, description="Stock volume for the current trading day."
- )
- iv30: Optional[float] = Field(
- default=None, description="The 30-day implied volatility of the stock."
- )
- iv30_change: Optional[float] = Field(
- default=None, description="Change in 30-day implied volatility of the stock."
- )
- last_trade_timestamp: Optional[datetime] = Field(
- default=None, description="Last trade timestamp for the stock."
- )
- iv30_annual_high: Optional[float] = Field(
- default=None, description="The 1-year high of implied volatility."
- )
- hv30_annual_high: Optional[float] = Field(
- default=None, description="The 1-year high of realized volatility."
- )
- iv30_annual_low: Optional[float] = Field(
- default=None, description="The 1-year low of implied volatility."
- )
- hv30_annual_low: Optional[float] = Field(
- default=None, description="The 1-year low of realized volatility."
- )
- iv60_annual_high: Optional[float] = Field(
- default=None, description="The 60-day high of implied volatility."
- )
- hv60_annual_high: Optional[float] = Field(
- default=None, description="The 60-day high of realized volatility."
- )
- iv60_annual_low: Optional[float] = Field(
- default=None, description="The 60-day low of implied volatility."
- )
- hv60_annual_low: Optional[float] = Field(
- default=None, description="The 60-day low of realized volatility."
- )
- iv90_annual_high: Optional[float] = Field(
- default=None, description="The 90-day high of implied volatility."
- )
- hv90_annual_high: Optional[float] = Field(
- default=None, description="The 90-day high of realized volatility."
- )
-
-
-class CboeEquityInfoFetcher(
- Fetcher[
- CboeEquityInfoQueryParams,
- List[CboeEquityInfoData],
- ]
-):
- """Transform the query, extract and transform the data from the CBOE endpoints."""
-
- @staticmethod
- def transform_query(params: Dict[str, Any]) -> CboeEquityInfoQueryParams:
- """Transform the query."""
- return CboeEquityInfoQueryParams(**params)
-
- @staticmethod
- def extract_data(
- query: CboeEquityInfoQueryParams, # pylint: disable=unused-argument
- credentials: Optional[Dict[str, str]],
- **kwargs: Any,
- ) -> List[Dict]:
- """Return the raw data from the CBOE endpoint."""
- results = []
-
- INDEXES = get_cboe_index_directory().index.to_list()
- SYMBOLS = get_cboe_directory()
-
- def get_one(symbol):
- data = pd.Series(dtype="object")
- if symbol in SYMBOLS.index or symbol in INDEXES:
- _info = pd.Series(get_ticker_info(symbol))
- _iv = pd.Series(get_ticker_iv(symbol))
- data = (
- pd.DataFrame(pd.concat([_info, _iv]))
- .transpose()
- .drop(columns="seqno")
- .iloc[0]
- )
-
- results.append(data.to_dict())
-
- with concurrent.futures.ThreadPoolExecutor() as executor:
- executor.map(get_one, query.symbol.split(","))
-
- return (
- pd.DataFrame.from_records(results)
- .sort_values(by="symbol")
- .to_dict("records")
- )
-
- @staticmethod
- def transform_data(
- query: CboeEquityInfoQueryParams, data: List[Dict], **kwargs: Any
- ) -> List[CboeEquityInfoData]:
- """Transform the data to the standard format."""
- return [CboeEquityInfoData.model_validate(d) for d in data]