import streamlit as st import time from datetime import datetime from services.broker_apis import BrokerManager def display_animated_broker_status(): """Display animated broker connection status widget""" # Initialize broker manager if 'broker_manager' not in st.session_state: st.session_state.broker_manager = BrokerManager() broker_manager = st.session_state.broker_manager # Get broker info active_broker = broker_manager.get_active_broker_name() # Create container for the widget status_container = st.container() with status_container: # CSS for animations and styling st.markdown(""" """, unsafe_allow_html=True) # Test connection try: account_info = broker_manager.get_account_info() is_connected = 'error' not in account_info if is_connected: status_class = "connected" status_text = "Connected" status_icon = "🟢" else: status_class = "disconnected" status_text = "Disconnected" status_icon = "🔴" except Exception: status_class = "disconnected" status_text = "Connection Error" status_icon = "🔴" account_info = {} is_connected = False # Format broker name for display broker_display_names = { 'alpaca_paper': 'Alpaca Paper Trading', 'alpaca_live': 'Alpaca Live Trading', 'tradier_paper': 'Tradier Sandbox', 'tradier_live': 'Tradier Live', 'robinhood': 'Robinhood' } broker_display = broker_display_names.get(active_broker, active_broker.title()) # Create the widget HTML widget_html = f"""
LIVE
{status_icon} {broker_display} LIVE
Status: {status_text} | Mode: {'Paper' if 'paper' in active_broker else 'Live'} Trading
""" if is_connected: portfolio_value = account_info.get('portfolio_value', 0) cash = account_info.get('cash', 0) buying_power = account_info.get('buying_power', 0) day_trades = account_info.get('day_trade_count', 0) widget_html += f"""
${portfolio_value:,.0f}
Portfolio
${cash:,.0f}
Cash
${buying_power:,.0f}
Buying Power
{day_trades}/3
Day Trades
""" current_time = datetime.now().strftime("%H:%M:%S") widget_html += f"""
Last updated: {current_time}
""" # Display the widget st.markdown(widget_html, unsafe_allow_html=True) # Refresh button col1, col2, col3 = st.columns([1, 1, 1]) with col2: if st.button("🔄 Refresh Status", key="refresh_broker_status"): # Force refresh broker manager if 'broker_manager' in st.session_state: st.session_state.broker_manager.reload_configuration() st.rerun() return is_connected, account_info def display_mini_broker_status(): """Display a compact version of the broker status widget""" if 'broker_manager' not in st.session_state: st.session_state.broker_manager = BrokerManager() broker_manager = st.session_state.broker_manager active_broker = broker_manager.get_active_broker_name() try: account_info = broker_manager.get_account_info() is_connected = 'error' not in account_info status_icon = "🟢" if is_connected else "🔴" status_text = "Connected" if is_connected else "Disconnected" except Exception: status_icon = "🔴" status_text = "Error" is_connected = False # Compact status display broker_display_names = { 'alpaca_paper': 'Alpaca Paper', 'alpaca_live': 'Alpaca Live', 'tradier_paper': 'Tradier Sandbox', 'tradier_live': 'Tradier Live', 'robinhood': 'Robinhood' } broker_display = broker_display_names.get(active_broker, active_broker) st.markdown(f"""
{status_icon} {broker_display} - {status_text}
""", unsafe_allow_html=True) return is_connected def display_connection_health_chart(): """Display a connection health monitoring chart""" # Store connection history in session state if 'connection_history' not in st.session_state: st.session_state.connection_history = [] if 'broker_manager' not in st.session_state: st.session_state.broker_manager = BrokerManager() broker_manager = st.session_state.broker_manager # Test connection try: account_info = broker_manager.get_account_info() is_connected = 'error' not in account_info response_time = 0.1 # Mock response time except Exception: is_connected = False response_time = None # Add to history current_time = datetime.now() st.session_state.connection_history.append({ 'timestamp': current_time, 'connected': is_connected, 'response_time': response_time }) # Keep only last 20 data points if len(st.session_state.connection_history) > 20: st.session_state.connection_history = st.session_state.connection_history[-20:] # Create chart if we have data if len(st.session_state.connection_history) >= 2: import plotly.graph_objects as go timestamps = [h['timestamp'] for h in st.session_state.connection_history] connected_status = [1 if h['connected'] else 0 for h in st.session_state.connection_history] fig = go.Figure() fig.add_trace(go.Scatter( x=timestamps, y=connected_status, mode='lines+markers', name='Connection Status', line=dict(color='green', width=3), fill='tozeroy', fillcolor='rgba(0, 255, 0, 0.2)' )) fig.update_layout( title="Broker Connection Health", xaxis_title="Time", yaxis_title="Status", yaxis=dict(tickmode='array', tickvals=[0, 1], ticktext=['Disconnected', 'Connected']), height=200, margin=dict(l=0, r=0, t=30, b=0), showlegend=False ) st.plotly_chart(fig, use_container_width=True)