app.py

"""Gradio UI for the US Stock Institutional Flow Scanner.

Deployable to Hugging Face Spaces (Gradio SDK) or run locally with
``python app.py``.
"""

from __future__ import annotations

import os
import sys
import threading
import time
from datetime import datetime, timedelta
from typing import Optional

import numpy as np
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots

import gradio as gr

# ---------------------------------------------------------------------------
# Workaround for gradio-client==1.3.0 + Gradio 4.44.0 incompatibility on
# Python 3.13.  ``gradio_client.utils.get_type`` does ``if "const" in schema:``
# where ``schema`` is sometimes a bool, raising
# ``TypeError: argument of type 'bool' is not iterable`` when Gradio tries
# to build its API schema at launch time.  We monkey-patch the function to
# be a no-op for non-dict inputs; the resulting API docs lose the affected
# field's type annotation, but the app launches and the UI works fine.
# ---------------------------------------------------------------------------
try:
    import gradio_client.utils as _gc_utils
    _orig_get_type = _gc_utils.get_type
    def _safe_get_type(schema, *args, **kwargs):
        if not isinstance(schema, dict):
            return "Any"
        return _orig_get_type(schema, *args, **kwargs)
    _gc_utils.get_type = _safe_get_type
    # Also patch the function the recursive walker uses
    _orig_j2pt = getattr(_gc_utils, "_json_schema_to_python_type", None)
    if _orig_j2pt is not None:
        def _safe_j2pt(schema, defs=None):
            if not isinstance(schema, dict):
                return "Any"
            return _orig_j2pt(schema, defs)
        _gc_utils._json_schema_to_python_type = _safe_j2pt
except Exception as _patch_err:  # never let this crash startup
    print(f"[gradio-client patch skipped: {_patch_err}]", file=sys.stderr)

from scanner.data_fetcher import fetch_ohlcv
from scanner.factor_sources import get_data_source
from scanner.flow_algo import compute_factors
from scanner.history import (
    latest_snapshot, save_snapshot, snapshot_summary, with_delta,
)
from scanner.intraday_factor import compute_intraday_factors_batch
from scanner.l2_factor import compute_l2_factors
from scanner.options_factor import compute_options_factors
from scanner.tick_factor import compute_tick_factors_batch
from scanner import paths
from scanner.performance import (
    HORIZON_DAYS, auto_improve, load_learned_meta,
    load_learned_weights, load_performance_log,
)
from scanner.persistence import pull_remote_cache, push_remote_cache
from scanner.scorer import DEFAULT_WEIGHTS, FACTOR_KEYS, score_factors, top_n
from scanner.universe import (
    SECTORS, apply_liquidity_filter, cached_sectors_for, get_sectors_for,
    load_universe,
)
from scanner.watchlist import load_watchlist, parse_tickers, save_watchlist


# ---------------------------------------------------------------------------
# State
# ---------------------------------------------------------------------------

RESULTS_STATE: dict = {
    "df": None,            # last scan result DataFrame (with deltas)
    "frames": None,        # last scan OHLCV frames
    "weights": dict(DEFAULT_WEIGHTS),
    "last_run": None,
    "last_msg": "No scan run yet.",
    "scanned_universe": [],
    "next_daily_run": None,  # datetime of the next scheduled auto-scan
}


# Daily auto-scan configuration
DAILY_SCAN_INITIAL_DELAY_SEC = 20        # let the app + Gradio queue boot
DAILY_SCAN_PERIOD_SEC = 24 * 3600        # 24 hours between scheduled scans


class _NoOpProgress:
    """No-op stand-in for ``gr.Progress()`` used by background scans."""
    def __call__(self, frac, desc=None):
        pass


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _format_status() -> str:
    n = 0 if RESULTS_STATE["df"] is None else len(RESULTS_STATE["df"])
    last = RESULTS_STATE["last_run"].strftime("%Y-%m-%d %H:%M:%S") if RESULTS_STATE["last_run"] else "never"
    nxt = RESULTS_STATE.get("next_daily_run")
    nxt_str = ""
    if nxt is not None:
        nxt_str = f"  •  **Next auto-scan:** {nxt.strftime('%Y-%m-%d %H:%M:%S')} UTC"
    return f"**Last run:** {last}  •  {n} stocks  •  {RESULTS_STATE['last_msg']}{nxt_str}"


