path: root/openbb_terminal/stocks/options/tradier_model.py

# mypy: disable-error-code=attr-defined

"""Tradier options model"""
__docformat__ = "numpy"

import logging
from datetime import datetime
from typing import List, Optional

import pandas as pd
import requests

from openbb_terminal.core.session.current_user import get_current_user
from openbb_terminal.decorators import check_api_key, log_start_end
from openbb_terminal.helper_funcs import request
from openbb_terminal.rich_config import console, optional_rich_track
from openbb_terminal.stocks.options.op_helpers import Options, PydanticOptions

logger = logging.getLogger(__name__)

option_columns = [
    "symbol",
    "bid",
    "ask",
    "strike",
    "bidsize",
    "asksize",
    "volume",
    "open_interest",
    "option_type",
]
greek_columns = ["delta", "gamma", "theta", "vega", "ask_iv", "bid_iv", "mid_iv"]
df_columns = option_columns + greek_columns

default_columns = [
    "mid_iv",
    "vega",
    "delta",
    "gamma",
    "theta",
    "volume",
    "open_interest",
    "bid",
    "ask",
]

sorted_chain_columns = [
    "symbol",
    "option_type",
    "expiration",
    "strike",
    "bid",
    "ask",
    "open_interest",
    "volume",
    "mid_iv",
    "delta",
    "gamma",
    "theta",
    "vega",
]


@check_api_key(["API_TRADIER_TOKEN"])
def lookup_company(symbol: str):
    response = request(
        "https://sandbox.tradier.com/v1/markets/lookup",
        params={"q": f"{symbol}"},
        headers={
            "Authorization": f"Bearer {get_current_user().credentials.API_TRADIER_TOKEN }",  # type: ignore[attr-defined]
            "Accept": "application/json",
        },
    )
    return response.json()


@log_start_end(log=logger)
@check_api_key(["API_TRADIER_TOKEN"])
def get_historical_options(
    symbol: str,
    expiry: str,
    strike: float = 0,
    put: bool = False,
    chain_id: Optional[str] = None,
) -> pd.DataFrame:
    """
    Gets historical option pricing.  This inputs either ticker, expiration, strike or the OCC chain ID and processes
    the request to tradier for historical premiums.

    Parameters
    ----------
    symbol: str
        Stock ticker symbol
    expiry: str
        Option expiration date
    strike: int
        Option strike price
    put: bool
        Is this a put option?
    chain_id: Optional[str]
        OCC chain ID

    Returns
    -------
    df_hist: pd.DataFrame
        Dataframe of historical option prices
    """
    if not chain_id:
        op_type = ["call", "put"][put]
        chain = get_option_chain(symbol, expiry)
        chain = chain[chain["option_type"] == op_type]

        try:
            symbol = chain[(chain.strike == strike)]["symbol"].values[0]
        except IndexError:
            error = f"Strike: {strike}, Option type: {op_type} not found"
            logging.exception(error)
            console.print(f"{error}\n")
            return pd.DataFrame()
    else:
        symbol = chain_id

    try:
        response = request(
            "https://sandbox.tradier.com/v1/markets/history",
            params={"symbol": {symbol}, "interval": "daily"},
            headers={
                "Authorization": f"Bearer {get_current_user().credentials.API_TRADIER_TOKEN}",
                "Accept": "application/json",
            },
        )
    except requests.exceptions.ReadTimeout:
        return pd.DataFrame()

    if response.status_code != 200:
        console.print("Error with request")
        return pd.DataFrame()

    data = response.json()["history"]
    if not data:
        console.print("No historical data available")
        return pd.DataFrame()

    df_hist = pd.DataFrame(data["day"])
    df_hist = df_hist.set_index("date")
    df_hist.index = pd.DatetimeIndex(df_hist.index)
    return df_hist


# pylint: disable=no-else-return

option_cols = [
    "strike",
    "bid",
    "ask",
    "volume",
    "open_interest",
    "mid_iv",
]

option_col_map = {"open_interest": "openinterest", "mid_iv": "iv"}


@log_start_end(log=logger)
@check_api_key(["API_TRADIER_TOKEN"])
def get_full_option_chain(symbol: str, quiet: bool = False) -> pd.DataFrame:
    """Get available expiration dates for given ticker

    Parameters
    ----------
    symbol: str
        Ticker symbol to get expirations for
    quiet: bool
        Suppress output of progress bar

    Returns
    -------
    pd.DataFrame
        Dataframe of all option chains
    """

    expirations = option_expirations(symbol)
    options_dfs: pd.DataFrame = []

    for expiry in optional_rich_track(
        expirations, suppress_output=quiet, desc="Getting Option Chain"
    ):
        chain = get_option_chain(symbol, expiry)
        options_dfs.append(chain)
    chain = pd.concat(options_dfs)
    chain = chain[sorted_chain_columns].rename(
        columns={
            "mid_iv": "impliedVolatility",
            "open_interest": "openInterest",
            "option_type": "optionType",
            "symbol": "optionSymbol",
        }
    )
    chain["openInterest"] = chain["openInterest"].astype(int)
    chain["volume"] = chain["volume"].astype(int)
    chain = chain.set_index(["expiration", "strike", "optionType"]).sort_index()

    return chain.reset_index()


@log_start_end(log=logger)
@check_api_key(["API_TRADIER_TOKEN"])
def option_expirations(symbol: str) -> List[str]:
    """Get available expiration dates for given ticker

    Parameters
    ----------
    symbol: str
        Ticker symbol to get expirations for

    Returns
    -------
    dates: List[str]
        List of of available expirations
    """
    r = request(
        "https://sandbox.tradier.com/v1/markets/options/expirations",
        params={"symbol": symbol, "includeAllRoots": "true", "strikes": "false"},
        headers={
            "Authorization": f"Bearer {get_current_user().credentials.API_TRADIER_TOKEN}",  # type: ignore[attr-defined]
            "Accept": "application/json",
        },
    )
    if r.status_code == 200:
        try:
            dates