"""AlphaForge Live Dashboard — Gradio-based real-time monitoring. Tabs: Overview | Portfolio | Risk | Sentiment | Signals | Logs Pattern: gr.Timer(5) + @gr.on(triggers=[demo.load, timer.tick]) """ import gradio as gr import plotly.graph_objects as go from plotly.subplots import make_subplots import pandas as pd import numpy as np import json import time import random import threading from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple # ── Dark theme ─────────────────────────────────────────────────────────────── DARK = { "bg": "#0a0e17", "card": "#111827", "accent": "#38bdf8", "green": "#10b981", "red": "#ef4444", "yellow": "#f59e0b", "text": "#94a3b8", "text_bright": "#e2e8f0", "border": "#1e293b", } PLOTLY_TEMPLATE = "plotly_dark" PLOTLY_LAYOUT = dict( template=PLOTLY_TEMPLATE, paper_bgcolor=DARK["bg"], plot_bgcolor=DARK["bg"], font=dict(color=DARK["text"], size=11), hovermode="x unified", margin=dict(l=8, r=24, t=32, b=8), height=340, ) # ── Data Store ─────────────────────────────────────────────────────────────── class DashboardDataStore: """Thread-safe data store for the dashboard.""" def __init__(self): self._lock = threading.Lock() self.reset() def reset(self): with self._lock: self.pnl = [] self.dates = [] self.returns = [] self.weights = {} self.positions = {} self.alerts = deque(maxlen=50) self.regime = "neutral" self.sharpe = 0.0 self.vol = 0.0 self.max_dd = 0.0 self.var_95 = 0.0 self.sortino = 0.0 self.calmar = 0.0 self.current_drawdown = 0.0 self.alpha = 0.0 self.beta = 0.0 self.turnover = 0.0 self.sentiment_scores = {} self.feature_importance = {} self.ic_history = [] self.regime_probs = {} def update_metrics(self, metrics: Dict): with self._lock: for k, v in metrics.items(): if hasattr(self, k): setattr(self, k, v) def update_pnl(self, date: datetime, pnl_value: float, ret: float): with self._lock: self.dates.append(date) self.pnl.append(pnl_value) def add_alert(self, level: str, title: str, text: str): with self._lock: self.alerts.append({ "time": datetime.now().strftime("%H:%M:%S"), "level": level, "title": title, "text": text }) def snapshot(self) -> Dict: with self._lock: return { "pnl": list(self.pnl), "dates": [str(d) for d in self.dates], "returns": list(self.returns), "regime": self.regime, "sharpe": self.sharpe, "vol": self.vol, "max_dd": self.max_dd, "var_95": self.var_95, "sortino": self.sortino, "calmar": self.calmar, "current_drawdown": self.current_drawdown, "alpha": self.alpha, "beta": self.beta, "turnover": self.turnover, "positions": self.positions, "sentiment_scores": self.sentiment_scores, "feature_importance": self.feature_importance, "ic_history": list(self.ic_history), "regime_probs": self.regime_probs, "alerts": list(self.alerts), } _store = DashboardDataStore() # ── Synthetic Data Generator ───────────────────────────────────────────────── def _generate_synthetic_run(): """Generate realistic synthetic data for demo purposes.""" np.random.seed(42) n_days = 252 dates = pd.bdate_range(start="2024-01-01", periods=n_days) # Correlated returns daily_mean = 0.0008 daily_vol = 0.012 returns = np.random.normal(daily_mean, daily_vol, n_days) # Add a drawdown period returns[100:140] = np.random.normal(-0.003, 0.02, 40) # Add recovery returns[140:180] = np.random.normal(0.0025, 0.012, 40) pnl = np.cumsum(returns) * 1_000_000 # scale to $1M # Compute metrics cumulative = np.cumprod(1 + returns) running_max = np.maximum.accumulate(cumulative) drawdowns = (cumulative - running_max) / running_max sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252) vol = np.std(returns) * np.sqrt(252) max_dd = np.min(drawdowns) sortino = np.mean(returns) / (np.std(returns[returns < 0]) + 1e-8) * np.sqrt(252) var_95 = -np.percentile(returns, 5) * 1_000_000 # Store _store.reset() for i, d in enumerate(dates): _store.update_pnl(d.to_pydatetime(), pnl[i], returns[i]) _store.update_metrics({ "sharpe": sharpe, "vol": vol, "max_dd": max_dd, "var_95": var_95, "sortino": sortino, "calmar": np.mean(returns) * 252 / abs(max_dd), "current_drawdown": drawdowns[-1], "alpha": 0.12, "beta": 0.95, "turnover": 0.15, }) # Positions _store.positions = { "SPY": 0.18, "QQQ": 0.15, "AAPL": 0.10, "MSFT": 0.12, "GOOGL": 0.08, "AMZN": 0.07, "META": 0.06, "NVDA": 0.14, "TSLA": 0.05, "JPM": 0.05, } _store.sentiment_scores = {"AAPL": 0.72, "MSFT": 0.65, "NVDA": 0.88, "TSLA": -0.34, "SPY": 0.15} _store.feature_importance = { "micro_amihud_illiquidity": 0.08, "macro_vix_level": 0.07, "ta_supertrend_dist": 0.06, "return_5d": 0.05, "cs_mom_63d": 0.05, "vol_regime_21d": 0.04, "regime_composite": 0.04, "ta_ichimoku_tk_cross": 0.04, "mr_signal": 0.03, "rvol_21d": 0.03, "yc_spread": 0.03, "volume_delta": 0.03, "sma_20d": 0.02, "macro_credit_spread": 0.02, "kyle_lambda": 0.02, "ta_keltner_position": 0.02, "cs_skew": 0.02, "trend_63d": 0.02, "vix_regime": 0.02, "return_21d": 0.01, } _store.regime_probs = {"bull": 0.45, "bear": 0.12, "high_vol": 0.28, "neutral": 0.15} _store.regime = "bull" _store.ic_history = [0.05 + np.random.normal(0, 0.02) for _ in range(60)] # Alerts _store.add_alert("info", "Backtest Complete", "Sharpe: 1.82, Max DD: -4.3%") _store.add_alert("warn", "Volatility Spike", "VIX at 28.5, reducing exposure to 80%") _store.add_alert("info", "Regime Change", "Switched from neutral to bull — increasing equity allocation") # ── Chart Builders ─────────────────────────────────────────────────────────── def build_pnl_chart() -> str: """Build PnL + Drawdown overlay chart.""" snap = _store.snapshot() if not snap["pnl"]: return "
No PnL data yet
" df = pd.DataFrame({"date": pd.to_datetime(snap["dates"]), "pnl": snap["pnl"]}) cumulative = np.cumprod(1 + np.array(snap["returns"])) running_max = np.maximum.accumulate(cumulative) dd = (cumulative - running_max) / running_max fig = make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.03, row_heights=[0.7, 0.3], subplot_titles=["Cumulative PnL", "Drawdown"]) # PnL fig.add_trace(go.Scatter( x=df["date"], y=df["pnl"], name="PnL", line=dict(color=DARK["accent"], width=2.2), fill="tozeroy", fillcolor="rgba(56,189,248,0.08)", ), row=1, col=1) # Underwater marker fig.add_trace(go.Scatter( x=df["date"], y=np.full(len(df), 0), line=dict(color=DARK["green"], width=1, dash="dot"), name="Breakeven", ), row=1, col=1) # Drawdown fig.add_trace(go.Scatter( x=df["date"], y=dd * 100, name="Drawdown %", line=dict(color=DARK["red"], width=1.5), fill="tozeroy", fillcolor="rgba(239,68,68,0.08)", ), row=2, col=1) fig.update_layout(**PLOTLY_LAYOUT, height=440, showlegend=False, yaxis=dict(title="$", tickprefix="$", tickformat=",.0f"), yaxis2=dict(title="%", ticksuffix="%"), ) return fig.to_html(include_plotlyjs="cdn", full_html=False, config={"responsive": True, "displayModeBar": False}) def build_risk_chart() -> str: """Build rolling risk metrics chart.""" snap = _store.snapshot() if not snap["returns"]: return "
No data
" rets = np.array(snap["returns"]) dates = pd.to_datetime(snap["dates"]) w = min(21, len(rets) // 4) rolling_sharpe = pd.Series(rets).rolling(w).apply( lambda x: np.mean(x) / (np.std(x) + 1e-10) * np.sqrt(252) ) rolling_vol = pd.Series(rets).rolling(w).std() * np.sqrt(252) * 100 fig = make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.06, row_heights=[0.5, 0.5], subplot_titles=["Rolling Sharpe (63d)", "Rolling Volatility %"]) fig.add_trace(go.Scatter(x=dates, y=rolling_sharpe, name="Sharpe", line=dict(color=DARK["accent"], width=2)), row=1, col=1) fig.add_trace(go.Scatter(x=dates, y=np.full(len(dates), snap["sharpe"]), line=dict(color="rgba(239,68,68,0.4)", dash="dot", width=1), name="Overall Sharpe"), row=1, col=1) fig.add_trace(go.Scatter(x=dates, y=rolling_vol, name="Vol %", line=dict(color=DARK["yellow"], width=2)), row=2, col=1) fig.update_layout(**PLOTLY_LAYOUT, height=400, showlegend=False) return fig.to_html(include_plotlyjs="cdn", full_html=False, config={"responsive": True, "displayModeBar": False}) def build_weights_chart() -> str: """Build portfolio weight treemap as bar chart.""" pos = _store.positions if not pos: return "
No positions
" items = sorted(pos.items(), key=lambda x: x[1], reverse=True) labels = [f"{k}" for k, v in items] values = [v * 100 for k, v in items] fig = go.Figure(go.Bar( x=labels, y=values, marker=dict(color=[DARK["accent"]]*3 + ["#6366f1"]*3 + [DARK["green"]]*2 + ["#8b5cf6"]*2 if len(labels) > 4 else [DARK["accent"]]*len(labels)), text=[f"{v:.1f}%" for v in values], textposition="outside", marker_line=dict(width=0), )) fig.update_layout( **PLOTLY_LAYOUT, height=260, showlegend=False, yaxis=dict(title="Weight %", ticksuffix="%", range=[0, max(values)*1.3]), xaxis=dict(tickangle=-30), ) return fig.to_html(include_plotlyjs="cdn", full_html=False, config={"responsive": True, "displayModeBar": False}) def build_feature_importance_chart() -> str: """Build top-10 feature importance chart.""" fi = _store.feature_importance if not fi: return "
No data
" items = sorted(fi.items(), key=lambda x: x[1], reverse=True)[:12] labels = [k.replace("_"," ").title()[:30] for k, v in items] values = [v * 100 for k, v in items] fig = go.Figure(go.Bar( y=labels[::-1], x=values[::-1], orientation="h", marker=dict(color=[DARK["accent"]]*len(labels)), text=[f"{v:.1f}%" for v in values[::-1]], textposition="outside", )) fig.update_layout(**PLOTLY_LAYOUT, height=340, showlegend=False, xaxis=dict(title="Importance %", ticksuffix="%"), margin=dict(l=120, r=24, t=32, b=8), ) return fig.to_html(include_plotlyjs="cdn", full_html=False, config={"responsive": True, "displayModeBar": False}) def build_regime_chart() -> str: """Build regime probability donut chart.""" rp = _store.regime_probs if not rp: return "
Loading...
" colors_map = {"bull": DARK["green"], "bear": DARK["red"], "high_vol": DARK["yellow"], "neutral": "#64748b"} fig = go.Figure(go.Pie( labels=list(rp.keys()), values=list(rp.values()), hole=0.55, marker=dict(colors=[colors_map.get(k, DARK["accent"]) for k in rp.keys()]), textinfo="label+percent", textfont=dict(size=12, color=DARK["text_bright"]), )) fig.update_layout(**PLOTLY_LAYOUT, height=260, showlegend=False) return fig.to_html(include_plotlyjs="cdn", full_html=False, config={"responsive": True, "displayModeBar": False}) def build_sentiment_chart() -> str: """Build sentiment bar chart.""" ss = _store.sentiment_scores if not ss: return "
No sentiment data
" items = sorted(ss.items(), key=lambda x: abs(x[1]), reverse=True) labels = [k for k, v in items] values = [v for k, v in items] colors = [DARK["green"] if v > 0 else DARK["red"] for v in values] fig = go.Figure(go.Bar( x=labels, y=values, marker=dict(color=colors), text=[f"{v:+.2f}" for v in values], textposition="outside", )) fig.update_layout( **PLOTLY_LAYOUT, height=240, showlegend=False, yaxis=dict(title="Sentiment Score", range=[-1.1, 1.1]), xaxis=dict(tickangle=-30), ) fig.add_hline(y=0, line=dict(color=DARK["text"], width=1, dash="dot")) return fig.to_html(include_plotlyjs="cdn", full_html=False, config={"responsive": True, "displayModeBar": False}) # ── HTML Components ────────────────────────────────────────────────────────── def _kpi_card(value: str, label: str, color: str = DARK["accent"], sub: str = "", sub_color: str = "") -> str: """Render a KPI card.""" sub_html = f'
{sub}
' if sub else "" return f'''
{label}
{value}
{sub_html}
''' def _ticker_pill(symbol: str, weight: float, pnl_pct: float = 0) -> str: """Render a ticker pill badge.""" pnl_color = DARK["green"] if pnl_pct >= 0 else DARK["red"] pnl_sign = "+" if pnl_pct >= 0 else "" return f'''
{symbol} {weight:.1%} {pnl_sign}{pnl_pct:.1%}
''' def _alert_badge(level: str, title: str, text: str, time_str: str) -> str: """Render an alert badge.""" colors = {"error": DARK["red"], "warn": DARK["yellow"], "info": DARK["accent"]} color = colors.get(level, DARK["text"]) return f'''
[{time_str}] {title} {text}
''' def build_top_bar() -> str: """Build top status bar.""" snap = _store.snapshot() regime_color = {"bull": DARK["green"], "bear": DARK["red"], "high_vol": DARK["yellow"], "neutral": DARK["text"]} rc = regime_color.get(snap.get("regime","neutral"), DARK["text"]) return f'''
🧠 AlphaForge v2.0 PRO
Regime: {snap.get("regime","--").upper()} Last Refresh: {datetime.now().strftime("%H:%M:%S")} ● LIVE
''' def build_kpi_row() -> str: """Build row of KPI cards.""" snap = _store.snapshot() pnl = snap.get("pnl", []) last_pnl = pnl[-1] if pnl else 0 return ( _kpi_card(f"${last_pnl:,.0f}", "Cumulative PnL", DARK["accent"]) + _kpi_card(f"{snap.get('sharpe',0):.2f}", "Sharpe", DARK["green"] if snap.get("sharpe",0) >= 1 else DARK["yellow"]) + _kpi_card(f"{snap.get('sortino',0):.2f}", "Sortino", DARK["accent"]) + _kpi_card(f"{snap.get('max_dd',0)*100:.1f}%", "Max Drawdown", DARK["red"]) + _kpi_card(f"${snap.get('var_95',0):,.0f}", "VaR 95%", DARK["yellow"]) + _kpi_card(f"{snap.get('calmar',0):.2f}", "Calmar", DARK["accent"]) + _kpi_card(f"{snap.get('alpha',0)*100:.1f}%", "Alpha", DARK["green"] if snap.get("alpha",0) >= 0 else DARK["red"]) + _kpi_card(f"{snap.get('beta',0):.2f}", "Beta", DARK["accent"]) ) def build_positions_html() -> str: """Build positions ticker pills.""" pos = _store.positions if not pos: return "

No positions

" items = sorted(pos.items(), key=lambda x: x[1], reverse=True) return "
" + \ "".join(_ticker_pill(k, v, np.random.normal(0, 0.03)) for k, v in items) + "
" def build_alerts_html() -> str: """Build alerts feed.""" alerts = _store.alerts return "".join(_alert_badge(a["level"], a["title"], a["text"], a["time"]) for a in alerts) # ── Dashboard App ──────────────────────────────────────────────────────────── custom_css = """ .gradio-container { max-width: 1400px !important; margin: 0 auto; } footer { display: none !important; } .tab-nav { margin-bottom: 12px !important; } """ with gr.Blocks( title="AlphaForge Pro Dashboard", css=custom_css, theme=gr.themes.Soft(primary_hue="blue", neutral_hue="slate"), fill_width=True, ) as demo: # Timer for auto-refresh (every 5 seconds) timer = gr.Timer(5) # Top status bar top_bar = gr.HTML(value=build_top_bar(), every=5) with gr.Row(): with gr.Column(scale=1, min_width=220): gr.Markdown("### 📊 KPIs") kpi_html = gr.HTML(value=build_kpi_row, every=5) gr.Markdown("### 💼 Positions") positions_html = gr.HTML(value=build_positions_html, every=5) gr.Markdown("### 🎯 Regime") regime_chart = gr.HTML(value=build_regime_chart, every=5) with gr.Column(scale=4): with gr.Tabs(): with gr.Tab("📈 PnL & Risk"): with gr.Row(): pnl_chart = gr.HTML(value=build_pnl_chart, every=5) with gr.Row(): risk_chart = gr.HTML(value=build_risk_chart, every=5) with gr.Tab("⚖️ Weights & Factors"): with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Portfolio Weights") weights_chart = gr.HTML(value=build_weights_chart, every=5) with gr.Column(scale=1): gr.Markdown("### Feature Importance") feature_chart = gr.HTML(value=build_feature_importance_chart, every=5) with gr.Tab("🗞️ Sentiment"): with gr.Row(): sentiment_chart = gr.HTML(value=build_sentiment_chart, every=5) with gr.Tab("⚠️ Alerts & Logs"): alerts_html = gr.HTML(value=build_alerts_html, every=5) # Auto refresh hook @gr.on(triggers=[demo.load, timer.tick], outputs=[top_bar, kpi_html, positions_html, regime_chart, pnl_chart, risk_chart, weights_chart, feature_chart, sentiment_chart, alerts_html]) def refresh_all(): return (build_top_bar(), build_kpi_row(), build_positions_html(), build_regime_chart(), build_pnl_chart(), build_risk_chart(), build_weights_chart(), build_feature_importance_chart(), build_sentiment_chart(), build_alerts_html()) if __name__ == "__main__": # Generate synthetic data for demo _generate_synthetic_run() demo.launch()