def _result_columns() -> list[str]:
    return ["ticker", "name", "rating", "score", "score_delta", "last_close",
            "adv_dollar", "cmf", "obv_slope", "big_bar_ratio", "vwap_dev",
            "rvol_signed", "l2_imbalance", "unusual_options",
            "block_aggression", "buy_persistence"]


def _df_for_display(df: Optional[pd.DataFrame]) -> Optional[pd.DataFrame]:
    if df is None or df.empty:
        return None
    cols = [c for c in _result_columns() if c in df.columns]
    return df[cols].copy()


def _watchlist_df(df: Optional[pd.DataFrame]) -> Optional[pd.DataFrame]:
    wl = set(load_watchlist())
    if df is None or df.empty or not wl:
        return None
    sub = df[df["ticker"].isin(wl)]
    if sub.empty:
        return None
    return _df_for_display(sub)


def _plot_detail(ticker: str, frames: dict | None) -> Optional[go.Figure]:
    if not ticker or not frames or ticker not in frames or frames[ticker].empty:
        return None
    df = frames[ticker].copy()
    if "Date" not in df.columns and df.index.name != "Date":
        df = df.reset_index()
    date_col = "Date" if "Date" in df.columns else df.columns[0]
    df[date_col] = pd.to_datetime(df[date_col])

    # Re-compute CMF for the plot
    close = df["Close"].astype(float)
    high = df["High"].astype(float)
    low = df["Low"].astype(float)
    vol = df["Volume"].astype(float)
    rng = (high - low).replace(0, np.nan)
    mfm = ((close - low) - (high - close)) / rng
    mfm = mfm.fillna(0.0)
    mfv = mfm * vol
    cmf = (mfv.rolling(20).sum() / vol.rolling(20).sum()).fillna(0)

    fig = make_subplots(
        rows=3, cols=1, shared_xaxes=True, vertical_spacing=0.03,
        row_heights=[0.6, 0.2, 0.2],
    )
    fig.add_trace(go.Candlestick(
        x=df[date_col], open=df["Open"], high=df["High"],
        low=df["Low"], close=df["Close"], name="Price",
    ), row=1, col=1)
    colors = ["#26a69a" if c >= o else "#ef5350" for c, o in zip(df["Close"], df["Open"])]
    fig.add_trace(go.Bar(
        x=df[date_col], y=df["Volume"], marker_color=colors, name="Volume",
    ), row=2, col=1)
    cmf_colors = ["#26a69a" if v >= 0 else "#ef5350" for v in cmf]
    fig.add_trace(go.Bar(
        x=df[date_col], y=cmf, marker_color=cmf_colors, name="CMF(20)",
    ), row=3, col=1)
    fig.update_layout(
        height=650,
        title=f"{ticker} — Price, Volume, CMF(20)",
        xaxis_rangeslider_visible=False,
        showlegend=False,
        margin=dict(l=40, r=20, t=40, b=20),
    )
    fig.update_yaxes(title_text="Price", row=1, col=1)
    fig.update_yaxes(title_text="Vol", row=2, col=1)
    fig.update_yaxes(title_text="CMF", row=3, col=1)
    return fig


def _build_csv(df: Optional[pd.DataFrame]) -> Optional[str]:
    if df is None or df.empty:
        return None
    df.to_csv(paths.RESULTS_CSV_PATH, index=False)
    return paths.RESULTS_CSV_PATH


def _sector_breakdown_figure(df: Optional[pd.DataFrame]) -> Optional[go.Figure]:
    if df is None or df.empty:
        return None
    sec_map = cached_sectors_for(df["ticker"].astype(str).tolist())
    if not sec_map:
        return None
    work = df[df["ticker"].isin(sec_map)].copy()
    if work.empty:
        return None
    work["sector"] = work["ticker"].map(sec_map)
    agg = (work.groupby("sector")
                .agg(mean_score=("score", "mean"),
                     count=("ticker", "count"))
                .reset_index())
    agg = agg.sort_values("mean_score", ascending=True)
    colors = ["#26a69a" if v >= 0 else "#ef5350" for v in agg["mean_score"]]
    fig = go.Figure(go.Bar(
        x=agg["mean_score"],
        y=agg["sector"],
        orientation="h",
        marker_color=colors,
        text=[f"n={n}" for n in agg["count"]],
        textposition="auto",
    ))
    fig.update_layout(
        title=f"Average flow score by sector ({len(work)} tickers with cached sector)",
        xaxis_title="Mean composite score",
        yaxis_title="",
        height=420,
        margin=dict(l=20, r=20, t=50, b=20),
    )
    fig.add_vline(x=0, line_color="#888", line_width=1)
    return fig


