cboe helpers

1 files changed, 87 insertions, 15 deletions

diff --git a/openbb_platform/providers/cboe/openbb_cboe/utils/helpers.py b/openbb_platform/providers/cboe/openbb_cboe/utils/helpers.py
index 7bb2965dca6..65c21b73c55 100644
--- a/openbb_platform/providers/cboe/openbb_cboe/utils/helpers.py
+++ b/openbb_platform/providers/cboe/openbb_cboe/utils/helpers.py
@@ -3,23 +3,78 @@
 # pylint: disable=expression-not-assigned, unused-argument
 
 from datetime import date as dateType
-from io import BytesIO, StringIO
-from typing import Any, List, Optional
+from typing import TYPE_CHECKING, Any, List, Literal, Optional
 
-from aiohttp_client_cache import SQLiteBackend
-from aiohttp_client_cache.session import CachedSession
-from openbb_core.app.utils import get_user_cache_directory
-from openbb_core.provider.utils.client import ClientResponse
 from openbb_core.provider.utils.helpers import amake_request, to_snake_case
-from pandas import DataFrame, read_csv
 
-TICKER_EXCEPTIONS = ["NDX", "RUT"]
-
-cache_dir = get_user_cache_directory()
-backend = SQLiteBackend(f"{cache_dir}/http/cboe_directories", expire_after=3600 * 24)
+if TYPE_CHECKING:
+    from pandas import DataFrame
 
+TICKER_EXCEPTIONS = ["NDX", "RUT"]
 
-async def response_callback(response: ClientResponse, _: Any):
+CONSTITUENTS_EU = Literal[  # pylint: disable=invalid-name
+    "BAT20P",
+    "BBE20P",
+    "BCH20P",
+    "BCHM30P",
+    "BDE40P",
+    "BDEM50P",
+    "BDES50P",
+    "BDK25P",
+    "BEP50P",
+    "BEPACP",
+    "BEPBUS",
+    "BEPCNC",
+    "BEPCONC",
+    "BEPCONS",
+    "BEPENGY",
+    "BEPFIN",
+    "BEPHLTH",
+    "BEPIND",
+    "BEPNEM",
+    "BEPTEC",
+    "BEPTEL",
+    "BEPUTL",
+    "BEPXUKP",
+    "BES35P",
+    "BEZ50P",
+    "BEZACP",
+    "BFI25P",
+    "BFR40P",
+    "BFRM20P",
+    "BIE20P",
+    "BIT40P",
+    "BNL25P",
+    "BNLM25P",
+    "BNO25G",
+    "BNORD40P",
+    "BPT20P",
+    "BSE30P",
+    "BUK100P",
+    "BUK250P",
+    "BUK350P",
+    "BUKAC",
+    "BUKBISP",
+    "BUKBUS",
+    "BUKCNC",
+    "BUKCONC",
+    "BUKCONS",
+    "BUKENGY",
+    "BUKFIN",
+    "BUKHI50P",
+    "BUKHLTH",
+    "BUKIND",
+    "BUKLO50P",
+    "BUKMINP",
+    "BUKNEM",
+    "BUKSC",
+    "BUKTEC",
+    "BUKTEL",
+    "BUKUTL",
+]
+
+
+async def response_callback(response, _):
     """Use callback for HTTP Client Response."""
     content_type = response.headers.get("Content-Type", "")
     if "application/json" in content_type:
@@ -31,6 +86,13 @@ async def response_callback(response: ClientResponse, _: Any):
 
 async def get_cboe_data(url, use_cache: bool = True, **kwargs) -> Any:
     """Use the generic Cboe HTTP request."""
+    # pylint: disable=import-outside-toplevel
+    from aiohttp_client_cache import SQLiteBackend
+    from aiohttp_client_cache.session import CachedSession
+    from openbb_core.app.utils import get_user_cache_directory
+
+    cache_dir = get_user_cache_directory()
+    backend = SQLiteBackend(f"{cache_dir}/http/cboe_directories", expire_after=3600 * 24)
     data: Any = None
     if use_cache is True:
         async with CachedSession(cache=backend) as cached_session:
@@ -45,7 +107,7 @@ async def get_cboe_data(url, use_cache: bool = True, **kwargs) -> Any:
     return data
 
 
-async def get_company_directory(use_cache: bool = True, **kwargs) -> DataFrame:
+async def get_company_directory(use_cache: bool = True, **kwargs) -> "DataFrame":
     """Get the US Company Directory for Cboe options.
 
     If use_cache is True, the data will be cached for 24 hours.
@@ -54,6 +116,9 @@ async def get_company_directory(use_cache: bool = True, **kwargs) -> DataFrame:
     -------
     DataFrame: Pandas DataFrame of the Cboe listings directory
     """
+    # pylint: disable=import-outside-toplevel
+    from io import BytesIO  # noqa
+    from pandas import read_csv  # noqa
     url = "https://www.cboe.com/us/options/symboldir/equity_index_options/?download=csv"
 
     results = await get_cboe_data(url, use_cache)
@@ -74,7 +139,7 @@ async def get_company_directory(use_cache: bool = True, **kwargs) -> DataFrame:
     return directory.astype(str)
 
 
-async def get_index_directory(use_cache: bool = True, **kwargs) -> DataFrame:
+async def get_index_directory(use_cache: bool = True, **kwargs) -> "DataFrame":
     """Get the Cboe Index Directory.
 
     If use_cache is True, the data will be cached for 24 hours.
@@ -83,6 +148,9 @@ async def get_index_directory(use_cache: bool = True, **kwargs) -> DataFrame:
     -------
     List[Dict]: A list of dictionaries containing the index information.
     """
+    # pylint: disable=import-outside-toplevel
+    from pandas import DataFrame
+
     url = "https://cdn.cboe.com/api/global/us_indices/definitions/all_indices.json"
 
     results = await get_cboe_data(url, use_cache=use_cache)
@@ -119,7 +187,7 @@ async def get_settlement_prices(
     archives: bool = False,
     final_settlement: bool = False,
     **kwargs,
-) -> DataFrame:
+) -> "DataFrame":
     """Get the settlement prices of CBOE futures.
 
     Parameters
@@ -138,6 +206,10 @@ async def get_settlement_prices(
     DataFrame
         Pandas DataFrame with results.
     """
+    # pylint: disable=import-outside-toplevel
+    from io import StringIO  # noqa
+    from pandas import DataFrame, read_csv  # noqa
+
     url = ""
     if settlement_date is not None:
         url = (