"""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)), )