Spaces:
Sleeping
Sleeping
File size: 4,143 Bytes
558db1e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | import streamlit as st
import pandas as pd
import json
@st.cache_resource
def get_run_engine():
from main import run_engine
return run_engine
run_engine = get_run_engine()
st.set_page_config(page_title="Portfolio Engine Dashboard", layout="wide")
st.title("Quantitative Portfolio Production Engine")
st.sidebar.header("Portfolio Inputs")
tickers_input = st.sidebar.text_input("Tickers (comma separated)", "SPY, TLT, GLD")
capital = st.sidebar.number_input("Capital ($)", value=100000.0)
risk = st.sidebar.slider("Risk Aversion (1=Aggressive, 10=Conservative)", min_value=1, max_value=10, value=5)
model = st.sidebar.selectbox("Return Model", options=[
(1, "CAPM"),
(2, "Black-Litterman"),
(3, "James-Stein"),
(4, "Fama-French"),
(5, "ML Stacking/BSTS")
], format_func=lambda x: x[1])[0]
allocation_engine = st.sidebar.selectbox("Allocation Engine", options=[
(1, "Mean-Variance"),
(2, "Hierarchical Risk Parity")
], format_func=lambda x: x[1])[0]
st.sidebar.subheader("Advanced Settings")
monthly = st.sidebar.checkbox("Monthly Frequency", value=False)
tax = st.sidebar.checkbox("Enable Tax Calculation", value=False)
with_futures = st.sidebar.checkbox("Enable Futures Overlay", value=False)
import requests
import time
def get_risk_factor(risk_level: int) -> float:
risk_map = {
1: 0.1, 2: 0.5, 3: 1.0, 4: 2.0, 5: 3.0,
6: 5.0, 7: 7.5, 8: 10.0, 9: 15.0, 10: 25.0
}
return risk_map.get(risk_level, 3.0)
if st.button("Run Optimization via API"):
with st.spinner("Calling remote API to run global quantitative optimization..."):
tickers_list = [t.strip() for t in tickers_input.split(",") if t.strip()]
payload = {
"tickers": tickers_list,
"capital": float(capital),
"risk_level": int(risk),
"model": int(model),
"engine": int(allocation_engine),
"currency": "$",
"days": 252,
"current_weights": {}
}
try:
import os
api_key = os.getenv("API_KEY", "SECRET_API_KEY_123")
headers = {"X-API-Key": api_key}
res = requests.post("http://127.0.0.1:8080/run_optimization", json=payload, headers=headers, timeout=300)
res.raise_for_status()
st.success("Optimization complete! Connect to live feed below.")
except Exception as e:
st.error(f"Error during API call: {e}")
st.header("Live Risk Dashboard")
if st.button("Start Live Data Feed"):
import websocket
import json
col1, col2, col3 = st.columns(3)
pnl_metric = col1.empty()
capital_metric = col2.empty()
var_breach_metric = col3.empty()
chart_holder = st.empty()
positions_holder = st.empty()
pnl_history = []
breaches = 0
var_limit = - (float(capital) * 0.01) # 1% 1-Day VaR limit for display
try:
ws = websocket.WebSocket()
import os
api_key = os.getenv("API_KEY", "SECRET_API_KEY_123")
ws.connect(f"ws://127.0.0.1:8080/ws?api_key={api_key}")
while True:
data = json.loads(ws.recv())
if data.get("type") == "live_update":
pnl = data["pnl"]
cap = data["capital"]
prices = data["prices"]
pnl_history.append(pnl)
if pnl < var_limit:
breaches += 1
pnl_metric.metric("Live P&L", f"${pnl:,.2f}", delta=f"${pnl - (pnl_history[-2] if len(pnl_history) > 1 else 0):,.2f}")
capital_metric.metric("Total Capital", f"${cap:,.2f}")
var_breach_metric.metric("VaR Limit Breaches", breaches)
chart_holder.line_chart(pnl_history)
pdf = pd.DataFrame(list(prices.items()), columns=["Ticker", "Live Price ($)"])
positions_holder.dataframe(pdf, use_container_width=True)
except Exception as e:
st.error(f"WebSocket disconnected or failed: {e}")
|