update

app.py

@@ -1,14 +1,69 @@
 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
 
+MAX_AUTO_EXPIRATIONS = 3
+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]] = {}
 
-def conf_int_ind(symb: str, target_time: str | None = None) -> pd.DataFrame:
-    ticker = yf.Ticker(symb)
-    expirations = ticker.options
+
+@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}.")
 
@@ -28,8 +83,8 @@ def conf_int_ind(symb: str, target_time: str | None = None) -> pd.DataFrame:
     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 = ticker.option_chain(target_time_r)
-    current_price = ticker.fast_info["lastPrice"]
+    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]
@@ -57,27 +112,37 @@ def conf_int_ind(symb: str, target_time: str | None = None) -> pd.DataFrame:
         "+2.5%(c)": [current_price * (1 + 2 * sdc)],
     }
 
-    return pd.DataFrame(row).set_index("symbol")
+    return pd.DataFrame(row).set_index("symbol"), [option_downloaded_at, price_downloaded_at]
 
 
-def conf_int_duo(symb: str, target_time: str | None = None) -> pd.DataFrame:
+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 = yf.Ticker(symb).options
+        expirations = get_expirations(symb)
         df_time = pd.to_datetime(expirations)
-        recent_count = (df_time - datetime.now() <= timedelta(days=14)).sum()
-        dlist = expirations[:recent_count]
+        recent_mask = (df_time - datetime.now() <= timedelta(days=14)) & (df_time >= datetime.now())
+        dlist = list(df_time[recent_mask].strftime("%Y-%m-%d"))[:MAX_AUTO_EXPIRATIONS]
         if not dlist:
-            return conf_int_ind(symb, None)
-        return pd.concat([conf_int_ind(symb, d) for d in 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)
+    return conf_int_ind(symb, target_time, force_refresh)
 
 
-def conf_int(symblist: str | list[str], target_time: str | None = None) -> pd.DataFrame:
+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)
+        return conf_int_duo(symblist, target_time, force_refresh)
     if isinstance(symblist, list):
-        return pd.concat([conf_int_duo(symb, target_time) for symb in symblist])
+        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.")
 
 
@@ -89,18 +154,19 @@ def parse_symbols(symbs_text: str) -> list[str]:
     return symbs
 
 
-def run_app(symbs_text: str, disable_target_time: bool, target_time: str) -> pd.DataFrame:
+def run_app(symbs_text: str, disable_target_time: bool, target_time: str, force_refresh: bool):
     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 = conf_int(symbs, resolved_target_time)
+        df, download_timestamps = conf_int(symbs, resolved_target_time, force_refresh)
     except Exception as exc:
         raise gr.Error(str(exc)) from exc
 
-    return df.reset_index()
+    latest_download = max(download_timestamps) if download_timestamps else "Unknown"
+    return df.reset_index(), f"Last data download: {latest_download}", gr.update(value=False)
 
 
 def toggle_target_time(disable_target_time: bool):
@@ -123,6 +189,10 @@ with gr.Blocks(title="Options Confidence Interval") as demo:
                 label="Disable target_time to include all expiration dates within 14 days",
                 value=True,
             )
+            force_refresh_input = gr.Checkbox(
+                label="Force refresh",
+                value=False,
+            )
             target_time_input = gr.Textbox(
                 label="target_time",
                 placeholder="YYYY-MM-DD",
@@ -131,6 +201,7 @@ with gr.Blocks(title="Options Confidence Interval") as demo:
 
     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,
@@ -139,8 +210,8 @@ with gr.Blocks(title="Options Confidence Interval") as demo:
     )
     submit_btn.click(
         run_app,
-        inputs=[symbs_input, disable_target_time_input, target_time_input],
-        outputs=output_df,
+        inputs=[symbs_input, disable_target_time_input, target_time_input, force_refresh_input],
+        outputs=[output_df, refresh_timestamp, force_refresh_input],
     )