Spaces:
Sleeping
Sleeping
| # At the top of app.py, make sure you have all these imports: | |
| import streamlit as st | |
| import pandas as pd | |
| import numpy as np | |
| from datetime import datetime, timedelta | |
| import plotly.graph_objects as go | |
| import plotly.express as px | |
| from typing import Dict, List, Optional, Tuple | |
| import yfinance as yf | |
| from prophet import Prophet | |
| import tensorflow as tf | |
| from scipy import stats | |
| import seaborn as sns | |
| import ta | |
| from concurrent.futures import ThreadPoolExecutor | |
| import logging | |
| logging.basicConfig(level=logging.INFO) | |
| import joblib | |
| import warnings | |
| warnings.filterwarnings('ignore') | |
| import concurrent.futures | |
| import time | |
| import requests | |
| from data_fetcher import EnhancedDataFetcher | |
| # Import custom modules | |
| from models.trading_models import EnhancedTradingModels | |
| from config import Config | |
| from models.technical_analysis import EnhancedTechnicalAnalysis | |
| from models.portfolio_optimization import PortfolioOptimizer | |
| from models.risk_metrics import RiskMetrics | |
| from models.pattern_recognition import PatternRecognition | |
| class DataProcessor: | |
| def prepare_data(data: pd.DataFrame) -> pd.DataFrame: | |
| """Prepare and validate data for analysis.""" | |
| try: | |
| # Ensure data has the correct columns | |
| required_columns = ['Date', 'Open', 'High', 'Low', 'Close', 'Volume'] | |
| # Reset index if Date is in index | |
| if isinstance(data.index, pd.DatetimeIndex): | |
| data = data.reset_index() | |
| # Rename columns if they're uppercase | |
| data.columns = [col.capitalize() if col.isupper() else col for col in data.columns] | |
| # Ensure Date column is datetime | |
| data['Date'] = pd.to_datetime(data['Date']) | |
| # Convert price columns to float | |
| numeric_columns = ['Open', 'High', 'Low', 'Close', 'Volume'] | |
| for col in numeric_columns: | |
| if col in data.columns: | |
| data[col] = pd.to_numeric(data[col], errors='coerce') | |
| # Forward fill missing values | |
| data = data.fillna(method='ffill') | |
| # Drop any remaining NaN values | |
| data = data.dropna() | |
| return data | |
| except Exception as e: | |
| print(f"Error preparing data: {str(e)}") | |
| return pd.DataFrame() | |
| # Add this to your app.py file | |
| # Key fixes in the DataManager class | |
| class DataManager: | |
| def load_data(ticker: str, start_date: str) -> Optional[pd.DataFrame]: | |
| """Load financial data with improved error handling and retry logic.""" | |
| try: | |
| max_retries = 3 | |
| for attempt in range(max_retries): | |
| try: | |
| # Handle crypto tickers | |
| if 'USD' in ticker: | |
| symbol = ticker | |
| else: | |
| symbol = ticker | |
| # Download data | |
| data = yf.download( | |
| symbol, | |
| start=start_date, | |
| end=datetime.now().strftime('%Y-%m-%d'), | |
| progress=False, | |
| interval='1d' | |
| ) | |
| if data.empty: | |
| print(f"Attempt {attempt + 1}: No data received for {ticker}") | |
| if attempt < max_retries - 1: | |
| time.sleep(2) | |
| continue | |
| return None | |
| # Process the data | |
| data = data.reset_index() | |
| # Standardize column names | |
| data.columns = [col if col == 'Date' else col.title() for col in data.columns] | |
| # Ensure datetime format | |
| data['Date'] = pd.to_datetime(data['Date']) | |
| # Add Adj Close if missing | |
| if 'Adj Close' not in data.columns: | |
| data['Adj Close'] = data['Close'] | |
| # Forward fill any missing values | |
| data = data.fillna(method='ffill').fillna(method='bfill') | |
| return data | |
| except Exception as e: | |
| print(f"Attempt {attempt + 1} failed for {ticker}: {str(e)}") | |
| if attempt < max_retries - 1: | |
| time.sleep(2) | |
| continue | |
| raise e | |
| except Exception as e: | |
| print(f"Failed to load data for {ticker}: {str(e)}") | |
| return None | |
| def validate_data(data: pd.DataFrame) -> bool: | |
| """Validate the loaded data.""" | |
| required_columns = ['Date', 'Open', 'High', 'Low', 'Close', 'Volume'] | |
| # Check for required columns | |
| if not all(col in data.columns for col in required_columns): | |
| print(f"Missing columns. Available columns: {data.columns.tolist()}") | |
| return False | |
| # Check for minimum data points | |
| if len(data) < 30: # Require at least 30 days of data | |
| print(f"Insufficient data points: {len(data)}") | |
| return False | |
| # Check for missing values | |
| if data[required_columns].isnull().any().any(): | |
| print("Data contains missing values") | |
| return False | |
| return True | |
| class AdvancedVisualization: | |
| def plot_candlestick_with_patterns(data: pd.DataFrame, patterns: pd.DataFrame, title: str) -> go.Figure: | |
| """Create advanced candlestick chart with pattern annotations.""" | |
| fig = go.Figure() | |
| # Add candlestick | |
| fig.add_trace(go.Candlestick( | |
| x=data['Date'], | |
| open=data['Open'], | |
| high=data['High'], | |
| low=data['Low'], | |
| close=data['Close'], | |
| name='Price' | |
| )) | |
| # Add patterns | |
| for pattern in ['Double_Top', 'Double_Bottom', 'Head_And_Shoulders', 'Triangle', 'Channel']: | |
| if pattern in patterns.columns: | |
| pattern_points = patterns[patterns[pattern] == 1] | |
| if not pattern_points.empty: | |
| fig.add_trace(go.Scatter( | |
| x=pattern_points['Date'], | |
| y=pattern_points['High'], | |
| mode='markers', | |
| name=pattern, | |
| marker=dict(size=10, symbol='triangle-down') | |
| )) | |
| # Update layout | |
| fig.update_layout( | |
| title=title, | |
| yaxis_title='Price', | |
| template='plotly_dark' if st.session_state.theme == 'dark' else 'plotly_white', | |
| xaxis_rangeslider_visible=False, | |
| height=600 | |
| ) | |
| return fig | |
| def plot_correlation_matrix(returns: pd.DataFrame) -> go.Figure: | |
| """Create correlation matrix heatmap.""" | |
| corr = returns.corr() | |
| fig = px.imshow( | |
| corr, | |
| labels=dict(color="Correlation"), | |
| x=corr.columns, | |
| y=corr.columns, | |
| color_continuous_scale='RdBu' | |
| ) | |
| fig.update_layout( | |
| title='Asset Correlation Matrix', | |
| template='plotly_dark' if st.session_state.theme == 'dark' else 'plotly_white' | |
| ) | |
| return fig | |
| class TradingStrategy: | |
| def __init__(self, data: pd.DataFrame): | |
| self.data = DataProcessor.prepare_data(data.copy()) | |
| self.signals = pd.DataFrame(index=self.data.index) | |
| def generate_signals(self) -> pd.DataFrame: | |
| """Generate trading signals based on multiple strategies.""" | |
| try: | |
| # Calculate technical indicators | |
| self.data['SMA20'] = self.data['Close'].rolling(window=20).mean() | |
| self.data['SMA50'] = self.data['Close'].rolling(window=50).mean() | |
| self.data['RSI'] = self._calculate_rsi() | |
| # Generate individual signals | |
| self.signals['MA_Cross'] = self._moving_average_crossover() | |
| self.signals['RSI'] = self._rsi_signals() | |
| self.signals['Trend'] = self._trend_signals() | |
| # Combine signals | |
| self.signals['Position'] = 0 | |
| # Generate combined signals | |
| buy_condition = ( | |
| (self.signals['MA_Cross'] == 1) & | |
| (self.signals['RSI'] == 1) | | |
| (self.signals['Trend'] == 1) | |
| ) | |
| sell_condition = ( | |
| (self.signals['MA_Cross'] == -1) & | |
| (self.signals['RSI'] == -1) | | |
| (self.signals['Trend'] == -1) | |
| ) | |
| self.signals.loc[buy_condition, 'Position'] = 1 | |
| self.signals.loc[sell_condition, 'Position'] = -1 | |
| return self.signals | |
| except Exception as e: | |
| print(f"Error generating signals: {str(e)}") | |
| return pd.DataFrame() | |
| def _calculate_rsi(self, periods=14): | |
| """Calculate RSI indicator.""" | |
| delta = self.data['Close'].diff() | |
| gain = (delta.where(delta > 0, 0)).rolling(window=periods).mean() | |
| loss = (-delta.where(delta < 0, 0)).rolling(window=periods).mean() | |
| rs = gain / loss | |
| rsi = 100 - (100 / (1 + rs)) | |
| return rsi | |
| def _moving_average_crossover(self) -> pd.Series: | |
| """Generate signals based on MA crossover.""" | |
| signals = pd.Series(0, index=self.data.index) | |
| signals.loc[self.data['SMA20'] > self.data['SMA50']] = 1 | |
| signals.loc[self.data['SMA20'] < self.data['SMA50']] = -1 | |
| return signals | |
| def _rsi_signals(self) -> pd.Series: | |
| """Generate signals based on RSI.""" | |
| signals = pd.Series(0, index=self.data.index) | |
| signals.loc[self.data['RSI'] < 30] = 1 # Oversold | |
| signals.loc[self.data['RSI'] > 70] = -1 # Overbought | |
| return signals | |
| def _trend_signals(self) -> pd.Series: | |
| """Generate signals based on trend.""" | |
| signals = pd.Series(0, index=self.data.index) | |
| price_sma = self.data['Close'] > self.data['SMA50'] | |
| signals.loc[price_sma] = 1 | |
| signals.loc[~price_sma] = -1 | |
| return signals | |
| # In the TradingStrategy class, update the _bollinger_signals method: | |
| def _bollinger_signals(self) -> pd.Series: | |
| """Generate signals based on Bollinger Bands.""" | |
| # Calculate Bollinger Bands using ta library | |
| indicator_bb = ta.volatility.BollingerBands(close=self.data['Close'], window=20, window_dev=2) | |
| upper = indicator_bb.bollinger_hband() | |
| lower = indicator_bb.bollinger_lband() | |
| signals = pd.Series(0, index=self.data.index) | |
| signals[self.data['Close'] > upper] = -1 # Sell signal | |
| signals[self.data['Close'] < lower] = 1 # Buy signal | |
| return signals | |
| def _pattern_signals(self) -> pd.Series: | |
| """Generate signals based on pattern recognition.""" | |
| pattern_recognition = PatternRecognition(self.data) | |
| patterns = pattern_recognition.detect_patterns() | |
| signals = pd.Series(0, index=self.data.index) | |
| signals[patterns['Double_Top'] == 1] = -1 | |
| signals[patterns['Double_Bottom'] == 1] = 1 | |
| return signals | |
| def initialize_session_state(): | |
| """Initialize session state variables.""" | |
| if 'theme' not in st.session_state: | |
| st.session_state.theme = Config.THEME | |
| if 'selected_assets' not in st.session_state: | |
| st.session_state.selected_assets = [] | |
| if 'optimization_results' not in st.session_state: | |
| st.session_state.optimization_results = None | |
| def render_sidebar() -> Tuple[str, str, List[str]]: | |
| """Render enhanced sidebar with additional controls.""" | |
| st.sidebar.title("โ๏ธ Configuration") | |
| # Theme selection | |
| theme = st.sidebar.selectbox( | |
| "Theme", | |
| ["light", "dark"], | |
| key="theme" | |
| ) | |
| # Asset selection | |
| asset_type = st.sidebar.radio("Select Asset Type", list(Config.ASSETS.keys())) | |
| # Multi-asset selection | |
| selected_assets = st.sidebar.multiselect( | |
| 'Select Assets for Analysis', | |
| options=list(Config.ASSETS[asset_type].keys()), | |
| format_func=lambda x: f"{x} ({Config.ASSETS[asset_type][x]})" | |
| ) | |
| # Analysis Options | |
| st.sidebar.subheader("Analysis Options") | |
| st.sidebar.checkbox("Technical Analysis", value=True) | |
| st.sidebar.checkbox("Pattern Recognition", value=True) | |
| st.sidebar.checkbox("Portfolio Optimization", value=True) | |
| st.sidebar.checkbox("Risk Metrics", value=True) | |
| return theme, asset_type, selected_assets | |
| # Add this function before the main() function: | |
| def plot_signals(data: pd.DataFrame, signals: pd.DataFrame, asset: str) -> go.Figure: | |
| """Plot trading signals on candlestick chart.""" | |
| fig = go.Figure() | |
| # Add candlestick | |
| fig.add_trace(go.Candlestick( | |
| x=data['Date'], | |
| open=data['Open'], | |
| high=data['High'], | |
| low=data['Low'], | |
| close=data['Close'], | |
| name='Price' | |
| )) | |
| # Add buy signals | |
| buy_points = data[signals['Position'] == 1] | |
| if not buy_points.empty: | |
| fig.add_trace(go.Scatter( | |
| x=buy_points['Date'], | |
| y=buy_points['High'], | |
| mode='markers', | |
| name='Buy Signal', | |
| marker=dict( | |
| symbol='triangle-up', | |
| size=15, | |
| color='green', | |
| ) | |
| )) | |
| # Add sell signals | |
| sell_points = data[signals['Position'] == -1] | |
| if not sell_points.empty: | |
| fig.add_trace(go.Scatter( | |
| x=sell_points['Date'], | |
| y=sell_points['Low'], | |
| mode='markers', | |
| name='Sell Signal', | |
| marker=dict( | |
| symbol='triangle-down', | |
| size=15, | |
| color='red', | |
| ) | |
| )) | |
| # Update layout | |
| fig.update_layout( | |
| title=f"{asset} Trading Signals", | |
| yaxis_title='Price', | |
| xaxis_title='Date', | |
| template='plotly_dark' if st.session_state.theme == 'dark' else 'plotly_white', | |
| xaxis_rangeslider_visible=False, | |
| height=600 | |
| ) | |
| return fig | |
| def main(): | |
| data_manager = DataManager() | |
| # Initialize session state | |
| initialize_session_state() | |
| # Page config | |
| st.set_page_config( | |
| page_title=Config.APP_TITLE, | |
| page_icon=Config.APP_ICON, | |
| layout="wide" | |
| ) | |
| # Custom CSS | |
| st.markdown(""" | |
| <style> | |
| .stButton>button { | |
| width: 100%; | |
| } | |
| .reportview-container { | |
| background: var(--background-color) | |
| } | |
| .sidebar .sidebar-content { | |
| background: var(--background-color) | |
| } | |
| </style> | |
| """, unsafe_allow_html=True) | |
| # Sidebar | |
| theme, asset_type, selected_assets = render_sidebar() | |
| # Main content | |
| st.title(f"{Config.APP_TITLE} - Advanced Market Analysis") | |
| if not selected_assets: | |
| st.warning("Please select at least one asset to analyze.") | |
| return | |
| failed_assets = [] | |
| # Tabs for different analyses | |
| tabs = st.tabs([ | |
| "๐ Technical Analysis", | |
| "๐ฏ Pattern Recognition", | |
| "๐ผ Portfolio Optimization", | |
| "โ ๏ธ Risk Analysis", | |
| "๐ Trading Signals" | |
| ]) | |
| # Load data for selected assets | |
| with st.spinner('Loading data...'): | |
| data_dict = {} | |
| returns_dict = {} | |
| failed_assets = [] | |
| # Use shorter timeframe for initial load | |
| start_date = (datetime.now() - timedelta(days=365)).strftime('%Y-%m-%d') | |
| for asset in selected_assets: | |
| try: | |
| st.text(f"Loading data for {asset}...") # Debug output | |
| # Get the data | |
| data = data_manager.load_data(asset, start_date) | |
| if data is not None and not data.empty: | |
| # Basic validation | |
| required_cols = ['Date', 'Open', 'High', 'Low', 'Close', 'Volume'] | |
| if all(col in data.columns for col in required_cols): | |
| data_dict[asset] = data | |
| returns_dict[asset] = data['Close'].pct_change().dropna() | |
| st.success(f"Successfully loaded data for {asset}") | |
| else: | |
| failed_assets.append(asset) | |
| st.error(f"Missing required columns for {asset}") | |
| else: | |
| failed_assets.append(asset) | |
| st.error(f"No data received for {asset}") | |
| except Exception as e: | |
| failed_assets.append(asset) | |
| st.error(f"Error loading data for {asset}: {str(e)}") | |
| continue | |
| if not data_dict: | |
| st.error("No data available for analysis. Please try different assets or check your internet connection.") | |
| return | |
| # Portfolio Optimization Tab - FIX 5: Add proper error handling and data validation | |
| def handle_portfolio_optimization(tabs, data_dict, returns_dict): | |
| with tabs[2]: | |
| st.subheader("Portfolio Optimization") | |
| if len(data_dict) < 2: | |
| st.warning("Please select at least 2 assets for portfolio optimization.") | |
| return | |
| try: | |
| # Create returns DataFrame with proper index | |
| returns_df = pd.DataFrame(returns_dict) | |
| # Initialize optimizer | |
| optimizer = PortfolioOptimizer(returns_df) | |
| # Calculate reasonable bounds for target return | |
| min_return = max(returns_df.mean().min() * 252, -0.5) # Limit to -50% | |
| max_return = min(returns_df.mean().max() * 252, 0.5) # Limit to +50% | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| target_return = st.slider( | |
| "Target Annual Return (%)", | |
| min_value=int(min_return * 100), | |
| max_value=int(max_return * 100), | |
| value=10 | |
| ) / 100 | |
| # Run optimization | |
| optimization_results = optimizer.optimize_portfolio(target_return) | |
| if optimization_results['success']: | |
| st.success("Portfolio optimization completed successfully!") | |
| # Display optimization results | |
| st.subheader("Optimal Portfolio Allocation") | |
| # Plot allocation pie chart | |
| fig = px.pie( | |
| values=list(optimization_results['weights'].values()), | |
| names=list(optimization_results['weights'].keys()), | |
| title="Optimal Portfolio Weights" | |
| ) | |
| st.plotly_chart(fig, use_container_width=True) | |
| # Display metrics | |
| col1, col2, col3 = st.columns(3) | |
| col1.metric("Expected Annual Return", f"{optimization_results['return']:.2%}") | |
| col2.metric("Annual Volatility", f"{optimization_results['volatility']:.2%}") | |
| col3.metric("Sharpe Ratio", f"{optimization_results['sharpe_ratio']:.2f}") | |
| else: | |
| st.warning(f"Optimization failed: {optimization_results['message']}") | |
| except Exception as e: | |
| st.error(f"Portfolio optimization error: {str(e)}") | |
| st.info("Try selecting different assets or adjusting the target return.") | |
| # Trading Signals Tab - FIX 6: Add proper signal generation | |
| # In your main() function, replace the trading signals tab section with: | |
| # Inside the `Trading Signals` Tab | |
| with tabs[4]: | |
| st.subheader("Trading Signals") | |
| for asset, data in data_dict.items(): | |
| try: | |
| st.subheader(f"Trading Signals - {asset}") | |
| # Use EnhancedTradingModels to generate signals and metrics | |
| model = EnhancedTradingModels(data) | |
| signals, metrics = model.generate_signals() | |
| if signals.empty: | |
| st.warning(f"No trading signals generated for {asset}") | |
| continue | |
| # Plot signals | |
| fig = plot_signals(data, signals, asset) | |
| st.plotly_chart(fig, use_container_width=True) | |
| # Display signal statistics | |
| st.subheader("Signal Metrics") | |
| metrics_df = pd.DataFrame.from_dict(metrics, orient='index', columns=['Value']) | |
| st.dataframe(metrics_df) | |
| except Exception as e: | |
| st.error(f"Error analyzing {asset}: {str(e)}") | |
| # Footer | |
| st.markdown(""" | |
| --- | |
| ### Trading Guru Pro - Advanced Market Analysis Platform | |
| Features: | |
| - Real-time technical analysis | |
| - Pattern recognition with machine learning | |
| - Portfolio optimization | |
| - Risk metrics and analysis | |
| - Automated trading signals | |
| Created by Muhammad Shaheer | |
| # Continuing main.py... | |
| ๐ง For support and feature requests: [Your Contact Info] | |
| *Disclaimer: This tool is for educational purposes only. Always conduct your own research before making investment decisions.* | |
| """) | |
| # Add helpful tooltips | |
| st.sidebar.markdown(""" | |
| ### ๐ Analysis Guide | |
| #### Technical Indicators | |
| - **RSI**: Measures momentum (Oversold < 30, Overbought > 70) | |
| - **MACD**: Shows trend direction and momentum | |
| - **Bollinger Bands**: Indicates volatility and potential reversals | |
| #### Pattern Recognition | |
| - **Double Top/Bottom**: Potential reversal patterns | |
| - **Head & Shoulders**: Major reversal pattern | |
| - **Triangles**: Continuation or reversal patterns | |
| #### Portfolio Metrics | |
| - **Sharpe Ratio**: Risk-adjusted return measure | |
| - **Max Drawdown**: Largest peak-to-trough decline | |
| - **VaR**: Potential loss in value | |
| #### Trading Signals | |
| - **Green Triangle**: Buy signal | |
| - **Red Triangle**: Sell signal | |
| *Remember to use multiple indicators for confirmation* | |
| """) | |
| class BacktestEngine: | |
| """Backtesting engine for trading strategies.""" | |
| def __init__(self, data: pd.DataFrame, initial_capital: float = 100000): | |
| self.data = data | |
| self.initial_capital = initial_capital | |
| self.positions = pd.DataFrame(index=data.index).fillna(0) | |
| self.portfolio_value = pd.DataFrame(index=data.index).fillna(0) | |
| def run_backtest(self, signals: pd.DataFrame) -> Dict: | |
| """Run backtest simulation.""" | |
| # Initialize portfolio | |
| portfolio = self.initial_capital | |
| position = 0 | |
| trades = [] | |
| for i in range(len(self.data)): | |
| if signals['Position'].iloc[i] == 1 and position == 0: # Buy signal | |
| shares = portfolio / self.data['Close'].iloc[i] | |
| position = shares | |
| trades.append({ | |
| 'date': self.data['Date'].iloc[i], | |
| 'type': 'buy', | |
| 'price': self.data['Close'].iloc[i], | |
| 'shares': shares, | |
| 'value': portfolio | |
| }) | |
| elif signals['Position'].iloc[i] == -1 and position > 0: # Sell signal | |
| portfolio = position * self.data['Close'].iloc[i] | |
| position = 0 | |
| trades.append({ | |
| 'date': self.data['Date'].iloc[i], | |
| 'type': 'sell', | |
| 'price': self.data['Close'].iloc[i], | |
| 'shares': shares, | |
| 'value': portfolio | |
| }) | |
| # Update portfolio value | |
| self.portfolio_value.iloc[i] = portfolio if position == 0 else position * self.data['Close'].iloc[i] | |
| # Calculate performance metrics | |
| returns = self.portfolio_value.pct_change().dropna() | |
| metrics = { | |
| 'total_return': (self.portfolio_value.iloc[-1] - self.initial_capital) / self.initial_capital, | |
| 'annual_return': returns.mean() * 252, | |
| 'volatility': returns.std() * np.sqrt(252), | |
| 'sharpe_ratio': (returns.mean() - Config.RISK_FREE_RATE/252) / (returns.std()) * np.sqrt(252), | |
| 'max_drawdown': (self.portfolio_value / self.portfolio_value.cummax() - 1).min(), | |
| 'trades': trades | |
| } | |
| return metrics | |
| class MarketSentiment: | |
| """Analyze market sentiment using various indicators.""" | |
| def calculate_sentiment_score(data: pd.DataFrame) -> float: | |
| """Calculate overall market sentiment score.""" | |
| # Technical indicators sentiment | |
| rsi = ta.momentum.RSIIndicator(data['Close']).rsi().iloc[-1] | |
| macd = ta.trend.MACD(data['Close']).macd().iloc[-1] | |
| # Volume sentiment | |
| volume_ma = data['Volume'].rolling(window=20).mean().iloc[-1] | |
| current_volume = data['Volume'].iloc[-1] | |
| volume_sentiment = 1 if current_volume > volume_ma else -1 | |
| # Trend sentiment | |
| sma_20 = data['Close'].rolling(window=20).mean().iloc[-1] | |
| sma_50 = data['Close'].rolling(window=50).mean().iloc[-1] | |
| trend_sentiment = 1 if sma_20 > sma_50 else -1 | |
| # Combine sentiments | |
| technical_sentiment = ( | |
| (1 if 40 < rsi < 60 else -1) + | |
| (1 if macd > 0 else -1) + | |
| volume_sentiment + | |
| trend_sentiment | |
| ) / 4 | |
| return technical_sentiment | |
| def add_market_sentiment_analysis(data_dict: Dict[str, pd.DataFrame]) -> None: | |
| """Add market sentiment analysis to the app.""" | |
| st.subheader("๐ Market Sentiment Analysis") | |
| sentiment_scores = {} | |
| for asset, data in data_dict.items(): | |
| sentiment_score = MarketSentiment.calculate_sentiment_score(data) | |
| sentiment_scores[asset] = sentiment_score | |
| # Create sentiment gauge charts | |
| cols = st.columns(len(sentiment_scores)) | |
| for i, (asset, score) in enumerate(sentiment_scores.items()): | |
| with cols[i]: | |
| fig = go.Figure(go.Indicator( | |
| mode="gauge+number", | |
| value=score * 100, | |
| domain={'x': [0, 1], 'y': [0, 1]}, | |
| gauge={ | |
| 'axis': {'range': [-100, 100]}, | |
| 'bar': {'color': "darkblue"}, | |
| 'steps': [ | |
| {'range': [-100, -33], 'color': "red"}, | |
| {'range': [-33, 33], 'color': "gray"}, | |
| {'range': [33, 100], 'color': "green"} | |
| ], | |
| } | |
| )) | |
| fig.update_layout( | |
| title=f"{asset} Sentiment", | |
| height=250 | |
| ) | |
| st.plotly_chart(fig, use_container_width=True) | |
| def add_portfolio_metrics(returns_dict: Dict[str, pd.Series]) -> None: | |
| """Add detailed portfolio metrics analysis.""" | |
| st.subheader("๐ Portfolio Performance Metrics") | |
| # Calculate metrics for each asset | |
| metrics_dict = {} | |
| for asset, returns in returns_dict.items(): | |
| metrics = { | |
| 'Annual Return': returns.mean() * 252, | |
| 'Volatility': returns.std() * np.sqrt(252), | |
| 'Sharpe Ratio': (returns.mean() * 252 - Config.RISK_FREE_RATE) / (returns.std() * np.sqrt(252)), | |
| 'Sortino Ratio': (returns.mean() * 252 - Config.RISK_FREE_RATE) / (returns[returns < 0].std() * np.sqrt(252)), | |
| 'Max Drawdown': (returns.cumsum() - returns.cumsum().cummax()).min(), | |
| 'Win Rate': len(returns[returns > 0]) / len(returns) | |
| } | |
| metrics_dict[asset] = metrics | |
| # Create metrics table | |
| metrics_df = pd.DataFrame(metrics_dict).T | |
| # Style the dataframe | |
| styled_df = metrics_df.style.format({ | |
| 'Annual Return': '{:.2%}', | |
| 'Volatility': '{:.2%}', | |
| 'Sharpe Ratio': '{:.2f}', | |
| 'Sortino Ratio': '{:.2f}', | |
| 'Max Drawdown': '{:.2%}', | |
| 'Win Rate': '{:.2%}' | |
| }).background_gradient(cmap='RdYlGn') | |
| st.dataframe(styled_df) | |
| if __name__ == "__main__": | |
| main() |