app.py

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)