import streamlit as st import yfinance as yf import pandas as pd from datetime import datetime, timedelta import pytz import plotly.graph_objects as go from plotly.subplots import make_subplots # --------------------------- # Configuration and Constants # --------------------------- SECTORS = { 'Technology': 'XLK', 'Healthcare': 'XLV', 'Financials': 'XLF', 'Energy': 'XLE', 'Utilities': 'XLU', 'Consumer Discretionary': 'XLY', 'Consumer Staples': 'XLP', 'Industrials': 'XLI', 'Materials': 'XLB', 'Real Estate': 'XLRE', 'Telecommunication Services': 'XLC' } COLORS = [ '#2E7D32', '#1976D2', '#C62828', '#F57C00', '#6A1B9A', '#283593', '#00838F', '#558B2F', '#D84315', '#4527A0', '#00695C' ] # --------------------------- # Utility Functions # --------------------------- def get_appropriate_interval(start_date, end_date, market_status): """ Determines the appropriate data interval based on date range and market status. """ days_diff = (end_date - start_date).days if market_status and days_diff <= 7: # Only use 1m data for up to 7 days when market is open return '1m' elif days_diff <= 60: return '1h' else: return '1d' @st.cache_data(ttl=60*5) # Cache data for 5 minutes def fetch_sector_performance(start_date, end_date, interval='1d'): """ Fetches the performance data for each sector ETF with improved error handling. """ sector_data = [] adjusted_end_date = end_date + timedelta(days=1) india_tz = pytz.timezone('Asia/Kolkata') start_datetime = datetime.combine(start_date, datetime.min.time()).astimezone(india_tz) end_datetime = datetime.combine(adjusted_end_date, datetime.min.time()).astimezone(india_tz) for sector, ticker in SECTORS.items(): try: stock = yf.Ticker(ticker) # Add error handling for download hist = stock.history(start=start_datetime, end=end_datetime, interval=interval) if not hist.empty and len(hist['Close']) >= 2: hist = hist.sort_index() latest_close = hist['Close'][-1] oldest_close = hist['Close'][0] performance = ((latest_close - oldest_close) / oldest_close) * 100 color = '#66BB6A' if performance >= 0 else '#EF5350' # Store more detailed data for debugging sector_data.append({ 'Sector': sector, 'Performance': performance, 'Color': color, 'Ticker': ticker, 'Historical Data': hist, 'Data Points': len(hist), 'Start Price': oldest_close, 'End Price': latest_close }) else: st.warning(f"Insufficient data for {sector} ({ticker})") sector_data.append({ 'Sector': sector, 'Performance': 0.0, 'Color': 'gray', 'Ticker': ticker, 'Historical Data': pd.DataFrame(), 'Error': 'Insufficient data' }) except Exception as e: st.warning(f"Error fetching data for {sector} ({ticker}): {str(e)}") sector_data.append({ 'Sector': sector, 'Performance': 0.0, 'Color': 'gray', 'Ticker': ticker, 'Historical Data': pd.DataFrame(), 'Error': str(e) }) # Sort data by performance descending sector_data = [d for d in sector_data if d['Performance'] != 0.0] # Filter out failed fetches sector_data.sort(key=lambda x: x['Performance'], reverse=True) return sector_data def is_market_open(): """ Checks if the NSE market is currently open based on Asia/Kolkata timezone. Returns: bool: True if market is open, False otherwise. """ india_tz = pytz.timezone('Asia/Kolkata') now = datetime.now(india_tz) # Define market hours: 9:15 AM to 3:30 PM IST market_open = now.replace(hour=9, minute=15, second=0, microsecond=0) market_close = now.replace(hour=15, minute=30, second=0, microsecond=0) return market_open <= now <= market_close def get_summary_stats(sector_data): """ Calculate summary statistics from sector data. Parameters: sector_data (list of dict): The sector data. Returns: dict: Summary statistics. """ if not sector_data: return {'positive': 0, 'negative': 0, 'best': None, 'worst': None} positive = len([item for item in sector_data if item['Performance'] > 0]) negative = len(sector_data) - positive best = sector_data[0] if sector_data else None worst = sector_data[-1] if sector_data else None return { 'positive': positive, 'negative': negative, 'best': best, 'worst': worst } def format_performance(performance): """ Format the performance value with a sign and two decimal places. """ return f"{performance:+.2f}%" # --------------------------- # Streamlit UI # --------------------------- def main(): st.set_page_config(page_title="Market Sector Performance Dashboard", layout="wide") st.title("📈 Market Sector Performance Dashboard") # Initialize session state if 'refresh' not in st.session_state: st.session_state.refresh = 0 if 'last_updated' not in st.session_state: st.session_state.last_updated = None # Sidebar controls with st.sidebar: st.header("Controls") # Date Range Selection with improved validation st.markdown("### Select Date Range") india_tz = pytz.timezone('Asia/Kolkata') now = datetime.now(india_tz) default_start = (now - timedelta(days=30)).date() default_end = now.date() start_date = st.date_input("Start Date", default_start) end_date = st.date_input("End Date", default_end) if start_date > end_date: st.error("Error: Start Date must be before End Date.") st.stop() # Warn about long date ranges date_diff = (end_date - start_date).days if date_diff > 365: st.warning("Long date ranges may affect data granularity.") # Determine market status market_status = is_market_open() interval = get_appropriate_interval(start_date, end_date, market_status) st.info(f"Using `{interval}` data interval.") # Refresh Button if st.button("🔄 Refresh Data"): st.session_state.refresh += 1 st.session_state.last_updated = datetime.now(india_tz).strftime("%Y-%m-%d %H:%M:%S") st.cache_data.clear() # Main content try: with st.spinner('Fetching sector data...'): sector_data = fetch_sector_performance(start_date, end_date, interval=interval) if not sector_data: st.warning("No valid sector data available for the selected date range.") st.stop() # Calculate summary stats stats = get_summary_stats(sector_data) # Summary Cards col1, col2, col3, col4 = st.columns(4) col1.metric("Sectors Up", stats['positive'], "") col2.metric("Sectors Down", stats['negative'], "") if stats['best'] and stats['best']['Performance'] != 0.0: col3.metric( "Best Performer", f"{stats['best']['Sector']} ({format_performance(stats['best']['Performance'])})" ) else: col3.metric("Best Performer", "N/A", "") if stats['worst'] and stats['worst']['Performance'] != 0.0: col4.metric( "Worst Performer", f"{stats['worst']['Sector']} ({format_performance(stats['worst']['Performance'])})" ) else: col4.metric("Worst Performer", "N/A", "") st.markdown("---") # Charts fig = make_subplots( rows=1, cols=2, subplot_titles=("Performance by Sector", "Sector Distribution"), specs=[[{"type": "bar"}, {"type": "pie"}]] ) # Bar Chart fig.add_trace( go.Bar( x=[item['Sector'] for item in sector_data], y=[item['Performance'] for item in sector_data], marker_color=[item['Color'] for item in sector_data], name="Performance", text=[format_performance(x) for x in [item['Performance'] for item in sector_data]], textposition='auto' ), row=1, col=1 ) # Pie Chart fig.add_trace( go.Pie( labels=[item['Sector'] for item in sector_data], values=[abs(item['Performance']) for item in sector_data], marker_colors=COLORS[:len(sector_data)], name="Sector Distribution", hoverinfo="label+percent", textinfo="label+percent" ), row=1, col=2 ) fig.update_layout( height=600, showlegend=False, title_text="Sector Performance Overview" ) st.plotly_chart(fig, use_container_width=True) st.markdown("---") # Performance Table # Create DataFrame excluding 'Color' and 'Historical Data' df = pd.DataFrame([{ 'Sector': item['Sector'], 'Performance (%)': format_performance(item['Performance']), 'Ticker': item['Ticker'] } for item in sector_data]) def color_performance(val): """ Apply color formatting based on performance value. """ try: num = float(val.strip('%')) if num > 0: color = 'green' elif num < 0: color = 'red' else: color = 'gray' return f'color: {color}' except: return 'color: black' # Apply styling to the Performance column styled_df = df.style.applymap(color_performance, subset=['Performance (%)']) st.subheader("Sector Performance Table") st.dataframe(styled_df, use_container_width=True) st.markdown(f"**Last Updated:** {st.session_state.last_updated}") # Optional: Display historical data for debugging show_debug = st.checkbox("Show Debugging Information") if show_debug: with st.expander("View Historical Data for Each Sector (For Debugging)"): for item in sector_data: sector = item['Sector'] ticker = item['Ticker'] hist = item.get('Historical Data', pd.DataFrame()) st.write(f"### {sector} ({ticker})") if not hist.empty: # Display the last 10 rows for brevity st.dataframe(hist.tail(10)) else: st.write("No historical data available.") except Exception as e: st.error(f"An unexpected error occurred: {str(e)}") st.markdown(""" **Troubleshooting Steps:** 1. Try selecting a shorter date range. 2. Refresh the page. 3. Check your internet connection. 4. If the issue persists, try again later. """) if __name__ == "__main__": main()