def _sector_status() -> str:
    if RESULTS_STATE["df"] is None or RESULTS_STATE["df"].empty:
        return "_Run a scan first._"
    tickers = RESULTS_STATE["df"]["ticker"].astype(str).tolist()
    cached = cached_sectors_for(tickers)
    missing = len(tickers) - len(cached)
    return (f"**{len(cached)}** tickers have cached sector data, "
            f"**{missing}** are missing.  Hit *Fetch missing sectors* to fill them in "
            f"(uses yfinance `.info`, ~1 request per missing ticker).")


# ---------------------------------------------------------------------------
# Auto-tune helpers
# ---------------------------------------------------------------------------

def _learned_weights_table() -> pd.DataFrame:
    """Side-by-side view of default vs learned weights for the Performance tab."""
    learned = load_learned_weights() or {}
    rows = []
    for k in FACTOR_KEYS:
        rows.append({
            "factor": k,
            "default": round(float(DEFAULT_WEIGHTS.get(k, 0.0)), 3),
            "learned": round(float(learned[k]), 3) if k in learned else None,
        })
    return pd.DataFrame(rows)


def _performance_status() -> str:
    meta = load_learned_meta()
    if not meta:
        return ("_No auto-tuned weights yet._  After a few scans (across "
                f"at least {HORIZON_DAYS}+ trading days) the optimizer will "
                "start finding tuned weights.")
    m = meta.get("metrics", {}) or {}
    ic = m.get("mean_ic")
    base = m.get("baseline_ic")
    gain = m.get("ic_gain")
    n = m.get("n_periods", 0)
    saved = meta.get("saved_at", "?")
    parts = [f"**Saved:** {saved}"]
    if ic is not None and ic == ic:  # not NaN
        parts.append(f"**Mean IC:** {ic:+.4f}")
    if base is not None and base == base:
        parts.append(f"**Baseline IC:** {base:+.4f}")
    if gain is not None and gain == gain:
        parts.append(f"**Gain:** {gain:+.4f}")
    parts.append(f"**Periods:** {n}")
    if m.get("horizon_days"):
        parts.append(f"**Horizon:** {m['horizon_days']} trading days")
    return "  •  ".join(parts)


def _performance_chart() -> Optional[go.Figure]:
    log = load_performance_log()
    if log.empty or "mean_ic" not in log.columns:
        return None
    log = log.copy()
    log["saved_at"] = pd.to_datetime(log["saved_at"], errors="coerce")
    log = log.dropna(subset=["saved_at"]).sort_values("saved_at")
    if log.empty:
        return None
    fig = make_subplots(rows=2, cols=1, shared_xaxes=True,
                        vertical_spacing=0.08,
                        row_heights=[0.55, 0.45],
                        subplot_titles=("Mean IC after each scan",
                                        "Hit rate"))
    fig.add_trace(go.Scatter(
        x=log["saved_at"], y=log["mean_ic"], mode="lines+markers",
        line=dict(color="#26a69a"), name="Mean IC",
    ), row=1, col=1)
    if "baseline_ic" in log.columns:
        fig.add_trace(go.Scatter(
            x=log["saved_at"], y=log["baseline_ic"], mode="lines",
            line=dict(color="#888", dash="dot"), name="Baseline IC",
        ), row=1, col=1)
    if "hit_rate" in log.columns:
        fig.add_trace(go.Scatter(
            x=log["saved_at"], y=log["hit_rate"], mode="lines+markers",
            line=dict(color="#ef5350"), name="Hit rate",
        ), row=2, col=1)
    fig.add_hline(y=0.0, row=1, col=1, line_color="#888", line_width=1)
    fig.add_hline(y=0.5, row=2, col=1, line_color="#888", line_width=1,
                  line_dash="dot")
    fig.update_layout(height=420, margin=dict(l=40, r=20, t=40, b=20),
                      showlegend=True, legend=dict(orientation="h"))
    fig.update_yaxes(title_text="IC", row=1, col=1)
    fig.update_yaxes(title_text="Hit rate", range=[0.3, 0.7], row=2, col=1)
    return fig


def _start_auto_improve(base_weights: dict) -> None:
    """Fire-and-forget background tuning."""
    def _worker():
        try:
            auto_improve(base_weights)
        except Exception:
            pass
    threading.Thread(target=_worker, daemon=True).start()


# ---------------------------------------------------------------------------
# Daily auto-scan
# ---------------------------------------------------------------------------

