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"""
""" # 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"""