File size: 4,743 Bytes
e4e4574 0a6cbcb e4e4574 |
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 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 |
"""
Live Dashboard Tab - Real-time cryptocurrency price monitoring
Refactored from app.py with improved type hints and structure
"""
import pandas as pd
import logging
import traceback
from typing import Tuple
import database
import collectors
import utils
# Setup logging with error handling
try:
logger = utils.setup_logging()
except (AttributeError, ImportError) as e:
# Fallback logging setup if utils.setup_logging() is not available
print(f"Warning: Could not import utils.setup_logging(): {e}")
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('dashboard_live')
# Initialize database
db = database.get_database()
def get_live_dashboard(search_filter: str = "") -> pd.DataFrame:
"""
Get live dashboard data with top 100 cryptocurrencies
Args:
search_filter: Search/filter text for cryptocurrencies (searches name and symbol)
Returns:
DataFrame with formatted cryptocurrency data including:
- Rank, Name, Symbol
- Price (USD), 24h Change (%)
- Volume, Market Cap
"""
try:
logger.info("Fetching live dashboard data...")
# Get latest prices from database
prices = db.get_latest_prices(100)
if not prices:
logger.warning("No price data available")
return _empty_dashboard_dataframe()
# Convert to DataFrame with filtering
df_data = []
for price in prices:
# Apply search filter if provided
if search_filter and not _matches_filter(price, search_filter):
continue
df_data.append(_format_price_row(price))
df = pd.DataFrame(df_data)
if df.empty:
logger.warning("No data matches filter criteria")
return _empty_dashboard_dataframe()
# Sort by rank
df = df.sort_values('Rank')
logger.info(f"Dashboard loaded with {len(df)} cryptocurrencies")
return df
except Exception as e:
logger.error(f"Error in get_live_dashboard: {e}\n{traceback.format_exc()}")
return pd.DataFrame({
"Error": [f"Failed to load dashboard: {str(e)}"]
})
def refresh_price_data() -> Tuple[pd.DataFrame, str]:
"""
Manually trigger price data collection and refresh dashboard
Returns:
Tuple of (updated DataFrame, status message string)
"""
try:
logger.info("Manual refresh triggered...")
# Collect fresh price data
success, count = collectors.collect_price_data()
if success:
message = f"✅ Successfully refreshed! Collected {count} price records."
else:
message = f"⚠️ Refresh completed with warnings. Collected {count} records."
# Return updated dashboard
df = get_live_dashboard()
return df, message
except Exception as e:
logger.error(f"Error in refresh_price_data: {e}")
return get_live_dashboard(), f"❌ Refresh failed: {str(e)}"
# ==================== PRIVATE HELPER FUNCTIONS ====================
def _empty_dashboard_dataframe() -> pd.DataFrame:
"""Create empty DataFrame with proper column structure"""
return pd.DataFrame({
"Rank": [],
"Name": [],
"Symbol": [],
"Price (USD)": [],
"24h Change (%)": [],
"Volume": [],
"Market Cap": []
})
def _matches_filter(price: dict, search_filter: str) -> bool:
"""
Check if price record matches search filter
Args:
price: Price data dictionary
search_filter: Search text
Returns:
True if matches, False otherwise
"""
search_lower = search_filter.lower()
name_lower = (price.get('name') or '').lower()
symbol_lower = (price.get('symbol') or '').lower()
return search_lower in name_lower or search_lower in symbol_lower
def _format_price_row(price: dict) -> dict:
"""
Format price data for dashboard display
Args:
price: Raw price data dictionary
Returns:
Formatted dictionary with display-friendly values
"""
return {
"Rank": price.get('rank', 999),
"Name": price.get('name', 'Unknown'),
"Symbol": price.get('symbol', 'N/A').upper(),
"Price (USD)": f"${price.get('price_usd', 0):,.2f}" if price.get('price_usd') else "N/A",
"24h Change (%)": f"{price.get('percent_change_24h', 0):+.2f}%" if price.get('percent_change_24h') is not None else "N/A",
"Volume": utils.format_number(price.get('volume_24h', 0)),
"Market Cap": utils.format_number(price.get('market_cap', 0))
}
|