app.py

from datetime import datetime, timedelta
from functools import lru_cache
import time

import gradio as gr
import numpy as np
import pandas as pd
import yfinance as yf

APP_VERSION = "0.1.1"
RETRY_ATTEMPTS = 3
RETRY_DELAY_SECONDS = 1.5
option_chain_cache: dict[tuple[str, str], tuple[object, str]] = {}
price_cache: dict[str, tuple[float, str]] = {}


@lru_cache(maxsize=128)
def get_ticker(symb: str) -> yf.Ticker:
    return yf.Ticker(symb)


@lru_cache(maxsize=128)
def get_expirations(symb: str) -> tuple[str, ...]:
    return tuple(get_ticker(symb).options)


def get_option_chain_with_retry(ticker: yf.Ticker, target_time: str):
    last_error = None
    for attempt in range(RETRY_ATTEMPTS):
        try:
            return ticker.option_chain(target_time)
        except Exception as exc:
            last_error = exc
            if attempt < RETRY_ATTEMPTS - 1:
                time.sleep(RETRY_DELAY_SECONDS * (attempt + 1))
    raise last_error


def get_option_chain(symb: str, target_time: str, force_refresh: bool = False):
    cache_key = (symb, target_time)
    if not force_refresh and cache_key in option_chain_cache:
        return option_chain_cache[cache_key]

    ticker = get_ticker(symb)
    opt_chain = get_option_chain_with_retry(ticker, target_time)
    downloaded_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    option_chain_cache[cache_key] = (opt_chain, downloaded_at)
    return opt_chain, downloaded_at


def get_current_price(symb: str, force_refresh: bool = False) -> tuple[float, str]:
    if not force_refresh and symb in price_cache:
        return price_cache[symb]

    ticker = get_ticker(symb)
    current_price = ticker.fast_info["lastPrice"]
    downloaded_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    price_cache[symb] = (current_price, downloaded_at)
    return current_price, downloaded_at


def conf_int_ind(
    symb: str, target_time: str | None = None, force_refresh: bool = False
) -> tuple[pd.DataFrame, list[str]]:
    ticker = get_ticker(symb)
    expirations = get_expirations(symb)
    if not expirations:
        raise ValueError(f"No option expirations found for {symb}.")

    df_time = pd.to_datetime(expirations)
    if target_time is None:
        target_time_dt = df_time[0]
    else:
        target_time_dt = pd.to_datetime(target_time)
        valid_times = df_time[df_time <= target_time_dt]
        if valid_times.empty:
            raise ValueError(
                f"No expiration on or before {target_time_dt.strftime('%Y-%m-%d')} for {symb}."
            )
        target_time_dt = valid_times[-1]

    target_time_r = target_time_dt.strftime("%Y-%m-%d")
    time_now = datetime.now().strftime("%Y-%m-%d")
    t_days = (pd.to_datetime(target_time_r) - pd.to_datetime(time_now)).days + 1

    opt_chain, option_downloaded_at = get_option_chain(symb, target_time_r, force_refresh)
    current_price, price_downloaded_at = get_current_price(symb, force_refresh)

    lop = opt_chain.puts.strike[~opt_chain.puts.inTheMoney].iloc[-1]
    hip = opt_chain.puts.strike[opt_chain.puts.inTheMoney].iloc[0]
    iv1p = opt_chain.puts.loc[opt_chain.puts.strike == lop, "impliedVolatility"].item()
    iv2p = opt_chain.puts.loc[opt_chain.puts.strike == hip, "impliedVolatility"].item()
    ivp = (iv1p + iv2p) / 2
    sdp = ivp * np.sqrt(t_days / 365)

    loca = opt_chain.calls.strike[opt_chain.calls.inTheMoney].iloc[-1]
    hica = opt_chain.calls.strike[~opt_chain.calls.inTheMoney].iloc[0]
    iv1c = opt_chain.calls.loc[opt_chain.calls.strike == loca, "impliedVolatility"].item()
    iv2c = opt_chain.calls.loc[opt_chain.calls.strike == hica, "impliedVolatility"].item()
    ivc = (iv1c + iv2c) / 2
    sdc = ivc * np.sqrt(t_days / 365)

    row = {
        "symbol": [ticker.info["symbol"]],
        "days": [f"{target_time_r}: ({t_days}) days"],
        "-2.5%(p)": [current_price * (1 - 2 * sdp)],
        "-6.5%(p)": [current_price * (1 - 1.5 * sdp)],
        "-16.5%(p)": [current_price * (1 - 1 * sdp)],
        "current": [current_price],
        "+16.5%(c)": [current_price * (1 + 1 * sdc)],
        "+6.5%(c)": [current_price * (1 + 1.5 * sdc)],
        "+2.5%(c)": [current_price * (1 + 2 * sdc)],
    }

    return pd.DataFrame(row).set_index("symbol"), [option_downloaded_at, price_downloaded_at]


