[BugFix] - Fix lowercase symbols (#6342)

* Fix lowercase symbols

* fix decorator for to_upper

* add annotated results with underlying symbol info

* ?

* Improvements

---------

Co-authored-by: Danglewood <85772166+deeleeramone@users.noreply.github.com>
Co-authored-by: Henrique Joaquim <henriquecjoaquim@gmail.com>
This commit is contained in:
Igor Radovanovic
2024-05-07 14:22:44 +02:00
committed by GitHub
parent d3b9ce6d37
commit 5c0b36cddb
2 changed files with 55 additions and 16 deletions

View File

@@ -21,8 +21,8 @@ class OptionsChainsQueryParams(QueryParams):
symbol: str = Field(description=QUERY_DESCRIPTIONS.get("symbol", ""))
@classmethod
@field_validator("symbol", mode="before", check_fields=False)
@classmethod
def to_upper(cls, v: str) -> str:
"""Return the symbol in uppercase."""
return v.upper()

View File

@@ -10,6 +10,7 @@ from openbb_cboe.utils.helpers import (
get_company_directory,
get_index_directory,
)
from openbb_core.provider.abstract.annotated_result import AnnotatedResult
from openbb_core.provider.abstract.fetcher import Fetcher
from openbb_core.provider.standard_models.options_chains import (
OptionsChainsData,
@@ -62,10 +63,9 @@ class CboeOptionsChainsFetcher(
query: CboeOptionsChainsQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
"""Return the raw data from the Cboe endpoint"""
symbol = query.symbol.replace("^", "").split(",")[0]
) -> Dict:
"""Return the raw data from the Cboe endpoint."""
symbol = query.symbol.replace("^", "").split(",")[0].upper()
INDEXES = await get_index_directory(use_cache=query.use_cache)
SYMBOLS = await get_company_directory(use_cache=query.use_cache)
INDEXES = INDEXES.set_index("index_symbol")
@@ -78,21 +78,57 @@ class CboeOptionsChainsFetcher(
if symbol in TICKER_EXCEPTIONS or symbol in INDEXES.index
else f"https://cdn.cboe.com/api/global/delayed_quotes/options/{symbol}.json"
)
return await amake_request(quotes_url)
results = await amake_request(quotes_url)
return results # type: ignore
@staticmethod
def transform_data(
query: CboeOptionsChainsQueryParams,
data: List[Dict],
data: Dict,
**kwargs: Any,
) -> List[CboeOptionsChainsData]:
"""Transform the data to the standard format"""
) -> AnnotatedResult[List[CboeOptionsChainsData]]:
"""Transform the data to the standard format."""
if not data:
raise EmptyDataError()
results_metadata = {}
options = data.get("data", {}).pop("options", [])
change_percent = data["data"].get("percent_change", None)
iv30_percent = data["data"].get("iv30_change_percent", None)
if change_percent:
change_percent = change_percent / 100
if iv30_percent:
iv30_percent = iv30_percent / 100
last_timestamp = data["data"].get("last_trade_time", None)
if last_timestamp:
last_timestamp = to_datetime(
last_timestamp, format="%Y-%m-%dT%H:%M:%S"
).strftime("%Y-%m-%d %H:%M:%S")
results_metadata.update(
{
"symbol": data["data"].get("symbol"),
"security_type": data["data"].get("security_type", None),
"bid": data["data"].get("bid", None),
"bid_size": data["data"].get("bid_size", None),
"ask": data["data"].get("ask", None),
"ask_size": data["data"].get("ask_size", None),
"open": data["data"].get("open", None),
"high": data["data"].get("high", None),
"low": data["data"].get("low", None),
"close": data["data"].get("close", None),
"volume": data["data"].get("volume", None),
"current_price": data["data"].get("current_price", None),
"prev_close": data["data"].get("prev_day_close", None),
"change": data["data"].get("price_change", None),
"change_percent": change_percent,
"iv30": data["data"].get("iv30", None),
"iv30_change": data["data"].get("iv30_change", None),
"iv30_change_percent": iv30_percent,
"last_tick": data["data"].get("tick", None),
"last_trade_timestamp": last_timestamp,
}
)
options_df = DataFrame.from_records(data["data"]["options"])
options_df = DataFrame.from_records(options)
options_df = options_df.rename(
columns={
@@ -152,7 +188,10 @@ class CboeOptionsChainsFetcher(
quotes["prev_close"] = round(quotes["prev_close"], 2)
quotes["change_percent"] = round(quotes["change_percent"] / 100, 4)
return [
CboeOptionsChainsData.model_validate(d)
for d in quotes.reset_index().to_dict("records")
]
return AnnotatedResult(
result=[
CboeOptionsChainsData.model_validate(d)
for d in quotes.reset_index().to_dict("records")
],
metadata=results_metadata,
)