def _run_scheduled_scan(reason: str) -> None:
    """Run a single full-universe-Filtered scan with the slider defaults
    and update :data:`RESULTS_STATE`.  Used by :func:`_start_daily_scan`.
    """
    weights, _ = _resolve_weights(
        False,
        DEFAULT_WEIGHTS["cmf"],
        DEFAULT_WEIGHTS["obv_slope"],
        DEFAULT_WEIGHTS["big_bar_ratio"],
        DEFAULT_WEIGHTS["vwap_dev"],
        DEFAULT_WEIGHTS["rvol_signed"],
    )
    try:
        df, frames, scanned, msg = _do_scan(
            "Filtered (default)", 5.0, 5_000_000.0, "All", "All",
            weights, _NoOpProgress(),
        )
    except Exception as e:
        RESULTS_STATE["last_msg"] = f"Error ({reason}): {e}"
        return

    suffix = f" (auto: {reason})"
    RESULTS_STATE["df"] = df
    RESULTS_STATE["frames"] = frames
    RESULTS_STATE["weights"] = weights
    RESULTS_STATE["last_run"] = datetime.utcnow()
    RESULTS_STATE["last_msg"] = f"{msg}{suffix}"
    RESULTS_STATE["scanned_universe"] = scanned
    if df is not None and not df.empty:
        _start_auto_improve(weights)


def _start_daily_scan() -> None:
    """Background thread: run a scan on startup (if the last one is
    older than :data:`DAILY_SCAN_PERIOD_SEC`), then re-run once every
    24 hours.  HF Spaces can sleep after 48h of inactivity, so the
    startup gate is what actually delivers the "one scan per day"
    guarantee: a fresh visit always triggers a scan if the cached one
    is stale.
    """
    def _worker():
        time.sleep(DAILY_SCAN_INITIAL_DELAY_SEC)
        # Initial scan: only if last successful scan is stale
        last = RESULTS_STATE.get("last_run")
        stale = (
            last is None
            or (datetime.utcnow() - last).total_seconds() > DAILY_SCAN_PERIOD_SEC
        )
        if stale:
            try:
                _run_scheduled_scan("startup")
            except Exception as e:
                RESULTS_STATE["last_msg"] = f"Auto-scan (startup) failed: {e}"
        # Then loop, scheduling a scan every 24h
        while True:
            RESULTS_STATE["next_daily_run"] = (
                datetime.utcnow() + timedelta(seconds=DAILY_SCAN_PERIOD_SEC)
            )
            time.sleep(DAILY_SCAN_PERIOD_SEC)
            try:
                _run_scheduled_scan("daily")
            except Exception as e:
                RESULTS_STATE["last_msg"] = f"Auto-scan (daily) failed: {e}"
    threading.Thread(target=_worker, daemon=True).start()


# ---------------------------------------------------------------------------
# Scan logic
# ---------------------------------------------------------------------------

