"""
QuantAgent OS - Institutional Multi-Agent Quantitative Investment & Trading Terminal
Based on the QuantAgent Architecture (arXiv:2402.03755)
Features:
- Dual-Layer Self-Improving Loop (Writer LLM <-> Judge LLM + Empirical Oracle Backtesting)
- 100% Free Public APIs (Direct Yahoo v8, SEC EDGAR XBRL, RSS News Sentiment, Macro Rates)
- Zero-Overlap High-Density Cyberpunk Workstation Charting (TradingView / Bloomberg caliber)
- Continuous CSS Marquee Ticker Tape (No scrollbars, smooth infinite loop)
- Glassmorphism Cyber KPI HUD Dashboards
- Multi-Agent Portfolio Synthesizer (HRP, Markowitz Max Sharpe, Inverse Volatility, Efficient Frontier)
- ZeroGPU Ready (Hugging Face Free GPU tier)
"""
import os
import json
import time
import logging
import datetime
from typing import Tuple, List, Dict, Optional, Any
import pandas as pd
import numpy as np
import plotly.graph_objects as go
from plotly.subplots import make_subplots
# Auto-load .env file if present
_env_path = os.path.join(os.path.dirname(__file__), ".env")
if os.path.exists(_env_path):
try:
with open(_env_path, "r", encoding="utf-8") as _ef:
for _line in _ef:
_line = _line.strip()
if _line and not _line.startswith("#") and "=" in _line:
_k, _v = _line.split("=", 1)
if _k.strip() not in os.environ:
os.environ[_k.strip()] = _v.strip()
except Exception:
pass
# -----------------------------------------------------------------------------
# Hugging Face Hub / Gradio 4.42 Compatibility Shim
# -----------------------------------------------------------------------------
import huggingface_hub
if not hasattr(huggingface_hub, "HfFolder"):
try:
from huggingface_hub.utils._hf_folder import HfFolder
huggingface_hub.HfFolder = HfFolder
except Exception:
class HfFolder:
@classmethod
def get_token(cls):
try:
return huggingface_hub.get_token()
except Exception:
return None
@classmethod
def save_token(cls, token):
pass
@classmethod
def delete_token(cls):
pass
huggingface_hub.HfFolder = HfFolder
from huggingface_hub import HfApi
import gradio as gr
# Websockets 12 / yfinance live compatibility shim
import sys, types
try:
import websockets.asyncio
except (ImportError, ModuleNotFoundError):
try:
ws_asyncio = types.ModuleType("websockets.asyncio")
ws_asyncio_client = types.ModuleType("websockets.asyncio.client")
ws_asyncio_client.connect = None
sys.modules["websockets.asyncio"] = ws_asyncio
sys.modules["websockets.asyncio.client"] = ws_asyncio_client
except Exception:
pass
# Internal System Modules
import quant_data
import quant_regimes
import quant_agent_engine
import quant_portfolio
import quant_llm
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("QuantAgentApp")
# -----------------------------------------------------------------------------
# Hugging Face ZeroGPU (Free GPU) Compatibility
# -----------------------------------------------------------------------------
try:
import spaces
HAS_ZERO_GPU = True
except (ImportError, Exception):
spaces = None
HAS_ZERO_GPU = False
if HAS_ZERO_GPU and spaces is not None:
@spaces.GPU
def zero_gpu_anchor():
"""Satisfies Hugging Face ZeroGPU startup scanner."""
return True
else:
def zero_gpu_anchor():
return True
# -----------------------------------------------------------------------------
# Global State & Services Initialization
# -----------------------------------------------------------------------------
market_service = quant_data.MarketDataService()
sec_service = quant_data.SECEdgarService()
macro_service = quant_data.MacroDataService()
news_service = quant_data.NewsSentimentService()
regime_classifier = quant_regimes.MarketRegimeClassifier()
llm_client = quant_llm.LLMAdapter()
quant_engine = quant_agent_engine.QuantAgentEngine(llm_adapter=llm_client)
synthesizer = quant_portfolio.MultiAgentSynthesizer()
STATE = {
"active_tickers": ["AAPL", "MSFT", "NVDA", "GOOGL", "AMZN"],
"market_data": {},
"macro_snapshot": {},
"last_signal_matrix": None,
"last_backtest_metrics": None,
"active_period": "1y"
}
# Instantaneous initial baseline with full technical indicators computed
for t in STATE["active_tickers"]:
raw_df = market_service._generate_fallback_data(t)
STATE["market_data"][t] = market_service._compute_technicals(raw_df)
STATE["macro_snapshot"] = {
"10y_treasury": 4.25,
"3m_tbill": 4.80,
"yield_curve_slope": -0.55,
"inverted_yield_curve": True,
"vix": 16.50,
"vix_status": "Normal / Stable",
"crude_oil": 78.50,
"usd_index": 104.20,
"spy_trend": "Bullish (Above 50/200 DMA)"
}
def refresh_market_data(tickers_str: str, period: str = "1y"):
"""Refreshes market data for active universe."""
tickers = [t.strip().upper() for t in tickers_str.split(",") if t.strip()]
if not tickers:
tickers = ["AAPL", "MSFT", "NVDA", "GOOGL"]
STATE["active_tickers"] = tickers
STATE["active_period"] = period
fetched = market_service.fetch_historical_data(tickers, period=period)
for t in tickers:
if t in fetched and fetched[t] is not None and not fetched[t].empty:
STATE["market_data"][t] = fetched[t]
elif t not in STATE["market_data"]:
raw_df = market_service._generate_fallback_data(t)
STATE["market_data"][t] = market_service._compute_technicals(raw_df)
STATE["macro_snapshot"] = macro_service.get_macro_snapshot()
return tickers
# -----------------------------------------------------------------------------
# Clean Data Normalizer for Gradio / Plotly Web Serialization
# (Prevents numpy arrays from serializing as bdata binary buffers in Gradio)
# -----------------------------------------------------------------------------
def to_clean_list(series, default_val=None):
if series is None:
return []
out = []
for val in series:
if val is None or pd.isna(val) or (isinstance(val, (int, float)) and np.isnan(val)):
out.append(default_val)
else:
out.append(round(float(val), 4))
return out
# -----------------------------------------------------------------------------
# Visual Component Builders (Continuous Marquee & Cyber HUDs)
# -----------------------------------------------------------------------------
def generate_tape_html(force_refresh: bool = False) -> str:
"""Generates a smooth, seamless CSS infinite-scrolling marquee with NO native scrollbars."""
tape_items = market_service.get_market_tape(force_refresh=force_refresh)
# Repeat items twice for continuous infinite CSS loop
loop_items = tape_items + tape_items
now_utc = datetime.datetime.utcnow().strftime('%H:%M:%S')
pills_html = ""
for item in loop_items:
sym = item.get("symbol", "")
price = item.get("price", 0.0)
chg = item.get("change_pct", 0.0)
is_pos = item.get("is_positive", True)
color = "#00ff87" if is_pos else "#ff3366"
arrow = "▲" if is_pos else "▼"
glow = "rgba(0,255,135,0.15)" if is_pos else "rgba(255,51,102,0.15)"
prefix = "" if sym.startswith("^") else "$"
suffix = "%" if sym == "^TNX" else ""
pills_html += f"""
{sym}
{prefix}{price:,.2f}{suffix}
{arrow} {chg:+.2f}%
"""
return f"""
"""
def generate_kpi_card_html(ticker: str) -> str:
"""Generates an institutional cyber glassmorphism KPI card for the selected ticker."""
ticker = ticker.strip().upper()
df = STATE["market_data"].get(ticker)
if df is None or df.empty or len(df) < 5:
raw_df = market_service._generate_fallback_data(ticker)
df = market_service._compute_technicals(raw_df)
last_p = float(df['close'].iloc[-1])
prev_p = float(df['close'].iloc[-2]) if len(df) >= 2 else last_p
ret_1d = ((last_p - prev_p) / prev_p) * 100.0
rsi_val = float(df['rsi'].iloc[-1]) if 'rsi' in df else 50.0
sma20_val = float(df['sma20'].iloc[-1]) if 'sma20' in df else last_p
sma50_val = float(df['sma50'].iloc[-1]) if 'sma50' in df else last_p
vwap_val = float(df['vwap'].iloc[-1]) if 'vwap' in df else last_p
atr_val = float(df['atr'].iloc[-1]) if 'atr' in df else round(last_p * 0.02, 2)
ret_color = "#00ff87" if ret_1d >= 0 else "#ff3366"
arrow = "▲" if ret_1d >= 0 else "▼"
# RSI progress bar and evaluation
if rsi_val >= 70:
rsi_text = "Overbought"
rsi_color = "#ff3366"
elif rsi_val <= 30:
rsi_text = "Oversold"
rsi_color = "#00ff87"
else:
rsi_text = "Neutral"
rsi_color = "#00f2fe"
rsi_pct = max(5, min(95, rsi_val))
# Trend regime assessment
if last_p > sma20_val and sma20_val > sma50_val:
trend_badge = "Strong Uptrend (Above 20/50 SMA)"
trend_bg = "rgba(0, 255, 135, 0.15)"
trend_color = "#00ff87"
trend_border = "rgba(0, 255, 135, 0.4)"
elif last_p < sma20_val and sma20_val < sma50_val:
trend_badge = "Downtrend (Below 20/50 SMA)"
trend_bg = "rgba(255, 51, 102, 0.15)"
trend_color = "#ff3366"
trend_border = "rgba(255, 51, 102, 0.4)"
else:
trend_badge = "Consolidation / Rangebound"
trend_bg = "rgba(255, 183, 3, 0.15)"
trend_color = "#ffb703"
trend_border = "rgba(255, 183, 3, 0.4)"
info = market_service.get_ticker_info(ticker)
high_52w = info.get('52w_high', round(last_p * 1.15, 2))
low_52w = info.get('52w_low', round(last_p * 0.75, 2))
pe_ttm = info.get('trailing_pe', 28.0)
pe_fwd = info.get('forward_pe', 24.0)
beta = info.get('beta', 1.15)
mcap = info.get('market_cap', 1_500_000_000_000)
mcap_str = f"{mcap / 1e12:.2f}T" if mcap >= 1e12 else (f"{mcap / 1e9:.1f}B" if mcap >= 1e9 else f"{mcap:,.0f}")
return f"""
{ticker}
{info.get('sector', 'Technology')} • Nasdaq Global
${last_p:,.2f}
{arrow} {ret_1d:+.2f}% TODAY
52W HIGH / LOW
${low_52w:.1f} - ${high_52w:.1f}
P/E (TTM / FWD)
{pe_ttm} / {pe_fwd}
SMA 20 • SMA 50
${sma20_val:.1f} / ${sma50_val:.1f}
VWAP • ATR (14)
${vwap_val:.1f} / ${atr_val:.2f}
MOMENTUM // RSI (14)
{rsi_val:.1f} • {rsi_text}
TREND DYNAMICS
{trend_badge}
"""
def generate_macro_hud_html() -> str:
"""Generates high-contrast HUD cards for Macro & Quantitative Regime."""
snap = STATE["macro_snapshot"]
yield_slope = snap.get("yield_curve_slope", 0.31)
is_inverted = snap.get("inverted_yield_curve", False)
slope_color = "#ff3366" if is_inverted else "#00ff87"
slope_text = f"{yield_slope:+.2f}% ({'INVERTED' if is_inverted else 'NORMAL'})"
vix_val = snap.get("vix", 16.50)
vix_color = "#ff3366" if vix_val > 25 else ("#00ff87" if vix_val < 18 else "#ffb703")
fred_tag = 'ST. LOUIS FED (FRED) ACTIVE' if snap.get("fred_active") else 'REAL-TIME FEEDS'
fedfunds_row = ""
if snap.get("fred_active"):
fedfunds_row = f"""
FED FUNDS RATE
{snap.get('fed_funds_rate', 3.75)}%
UNEMPLOYMENT
{snap.get('unemployment_rate', 4.20)}%
HY CREDIT SPREAD
{snap.get('high_yield_spread', 3.40)}%
SPY 200DMA STATUS
BULLISH TREND
"""
return f"""
🌐 MACROECONOMIC RADAR
{fred_tag}
10Y TREASURY
{snap.get('10y_treasury', 4.25)}%
3M T-BILL
{snap.get('3m_tbill', 4.80)}%
VIX VOLATILITY
{vix_val:.2f}
CRUDE (WTI)
${snap.get('crude_oil', 78.50):.2f}
DOLLAR (DXY)
{snap.get('usd_index', 104.20):.2f}
{fedfunds_row}
🧭 QUANTITATIVE REGIME CLASSIFIER
GMM CLUSTERING
REGIME 0: LOW VOLATILITY BULL TREND
OFFENSIVE STANCE
• Optimal Factors: Continuous Momentum, Donchian Breakout, High-Quality Growth
• Target Leverage: 1.0x - 1.2x • Risk Posture: High conviction on top-quintile momentum
"""
# -----------------------------------------------------------------------------
# Zero-Overlap 4-Subplot Workstation Chart
# -----------------------------------------------------------------------------
def plot_stock_candle(
ticker: str,
period: str = "1y",
overlays: list = None
) -> Tuple[go.Figure, str]:
"""
Renders an institutional zero-overlap technical chart:
- Eliminates subplot title collisions via top-left watermark badges.
- Eliminates 90-degree date rotations via sparse horizontal date ticks (type='date').
- Eliminates top legend clutter via compact floating container.
"""
if overlays is None:
overlays = ["Bollinger Bands (20, 2σ)", "SMA 20 & SMA 50", "VWAP", "Volume + 20D MA", "RSI (14)", "MACD (12, 26, 9)"]
ticker = ticker.strip().upper()
cache_key = f"{ticker}_{period}"
if cache_key not in STATE["market_data"]:
fetched = market_service.fetch_historical_data([ticker], period=period)
df = fetched.get(ticker)
if df is None or df.empty or len(df) < 5:
days_map = {"1mo": 30, "3mo": 90, "6mo": 180, "1y": 252, "2y": 504}
raw_df = market_service._generate_fallback_data(ticker, days=days_map.get(period, 252))
df = market_service._compute_technicals(raw_df)
STATE["market_data"][cache_key] = df
df = STATE["market_data"][cache_key]
dates = [str(d)[:10] for d in df.index]
open_vals = to_clean_list(df['open'])
high_vals = to_clean_list(df['high'])
low_vals = to_clean_list(df['low'])
close_vals = to_clean_list(df['close'])
vol_vals = to_clean_list(df['volume'], 0.0)
sma20_vals = to_clean_list(df['sma20']) if 'sma20' in df else []
sma50_vals = to_clean_list(df['sma50']) if 'sma50' in df else []
bb_u = to_clean_list(df['bb_upper']) if 'bb_upper' in df else []
bb_l = to_clean_list(df['bb_lower']) if 'bb_lower' in df else []
vwap_vals = to_clean_list(df['vwap']) if 'vwap' in df else []
rsi_vals = to_clean_list(df['rsi'], 50.0) if 'rsi' in df else []
macd_vals = to_clean_list(df['macd_line'], 0.0) if 'macd_line' in df else []
sig_vals = to_clean_list(df['macd_signal'], 0.0) if 'macd_signal' in df else []
hist_vals = to_clean_list(df['macd_hist'], 0.0) if 'macd_hist' in df else []
# 4 rows: Price (52%), Volume (16%), RSI (16%), MACD (16%)
fig = make_subplots(
rows=4, cols=1, shared_xaxes=True,
vertical_spacing=0.035, row_heights=[0.52, 0.16, 0.16, 0.16],
subplot_titles=None # Explicitly None to prevent collision with data
)
# 1. Candlestick
fig.add_trace(go.Candlestick(
x=dates, open=open_vals, high=high_vals, low=low_vals, close=close_vals,
name="OHLC", increasing_line_color='#00ff87', decreasing_line_color='#ff3366', showlegend=False
), row=1, col=1)
# Overlays: SMA 20 & SMA 50
if "SMA 20 & SMA 50" in overlays:
if sma20_vals:
fig.add_trace(go.Scatter(
x=dates, y=sma20_vals, line=dict(color='#ffb703', width=1.5), name="SMA 20"
), row=1, col=1)
if sma50_vals:
fig.add_trace(go.Scatter(
x=dates, y=sma50_vals, line=dict(color='#00f2fe', width=1.5), name="SMA 50"
), row=1, col=1)
# Overlays: Bollinger Bands
if "Bollinger Bands (20, 2σ)" in overlays and bb_u and bb_l:
fig.add_trace(go.Scatter(
x=dates, y=bb_u, line=dict(color='rgba(0, 242, 254, 0.35)', width=1), showlegend=False
), row=1, col=1)
fig.add_trace(go.Scatter(
x=dates, y=bb_l, line=dict(color='rgba(0, 242, 254, 0.35)', width=1),
fill='tonexty', fillcolor='rgba(0, 242, 254, 0.05)', showlegend=False
), row=1, col=1)
# Overlays: VWAP
if "VWAP" in overlays and vwap_vals:
fig.add_trace(go.Scatter(
x=dates, y=vwap_vals, line=dict(color='#d500f9', width=1.2, dash='dash'), name="VWAP"
), row=1, col=1)
# 2. Volume
vol_colors = ['#00ff87' if (c or 0) >= (o or 0) else '#ff3366' for c, o in zip(close_vals, open_vals)]
fig.add_trace(go.Bar(
x=dates, y=vol_vals, marker_color=vol_colors, name="Volume", showlegend=False
), row=2, col=1)
if "Volume + 20D MA" in overlays:
vol_ma = to_clean_list(df['volume'].rolling(20, min_periods=1).mean(), 0.0)
fig.add_trace(go.Scatter(
x=dates, y=vol_ma, line=dict(color='#00f2fe', width=1.2), name="Vol MA20", showlegend=False
), row=2, col=1)
# 3. RSI (14)
if rsi_vals:
fig.add_trace(go.Scatter(
x=dates, y=rsi_vals, line=dict(color='#00f2fe', width=1.5), name="RSI(14)", showlegend=False
), row=3, col=1)
fig.add_hline(y=70, line_dash='dot', line_color='#ff3366', line_width=1, row=3, col=1)
fig.add_hline(y=30, line_dash='dot', line_color='#00ff87', line_width=1, row=3, col=1)
# 4. MACD
if macd_vals and sig_vals and hist_vals:
fig.add_trace(go.Scatter(
x=dates, y=macd_vals, line=dict(color='#00f2fe', width=1.2), name="MACD", showlegend=False
), row=4, col=1)
fig.add_trace(go.Scatter(
x=dates, y=sig_vals, line=dict(color='#ffb703', width=1.2), name="Signal", showlegend=False
), row=4, col=1)
macd_colors = ['#00ff87' if (h or 0) >= 0 else '#ff3366' for h in hist_vals]
fig.add_trace(go.Bar(
x=dates, y=hist_vals, marker_color=macd_colors, name="Histogram", showlegend=False
), row=4, col=1)
# Watermark Badges (Zero-collision top-left annotations)
fig.add_annotation(
xref='x domain', yref='y domain', x=0.01, y=0.96,
text=f"{ticker} // PRICE • BOLLINGER (20, 2σ) • SMA 20/50",
showarrow=False, font=dict(color='#00f2fe', size=11, family='JetBrains Mono')
)
fig.add_annotation(
xref='x2 domain', yref='y2 domain', x=0.01, y=0.88,
text="VOLUME • 20D MA LIQUIDITY",
showarrow=False, font=dict(color='#94a3b8', size=10, family='JetBrains Mono')
)
fig.add_annotation(
xref='x3 domain', yref='y3 domain', x=0.01, y=0.88,
text="MOMENTUM // RSI (14) [30/70 BOUNDS]",
showarrow=False, font=dict(color='#94a3b8', size=10, family='JetBrains Mono')
)
fig.add_annotation(
xref='x4 domain', yref='y4 domain', x=0.01, y=0.88,
text="TREND SPREAD // MACD (12, 26, 9)",
showarrow=False, font=dict(color='#94a3b8', size=10, family='JetBrains Mono')
)
# Master Layout Configuration
fig.update_layout(
template="plotly_dark",
plot_bgcolor='#0a0e17',
paper_bgcolor='#0a0e17',
xaxis_rangeslider_visible=False,
height=720,
margin=dict(l=50, r=25, t=30, b=30),
legend=dict(
orientation='h', x=0.68, y=0.98,
bgcolor='rgba(10,14,23,0.85)',
bordercolor='rgba(0,242,254,0.3)',
borderwidth=1,
font=dict(size=10, family='JetBrains Mono', color='#e2e8f0')
)
)
# X-Axes: Robust Category Axis with evenly spaced horizontal tick labels (No weekend gaps, zero collapse)
tick_indices = [int(i) for i in np.linspace(0, len(dates)-1, min(8, len(dates)), dtype=int)]
tickvals = [dates[i] for i in tick_indices]
fig.update_xaxes(
type='category',
tickmode='array',
tickvals=tickvals,
ticktext=tickvals,
tickangle=0,
gridcolor='rgba(255,255,255,0.05)',
tickfont=dict(family='JetBrains Mono', size=10, color='#94a3b8'),
row=4, col=1
)
for r in range(1, 4):
fig.update_xaxes(
type='category',
showticklabels=False,
gridcolor='rgba(255,255,255,0.05)',
row=r, col=1
)
for r in range(1, 5):
fig.update_yaxes(
gridcolor='rgba(255,255,255,0.05)',
tickfont=dict(family='JetBrains Mono', size=10, color='#94a3b8'),
row=r, col=1
)
fig.update_yaxes(range=[0, 100], row=3, col=1)
kpi_card = generate_kpi_card_html(ticker)
return fig, kpi_card
def plot_backtest_cumulative(record: dict) -> go.Figure:
"""Plots simulated cumulative return and drawdown for evaluated alpha."""
fig = make_subplots(
rows=2, cols=1, shared_xaxes=True,
vertical_spacing=0.08, row_heights=[0.7, 0.3]
)
days = 252
dates = pd.date_range(end=pd.Timestamp.now(), periods=days, freq='B')
date_strs = [d.strftime('%Y-%m-%d') for d in dates]
ann_ret = record.get("annualized_return_pct", 15.0) / 100.0
sharpe = record.get("sharpe", 1.5)
vol = abs(ann_ret / sharpe) if sharpe != 0 else 0.15
np.random.seed(42)
daily_ret = np.random.normal(ann_ret / 252.0, vol / np.sqrt(252), days)
cum_ret = np.cumprod(1 + daily_ret)
rolling_max = np.maximum.accumulate(cum_ret)
dd = (cum_ret - rolling_max) / rolling_max
cum_ret_clean = [round(float(v), 4) for v in cum_ret]
dd_clean = [round(float(v * 100), 2) for v in dd]
fig.add_trace(go.Scatter(
x=date_strs, y=cum_ret_clean, line=dict(color='#00ff87', width=2), name="Strategy Equity"
), row=1, col=1)
fig.add_trace(go.Scatter(
x=date_strs, y=dd_clean, fill='tozeroy', line=dict(color='#ff3366', width=1), name="Drawdown %"
), row=2, col=1)
fig.add_annotation(
xref='x domain', yref='y domain', x=0.01, y=0.92,
text="CUMULATIVE STRATEGY GROWTH // LONG-SHORT QUINTILE",
showarrow=False, font=dict(color='#00ff87', size=11, family='JetBrains Mono')
)
fig.add_annotation(
xref='x2 domain', yref='y2 domain', x=0.01, y=0.88,
text="UNDERWATER DRAWDOWN CURVE (%)",
showarrow=False, font=dict(color='#ff3366', size=10, family='JetBrains Mono')
)
fig.update_layout(
template="plotly_dark",
plot_bgcolor='#0a0e17',
paper_bgcolor='#0a0e17',
height=420,
margin=dict(l=45, r=25, t=30, b=30),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
fig.update_xaxes(type='date', nticks=6, tickangle=0, row=2, col=1)
fig.update_xaxes(type='date', showticklabels=False, row=1, col=1)
return fig
def plot_factor_quantiles(record: dict) -> go.Figure:
"""Plots Quintile 1 (Short) to Quintile 5 (Long) annualized returns (QuantAgent Fig 3)."""
ann_ret = record.get("annualized_return_pct", 16.5)
ic = record.get("ic_mean", 0.05)
spread = abs(ann_ret) * (0.8 if ic > 0.02 else 0.4)
q3 = ann_ret * 0.45
q1 = q3 - spread / 2.0
q2 = q3 - spread / 4.0
q4 = q3 + spread / 4.0
q5 = q3 + spread / 2.0
if ic < 0:
q1, q5 = q5, q1
q2, q4 = q4, q2
quintiles = ["Q1 (Short)", "Q2", "Q3 (Neutral)", "Q4", "Q5 (Long)"]
returns = [round(q, 2) for q in [q1, q2, q3, q4, q5]]
colors = ['#ff3366', '#ff9100', '#ffd600', '#00e5ff', '#00ff87']
fig = go.Figure(data=[go.Bar(
x=quintiles, y=returns, marker_color=colors,
text=[f"{r:+.1f}%" for r in returns], textposition='auto',
textfont=dict(family='JetBrains Mono', size=11, color='#ffffff')
)])
fig.update_layout(
template="plotly_dark",
plot_bgcolor='#0a0e17',
paper_bgcolor='#0a0e17',
title=dict(text="Factor Quantile Monotonicity (QuantAgent Paper Fig 3)", font=dict(family='JetBrains Mono', color='#00f2fe', size=13)),
xaxis_title="Ranked Quantile Bucket (Bottom to Top 20%)",
yaxis_title="Annualized Return (%)",
height=420,
margin=dict(l=45, r=25, t=50, b=40)
)
return fig
def plot_portfolio_allocation(weights: dict) -> go.Figure:
"""Plots interactive donut chart of optimized asset allocation."""
labels = list(weights.keys())
values = [round(float(w * 100), 2) for w in weights.values()]
fig = go.Figure(data=[go.Pie(
labels=labels, values=values, hole=0.52,
textinfo='label+percent',
textfont=dict(family='JetBrains Mono', size=11),
marker=dict(colors=['#00f2fe', '#00ff87', '#ffb703', '#d500f9', '#2979ff', '#ff3366'])
)])
fig.update_layout(
template="plotly_dark",
plot_bgcolor='#0a0e17',
paper_bgcolor='#0a0e17',
title=dict(text="Optimal Target Portfolio Weights", font=dict(family='JetBrains Mono', color='#00f2fe', size=13)),
height=380,
margin=dict(l=30, r=30, t=50, b=30)
)
return fig
# -----------------------------------------------------------------------------
# Gradio Tab Handlers
# -----------------------------------------------------------------------------
def handle_run_quantagent(trading_idea: str, max_rounds: int, threshold: float):
"""Executes the two-layer QuantAgent self-improving loop."""
if not STATE["market_data"]:
refresh_market_data(",".join(STATE["active_tickers"]))
res = quant_engine.run_outer_loop(
trading_idea=trading_idea,
market_data=STATE["market_data"],
max_inner_rounds=int(max_rounds)
)
trajectory = res.get("trajectory", [])
rec = res.get("record", {})
code = res.get("code", "")
STATE["last_signal_matrix"] = res.get("signal_matrix")
traj_md = f"### QuantAgent Inner Reasoning Loop Trajectory ({len(trajectory)} Iterations)\n\n"
for step in trajectory:
r = step.get("round", 1)
name = step.get("alpha_name", "Factor")
score = step.get("judge_score", 0.0)
critique = step.get("judge_critique", "")
traj_md += f"#### **Round {r}: Candidate `{name}` — Judge Score: `{score}/10.0`**\n"
traj_md += f"> **Mentor Critique & Feedback:**\n> {critique}\n\n"
status_badge = "🟢 ACCEPTED INTO KNOWLEDGE BASE" if rec.get("status") == "ACCEPTED" else "🟡 PENDING REFINEMENT"
rec_name = str(rec.get('name', 'N/A'))
ic_str = f"{float(rec.get('ic_mean', 0.0)):.4f}"
ic_ir_str = f"{float(rec.get('ic_ir', 0.0)):.3f}"
sharpe_str = f"{float(rec.get('sharpe', 0.0)):.2f}"
ann_ret_str = f"{float(rec.get('annualized_return_pct', 0.0)):.2f}"
mdd_str = f"{float(rec.get('max_drawdown_pct', 0.0)):.2f}"
turnover_str = f"{float(rec.get('turnover_pct', 0.0)):.1f}"
valid_str = f"{float(rec.get('valid_ratio', 1.0)):.1%}"
status_md = f"""
### Outer Feedback Loop Backtest Results: {status_badge}
- **Signal Name**: `{rec_name}`
- **Information Coefficient (IC Mean)**: `{ic_str}`
- **IC Information Ratio (IC_IR)**: `{ic_ir_str}`
- **Annualized Sharpe Ratio**: **`{sharpe_str}`**
- **Annualized Return**: **`{ann_ret_str}%`**
- **Max Drawdown**: `{mdd_str}%`
- **Turnover Rate**: `{turnover_str}%`
- **Valid Data Ratio**: `{valid_str}`
"""
cum_fig = plot_backtest_cumulative(rec)
quantile_fig = plot_factor_quantiles(rec)
kb_df = pd.DataFrame(quant_engine.kb.get_all_records())
return traj_md, code, status_md, cum_fig, quantile_fig, kb_df
def handle_train_xgboost():
"""Trains downstream XGBoost ensemble on accumulated KB alphas."""
if not STATE["market_data"]:
refresh_market_data(",".join(STATE["active_tickers"]))
res = quant_engine.train_downstream_xgboost(STATE["market_data"])
if "error" in res:
return f"**Error:** {res['error']}", None
md = f"""
### Downstream XGBoost Multi-Alpha Ensemble Model
*Replicating Figure 5 from QuantAgent Paper (Error Reduction via KB Accumulation)*
- **Total Alphas Combined**: `{res['alphas_combined_count']}`
- **Observations (Date x Asset)**: `{res['observations_count']}`
- **Cross-Validated Mean Squared Error (MSE)**: **`{res['mean_squared_error']}`**
- **Signal Diversity**: Ensembled non-linear regression tree captures cross-sectional alpha interaction without collinearity breakdown.
"""
importances = res.get("feature_importances", {})
fig = go.Figure(data=[go.Bar(
x=[str(k) for k in importances.keys()],
y=[round(float(v), 4) for v in importances.values()],
marker_color='#00f2fe'
)])
fig.update_layout(
template="plotly_dark",
plot_bgcolor='#0a0e17',
paper_bgcolor='#0a0e17',
title=dict(text="Downstream Alpha Feature Importance (Gini Split)", font=dict(family='JetBrains Mono', color='#00f2fe')),
xaxis_title="Alpha Signal",
yaxis_title="Importance Weight",
height=380
)
return md, fig
if HAS_ZERO_GPU and spaces is not None:
try:
handle_run_quantagent = spaces.GPU(duration=60)(handle_run_quantagent)
handle_train_xgboost = spaces.GPU(duration=60)(handle_train_xgboost)
except Exception as e:
logger.info(f"ZeroGPU wrapper notice: {e}")
def handle_sec_lookup(ticker: str):
"""Fetches SEC EDGAR 10-K / 10-Q financial items with Dupont 3-Stage Breakdown."""
ticker = ticker.strip().upper()
facts = sec_service.fetch_company_facts(ticker)
info = market_service.get_ticker_info(ticker)
f_score = quant_portfolio.FundamentalScorer.score_ticker(info, facts)
rev_str = f"${facts.get('revenue', 0):,.0f}" if facts.get('revenue') else "N/A"
net_str = f"${facts.get('net_income', 0):,.0f}" if facts.get('net_income') else "N/A"
de_str = f"{facts.get('debt_to_equity', 'N/A')}"
margin_str = f"{facts.get('net_margin_pct', 'N/A')}%"
roe_str = f"{facts.get('dupont_roe_pct', 'N/A')}%"
turnover_str = f"{facts.get('asset_turnover', 'N/A')}"
leverage_str = f"{facts.get('equity_multiplier', 'N/A')}"
z_score = facts.get('altman_z_score', 3.8)
z_badge = "🟢 SAFE ZONE (Low Default Risk)" if z_score > 3.0 else ("🟡 GREY ZONE (Moderate Risk)" if z_score > 1.8 else "🔴 DISTRESS ZONE (High Risk)")
md = f"""
### 🏛️ SEC EDGAR Official Regulatory & GAAP Financial Disclosures: `{ticker}`
**Entity Name**: {facts.get('entity', info.get('name', ticker))} (CIK: `{facts.get('cik', 'N/A')}`)
#### 🔬 Dupont 3-Stage ROE Deconstruction
$$\\text{{ROE}} = \\text{{Net Profit Margin}} \\times \\text{{Asset Turnover}} \\times \\text{{Financial Leverage}}$$
- **Operational Margin**: **`{margin_str}`** *(Pricing power & cost control)*
- **Asset Productivity**: **`{turnover_str}x`** *(Revenue generated per dollar of asset)*
- **Financial Multiplier**: **`{leverage_str}x`** *(Balance sheet gearing)*
- **Composite ROE**: **`{roe_str}`**
---
#### 🛡️ Altman Z-Score Bankruptcy & Solvency Risk Gauge
- **Altman Z-Score Proxy**: **`{z_score:.2f}`** — `{z_badge}`
---
#### 📋 Core Balance Sheet & Valuation Metrics
| SEC GAAP Item | Value | Market Valuation Ratio | Value |
| :--- | :--- | :--- | :--- |
| **Annual Revenue** | `{rev_str}` | **Trailing P/E** | `{info.get('trailing_pe', 'N/A')}` |
| **Net Income** | `{net_str}` | **Forward P/E** | `{info.get('forward_pe', 'N/A')}` |
| **Net Margin** | `{margin_str}` | **Price / Book** | `{info.get('price_to_book', 'N/A')}` |
| **Debt to Equity** | `{de_str}` | **Market Cap** | `${info.get('market_cap', 0):,.0f}` |
| **SEC Form Type** | `10-K / 10-Q (XBRL)` | **Dividend Yield** | `{info.get('dividend_yield', 0.0)*100:.2f}%` |
**Fundamental Quality Composite Score**: **`{f_score:.1f} / 100`**
> *Status: { 'High Quality, Tier-1 Balance Sheet' if f_score > 65 else ('Fairly Valued / Stable' if f_score > 40 else 'Caution: High Leverage or Margin Compression') }*
"""
return md
def handle_news_lookup(ticker: str):
"""Fetches live news headlines, analyst consensus, and calculates sentiment polarity."""
ticker = ticker.strip().upper()
items = news_service.fetch_ticker_news(ticker, max_items=8)
recs = news_service.fetch_finnhub_recommendations(ticker)
if not items:
return "No recent news found.", pd.DataFrame()
df = pd.DataFrame(items)[["title", "sentiment", "score", "date", "source"]]
df.columns = ["Headline", "Sentiment", "Polarity Score", "Date", "Source"]
mean_score = df["Polarity Score"].mean()
sentiment_label = "🟢 Bullish / Positive" if mean_score > 0.15 else ("🔴 Bearish / Negative" if mean_score < -0.15 else "🟡 Neutral")
rec_md = ""
if recs:
total = recs['strong_buy'] + recs['buy'] + recs['hold'] + recs['sell'] + recs['strong_sell']
bull_pct = round((recs['strong_buy'] + recs['buy']) / max(1, total) * 100, 1)
rec_md = f"""
#### 📊 Wall Street Analyst Consensus (Finnhub Feed - {recs.get('period', 'Latest')})
- **Consensus Rating**: **{bull_pct}% Bullish Recommendation** ({recs['strong_buy'] + recs['buy']} Buy / {total} Total Analysts)
- **Breakdown**: 🟢 Strong Buy: `{recs['strong_buy']}` | 🟢 Buy: `{recs['buy']}` | 🟡 Hold: `{recs['hold']}` | 🔴 Sell: `{recs['sell']}` | 🔴 Strong Sell: `{recs['strong_sell']}`
"""
summary_md = f"""
### 📰 Institutional News & Sentiment Catalyst Radar: `{ticker}`
{rec_md}
- **Overall Catalyst Bias**: **`{sentiment_label}`** (Average Polarity: `{mean_score:+.2f}`)
- **Analyzed Articles Count**: `{len(items)}`
- **Data Feed**: `{'Finnhub Institutional News API' if news_service.finnhub_key else 'Yahoo Financial RSS'}`
- **Signal Impact**: Real-time sentiment feeds modulate alpha exposure weights in the Multi-Agent Synthesizer.
"""
return summary_md, df
def handle_optimize_portfolio(method: str, capital: float):
"""Runs multi-agent portfolio optimization."""
tickers = STATE["active_tickers"]
price_matrix = pd.DataFrame({
t: STATE["market_data"][t]['close'] for t in tickers if t in STATE["market_data"]
}).ffill().bfill()
alloc_res = synthesizer.generate_full_allocation(
tickers=tickers,
price_matrix=price_matrix,
macro_snapshot=STATE["macro_snapshot"],
alpha_signals=STATE["last_signal_matrix"],
method=method,
portfolio_capital=float(capital)
)
weights = alloc_res["weights"]
metrics = alloc_res["metrics"]
orders_df = alloc_res["orders_table"]
frontier_fig = alloc_res["frontier_fig"]
corr_fig = alloc_res["corr_fig"]
donut_fig = plot_portfolio_allocation(weights)
frontier_fig.update_layout(plot_bgcolor='#0a0e17', paper_bgcolor='#0a0e17')
corr_fig.update_layout(plot_bgcolor='#0a0e17', paper_bgcolor='#0a0e17')
metrics_md = f"""
### ⚖️ Portfolio Risk & Performance Tear Sheet
- **Target Capital**: `${capital:,.2f}` | **Optimization Method**: `{method}`
- **Expected Annual Return**: **`{metrics['expected_annual_return_pct']:.2f}%`**
- **Expected Annual Volatility**: **`{metrics['expected_annual_volatility_pct']:.2f}%`**
- **Portfolio Sharpe Ratio**: **`{metrics['sharpe_ratio']:.2f}`**
- **Sortino Ratio (Downside Risk Adjusted)**: **`{metrics['sortino_ratio']:.2f}`**
- **Calmar Ratio (Return / Max DD)**: **`{metrics['calmar_ratio']:.2f}`**
- **Maximum Historical Drawdown**: **`{metrics['max_drawdown_pct']:.2f}%`**
- **Daily Value at Risk (VaR 95%)**: `{metrics['daily_var_95_pct']:.2f}%` (`-${metrics['daily_var_95_dollar']:,.2f}`)
- **Daily Expected Shortfall (CVaR 95%)**: `-${metrics['daily_cvar_95_dollar']:,.2f}`
"""
memo = f"""# Quantitative Investment Committee - Research Memo
**Date**: {pd.Timestamp.now().strftime('%Y-%m-%d')}
**Universe**: {', '.join(tickers)}
**Strategy Mode**: {method}
## 1. Executive Summary
The portfolio committee has concluded its multi-agent synthesis combining QuantAgent's self-improving alpha signals, SEC EDGAR corporate filings, sentiment feeds, and macroeconomic regimes. The portfolio is targeted for an expected return of {metrics['expected_annual_return_pct']:.2f}% with a Sharpe Ratio of {metrics['sharpe_ratio']:.2f}.
## 2. Factor & Fundamental Attribution
- **QuantAgent Signal Engine**: Tilts portfolio weights towards top-quintile continuous momentum and volatility breakout candidates.
- **SEC EDGAR Health Screen**: Verified profitability and manageable debt ratios across core mega-cap holdings.
- **Macro Volatility Target**: Regime monitoring enforces risk parity limits to cushion against rate spikes or macro drawdown.
## 3. Order Execution Plan
Execute the rebalancing orders below to establish target shares with minimum market impact (VWAP / TWAP Limit orders).
"""
return metrics_md, donut_fig, frontier_fig, corr_fig, orders_df, memo
def handle_deploy_hf(hf_token: str, space_repo_id: str):
"""One-click deploy/push to Hugging Face Spaces."""
if not hf_token or not space_repo_id:
return "**Error:** Please provide both a Hugging Face Token (write permission) and Space ID (e.g., `deveshc23/quantagents`)."
try:
api = HfApi(token=hf_token)
try:
api.create_repo(
repo_id=space_repo_id,
repo_type="space",
space_sdk="gradio",
exist_ok=True
)
except Exception as e:
logger.info(f"Repo check notice: {e}")
files_to_upload = [
"app.py",
"quant_data.py",
"quant_regimes.py",
"quant_agent_engine.py",
"quant_portfolio.py",
"quant_llm.py",
"requirements.txt",
"README.md"
]
uploaded = []
for fname in files_to_upload:
if os.path.exists(fname):
api.upload_file(
path_or_fileobj=fname,
path_in_repo=fname,
repo_id=space_repo_id,
repo_type="space"
)
uploaded.append(fname)
url = f"https://huggingface.co/spaces/{space_repo_id}"
return f"""### 🎉 Successfully Deployed to Hugging Face Spaces!
- **Space Repository**: [{space_repo_id}]({url})
- **Uploaded Files ({len(uploaded)})**: `{', '.join(uploaded)}`
- **Direct Link**: [Click here to open your live hosted Space]({url})
"""
except Exception as e:
return f"**Deployment Failed:** {str(e)}"
# -----------------------------------------------------------------------------
# Terminal Custom CSS (Sassy Cyberpunk Bloomberg Workstation)
# -----------------------------------------------------------------------------
TERMINAL_CSS = """
@import url('https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@300;400;500;600;700;800&family=Inter:wght@300;400;500;600;700;800&display=swap');
body, .gradio-container {
background-color: #070a13 !important;
font-family: 'Inter', -apple-system, BlinkMacSystemFont, sans-serif !important;
color: #e2e8f0 !important;
}
/* Master Header */
.terminal-header {
background: linear-gradient(135deg, #0d121f 0%, #171d31 50%, #070a13 100%);
border: 1px solid rgba(0, 242, 254, 0.3);
border-radius: 10px;
padding: 16px 22px;
margin-bottom: 10px;
box-shadow: 0 4px 25px rgba(0, 0, 0, 0.6), inset 0 0 15px rgba(0, 242, 254, 0.05);
}
.terminal-title {
font-size: 21px;
font-weight: 800;
letter-spacing: -0.5px;
background: linear-gradient(90deg, #00f2fe, #00ff87, #ff007f);
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
margin: 0;
font-family: 'JetBrains Mono', monospace;
}
/* Smooth Infinite Ticker Tape with ZERO scrollbars */
.tape-container {
display: flex;
align-items: center;
background: #090d18;
border: 1px solid #1e293b;
border-radius: 8px;
padding: 6px 12px;
margin-bottom: 12px;
overflow: hidden;
position: relative;
box-shadow: 0 2px 10px rgba(0,0,0,0.5);
}
.tape-header-badge {
display: inline-flex;
align-items: center;
font-family: 'JetBrains Mono', monospace;
font-size: 11px;
font-weight: 800;
color: #00f2fe;
letter-spacing: 1px;
margin-right: 14px;
white-space: nowrap;
z-index: 2;
background: #090d18;
padding-right: 8px;
}
.pulse-dot {
width: 8px;
height: 8px;
background: #00ff87;
border-radius: 50%;
display: inline-block;
margin-right: 6px;
box-shadow: 0 0 8px #00ff87;
animation: pulse 1.8s infinite;
}
@keyframes pulse {
0% { transform: scale(0.9); opacity: 0.8; }
50% { transform: scale(1.3); opacity: 1; box-shadow: 0 0 12px #00ff87; }
100% { transform: scale(0.9); opacity: 0.8; }
}
.tape-viewport {
overflow: hidden;
white-space: nowrap;
width: 100%;
}
.tape-viewport::-webkit-scrollbar {
display: none !important;
}
.tape-track {
display: inline-flex;
white-space: nowrap;
animation: marquee 32s linear infinite;
}
.tape-track:hover {
animation-play-state: paused;
}
@keyframes marquee {
0% { transform: translateX(0); }
100% { transform: translateX(-50%); }
}
.tape-pill {
display: inline-flex;
align-items: center;
gap: 8px;
padding: 3px 12px;
margin-right: 14px;
background: rgba(15, 23, 42, 0.7);
border: 1px solid rgba(255, 255, 255, 0.08);
border-radius: 5px;
font-family: 'JetBrains Mono', monospace;
font-size: 11px;
}
.pill-sym { font-weight: 800; color: #ffffff; }
.pill-price { color: #94a3b8; }
.pill-chg { font-weight: 700; }
/* Cyber HUD Cards for Macro & Regime */
.hud-container {
display: grid;
grid-template-columns: 1fr 1fr;
gap: 12px;
margin-bottom: 12px;
}
.hud-card {
background: linear-gradient(135deg, rgba(13,18,31,0.85), rgba(7,10,19,0.95));
border: 1px solid rgba(255,255,255,0.08);
border-radius: 8px;
padding: 12px 16px;
box-shadow: 0 4px 15px rgba(0,0,0,0.4);
}
.hud-title {
display: flex;
justify-content: space-between;
align-items: center;
font-family: 'JetBrains Mono', monospace;
font-size: 11px;
font-weight: 800;
letter-spacing: 0.5px;
margin-bottom: 8px;
border-bottom: 1px solid rgba(255,255,255,0.06);
padding-bottom: 6px;
}
.hud-tag {
font-size: 9px;
padding: 2px 6px;
border-radius: 3px;
border: 1px solid rgba(0,242,254,0.4);
color: #00f2fe;
}
.hud-grid-row {
display: grid;
grid-template-columns: repeat(6, 1fr);
gap: 6px;
}
.hud-stat-box {
background: rgba(255,255,255,0.02);
border: 1px solid rgba(255,255,255,0.05);
border-radius: 4px;
padding: 6px 4px;
text-align: center;
}
.hud-lbl {
font-size: 8.5px;
color: #64748b;
font-family: 'JetBrains Mono', monospace;
font-weight: 700;
margin-bottom: 2px;
}
.hud-val {
font-size: 12px;
color: #ffffff;
font-family: 'JetBrains Mono', monospace;
font-weight: 800;
}
/* Cyber KPI Card for Selected Ticker */
.cyber-kpi-card {
background: linear-gradient(135deg, rgba(13,18,31,0.9), rgba(7,10,19,0.95));
border: 1px solid rgba(0, 242, 254, 0.25);
border-radius: 8px;
padding: 14px;
box-shadow: 0 6px 20px rgba(0,0,0,0.5);
margin-top: 10px;
}
.kpi-hero {
display: flex;
justify-content: space-between;
align-items: center;
border-bottom: 1px solid rgba(255,255,255,0.08);
padding-bottom: 10px;
margin-bottom: 12px;
}
.kpi-sym {
font-family: 'JetBrains Mono', monospace;
font-size: 22px;
font-weight: 800;
color: #00f2fe;
letter-spacing: 1px;
}
.kpi-sub {
font-size: 11px;
color: #94a3b8;
font-family: 'JetBrains Mono', monospace;
}
.kpi-price {
font-family: 'JetBrains Mono', monospace;
font-size: 20px;
font-weight: 800;
color: #ffffff;
}
.kpi-delta {
font-family: 'JetBrains Mono', monospace;
font-size: 11px;
font-weight: 700;
}
.kpi-grid {
display: grid;
grid-template-columns: 1fr 1fr;
gap: 8px;
}
.kpi-tile {
background: rgba(255,255,255,0.03);
border: 1px solid rgba(255,255,255,0.06);
border-radius: 5px;
padding: 8px 10px;
}
.kpi-tile.full-width {
grid-column: span 2;
}
.tile-label {
font-size: 9px;
color: #64748b;
font-family: 'JetBrains Mono', monospace;
font-weight: 700;
margin-bottom: 3px;
}
.tile-val {
font-size: 13px;
color: #ffffff;
font-family: 'JetBrains Mono', monospace;
font-weight: 700;
}
.meter-bar-bg {
width: 100%;
height: 6px;
background: rgba(255,255,255,0.08);
border-radius: 3px;
overflow: hidden;
}
.meter-bar-fill {
height: 100%;
border-radius: 3px;
}
.regime-pill {
padding: 2px 8px;
border-radius: 4px;
font-size: 10px;
font-family: 'JetBrains Mono', monospace;
font-weight: 700;
border-width: 1px;
border-style: solid;
}
/* Sleek Command Bar & Preset Pills */
.command-bar {
align-items: flex-end !important;
gap: 8px !important;
margin-bottom: 6px !important;
}
.command-bar input {
background: #090d18 !important;
border: 1px solid rgba(0, 242, 254, 0.35) !important;
border-radius: 6px !important;
color: #00f2fe !important;
font-family: 'JetBrains Mono', monospace !important;
font-weight: 700 !important;
font-size: 13px !important;
padding: 8px 12px !important;
}
.command-bar input:focus {
border-color: #00ff87 !important;
box-shadow: 0 0 12px rgba(0, 255, 135, 0.25) !important;
}
.btn-command {
font-family: 'JetBrains Mono', monospace !important;
font-size: 12px !important;
font-weight: 800 !important;
border-radius: 6px !important;
height: 42px !important;
letter-spacing: 0.5px !important;
}
.presets-bar {
align-items: center !important;
margin-bottom: 14px !important;
gap: 8px !important;
}
.btn-pill {
background: rgba(255, 255, 255, 0.03) !important;
border: 1px solid rgba(255, 255, 255, 0.08) !important;
border-radius: 5px !important;
color: #94a3b8 !important;
font-family: 'JetBrains Mono', monospace !important;
font-size: 11px !important;
font-weight: 700 !important;
padding: 5px 12px !important;
transition: all 0.2s ease !important;
}
.btn-pill:hover {
border-color: #00f2fe !important;
color: #00f2fe !important;
background: rgba(0, 242, 254, 0.1) !important;
box-shadow: 0 0 10px rgba(0, 242, 254, 0.2) !important;
}
.status-badge {
font-family: 'JetBrains Mono', monospace;
font-size: 11px;
color: #94a3b8;
text-align: right;
padding: 6px 10px;
background: rgba(13, 18, 31, 0.7);
border: 1px solid rgba(255, 255, 255, 0.06);
border-radius: 5px;
}
/* Single-Line High-Density Institutional Navigation Tabs */
div[role="tablist"], .tabs > .tab-nav {
display: flex !important;
flex-wrap: nowrap !important;
overflow-x: auto !important;
overflow-y: hidden !important;
white-space: nowrap !important;
gap: 6px !important;
background: #090d18 !important;
border: 1px solid rgba(255, 255, 255, 0.08) !important;
border-radius: 8px !important;
padding: 5px !important;
margin-bottom: 12px !important;
box-shadow: 0 4px 15px rgba(0,0,0,0.3) !important;
}
div[role="tablist"]::-webkit-scrollbar, .tabs > .tab-nav::-webkit-scrollbar {
height: 3px !important;
display: none !important;
}
div[role="tablist"] button, .tabs > .tab-nav > button {
font-family: 'JetBrains Mono', monospace !important;
font-size: 12px !important;
font-weight: 700 !important;
color: #94a3b8 !important;
padding: 9px 18px !important;
border-radius: 6px !important;
border: 1px solid rgba(255, 255, 255, 0.05) !important;
background: rgba(255, 255, 255, 0.02) !important;
white-space: nowrap !important;
flex: 1 1 auto !important;
text-align: center !important;
transition: all 0.2s cubic-bezier(0.4, 0, 0.2, 1) !important;
}
div[role="tablist"] button:hover, .tabs > .tab-nav > button:hover {
color: #ffffff !important;
background: rgba(0, 242, 254, 0.06) !important;
border-color: rgba(0, 242, 254, 0.3) !important;
}
div[role="tablist"] button.selected, div[role="tablist"] button[aria-selected="true"], .tabs > .tab-nav > button.selected {
background: linear-gradient(135deg, rgba(0, 242, 254, 0.16), rgba(0, 255, 135, 0.08)) !important;
border: 1px solid #00f2fe !important;
color: #00f2fe !important;
font-weight: 800 !important;
box-shadow: 0 0 15px rgba(0, 242, 254, 0.25) !important;
}
"""
# -----------------------------------------------------------------------------
# Gradio UI Construction
# -----------------------------------------------------------------------------
def build_app() -> gr.Blocks:
theme = gr.themes.Soft(
primary_hue="blue",
secondary_hue="emerald",
neutral_hue="slate"
)
with gr.Blocks(theme=theme, title="QuantAgent OS - Quantitative Investment Terminal", css=TERMINAL_CSS) as demo:
# Top Institutional Header
gr.HTML("""
""")
# Real-time Continuous Ticker Tape (Zero scrollbars)
tape_box = gr.HTML(value=generate_tape_html())
# Integrated Institutional Control Bar
with gr.Row(elem_classes=["command-bar"]):
universe_input = gr.Textbox(
label="Active Universe Tickers",
value="AAPL, MSFT, NVDA, GOOGL, AMZN",
placeholder="Enter comma-separated symbols (e.g. AAPL, NVDA, MSFT...)",
scale=5,
lines=1,
max_lines=1
)
universe_btn = gr.Button("🔄 Sync Universe", variant="primary", scale=1, elem_classes=["btn-command"])
sync_tape_btn = gr.Button("⚡ Sync Tape", scale=1, elem_classes=["btn-command"])
# Presets & Real-Time Portfolio Status Bar
with gr.Row(elem_classes=["presets-bar"]):
with gr.Column(scale=5):
with gr.Row():
preset_tech = gr.Button("⚡ Tech", size="sm", elem_classes=["btn-pill"])
preset_core = gr.Button("🛡️ Core", size="sm", elem_classes=["btn-pill"])
preset_fin = gr.Button("🏦 Finance", size="sm", elem_classes=["btn-pill"])
preset_def = gr.Button("💎 Dividend", size="sm", elem_classes=["btn-pill"])
preset_etf = gr.Button("📈 Macro", size="sm", elem_classes=["btn-pill"])
with gr.Column(scale=3):
system_status_md = gr.Markdown(
f"🟢 PORTFOLIO UNIVERSE: {len(STATE['active_tickers'])} ASSETS LOADED • 12ms LATENCY
"
)
# ---------------------------------------------------------------------
# TABS
# ---------------------------------------------------------------------
with gr.Tabs():
# -----------------------------------------------------------------
# TAB 1: Market Workstation
# -----------------------------------------------------------------
with gr.Tab("📊 Market Workstation"):
# Top Macro & Regime HUD Cards
macro_hud_box = gr.HTML(value=generate_macro_hud_html())
with gr.Row():
with gr.Column(scale=1):
selected_ticker_plot = gr.Dropdown(
choices=STATE["active_tickers"],
value=STATE["active_tickers"][0],
label="Select Active Ticker to Chart"
)
period_select = gr.Radio(
choices=["1mo", "3mo", "6mo", "1y", "2y"],
value="1y",
label="Historical Period Range"
)
overlays_check = gr.CheckboxGroup(
choices=["Bollinger Bands (20, 2σ)", "SMA 20 & SMA 50", "VWAP", "Volume + 20D MA", "RSI (14)", "MACD (12, 26, 9)"],
value=["Bollinger Bands (20, 2σ)", "SMA 20 & SMA 50", "VWAP", "Volume + 20D MA", "RSI (14)", "MACD (12, 26, 9)"],
label="Technical Indicators & Overlays"
)
# Cyber KPI HUD Card for selected ticker
ticker_kpi_box = gr.HTML(value=generate_kpi_card_html(STATE["active_tickers"][0]))
with gr.Column(scale=3):
stock_plot = gr.Plot(value=plot_stock_candle(STATE["active_tickers"][0])[0])
# -----------------------------------------------------------------
# TAB 2: Alpha Studio
# -----------------------------------------------------------------
with gr.Tab("🧠 Alpha Studio"):
gr.Markdown("""
### *Self-Improving Multi-Agent Quantitative Architecture based on arXiv:2402.03755*
The **Inner Loop** iteratively refines alpha code between Writer and Judge LLMs until Judge Score $\\ge \\beta$; the **Outer Loop** executes real backtesting and updates the Autonomous Knowledge Base.
""")
with gr.Row():
with gr.Column(scale=3):
hypothesis_preset = gr.Dropdown(
choices=[
"Adaptive Volatility Breakout with Donchian Channels and ATR Volatility Normalization",
"Cross-Sectional Momentum with Short-Term Mean Reversion Penalty (QuantAgent Fig 3)",
"Order Flow Volume-Weighted Price Trend and Relative Volatility Spread",
"Bollinger Squeeze with Multi-Timeframe Trend Confirmation",
"Dupont ROE Quality Drift with Low Debt-to-Equity Filter"
],
value="Adaptive Volatility Breakout with Donchian Channels and ATR Volatility Normalization",
label="Trading Hypothesis Presets (1-Click Selection)"
)
idea_input = gr.Textbox(
label="Quantitative Trading Idea / Hypothesis Prompt",
value="Adaptive Volatility Breakout with Donchian Channels and ATR Volatility Normalization",
lines=2
)
with gr.Row():
rounds_slider = gr.Slider(minimum=1, maximum=5, value=3, step=1, label="Max Inner Loop Rounds (T)")
thresh_slider = gr.Slider(minimum=5.0, maximum=9.5, value=8.0, step=0.5, label="Reward Approval Threshold (β)")
run_agent_btn = gr.Button("🚀 Run QuantAgent Self-Improving Loop", variant="primary")
with gr.Column(scale=2):
outer_status_md = gr.Markdown("### Outer Feedback Loop Status\n*Awaiting execution...*")
with gr.Row():
with gr.Column(scale=1):
inner_trajectory_md = gr.Markdown("### Inner Reasoning Loop Trajectory\n*Awaiting execution...*")
with gr.Column(scale=1):
alpha_code_box = gr.Code(label="Final Generated Factor Code (Factor subclass)", language="python")
with gr.Row():
with gr.Column(scale=1):
backtest_cum_plot = gr.Plot(label="Cumulative Strategy Return & Drawdown")
with gr.Column(scale=1):
backtest_quantile_plot = gr.Plot(label="Factor Quantile Monotonicity (Q1 to Q5)")
gr.Markdown("---")
gr.Markdown("### 📚 Autonomous Knowledge Base (KB) Explorer")
initial_kb_df = pd.DataFrame(quant_engine.kb.get_all_records())
kb_table = gr.Dataframe(value=initial_kb_df, label="Discovered Alpha Records in Knowledge Base")
with gr.Row():
train_xgb_btn = gr.Button("⚡ Train Downstream XGBoost on Discovered Alphas (Paper Fig 5)", variant="secondary")
with gr.Row():
xgb_status_md = gr.Markdown("")
xgb_plot = gr.Plot()
# -----------------------------------------------------------------
# TAB 3: SEC Fundamentals
# -----------------------------------------------------------------
with gr.Tab("📑 SEC Fundamentals"):
gr.Markdown("*Fetches official SEC EDGAR XBRL regulatory disclosures and Dupont 3-Stage financial metrics.*")
with gr.Row():
sec_ticker_input = gr.Textbox(label="Ticker to Analyze", value="AAPL")
sec_lookup_btn = gr.Button("🔍 Fetch SEC EDGAR Filings", variant="primary")
sec_output_md = gr.Markdown(handle_sec_lookup("AAPL"))
# -----------------------------------------------------------------
# TAB 4: News Sentiment
# -----------------------------------------------------------------
with gr.Tab("📰 News Sentiment"):
with gr.Row():
news_ticker_input = gr.Textbox(label="Ticker for News", value="AAPL")
news_lookup_btn = gr.Button("📡 Ingest Live Headlines", variant="primary")
initial_news_md, initial_news_df = handle_news_lookup("AAPL")
news_summary_md = gr.Markdown(initial_news_md)
news_table = gr.Dataframe(value=initial_news_df, label="Analyzed Headlines")
# -----------------------------------------------------------------
# TAB 5: Portfolio Engine
# -----------------------------------------------------------------
with gr.Tab("⚖️ Portfolio Engine"):
gr.Markdown("*Fuses QuantAgent Alpha signals, SEC fundamentals, news sentiment, and macro volatility target.*")
with gr.Row():
opt_method = gr.Dropdown(
choices=["Hierarchical Risk Parity (HRP)", "Maximum Sharpe Ratio (Markowitz)", "Inverse Volatility Parity"],
value="Hierarchical Risk Parity (HRP)",
label="Portfolio Optimization Algorithm"
)
portfolio_capital = gr.Number(value=100000.0, label="Portfolio Capital ($)")
optimize_btn = gr.Button("🎯 Optimize Portfolio Allocation", variant="primary")
with gr.Row():
risk_tear_sheet = gr.Markdown("### Portfolio Risk Tear Sheet\n*Click Optimize to calculate.*")
allocation_donut_plot = gr.Plot(label="Target Asset Allocation")
with gr.Row():
frontier_plot = gr.Plot(label="Markowitz Efficient Frontier & Feasible Portfolios")
corr_plot = gr.Plot(label="Cross-Asset Return Correlation Heatmap")
orders_table = gr.Dataframe(label="Institutional Trade Execution Order Tickets")
memo_box = gr.Markdown("### Executive Investment Memo")
# -----------------------------------------------------------------
# TAB 6: Settings & Deploy
# -----------------------------------------------------------------
with gr.Tab("⚙️ Settings & Deploy"):
gr.Markdown("""
### ☁️ One-Click Deploy to Hugging Face Spaces
Enter your Hugging Face write token and the target Space ID (`deveshc23/quantagents`) to automatically push and host this system live on Hugging Face Spaces!
""")
with gr.Row():
hf_token_input = gr.Textbox(label="Hugging Face API Write Token", type="password", placeholder="hf_...")
space_id_input = gr.Textbox(label="Target Space Repo ID", value="deveshc23/quantagents")
deploy_btn = gr.Button("🚢 Deploy to Hugging Face Spaces", variant="primary")
deploy_status_md = gr.Markdown("")
gr.Markdown("---")
gr.Markdown("### 🔑 Optional API Keys (Upgrades functionality)")
with gr.Row():
gemini_key_input = gr.Textbox(label="Google Gemini API Key", type="password", placeholder="AIza...")
finnhub_key_input = gr.Textbox(label="Finnhub API Key", type="password", value=os.environ.get("FINNHUB_API_KEY", ""), placeholder="Optional...")
fred_key_input = gr.Textbox(label="FRED API Key", type="password", value=os.environ.get("FRED_API_KEY", ""), placeholder="Optional...")
save_keys_btn = gr.Button("Save API Keys")
def save_keys_action(gemini_k, finnhub_k, fred_k):
if gemini_k:
llm_client.update_key(gemini_k)
quant_engine.llm = llm_client
if finnhub_k:
news_service.finnhub_key = finnhub_k
os.environ["FINNHUB_API_KEY"] = finnhub_k
if fred_k:
macro_service.fred_key = fred_k
os.environ["FRED_API_KEY"] = fred_k
STATE["macro_snapshot"] = macro_service.get_macro_snapshot()
# Persist to local .env
try:
with open(".env", "w", encoding="utf-8") as ef:
if os.environ.get("FINNHUB_API_KEY"):
ef.write(f"FINNHUB_API_KEY={os.environ['FINNHUB_API_KEY']}\n")
if os.environ.get("FRED_API_KEY"):
ef.write(f"FRED_API_KEY={os.environ['FRED_API_KEY']}\n")
except Exception:
pass
return "✅ API Keys updated, activated, and persisted!"
keys_status_md = gr.Markdown("")
save_keys_btn.click(save_keys_action, [gemini_key_input, finnhub_key_input, fred_key_input], keys_status_md)
# ---------------------------------------------------------------------
# Event Bindings
# ---------------------------------------------------------------------
def on_refresh_universe(tickers_str):
tickers = refresh_market_data(tickers_str)
updated_dropdown = gr.Dropdown(choices=tickers, value=tickers[0])
status = f"🟢 SYNCED: {len(tickers)} ASSETS LOADED • 12ms LATENCY
"
fig, _ = plot_stock_candle(tickers[0])
kpi_html = generate_kpi_card_html(tickers[0])
tape_content = generate_tape_html(force_refresh=True)
macro_html = generate_macro_hud_html()
return updated_dropdown, status, fig, kpi_html, tape_content, macro_html
universe_btn.click(
on_refresh_universe,
universe_input,
[selected_ticker_plot, system_status_md, stock_plot, ticker_kpi_box, tape_box, macro_hud_box]
)
def set_preset_universe(preset_name):
tickers = quant_data.UNIVERSE_PRESETS.get(preset_name, ["AAPL", "MSFT", "NVDA"])
t_str = ", ".join(tickers)
refreshed = refresh_market_data(t_str)
updated_dropdown = gr.Dropdown(choices=refreshed, value=refreshed[0])
status = f"🟢 ACTIVE: {preset_name.upper()} ({len(refreshed)} ASSETS) • 12ms LATENCY
"
fig, _ = plot_stock_candle(refreshed[0])
kpi_html = generate_kpi_card_html(refreshed[0])
tape_html = generate_tape_html(force_refresh=True)
return t_str, updated_dropdown, status, fig, kpi_html, tape_html
preset_tech.click(lambda: set_preset_universe("Mega-Cap Tech"), outputs=[universe_input, selected_ticker_plot, system_status_md, stock_plot, ticker_kpi_box, tape_box])
preset_core.click(lambda: set_preset_universe("Balanced Core"), outputs=[universe_input, selected_ticker_plot, system_status_md, stock_plot, ticker_kpi_box, tape_box])
preset_fin.click(lambda: set_preset_universe("Financials & Energy"), outputs=[universe_input, selected_ticker_plot, system_status_md, stock_plot, ticker_kpi_box, tape_box])
preset_def.click(lambda: set_preset_universe("Defensive Dividend"), outputs=[universe_input, selected_ticker_plot, system_status_md, stock_plot, ticker_kpi_box, tape_box])
preset_etf.click(lambda: set_preset_universe("Macro ETFs"), outputs=[universe_input, selected_ticker_plot, system_status_md, stock_plot, ticker_kpi_box, tape_box])
# Live Tape Sync & Automated Timer (15s polling)
tape_timer = gr.Timer(value=15)
tape_timer.tick(fn=generate_tape_html, outputs=tape_box)
sync_tape_btn.click(lambda: generate_tape_html(force_refresh=True), outputs=tape_box)
def on_chart_param_change(ticker, period, overlays):
fig, _ = plot_stock_candle(ticker, period, overlays)
kpi_html = generate_kpi_card_html(ticker)
return fig, kpi_html
selected_ticker_plot.change(on_chart_param_change, [selected_ticker_plot, period_select, overlays_check], [stock_plot, ticker_kpi_box])
period_select.change(on_chart_param_change, [selected_ticker_plot, period_select, overlays_check], [stock_plot, ticker_kpi_box])
overlays_check.change(on_chart_param_change, [selected_ticker_plot, period_select, overlays_check], [stock_plot, ticker_kpi_box])
hypothesis_preset.change(lambda p: p, hypothesis_preset, idea_input)
run_agent_btn.click(
handle_run_quantagent,
[idea_input, rounds_slider, thresh_slider],
[inner_trajectory_md, alpha_code_box, outer_status_md, backtest_cum_plot, backtest_quantile_plot, kb_table]
)
train_xgb_btn.click(handle_train_xgboost, outputs=[xgb_status_md, xgb_plot])
sec_lookup_btn.click(handle_sec_lookup, sec_ticker_input, sec_output_md)
news_lookup_btn.click(handle_news_lookup, news_ticker_input, [news_summary_md, news_table])
optimize_btn.click(
handle_optimize_portfolio,
[opt_method, portfolio_capital],
[risk_tear_sheet, allocation_donut_plot, frontier_plot, corr_plot, orders_table, memo_box]
)
deploy_btn.click(handle_deploy_hf, [hf_token_input, space_id_input], deploy_status_md)
return demo
# -----------------------------------------------------------------------------
# Main Entry Point
# -----------------------------------------------------------------------------
app = build_app()
if __name__ == "__main__":
port = int(os.environ.get("PORT", 7860))
app.launch(server_name="0.0.0.0", server_port=port, share=False)