import streamlit as st import pandas as pd import numpy as np import matplotlib.pyplot as plt import datetime from dateutil.relativedelta import relativedelta import plotly.express as px import plotly.graph_objects as go import plotly.figure_factory as ff from plotly.subplots import make_subplots import yfinance as yf import seaborn as sns from scipy import stats from typing import Dict, Optional, List import warnings warnings.filterwarnings('ignore') # Try importing mftool, handle if not available try: from mftool import Mftool mftool_available = True except ImportError: mftool_available = False try: from yahooquery import Ticker yahooquery_available = True except ImportError: yahooquery_available = False # Set page configuration st.set_page_config( page_title="Mutual Fund Analytics Suite", page_icon="📈", layout="wide", initial_sidebar_state="expanded" ) # Custom CSS styling st.markdown(""" """, unsafe_allow_html=True) # Cache data fetching functions @st.cache_data(ttl=3600) def fetch_mutual_fund_data(mutual_fund_code: str) -> Optional[pd.DataFrame]: """Fetch mutual fund data from mftool.""" try: mf = Mftool() df = (mf.get_scheme_historical_nav(mutual_fund_code, as_Dataframe=True) .reset_index() .assign(nav=lambda x: x['nav'].astype(float), date=lambda x: pd.to_datetime(x['date'], format='%d-%m-%Y')) .sort_values('date') .reset_index(drop=True)) return df except Exception as e: st.error(f"Error fetching mutual fund data: {str(e)}") return None @st.cache_data(ttl=3600) def load_yahoo_finance_data(ticker_symbol: str, start_date: datetime.date, end_date: datetime.date) -> Optional[pd.DataFrame]: """Fetch data from Yahoo Finance.""" try: data = yf.download(ticker_symbol, start=start_date, end=end_date) data = data.reset_index() data = data.rename(columns={'Date': 'date', 'Close': 'nav', 'Volume': 'volume'}) return data except Exception as e: st.error(f"Error fetching Yahoo Finance data: {str(e)}") return None def calculate_risk_metrics(returns: pd.Series) -> Dict[str, float]: """Calculate comprehensive risk metrics for the fund.""" try: metrics = { 'volatility': returns.std() * np.sqrt(252), 'sharpe_ratio': (returns.mean() * 252) / (returns.std() * np.sqrt(252)), 'sortino_ratio': (returns.mean() * 252) / (returns[returns < 0].std() * np.sqrt(252)), 'max_drawdown': (1 - (1 + returns).cumprod() / (1 + returns).cumprod().cummax()).max(), 'skewness': stats.skew(returns), 'kurtosis': stats.kurtosis(returns), 'var_95': np.percentile(returns, 5), 'cvar_95': returns[returns <= np.percentile(returns, 5)].mean(), 'positive_days': (returns > 0).mean() * 100, 'negative_days': (returns < 0).mean() * 100, 'avg_gain': returns[returns > 0].mean(), 'avg_loss': returns[returns < 0].mean() } return metrics except Exception as e: st.error(f"Error calculating risk metrics: {str(e)}") return {} def plot_price_volume_chart(df: pd.DataFrame) -> go.Figure: """Create an interactive price and volume chart.""" try: fig = make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.03, row_heights=[0.7, 0.3]) fig.add_trace(go.Candlestick(x=df['date'], open=df['Open'], high=df['High'], low=df['Low'], close=df['nav'], name='Price'), row=1, col=1) fig.add_trace(go.Bar(x=df['date'], y=df['volume'], name='Volume'), row=2, col=1) fig.update_layout( title='Price and Volume Analysis', yaxis_title='Price', yaxis2_title='Volume', height=800, template='plotly_white' ) return fig except Exception as e: st.error(f"Error creating price-volume chart: {str(e)}") return None def plot_returns_distribution(returns: pd.Series) -> go.Figure: """Create an interactive returns distribution plot.""" try: fig = go.Figure() # Actual returns distribution fig.add_trace(go.Histogram( x=returns, name='Actual Returns', nbinsx=50, histnorm='probability' )) # Normal distribution overlay x_range = np.linspace(returns.min(), returns.max(), 100) normal_dist = stats.norm.pdf(x_range, returns.mean(), returns.std()) fig.add_trace(go.Scatter( x=x_range, y=normal_dist, name='Normal Distribution', line=dict(color='red') )) fig.update_layout( title='Returns Distribution Analysis', xaxis_title='Returns', yaxis_title='Probability', barmode='overlay', showlegend=True, template='plotly_white' ) return fig except Exception as e: st.error(f"Error creating returns distribution plot: {str(e)}") return None def plot_rolling_metrics(df: pd.DataFrame, window: int = 30) -> go.Figure: """Create rolling metrics visualization with confidence bands.""" try: rolling_returns = df['daily_returns'].rolling(window=window) rolling_vol = rolling_returns.std() * np.sqrt(252) rolling_mean = rolling_returns.mean() * 252 rolling_sharpe = rolling_mean / (rolling_returns.std() * np.sqrt(252)) fig = go.Figure() # Add rolling volatility with confidence bands vol_std = rolling_vol.std() fig.add_trace(go.Scatter( x=df['date'], y=rolling_vol + 2*vol_std, fill=None, mode='lines', line_color='rgba(0,100,80,0.2)', name='Volatility Upper Band' )) fig.add_trace(go.Scatter( x=df['date'], y=rolling_vol - 2*vol_std, fill='tonexty', mode='lines', line_color='rgba(0,100,80,0.2)', name='Volatility Lower Band' )) fig.add_trace(go.Scatter( x=df['date'], y=rolling_vol, name='Rolling Volatility', line=dict(color='rgb(0,100,80)') )) fig.add_trace(go.Scatter( x=df['date'], y=rolling_sharpe, name='Rolling Sharpe Ratio', yaxis='y2', line=dict(color='rgb(200,30,30)') )) fig.update_layout( title=f'Rolling Metrics (Window: {window} days)', yaxis=dict(title='Annualized Volatility'), yaxis2=dict(title='Sharpe Ratio', overlaying='y', side='right'), showlegend=True, height=600, template='plotly_white' ) return fig except Exception as e: st.error(f"Error creating rolling metrics plot: {str(e)}") return None def plot_comparative_analysis(dfs: Dict[str, pd.DataFrame]) -> List[go.Figure]: """Create comparative analysis plots.""" try: # Normalize all fund values to 100 normalized_dfs = {} for name, df in dfs.items(): normalized_dfs[name] = df.copy() normalized_dfs[name]['normalized_nav'] = df['nav'] / df['nav'].iloc[0] * 100 # Create comparative performance plot perf_fig = go.Figure() for name, df in normalized_dfs.items(): perf_fig.add_trace(go.Scatter( x=df['date'], y=df['normalized_nav'], name=name, mode='lines' )) perf_fig.update_layout( title='Comparative Performance Analysis', xaxis_title='Date', yaxis_title='Normalized Value (Base=100)', template='plotly_white' ) # Create correlation heatmap returns_df = pd.DataFrame() for name, df in dfs.items(): returns_df[name] = df['nav'].pct_change() corr_matrix = returns_df.corr() corr_fig = go.Figure(data=go.Heatmap( z=corr_matrix, x=corr_matrix.columns, y=corr_matrix.columns, colorscale='RdBu', zmin=-1, zmax=1 )) corr_fig.update_layout( title='Returns Correlation Matrix', template='plotly_white' ) return [perf_fig, corr_fig] except Exception as e: st.error(f"Error creating comparative analysis plots: {str(e)}") return [] def plot_risk_analytics(df: pd.DataFrame) -> List[go.Figure]: """Create risk analytics plots.""" try: returns = df['nav'].pct_change() # Create drawdown plot cum_returns = (1 + returns).cumprod() rolling_max = cum_returns.cummax() drawdowns = (cum_returns - rolling_max) / rolling_max drawdown_fig = go.Figure() drawdown_fig.add_trace(go.Scatter( x=df['date'], y=drawdowns, fill='tozeroy', name='Drawdown' )) drawdown_fig.update_layout( title='Historical Drawdown Analysis', xaxis_title='Date', yaxis_title='Drawdown', template='plotly_white' ) # Create risk-return scatter plot rolling_windows = [30, 60, 90, 180, 252] risk_return_data = [] for window in rolling_windows: rolling_returns = returns.rolling(window=window) risk = rolling_returns.std() * np.sqrt(252) ret = rolling_returns.mean() * 252 risk_return_data.append({ 'window': f'{window} days', 'risk': risk.mean(), 'return': ret.mean() }) risk_return_df = pd.DataFrame(risk_return_data) risk_return_fig = px.scatter( risk_return_df, x='risk', y='return', text='window', title='Risk-Return Analysis Across Different Time Windows' ) risk_return_fig.update_traces(textposition='top center') risk_return_fig.update_layout(template='plotly_white') return [drawdown_fig, risk_return_fig] except Exception as e: st.error(f"Error creating risk analytics plots: {str(e)}") return [] def main(): st.title("📊 Advanced Mutual Fund Analytics Platform") st.markdown(""" ### Professional-Grade Investment Analysis Tool This platform provides comprehensive mutual fund analytics with advanced risk metrics, interactive visualizations, and comparative analysis capabilities. """) # Sidebar controls st.sidebar.header("Analysis Controls") analysis_type = st.sidebar.selectbox( "Select Analysis Type", ["Single Fund Analysis", "Comparative Analysis", "Risk Analytics"] ) # Date range selection col1, col2 = st.sidebar.columns(2) with col1: start_date = st.date_input( "Start Date", datetime.date.today() - relativedelta(years=3) ) with col2: end_date = st.date_input( "End Date", datetime.date.today() ) if analysis_type == "Single Fund Analysis": st.header("Single Fund Analysis") input_type = st.radio( "Select Input Type", ["Yahoo Finance Ticker", "Mutual Fund Code (Indian)"] ) if input_type == "Yahoo Finance Ticker": fund_id = st.text_input("Enter Yahoo Finance Ticker", "0P0000XW8F.BO") if st.button("Analyze Fund"): with st.spinner("Fetching and analyzing data..."): df = load_yahoo_finance_data(fund_id, start_date, end_date) if df is not None: df['daily_returns'] = df['nav'].pct_change() metrics = calculate_risk_metrics(df['daily_returns'].dropna()) # Display metrics in a clean format col1, col2, col3, col4 = st.columns(4) with col1: st.metric("Annualized Volatility", f"{metrics['volatility']:.2%}") st.metric("Sharpe Ratio", f"{metrics['sharpe_ratio']:.2f}") with col2: st.metric("Maximum Drawdown", f"{metrics['max_drawdown']:.2%}") st.metric("Value at Risk (95%)", f"{metrics['var_95']:.2%}") with col3: st.metric("Positive Days", f"{metrics['positive_days']:.1f}%") st.metric("Average Daily Gain", f"{metrics['avg_gain']:.2%}") with col4: st.metric("Negative Days", f"{metrics['negative_days']:.1f}%") st.metric("Average Daily Loss", f"{metrics['avg_loss']:.2%}") # Create tabs for different visualizations tab1, tab2, tab3 = st.tabs(["Price Analysis", "Returns Analysis", "Risk Metrics"]) with tab1: if 'Open' in df.columns: price_vol_fig = plot_price_volume_chart(df) if price_vol_fig: st.plotly_chart(price_vol_fig, use_container_width=True) with tab2: returns_dist_fig = plot_returns_distribution(df['daily_returns'].dropna()) if returns_dist_fig: st.plotly_chart(returns_dist_fig, use_container_width=True) with tab3: window = st.slider("Rolling Window (days)", 10, 252, 30) rolling_fig = plot_rolling_metrics(df, window) if rolling_fig: st.plotly_chart(rolling_fig, use_container_width=True) else: fund_code = st.text_input("Enter Mutual Fund Code", "118989") if st.button("Analyze Fund"): with st.spinner("Fetching and analyzing data..."): df = fetch_mutual_fund_data(fund_code) if df is not None: df['daily_returns'] = df['nav'].pct_change() # Perform the same analysis as above metrics = calculate_risk_metrics(df['daily_returns'].dropna()) # Display metrics and charts (same as above) col1, col2, col3, col4 = st.columns(4) with col1: st.metric("Annualized Volatility", f"{metrics['volatility']:.2%}") st.metric("Sharpe Ratio", f"{metrics['sharpe_ratio']:.2f}") with col2: st.metric("Maximum Drawdown", f"{metrics['max_drawdown']:.2%}") st.metric("Value at Risk (95%)", f"{metrics['var_95']:.2%}") with col3: st.metric("Positive Days", f"{metrics['positive_days']:.1f}%") st.metric("Average Daily Gain", f"{metrics['avg_gain']:.2%}") with col4: st.metric("Negative Days", f"{metrics['negative_days']:.1f}%") st.metric("Average Daily Loss", f"{metrics['avg_loss']:.2%}") tab1, tab2 = st.tabs(["Returns Analysis", "Risk Metrics"]) with tab1: returns_dist_fig = plot_returns_distribution(df['daily_returns'].dropna()) if returns_dist_fig: st.plotly_chart(returns_dist_fig, use_container_width=True) with tab2: window = st.slider("Rolling Window (days)", 10, 252, 30) rolling_fig = plot_rolling_metrics(df, window) if rolling_fig: st.plotly_chart(rolling_fig, use_container_width=True) elif analysis_type == "Comparative Analysis": st.header("Comparative Analysis") num_funds = st.number_input("Number of funds to compare", min_value=2, max_value=5, value=2) funds_data = {} for i in range(num_funds): st.subheader(f"Fund {i + 1}") input_type = st.radio( f"Select Input Type for Fund {i + 1}", ["Yahoo Finance Ticker", "Mutual Fund Code (Indian)"], key=f"input_type_{i}" ) if input_type == "Yahoo Finance Ticker": fund_id = st.text_input(f"Enter Yahoo Finance Ticker {i + 1}", value=f"0P0000XW8F.BO" if i == 0 else "", key=f"yahoo_{i}") fund_name = st.text_input(f"Enter Fund Name {i + 1}", value=f"Fund {i + 1}", key=f"name_{i}") funds_data[fund_name] = {'id': fund_id, 'type': 'yahoo'} else: fund_id = st.text_input(f"Enter Mutual Fund Code {i + 1}", value="118989" if i == 0 else "", key=f"mf_{i}") fund_name = st.text_input(f"Enter Fund Name {i + 1}", value=f"Fund {i + 1}", key=f"name_{i}") funds_data[fund_name] = {'id': fund_id, 'type': 'mf'} if st.button("Compare Funds"): with st.spinner("Fetching and comparing data..."): dfs = {} for name, info in funds_data.items(): if info['type'] == 'yahoo': df = load_yahoo_finance_data(info['id'], start_date, end_date) else: df = fetch_mutual_fund_data(info['id']) if df is not None: dfs[name] = df if len(dfs) > 1: comparison_figs = plot_comparative_analysis(dfs) if comparison_figs: st.subheader("Comparative Performance") st.plotly_chart(comparison_figs[0], use_container_width=True) st.subheader("Correlation Analysis") st.plotly_chart(comparison_figs[1], use_container_width=True) else: # Risk Analytics st.header("Risk Analytics") input_type = st.radio( "Select Input Type", ["Yahoo Finance Ticker", "Mutual Fund Code (Indian)"] ) if input_type == "Yahoo Finance Ticker": fund_id = st.text_input("Enter Yahoo Finance Ticker", "0P0000XW8F.BO") else: fund_id = st.text_input("Enter Mutual Fund Code", "118989") if st.button("Analyze Risk"): with st.spinner("Performing risk analysis..."): df = load_yahoo_finance_data(fund_id, start_date, end_date) if input_type == "Yahoo Finance Ticker" else fetch_mutual_fund_data(fund_id) if df is not None: risk_figs = plot_risk_analytics(df) if risk_figs: st.subheader("Drawdown Analysis") st.plotly_chart(risk_figs[0], use_container_width=True) st.subheader("Risk-Return Analysis") st.plotly_chart(risk_figs[1], use_container_width=True) if __name__ == "__main__": main()