def _do_scan(
    mode: str,
    min_price: float,
    min_adv: float,
    sector: str,
    cmf_filter: str,
    weights: dict,
    progress: gr.Progress,
) -> tuple[Optional[pd.DataFrame], dict, list[str], str]:
    """Run the full pipeline.  Returns (results_df, frames_dict, scanned, msg)."""
    progress(0, desc="Loading universe...")
    universe = load_universe()
    all_tickers = universe["ticker"].tolist()
    if not all_tickers:
        return None, {}, [], "Universe is empty."

    if mode.startswith("Filtered") or sector != "All":
        progress(0.05, desc=f"Liquidity filter (price>${min_price}, ADV>${min_adv:,.0f})...")
        tickers = apply_liquidity_filter(all_tickers, min_price=float(min_price),
                                          min_adv_usd=float(min_adv))
    else:
        tickers = all_tickers

    if not tickers:
        return None, {}, [], "No tickers left after liquidity filter."

    # Sector filter (uses disk cache; only hits the network for missing entries)
    if sector != "All":
        progress(0.1, desc=f"Resolving sectors for {len(tickers)} tickers...")

        def _sec_cb(done: int, total: int, t: str) -> None:
            if total > 0:
                progress(0.10 + 0.05 * (done / total),
                         desc=f"Sectors {done}/{total} ({t or ''})")

        sec_map = get_sectors_for(tickers, progress_cb=_sec_cb)
        keep = [t for t in tickers if sec_map.get(t) == sector]
        if not keep:
            return None, {}, tickers, f"No tickers matched sector={sector}."
        tickers = keep

    # Always include watchlist tickers, regardless of liquidity/sector filter.
    watch = load_watchlist()
    extras = [t for t in watch if t not in tickers]
    if extras:
        tickers = tickers + extras

    scanned_universe = tickers
    progress(0.15, desc=f"Pulling OHLCV for {len(tickers)} tickers...")

    def cb(done: int, total: int, t: str) -> None:
        if total > 0:
            frac = 0.15 + 0.70 * (done / total)
            progress(frac, desc=f"Pulled {done}/{total} ({t})")

    frames, failed = fetch_ohlcv(tickers, period="6mo", progress_cb=cb)

    progress(0.88, desc="Computing factors...")
    factors = []
    valid_frames = {}
    for t, f in frames.items():
        if f is None or f.empty or len(f) < 30:
            continue
        fs = compute_factors(f, t)
        if fs is None:
            continue
        factors.append(fs)
        valid_frames[t] = f

    if not factors:
        return None, valid_frames, scanned_universe, "No tickers had enough data."

    # ----------------------------------------------------------------
    # Institutional-flow factors (L2, options, ticks, intraday).
    # Computed from the configured FactorDataSource (StubDataSource by
    # default; set FSCANNER_DATA_SOURCE=futu for live Futu OpenD).  Any
    # data-source exception is swallowed - a missing real feed just
    # leaves the new factors at neutral (0.0) and the scan still
    # produces a valid ranking from the original 5 flow factors.
    # ----------------------------------------------------------------
    progress(0.92, desc="Computing institutional factors (L2/options/ticks/intraday)...")
    extra_factors = {"l2_imbalance": {}, "unusual_options": {},
                     "block_aggression": {}, "buy_persistence": {}}
    inst_tickers = [fs.ticker for fs in factors]
    try:
        source = get_data_source()
        l2 = compute_l2_factors(inst_tickers, source=source)
        extra_factors["l2_imbalance"] = l2
        opt = compute_options_factors(inst_tickers, source=source)
        extra_factors["unusual_options"] = opt
        ticks_df = compute_tick_factors_batch(inst_tickers, source=source)
        for t in inst_tickers:
            if t in ticks_df.index:
                extra_factors["block_aggression"][t] = float(
                    ticks_df.loc[t, "block_aggression"]
                )
        intraday_df = compute_intraday_factors_batch(inst_tickers, source=source)
        for t in inst_tickers:
            if t in intraday_df.index:
                extra_factors["buy_persistence"][t] = float(
                    intraday_df.loc[t, "aggression_persistence"]
                )
    except Exception as e:
        print(f"[institutional factors skipped: {e}]", file=sys.stderr)

    progress(0.94, desc="Scoring..." )
    df = score_factors(factors, weights=weights, extra_factors=extra_factors)

    # Attach names
    name_map = dict(zip(universe["ticker"], universe["name"]))
    df["name"] = df["ticker"].map(name_map).fillna("")

    # CMF-side filter (purely a UI filter on results).  Watchlist tickers
    # are still kept so users always see them.
    watch_set = set(watch)
    if cmf_filter == "Net buying":
        df = df[(df["cmf"] > 0) | (df["ticker"].isin(watch_set))]
    elif cmf_filter == "Net selling":
        df = df[(df["cmf"] < 0) | (df["ticker"].isin(watch_set))]

    # Attach delta vs. previous snapshot, then save current
    prev = latest_snapshot()
    df = with_delta(df, prev)
    save_snapshot(df)

    progress(1.0, desc="Done.")
    msg = f"Scanned {len(scanned_universe)} tickers; scored {len(df)}."
    if failed:
        msg += f" ({len(failed)} fetches failed.)"
    if extras:
        msg += f" Watchlist add-ons: {len(extras)}."

    # Async push to HF Dataset
    threading.Thread(target=push_remote_cache, daemon=True).start()

    return df, valid_frames, scanned_universe, msg


# ---------------------------------------------------------------------------
# Gradio callbacks
# ---------------------------------------------------------------------------

def _weights_from_sliders(w_cmf, w_obv, w_big, w_vwap, w_rvol):
    s = w_cmf + w_obv + w_big + w_vwap + w_rvol
    if s <= 0:
        return dict(DEFAULT_WEIGHTS)
    return {
        "cmf": w_cmf / s,
        "obv_slope": w_obv / s,
        "big_bar_ratio": w_big / s,
        "vwap_dev": w_vwap / s,
        "rvol_signed": w_rvol / s,
    }


