import json import warnings from datetime import datetime, time, timedelta from urllib.error import HTTPError, URLError from urllib.parse import quote from urllib.request import Request, urlopen from zoneinfo import ZoneInfo import gradio as gr import matplotlib import matplotlib.dates as mdates import matplotlib.pyplot as plt import numpy as np import pandas as pd import yfinance as yf from matplotlib.lines import Line2D from matplotlib.patches import Patch matplotlib.use("Agg") warnings.filterwarnings("ignore", message="urllib3 v2 only supports OpenSSL.*") DEFAULT_TICKER = "TSLA" BENCHMARK = "QQQ" YEARS = 5 MA_FAST = 50 MA_SLOW = 200 RS_MA20 = 20 RS_MA60 = 60 BB_WINDOW = 20 BB_STD_MULT = 2 KELT_EMA_WINDOW = 20 KELT_ATR_MULT = 2 KELT_ATR_WINDOW = 10 ADX_WINDOW = 14 DMI_CLEAR_GAP = 5 SIGMA = 1.5 BETA_WINDOW = 252 SLOPE_WINDOW = 60 TRADING_DAYS_PER_YEAR = 252 FIG_FACE = "#080912" AX_FACE = "#171724" GRID = "#343746" TEXT = "#E7E8EE" MUTED = "#B9BBC6" PRICE = "#E6E6E6" MA200 = "#D9787C" MA50 = "#73BDB6" BAND = "#7F7A34" POS = "#69C4C0" NEG = "#CF5B62" NEUTRAL = "#808493" GOLD = "#E8D268" ORANGE = "#D38343" def normalize_ticker(ticker): return (ticker or DEFAULT_TICKER).strip().upper() def extract_price_field(raw, ticker, field): if isinstance(raw.columns, pd.MultiIndex): level0 = raw.columns.get_level_values(0) level1 = raw.columns.get_level_values(1) if ticker in level0 and field in raw[ticker].columns: return raw[ticker][field].rename(ticker) if field in level0 and ticker in level1: return raw[field][ticker].rename(ticker) elif field in raw.columns: return raw[field].rename(ticker) raise RuntimeError(f"Could not find {field} column for {ticker}. Returned columns: {list(raw.columns)}") def extract_close(raw, ticker): return extract_price_field(raw, ticker, "Close") def extract_ohlc(raw, ticker): return pd.concat( [ extract_price_field(raw, ticker, "High").rename("high"), extract_price_field(raw, ticker, "Low").rename("low"), extract_price_field(raw, ticker, "Close").rename("close"), ], axis=1, ) def drop_unclosed_current_us_session(frame): ny_now = datetime.now(ZoneInfo("America/New_York")) today = pd.Timestamp(ny_now.date()) if ny_now.time() < time(16, 10): frame = frame[frame.index.normalize() < today] return frame def ny_midnight_epoch(date_value): return int(datetime.combine(date_value, time.min).replace(tzinfo=ZoneInfo("America/New_York")).timestamp()) def fetch_yahoo_chart_ohlc(symbol, start_date, end_date): symbol = normalize_ticker(symbol) period1 = ny_midnight_epoch(start_date) period2 = ny_midnight_epoch(end_date) url = ( f"https://query1.finance.yahoo.com/v8/finance/chart/{quote(symbol, safe='')}" f"?period1={period1}&period2={period2}&interval=1d&events=history&includeAdjustedClose=true" ) request = Request(url, headers={"User-Agent": "Mozilla/5.0"}) try: with urlopen(request, timeout=20) as response: payload = json.loads(response.read().decode("utf-8")) except (HTTPError, URLError, TimeoutError) as exc: raise RuntimeError(f"Yahoo chart API request failed for {symbol}: {exc}") from exc chart = payload.get("chart", {}) if chart.get("error"): raise RuntimeError(f"Yahoo chart API error for {symbol}: {chart['error']}") results = chart.get("result") or [] if not results: raise RuntimeError(f"Yahoo chart API returned no result for {symbol}.") result = results[0] timestamps = result.get("timestamp") or [] quotes = result.get("indicators", {}).get("quote") or [] if not timestamps or not quotes: raise RuntimeError(f"Yahoo chart API returned no OHLC rows for {symbol}.") quote_data = quotes[0] frame = pd.DataFrame( { "high": quote_data.get("high"), "low": quote_data.get("low"), "close": quote_data.get("close"), }, index=pd.to_datetime(timestamps, unit="s", utc=True).tz_convert("America/New_York").tz_localize(None).normalize(), ) adjclose = (result.get("indicators", {}).get("adjclose") or [{}])[0].get("adjclose") if adjclose: raw_close = pd.to_numeric(frame["close"], errors="coerce") adjusted_close = pd.to_numeric(pd.Series(adjclose, index=frame.index), errors="coerce") adjustment_ratio = adjusted_close / raw_close.replace(0, np.nan) frame["high"] = pd.to_numeric(frame["high"], errors="coerce") * adjustment_ratio frame["low"] = pd.to_numeric(frame["low"], errors="coerce") * adjustment_ratio frame["close"] = adjusted_close frame = frame.apply(pd.to_numeric, errors="coerce") frame = frame.groupby(frame.index).last() frame = frame.dropna(subset=["high", "low", "close"]) if frame.empty: raise RuntimeError(f"Yahoo chart API returned only empty OHLC rows for {symbol}.") return drop_unclosed_current_us_session(frame) def download_daily_market_data_from_yahoo_chart(ticker, benchmark, start_date, end_date): ticker_ohlc = fetch_yahoo_chart_ohlc(ticker, start_date, end_date) benchmark_ohlc = fetch_yahoo_chart_ohlc(benchmark, start_date, end_date) closes = pd.concat( [ ticker_ohlc["close"].rename(ticker), benchmark_ohlc["close"].rename(benchmark), ], axis=1, ).dropna(how="all") return closes, ticker_ohlc def download_daily_market_data(ticker, benchmark): ticker = normalize_ticker(ticker) benchmark = normalize_ticker(benchmark) tickers = list(dict.fromkeys([ticker, benchmark])) warmup_years = max(1.5, (max(MA_SLOW, BETA_WINDOW) + 80) / TRADING_DAYS_PER_YEAR) end_date = datetime.now(ZoneInfo("America/New_York")).date() + timedelta(days=1) start_date = end_date - timedelta(days=int((YEARS + warmup_years + 0.25) * 365.25)) yfinance_error = None try: raw = yf.download( tickers=tickers, start=start_date.isoformat(), end=end_date.isoformat(), interval="1d", auto_adjust=True, progress=False, group_by="ticker", threads=False, ) if raw.empty: raise RuntimeError(f"No data returned for {', '.join(tickers)}.") closes = pd.concat([extract_close(raw, symbol) for symbol in tickers], axis=1) closes = closes.sort_index() closes.index = pd.to_datetime(closes.index).tz_localize(None) closes = closes.dropna(how="all") closes = drop_unclosed_current_us_session(closes) ticker_ohlc = extract_ohlc(raw, ticker) ticker_ohlc = ticker_ohlc.sort_index() ticker_ohlc.index = pd.to_datetime(ticker_ohlc.index).tz_localize(None) ticker_ohlc = ticker_ohlc.dropna(how="all") ticker_ohlc = drop_unclosed_current_us_session(ticker_ohlc) ticker_ohlc = ticker_ohlc.dropna(subset=["high", "low", "close"]) except Exception as exc: yfinance_error = exc closes, ticker_ohlc = download_daily_market_data_from_yahoo_chart(ticker, benchmark, start_date, end_date) if ticker not in closes or closes[ticker].dropna().empty: detail = f" yfinance error: {yfinance_error}" if yfinance_error else "" raise RuntimeError(f"No usable close data found for {ticker}.{detail}") if benchmark not in closes or closes[benchmark].dropna().empty: detail = f" yfinance error: {yfinance_error}" if yfinance_error else "" raise RuntimeError(f"No usable close data found for benchmark {benchmark}.{detail}") if ticker_ohlc.empty: detail = f" yfinance error: {yfinance_error}" if yfinance_error else "" raise RuntimeError(f"No usable OHLC data found for {ticker}.{detail}") return closes, ticker_ohlc def build_indicators(closes, ticker, benchmark): ticker = normalize_ticker(ticker) benchmark = normalize_ticker(benchmark) price = closes[ticker].dropna() if len(price) < MA_SLOW + SLOPE_WINDOW + 10: raise RuntimeError(f"{ticker} only has {len(price)} daily rows. Need more history.") indicators = pd.DataFrame(index=price.index) indicators["close"] = price indicators["ma_fast"] = price.rolling(MA_FAST).mean() indicators["ma_slow"] = price.rolling(MA_SLOW).mean() rolling_std = price.rolling(MA_SLOW).std() indicators["upper_band"] = indicators["ma_slow"] + SIGMA * rolling_std indicators["lower_band"] = indicators["ma_slow"] - SIGMA * rolling_std indicators["z_score"] = (price - indicators["ma_slow"]) / rolling_std indicators["ma_slow_slope"] = ( (indicators["ma_slow"] / indicators["ma_slow"].shift(SLOPE_WINDOW) - 1.0) * (TRADING_DAYS_PER_YEAR / SLOPE_WINDOW) * 100.0 ) residuals = pd.DataFrame(index=closes.index) if ticker == benchmark: daily_residual = closes[ticker].pct_change() * 0.0 else: aligned = closes[[ticker, benchmark]].dropna() returns = aligned.pct_change() benchmark_variance = returns[benchmark].rolling(BETA_WINDOW).var() rolling_beta = returns[ticker].rolling(BETA_WINDOW).cov(returns[benchmark]) / benchmark_variance daily_residual = returns[ticker] - rolling_beta * returns[benchmark] residuals["residual_20"] = daily_residual.rolling(20).sum() * 100.0 residuals["residual_40"] = daily_residual.rolling(40).sum() * 100.0 return indicators, residuals def build_relative_strength(closes, ticker, benchmark): ticker = normalize_ticker(ticker) benchmark = normalize_ticker(benchmark) aligned = closes[[ticker, benchmark]].dropna() if aligned.empty: raise RuntimeError(f"No overlapping close data found for {ticker} and {benchmark}.") relative_strength = pd.DataFrame(index=aligned.index) relative_strength["rs"] = aligned[ticker] / aligned[benchmark] relative_strength["rs_ma20"] = relative_strength["rs"].rolling(RS_MA20).mean() relative_strength["rs_ma60"] = relative_strength["rs"].rolling(RS_MA60).mean() return relative_strength def wilder_smooth(series, window): return series.ewm(alpha=1 / window, adjust=False, min_periods=window).mean() def build_keltner_squeeze(ohlc): high = ohlc["high"] low = ohlc["low"] close = ohlc["close"] bb_middle = close.rolling(BB_WINDOW).mean() bb_std = close.rolling(BB_WINDOW).std() bb_upper = bb_middle + BB_STD_MULT * bb_std bb_lower = bb_middle - BB_STD_MULT * bb_std bb_width = bb_upper - bb_lower keltner_middle = close.ewm(span=KELT_EMA_WINDOW, adjust=False, min_periods=KELT_EMA_WINDOW).mean() previous_close = close.shift(1) true_range = pd.concat( [(high - low).abs(), (high - previous_close).abs(), (low - previous_close).abs()], axis=1, ).max(axis=1) atr = wilder_smooth(true_range, KELT_ATR_WINDOW) keltner_upper = keltner_middle + KELT_ATR_MULT * atr keltner_lower = keltner_middle - KELT_ATR_MULT * atr kc_width = keltner_upper - keltner_lower squeeze_ratio = bb_width / kc_width.replace(0, np.nan) keltner_squeeze = pd.DataFrame(index=ohlc.index) keltner_squeeze["close"] = close keltner_squeeze["bb_upper"] = bb_upper keltner_squeeze["bb_lower"] = bb_lower keltner_squeeze["bb_width"] = bb_width keltner_squeeze["keltner_upper"] = keltner_upper keltner_squeeze["keltner_lower"] = keltner_lower keltner_squeeze["kc_width"] = kc_width keltner_squeeze["squeeze_ratio"] = squeeze_ratio return keltner_squeeze.replace([np.inf, -np.inf], np.nan) def build_adx_dmi(ohlc): high = ohlc["high"] low = ohlc["low"] close = ohlc["close"] up_move = high.diff() down_move = -low.diff() plus_dm = pd.Series(np.where((up_move > down_move) & (up_move > 0), up_move, 0.0), index=ohlc.index) minus_dm = pd.Series(np.where((down_move > up_move) & (down_move > 0), down_move, 0.0), index=ohlc.index) previous_close = close.shift(1) true_range = pd.concat( [(high - low).abs(), (high - previous_close).abs(), (low - previous_close).abs()], axis=1, ).max(axis=1) atr = wilder_smooth(true_range, ADX_WINDOW).replace(0, np.nan) plus_di = 100.0 * wilder_smooth(plus_dm, ADX_WINDOW) / atr minus_di = 100.0 * wilder_smooth(minus_dm, ADX_WINDOW) / atr dx_denominator = (plus_di + minus_di).replace(0, np.nan) dx = 100.0 * (plus_di - minus_di).abs() / dx_denominator adx = wilder_smooth(dx, ADX_WINDOW) adx_dmi = pd.DataFrame(index=ohlc.index) adx_dmi["adx"] = adx adx_dmi["plus_di"] = plus_di adx_dmi["minus_di"] = minus_di adx_dmi["di_gap"] = plus_di - minus_di return adx_dmi.replace([np.inf, -np.inf], np.nan) def last_n_years(frame, years): clean_index = frame.dropna(how="all").index if clean_index.empty: return frame start = clean_index.max() - pd.DateOffset(years=years) return frame.loc[frame.index >= start] def setup_axis(ax, ylabel=None): ax.set_facecolor(AX_FACE) ax.grid(True, color=GRID, linewidth=1.0, alpha=0.55) ax.tick_params(colors=MUTED, labelsize=10) for spine in ax.spines.values(): spine.set_color("#0F1018") spine.set_linewidth(1.2) if ylabel: ax.set_ylabel(ylabel, color=TEXT, fontsize=11) ax.xaxis.set_major_locator(mdates.MonthLocator(interval=6)) ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m")) def fmt_value(value, decimals=2, suffix=""): if pd.isna(value): return "n/a" return f"{value:.{decimals}f}{suffix}" def latest_metric_line(title, row, metrics): latest_date = pd.Timestamp(row.name).date() values = " | ".join(f"**{label}:** `{value}`" for label, value in metrics) return f"**Latest {title} ({latest_date})**\n\n{values}" def dmi_direction_label(di_gap): if pd.isna(di_gap): return "n/a" if di_gap >= DMI_CLEAR_GAP: return "+DI clearly stronger" if di_gap <= -DMI_CLEAR_GAP: return "-DI clearly stronger" return "direction unclear" def adx_regime_label(adx): if pd.isna(adx): return "n/a" if adx >= 40: return "strong trend / exhaustion watch" if adx >= 25: return "trend confirmed" if adx >= 20: return "range / trend boundary" return "range" def plot_mean_reversion(data, ticker): plot_data = last_n_years(data, YEARS).dropna(subset=["close"]) fig, ax = plt.subplots(figsize=(18, 8), facecolor=FIG_FACE) setup_axis(ax, "Price ($)") band = plot_data.dropna(subset=["upper_band", "lower_band"]) ax.fill_between( band.index, band["lower_band"], band["upper_band"], color=BAND, alpha=0.34, label=f"{SIGMA:g} Sigma Band", linewidth=0, ) ax.plot(plot_data.index, plot_data["close"], color=PRICE, linewidth=1.35, label=f"{ticker} Price") ax.plot(plot_data.index, plot_data["ma_slow"], color=MA200, linestyle="--", linewidth=1.35, label=f"{MA_SLOW} DMA") ax.plot(plot_data.index, plot_data["ma_fast"], color=MA50, linestyle="--", linewidth=1.35, label=f"{MA_FAST} DMA") ax.set_title(f"{ticker} Mean Reversion Dashboard ({YEARS}-Year)", color=TEXT, fontsize=18, weight="bold") legend = ax.legend(loc="upper left", frameon=True, facecolor="#F1F1F4", edgecolor="#C8C9CF", fontsize=10) for text in legend.get_texts(): text.set_color("#151720") fig.tight_layout() latest = plot_data.iloc[-1] latest_markdown = latest_metric_line( "mean reversion data", latest, [ (f"{ticker} close", fmt_value(latest["close"])), (f"{MA_FAST} DMA", fmt_value(latest["ma_fast"])), (f"{MA_SLOW} DMA", fmt_value(latest["ma_slow"])), (f"{SIGMA:g} sigma band", f"{fmt_value(latest['lower_band'])} to {fmt_value(latest['upper_band'])}"), ], ) return fig, latest_markdown def plot_relative_strength(relative_strength, ticker, benchmark): plot_data = last_n_years(relative_strength, YEARS).dropna(subset=["rs"]) fig, ax = plt.subplots(figsize=(18, 4.8), facecolor=FIG_FACE) setup_axis(ax, "RS Ratio") ax.plot(plot_data.index, plot_data["rs"], color=PRICE, linewidth=1.35, label=f"RS = {ticker} / {benchmark}") ax.plot(plot_data.index, plot_data["rs_ma20"], color=MA50, linestyle="--", linewidth=1.35, label="RS_MA20") ax.plot(plot_data.index, plot_data["rs_ma60"], color=MA200, linestyle="--", linewidth=1.35, label="RS_MA60") ax.set_title(f"Relative Strength ({ticker} / {benchmark})", color=TEXT, fontsize=16) legend = ax.legend(loc="upper left", frameon=True, facecolor="#F1F1F4", edgecolor="#C8C9CF", fontsize=10) for text in legend.get_texts(): text.set_color("#151720") fig.tight_layout() latest = plot_data.iloc[-1] latest_markdown = latest_metric_line( "relative strength", latest, [ ("RS", fmt_value(latest["rs"], decimals=4)), ("RS_MA20", fmt_value(latest["rs_ma20"], decimals=4)), ("RS_MA60", fmt_value(latest["rs_ma60"], decimals=4)), ], ) return fig, latest_markdown def plot_keltner_squeeze(keltner_squeeze, ticker): plot_data = last_n_years(keltner_squeeze, YEARS).dropna(subset=["close"]) fig, (ax_price, ax_ratio) = plt.subplots( 2, 1, figsize=(18, 7.2), sharex=True, gridspec_kw={"height_ratios": [3, 1]}, facecolor=FIG_FACE, ) setup_axis(ax_price, "Price ($)") setup_axis(ax_ratio, "Ratio") keltner_band = plot_data.dropna(subset=["keltner_upper", "keltner_lower"]) ax_price.fill_between( keltner_band.index, keltner_band["keltner_lower"], keltner_band["keltner_upper"], color=BAND, alpha=0.24, label=f"KELT({KELT_EMA_WINDOW},{KELT_ATR_MULT:g},{KELT_ATR_WINDOW})", linewidth=0, ) ax_price.plot(plot_data.index, plot_data["close"], color=PRICE, linewidth=1.35, label=f"{ticker} Price") ax_price.plot(plot_data.index, plot_data["bb_upper"], color=MA50, linestyle="--", linewidth=1.15, label=f"BB Upper ({BB_WINDOW},{BB_STD_MULT:g})") ax_price.plot(plot_data.index, plot_data["bb_lower"], color=MA50, linestyle="--", linewidth=1.15, label=f"BB Lower ({BB_WINDOW},{BB_STD_MULT:g})") ax_price.plot(plot_data.index, plot_data["keltner_upper"], color=GOLD, linewidth=1.2, label="Keltner Upper") ax_price.plot(plot_data.index, plot_data["keltner_lower"], color=ORANGE, linewidth=1.2, label="Keltner Lower") ax_price.set_title(f"{ticker} KELT / Bollinger Squeeze", color=TEXT, fontsize=16) ratio_data = plot_data.dropna(subset=["squeeze_ratio"]) ax_ratio.plot(ratio_data.index, ratio_data["squeeze_ratio"], color=PRICE, linewidth=1.3, label="Squeeze_Ratio = BB_Width / KC_Width") ax_ratio.axhline(1.0, color=NEG, linestyle="--", linewidth=1.15, alpha=0.75, label="Squeeze threshold (1.0)") ax_ratio.set_xlabel("Date", color=TEXT, fontsize=11) for ax in (ax_price, ax_ratio): legend = ax.legend(loc="upper left", frameon=True, facecolor="#F1F1F4", edgecolor="#C8C9CF", fontsize=9) for text in legend.get_texts(): text.set_color("#151720") fig.tight_layout() latest = ratio_data.iloc[-1] squeeze_state = "squeeze on" if latest["squeeze_ratio"] < 1 else "squeeze off" latest_markdown = latest_metric_line( "KELT / squeeze", latest, [ ("BB_Width", fmt_value(latest["bb_width"])), ("KC_Width", fmt_value(latest["kc_width"])), ("Squeeze_Ratio", fmt_value(latest["squeeze_ratio"], decimals=3)), ("state", squeeze_state), ], ) return fig, latest_markdown def plot_adx_dmi_regime_map(adx_dmi): plot_data = last_n_years(adx_dmi, YEARS).dropna(subset=["adx", "plus_di", "minus_di", "di_gap"]) fig, ax = plt.subplots(figsize=(18, 5.4), facecolor=FIG_FACE) setup_axis(ax, "ADX") y_max = max(45.0, float(plot_data["adx"].max()) * 1.12) green_mask = (plot_data["di_gap"] >= DMI_CLEAR_GAP).to_numpy() red_mask = (plot_data["di_gap"] <= -DMI_CLEAR_GAP).to_numpy() gray_mask = ~(green_mask | red_mask) ax.fill_between(plot_data.index, 0, y_max, where=green_mask, color=POS, alpha=0.16, linewidth=0) ax.fill_between(plot_data.index, 0, y_max, where=red_mask, color=NEG, alpha=0.15, linewidth=0) ax.fill_between(plot_data.index, 0, y_max, where=gray_mask, color=NEUTRAL, alpha=0.12, linewidth=0) ax.plot(plot_data.index, plot_data["adx"], color=GOLD, linewidth=1.8, label=f"ADX({ADX_WINDOW}) smoothed") for level, label in [ (40, "40 Strong Trend / Exhaustion Watch"), (25, "25 Trend Confirmed"), (20, "20 Range / Trend Boundary"), ]: ax.axhline(level, color="#D7D9E4", linestyle="--", linewidth=1.0, alpha=0.62) ax.text(plot_data.index[0], level + 0.8, label, color=MUTED, fontsize=10, va="bottom", ha="left") ax.set_ylim(0, y_max) ax.set_title("ADX / DMI Regime Map", color=TEXT, fontsize=16) ax.set_xlabel("Date", color=TEXT, fontsize=11) handles = [ Line2D([0], [0], color=GOLD, linewidth=1.8, label=f"ADX({ADX_WINDOW}) smoothed"), Patch(facecolor=POS, alpha=0.35, label=f"+DI stronger by >= {DMI_CLEAR_GAP:g}"), Patch(facecolor=NEG, alpha=0.35, label=f"-DI stronger by >= {DMI_CLEAR_GAP:g}"), Patch(facecolor=NEUTRAL, alpha=0.35, label="Direction unclear"), ] legend = ax.legend(handles=handles, loc="upper left", frameon=True, facecolor="#F1F1F4", edgecolor="#C8C9CF", fontsize=9) for text in legend.get_texts(): text.set_color("#151720") fig.tight_layout() latest = plot_data.iloc[-1] latest_markdown = latest_metric_line( "ADX / DMI regime", latest, [ (f"ADX({ADX_WINDOW})", fmt_value(latest["adx"])), ("+DI", fmt_value(latest["plus_di"])), ("-DI", fmt_value(latest["minus_di"])), ("ADX zone", adx_regime_label(latest["adx"])), ("DMI background", dmi_direction_label(latest["di_gap"])), ], ) return fig, latest_markdown def plot_zscore(data): plot_data = last_n_years(data, YEARS).dropna(subset=["z_score"]) colors = np.where(plot_data["z_score"] >= SIGMA, POS, np.where(plot_data["z_score"] <= -SIGMA, NEG, NEUTRAL)) fig, ax = plt.subplots(figsize=(18, 4.6), facecolor=FIG_FACE) setup_axis(ax, "Z-Score") ax.bar(plot_data.index, plot_data["z_score"], width=2.6, color=colors, edgecolor=colors, alpha=0.72) ax.axhline(0, color="#D7D9E4", linewidth=1.0, alpha=0.55) ax.axhline(SIGMA, color=POS, linestyle="--", linewidth=1.4, alpha=0.75) ax.axhline(-SIGMA, color=NEG, linestyle="--", linewidth=1.4, alpha=0.75) ax.set_title("Price Z-Score vs 200 DMA", color=TEXT, fontsize=16) handles = [ Line2D([0], [0], color=NEG, linestyle="--", linewidth=1.4, label=f"Oversold (-{SIGMA:g})"), Line2D([0], [0], color=POS, linestyle="--", linewidth=1.4, label=f"Overbought (+{SIGMA:g})"), ] legend = ax.legend(handles=handles, loc="upper left", frameon=True, facecolor="#F1F1F4", edgecolor="#C8C9CF", fontsize=10) for text in legend.get_texts(): text.set_color("#151720") fig.tight_layout() latest = plot_data.iloc[-1] latest_markdown = latest_metric_line( "z-score", latest, [("Z-Score", fmt_value(latest["z_score"]))], ) return fig, latest_markdown def plot_residuals(residuals, ticker, benchmark): plot_data = last_n_years(residuals, YEARS).dropna(how="all") fig, ax = plt.subplots(figsize=(18, 4.8), facecolor=FIG_FACE) setup_axis(ax, "Residual (%)") ax.axhline(0, color="#D7D9E4", linewidth=1.1, alpha=0.60) ax.plot(plot_data.index, plot_data["residual_20"], color=GOLD, linewidth=1.4, label="20-Day Cum. Residual") ax.plot(plot_data.index, plot_data["residual_40"], color=ORANGE, linewidth=1.4, label="40-Day Cum. Residual") ax.set_title(f"Beta-Adjusted Residual ({ticker} vs {benchmark}-Predicted)", color=TEXT, fontsize=16) legend = ax.legend(loc="upper left", frameon=True, facecolor="#F1F1F4", edgecolor="#C8C9CF", fontsize=10) for text in legend.get_texts(): text.set_color("#151720") fig.tight_layout() latest = plot_data.iloc[-1] latest_markdown = latest_metric_line( "beta-adjusted residual", latest, [ ("20-day cumulative residual", fmt_value(latest["residual_20"], suffix="%")), ("40-day cumulative residual", fmt_value(latest["residual_40"], suffix="%")), ], ) return fig, latest_markdown def plot_slope(data): plot_data = last_n_years(data, YEARS).dropna(subset=["ma_slow_slope"]) colors = np.where(plot_data["ma_slow_slope"] >= 0, POS, NEG) fig, ax = plt.subplots(figsize=(18, 4.8), facecolor=FIG_FACE) setup_axis(ax, "Slope (%)") ax.bar(plot_data.index, plot_data["ma_slow_slope"], width=2.6, color=colors, edgecolor=colors, alpha=0.72) ax.axhline(0, color="#D7D9E4", linewidth=1.1, alpha=0.60) ax.set_title("200 DMA Slope (Regime Filter)", color=TEXT, fontsize=16) ax.set_xlabel("Date", color=TEXT, fontsize=11) fig.tight_layout() latest = plot_data.iloc[-1] latest_markdown = latest_metric_line( "200 DMA slope", latest, [("annualized slope", fmt_value(latest["ma_slow_slope"], suffix="%"))], ) return fig, latest_markdown def build_dashboard(ticker): ticker = normalize_ticker(ticker) benchmark = normalize_ticker(BENCHMARK) closes, ticker_ohlc = download_daily_market_data(ticker, benchmark) indicators, residuals = build_indicators(closes, ticker, benchmark) relative_strength = build_relative_strength(closes, ticker, benchmark) keltner_squeeze = build_keltner_squeeze(ticker_ohlc) adx_dmi = build_adx_dmi(ticker_ohlc) charts = [ plot_mean_reversion(indicators, ticker), plot_relative_strength(relative_strength, ticker, benchmark), plot_keltner_squeeze(keltner_squeeze, ticker), plot_adx_dmi_regime_map(adx_dmi), plot_zscore(indicators), plot_residuals(residuals, ticker, benchmark), plot_slope(indicators), ] latest_close = closes[ticker].dropna().iloc[-1] latest_date = closes[ticker].dropna().index[-1].date() status = f"### {ticker} results\nLatest closed daily bar: `{latest_date}` close=`{latest_close:.2f}`" outputs = [status] for fig, markdown in charts: outputs.extend([fig, markdown]) return outputs def safe_build_dashboard(ticker): try: return build_dashboard(ticker) except Exception as exc: empty_figs = [None, ""] * 7 return [f"### Error\n`{exc}`", *empty_figs] with gr.Blocks(title="Trading Dashboard", theme=gr.themes.Soft()) as demo: gr.Markdown("# Trading Dashboard") with gr.Row(): ticker_input = gr.Textbox(value=DEFAULT_TICKER, label="Ticker", placeholder="TSLA / AAPL / NVDA") run_button = gr.Button("Run", variant="primary") status_output = gr.Markdown() outputs = [status_output] plot_outputs = [] markdown_outputs = [] chart_titles = [ "Mean Reversion Dashboard", "Relative Strength", "KELT / Bollinger Squeeze", "ADX / DMI Regime Map", "Price Z-Score vs 200 DMA", "Beta-Adjusted Residual", "200 DMA Slope", ] for title in chart_titles: gr.Markdown(f"## {title}") plot_component = gr.Plot() markdown_component = gr.Markdown() plot_outputs.append(plot_component) markdown_outputs.append(markdown_component) outputs.extend([plot_component, markdown_component]) run_button.click(safe_build_dashboard, inputs=ticker_input, outputs=outputs) ticker_input.submit(safe_build_dashboard, inputs=ticker_input, outputs=outputs) demo.load(safe_build_dashboard, inputs=ticker_input, outputs=outputs) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)