def conf_int_duo(
    symb: str, target_time: str | None = None, force_refresh: bool = False
) -> tuple[pd.DataFrame, list[str]]:
    if target_time is None:
        expirations = get_expirations(symb)
        df_time = pd.to_datetime(expirations)
        recent_mask = (df_time - datetime.now() <= timedelta(days=14)) & (df_time >= datetime.now())
        dlist = list(df_time[recent_mask].strftime("%Y-%m-%d"))
        if not dlist:
            return conf_int_ind(symb, None, force_refresh)
        results = [conf_int_ind(symb, d, force_refresh) for d in dlist]
        frames = [result[0] for result in results]
        timestamps = [stamp for result in results for stamp in result[1]]
        return pd.concat(frames), timestamps

    return conf_int_ind(symb, target_time, force_refresh)


def conf_int(
    symblist: str | list[str], target_time: str | None = None, force_refresh: bool = False
) -> tuple[pd.DataFrame, list[str]]:
    if isinstance(symblist, str):
        return conf_int_duo(symblist, target_time, force_refresh)
    if isinstance(symblist, list):
        results = [conf_int_duo(symb, target_time, force_refresh) for symb in symblist]
        frames = [result[0] for result in results]
        timestamps = [stamp for result in results for stamp in result[1]]
        return pd.concat(frames), timestamps
    raise TypeError("symblist must be a string or a list of strings.")


def parse_symbols(symbs_text: str) -> list[str]:
    symbs = [item.strip().upper() for item in symbs_text.replace("\n", ",").split(",")]
    symbs = [item for item in symbs if item]
    if not symbs:
        raise gr.Error("Provide at least one symbol.")
    return symbs


def run_app(symbs_text: str, disable_target_time: bool, target_time: str):
    symbs = parse_symbols(symbs_text)
    resolved_target_time = None if disable_target_time else (target_time or None)
    if not disable_target_time and resolved_target_time is None:
        raise gr.Error("Provide target_time in YYYY-MM-DD format, or disable it.")

    try:
        df, download_timestamps = conf_int(symbs, resolved_target_time, True)
    except Exception as exc:
        raise gr.Error(str(exc)) from exc

    latest_download = max(download_timestamps) if download_timestamps else "Unknown"
    return df.reset_index(), f"Last data download: {latest_download}"


def toggle_target_time(disable_target_time: bool):
    return gr.update(interactive=not disable_target_time, value="" if disable_target_time else None)


with gr.Blocks(title="Options Confidence Interval") as demo:
    gr.Markdown("## Options Confidence Interval")
    gr.Markdown(f"Version: `{APP_VERSION}`")
    gr.Markdown("Enter one or more symbols. Use a comma or a new line to separate multiple symbols.")

    with gr.Row():
        symbs_input = gr.Textbox(
            label="symbs",
            lines=3,
            value="SOXL, SPY",
            placeholder="SOXL, SPY",
        )
        with gr.Column():
            disable_target_time_input = gr.Checkbox(
                label="Disable target_time to include all expiration dates within 14 days",
                value=True,
            )
            target_time_input = gr.Textbox(
                label="target_time",
                placeholder="YYYY-MM-DD",
                interactive=False,
            )

    submit_btn = gr.Button("Run")
    output_df = gr.Dataframe(label="Result")
    refresh_timestamp = gr.Textbox(label="Data download timestamp", interactive=False)

    disable_target_time_input.change(
        toggle_target_time,
        inputs=disable_target_time_input,
        outputs=target_time_input,
    )
    submit_btn.click(
        run_app,
        inputs=[symbs_input, disable_target_time_input, target_time_input],
        outputs=[output_df, refresh_timestamp],
    )


if __name__ == "__main__":
    demo.launch(ssr_mode=False, share=True)