def _resolve_weights(use_learned: bool, w_cmf, w_obv, w_big, w_vwap, w_rvol
                     ) -> tuple[dict, str]:
    """Pick slider weights or saved learned weights based on the checkbox."""
    if use_learned:
        learned = load_learned_weights()
        if learned:
            return learned, "auto-tuned"
    return _weights_from_sliders(w_cmf, w_obv, w_big, w_vwap, w_rvol), "manual"


def run_scan(mode, min_price, min_adv, sector, cmf_filter, use_learned,
             w_cmf, w_obv, w_big, w_vwap, w_rvol, progress=gr.Progress()):
    try:
        weights, source = _resolve_weights(use_learned, w_cmf, w_obv, w_big,
                                           w_vwap, w_rvol)
        df, frames, scanned, msg = _do_scan(
            mode, float(min_price), float(min_adv), sector, cmf_filter,
            weights, progress,
        )
        msg = f"{msg} Weights: {source}."
        RESULTS_STATE["df"] = df
        RESULTS_STATE["frames"] = frames
        RESULTS_STATE["weights"] = weights
        RESULTS_STATE["last_run"] = datetime.utcnow()
        RESULTS_STATE["last_msg"] = msg
        RESULTS_STATE["scanned_universe"] = scanned

        # Kick off background re-tuning; result picked up on the *next* scan
        _start_auto_improve(weights)

        status = _format_status()
        if df is None or df.empty:
            return (None, None, None, None, None, status, None,
                    _sector_status(), snapshot_summary(),
                    _learned_weights_table(), _performance_status(),
                    _performance_chart())
        return (
            _df_for_display(df),
            _df_for_display(top_n(df, 20, "buy")),
            _df_for_display(top_n(df, 20, "sell")),
            _watchlist_df(df),
            _sector_breakdown_figure(df),
            status,
            _build_csv(df),
            _sector_status(),
            snapshot_summary(),
            _learned_weights_table(),
            _performance_status(),
            _performance_chart(),
        )
    except Exception as e:
        RESULTS_STATE["last_msg"] = f"Error: {e}"
        return (None, None, None, None, None, _format_status(), None,
                _sector_status(), snapshot_summary(),
                _learned_weights_table(), _performance_status(),
                _performance_chart())


def refresh_cache(mode, min_price, min_adv, sector, cmf_filter, use_learned,
                  w_cmf, w_obv, w_big, w_vwap, w_rvol, progress=gr.Progress()):
    """Force-pull everything from yfinance and overwrite the cache."""
    progress(0, desc="Forcing cache refresh...")
    universe = load_universe()
    all_tickers = universe["ticker"].tolist()

    if mode.startswith("Filtered") or sector != "All":
        tickers = apply_liquidity_filter(all_tickers, min_price=float(min_price),
                                          min_adv_usd=float(min_adv))
    else:
        tickers = all_tickers

    def cb(done, total, t):
        if total > 0:
            progress(done / total, desc=f"Refreshing {done}/{total} ({t})")

    frames, failed = fetch_ohlcv(tickers, period="6mo", force_refresh=True, progress_cb=cb)
    threading.Thread(target=push_remote_cache, daemon=True).start()
    return f"Cache refreshed: {sum(1 for f in frames.values() if not f.empty)} tickers, {len(failed)} failed."


def show_detail(ticker: str):
    if not ticker:
        return None
    ticker = ticker.strip().upper()
    frames = RESULTS_STATE.get("frames") or {}
    fig = _plot_detail(ticker, frames)
    if fig is None:
        from scanner.data_fetcher import _cache_load
        cache = _cache_load()
        if ticker in cache and not cache[ticker].empty:
            fig = _plot_detail(ticker, {ticker: cache[ticker]})
    return fig


def save_watchlist_cb(text: str):
    cleaned = save_watchlist(parse_tickers(text))
    return (", ".join(cleaned) if cleaned else "",
            f"Saved **{len(cleaned)}** ticker(s).",
            _watchlist_df(RESULTS_STATE.get("df")))


def fetch_missing_sectors_cb(progress=gr.Progress()):
    df = RESULTS_STATE.get("df")
    if df is None or df.empty:
        return _sector_status(), None
    tickers = df["ticker"].astype(str).tolist()
    cached = cached_sectors_for(tickers)
    missing = [t for t in tickers if t not in cached]
    if not missing:
        return _sector_status(), _sector_breakdown_figure(df)

    def _cb(done: int, total: int, t: str) -> None:
        if total > 0:
            progress(done / total, desc=f"Fetching sectors {done}/{total} ({t})")

    get_sectors_for(missing, progress_cb=_cb)
    return _sector_status(), _sector_breakdown_figure(df)


def apply_learned_to_sliders_cb():
    """Copy currently-learned weights into the slider inputs."""
    learned = load_learned_weights() or dict(DEFAULT_WEIGHTS)
    return (
        float(learned.get("cmf", 0.0)),
        float(learned.get("obv_slope", 0.0)),
        float(learned.get("big_bar_ratio", 0.0)),
        float(learned.get("vwap_dev", 0.0)),
        float(learned.get("rvol_signed", 0.0)),
    )


def retune_now_cb(progress=gr.Progress()):
    """Run the optimizer synchronously and report the result."""
    progress(0, desc="Running auto-tune over snapshot history...")
    base = RESULTS_STATE.get("weights") or dict(DEFAULT_WEIGHTS)
    auto_improve(base)
    progress(1.0, desc="Done.")
    return (_learned_weights_table(), _performance_status(),
            _performance_chart())


# ---------------------------------------------------------------------------
# UI
# ---------------------------------------------------------------------------

def build_ui() -> gr.Blocks:
    # Try to hydrate cache from remote on startup (best-effort, silent on failure)
    try:
        pull_remote_cache()
    except Exception:
        pass

    initial_watch = ", ".join(load_watchlist())

    with gr.Blocks(title="US Institutional Flow Scanner") as demo:
        gr.Markdown(
            """
            # US Stock Institutional Flow Scanner

            Cross-sectional scan of the US common-stock universe using a 5-factor
            proxy for institutional buying / selling pressure.  Data is pulled
            from **yfinance** (with optional **Polygon** fallback) and cached
            on disk + (optionally) a private HF Dataset so subsequent runs are
            near-instant.

            **Disclaimer:** this is a public-data proxy, not actual 13F / block-trade
            / dark-pool data.  Not financial advice.
            """
        )

        with gr.Row():
            with gr.Column(scale=1, min_width=280):
                mode = gr.Radio(
                    ["Filtered (default)", "Full universe"],
                    value="Filtered (default)",
                    label="Scan mode",
                )
                min_price = gr.Slider(1, 100, value=5, step=1, label="Min price ($)")
                min_adv = gr.Number(value=5_000_000, label="Min 20-day ADV ($)",
                                    precision=0)
                sector = gr.Dropdown(SECTORS, value="All", label="Sector (slower when chosen)")
                cmf_filter = gr.Radio(
                    ["All", "Net buying", "Net selling"],
                    value="All",
                    label="CMF side filter",
                )
                with gr.Accordion("Scoring weights (auto-normalised)", open=False):
                    w_cmf = gr.Slider(0, 0.6, value=0.30, step=0.05, label="CMF")
                    w_obv = gr.Slider(0, 0.6, value=0.25, step=0.05, label="OBV slope")
                    w_big = gr.Slider(0, 0.6, value=0.20, step=0.05, label="Big-bar ratio")
                    w_vwap = gr.Slider(0, 0.6, value=0.15, step=0.05, label="VWAP dev")
                    w_rvol = gr.Slider(0, 0.6, value=0.10, step=0.05, label="RVOL")
                use_learned = gr.Checkbox(
                    value=bool(load_learned_weights()),
                    label="Use auto-tuned weights (overrides sliders)",
                    info=("After each scan the algorithm scores its own past "
                          "predictions and searches for weights with higher IC.  "
                          "Tick this to use the latest learned weights on the "
                          "next scan."),
                )
                run_btn = gr.Button("Run scan", variant="primary")
                refresh_btn = gr.Button("Force-refresh data cache")

            with gr.Column(scale=3):
                status = gr.Markdown(_format_status())
                refresh_msg = gr.Markdown("")

                with gr.Tab("All results"):
                    full_table = gr.Dataframe(
                        headers=_result_columns(),
                        interactive=False,
                        wrap=True,
                        show_label=False,
                    )
                    dl = gr.File(label="Download full results CSV")

                with gr.Tab("Top 20 buys / sells"):
                    top_buys = gr.Dataframe(
                        headers=_result_columns(),
                        interactive=False, wrap=True, show_label=False,
                    )
                    top_sells = gr.Dataframe(
                        headers=_result_columns(),
                        interactive=False, wrap=True, show_label=False,
                    )

                with gr.Tab("Watchlist"):
                    gr.Markdown(
                        "Tickers entered here are **always** included in the scan, "
                        "regardless of the liquidity or CMF-side filter.  "
                        "Comma, space, or newline separated.  Up to 100 tickers."
                    )
                    watch_in = gr.Textbox(
                        value=initial_watch,
                        label="Watchlist",
                        placeholder="AAPL, MSFT, NVDA",
                        lines=2,
                    )
                    with gr.Row():
                        save_watch_btn = gr.Button("Save watchlist", variant="primary")
                    watch_status = gr.Markdown(
                        f"Currently saved: **{len(load_watchlist())}** ticker(s)."
                    )
                    watch_table = gr.Dataframe(
                        headers=_result_columns(),
                        interactive=False, wrap=True, show_label=False,
                    )

                with gr.Tab("Sector breakdown"):
                    sector_status_md = gr.Markdown(_sector_status())
                    fetch_sec_btn = gr.Button("Fetch missing sectors (slow)")
                    sector_plot = gr.Plot(label="Mean score by sector")

                with gr.Tab("Per-stock detail"):
                    ticker_in = gr.Textbox(
                        label="Ticker (e.g. AAPL)",
                        placeholder="Type a ticker and press Enter",
                    )
                    detail_plot = gr.Plot(label="Price + Volume + CMF(20)")
                    ticker_in.submit(show_detail, inputs=ticker_in, outputs=[detail_plot])

                with gr.Tab("History"):
                    gr.Markdown(
                        "Each scan is saved as a parquet snapshot under a "
                        "temporary directory.  Most recent snapshot is used "
                        "to compute the `score_delta` column."
                    )
                    history_table = gr.Dataframe(
                        value=snapshot_summary(),
                        headers=["timestamp_utc", "tickers", "size_kb"],
                        interactive=False, wrap=True, show_label=False,
                    )

                with gr.Tab("Performance / auto-tune"):
                    gr.Markdown(
                        f"""After each scan we score the algorithm against
                        its own past predictions: for every saved snapshot,
                        we compute the realised forward return over the next
                        **{HORIZON_DAYS} trading days** and the Spearman rank
                        correlation (Information Coefficient) between scores
                        and those returns.  A small random search over
                        alternative weight vectors then runs in the
                        background; if it finds weights with higher mean IC
                        than the baseline, they're persisted and become
                        available on the next scan.

                        - **Mean IC** > 0 means scores are positively
                          predictive of forward returns.
                        - **Hit rate** is the fraction of tickers where the
                          sign of the score matched the sign of the realised
                          return (0.5 = no edge).
                        - You need at least **5 snapshots** spanning
                          **{HORIZON_DAYS}+ trading days** before auto-tuning
                          activates.
                        """
                    )
                    perf_status_md = gr.Markdown(_performance_status())
                    with gr.Row():
                        retune_btn = gr.Button("Re-tune now (sync)",
                                               variant="primary")
                        apply_learned_btn = gr.Button("Apply learned weights to sliders")
                    learned_table = gr.Dataframe(
                        value=_learned_weights_table(),
                        headers=["factor", "default", "learned"],
                        interactive=False, wrap=True, show_label=False,
                    )
                    perf_plot = gr.Plot(label="IC history")

        scan_inputs = [mode, min_price, min_adv, sector, cmf_filter,
                       use_learned, w_cmf, w_obv, w_big, w_vwap, w_rvol]
        scan_outputs = [full_table, top_buys, top_sells, watch_table,
                        sector_plot, status, dl, sector_status_md,
                        history_table, learned_table, perf_status_md,
                        perf_plot]
        run_btn.click(run_scan, inputs=scan_inputs, outputs=scan_outputs)
        refresh_btn.click(refresh_cache, inputs=scan_inputs, outputs=[refresh_msg])
        save_watch_btn.click(
            save_watchlist_cb,
            inputs=[watch_in],
            outputs=[watch_in, watch_status, watch_table],
        )
        fetch_sec_btn.click(
            fetch_missing_sectors_cb,
            outputs=[sector_status_md, sector_plot],
        )
        retune_btn.click(
            retune_now_cb,
            outputs=[learned_table, perf_status_md, perf_plot],
        )
        apply_learned_btn.click(
            apply_learned_to_sliders_cb,
            outputs=[w_cmf, w_obv, w_big, w_vwap, w_rvol],
        )

    return demo


demo = build_ui()


# Kick off the once-per-day auto-scan thread
_start_daily_scan()


if __name__ == "__main__":
    demo.queue(max_size=8).launch(
        server_name="0.0.0.0",
        server_port=int(os.environ.get("PORT", 7860)),
    )