import streamlit as st import requests import pandas as pd import plotly.graph_objects as go from datetime import datetime import dateutil.relativedelta import os # ---- PAGE CONFIG ---- st.set_page_config(layout="wide") # ---- GLOBALS ---- API_KEY = os.getenv("FMP_API_KEY") # ---- SIDEBAR INPUTS ---- st.sidebar.title("User Inputs") with st.sidebar.expander("Configuration", expanded=True): ticker = st.text_input("Ticker:", "ASML", help="Insert the stock ticker.") # Radio selection for Annual vs Quarterly data data_period = st.radio("Select Data Period", ("Annual", "Quarterly")) if data_period == "Annual": period_api = "annual" period_count = st.number_input( "Years of historical data:", min_value=1, max_value=50, value=15, help="Choose how many years of historical data to retrieve." ) cutoff_date = datetime.now() - dateutil.relativedelta.relativedelta(years=period_count) xaxis_title = "Year" tickformat = "%Y" dtick = "M12" HIST_KEY = "historical_df_annual" FORECAST_KEY = "forecast_df_annual" else: period_api = "quarter" period_count = st.number_input( "Quarters of historical data:", min_value=1, max_value=200, value=20, help="Choose how many quarters of historical data to retrieve." ) cutoff_date = datetime.now() - dateutil.relativedelta.relativedelta(months=period_count * 3) xaxis_title = "Quarter" tickformat = "%Y-%m" dtick = "M3" HIST_KEY = "historical_df_quarter" FORECAST_KEY = "forecast_df_quarter" run_button = st.sidebar.button("Run Analysis") # Initialize session state if not present if HIST_KEY not in st.session_state: st.session_state[HIST_KEY] = pd.DataFrame() if FORECAST_KEY not in st.session_state: st.session_state[FORECAST_KEY] = pd.DataFrame() # ---- HELPER FUNCTION: VALUE FORMATTING ---- def format_value(x): if abs(x) >= 1e9: return f"{x/1e9:.1f}B" elif abs(x) >= 1e6: return f"{x/1e6:.1f}M" elif abs(x) >= 1e3: return f"{x/1e3:.1f}K" else: return f"{x:.1f}" @st.cache_data def fetch_data(hist_url, forecast_url): hist_data = requests.get(hist_url, timeout=10).json() forecast_data = requests.get(forecast_url, timeout=10).json() return hist_data, forecast_data # ---- MAIN APP START ---- def main(): st.title("Analyst Forecasts & Estimates") st.write( "This tool fetches historical financial data and analyst forecasts. " "It helps you see past trends and future estimates over your selected period." ) if run_button: if not ticker.strip(): st.error("Please enter a valid ticker.") return hist_url = ( f"https://financialmodelingprep.com/api/v3/income-statement/{ticker}" f"?period={period_api}&limit={period_count}&apikey={API_KEY}" ) forecast_url = ( f"https://financialmodelingprep.com/api/v3/analyst-estimates/{ticker}" f"?period={period_api}&apikey={API_KEY}" ) try: hist_data, forecast_data = fetch_data(hist_url, forecast_url) except Exception: st.error("Could not retrieve data at this time.") return st.session_state[HIST_KEY] = pd.DataFrame(hist_data) st.session_state[FORECAST_KEY] = pd.DataFrame(forecast_data) if st.session_state[HIST_KEY].empty and st.session_state[FORECAST_KEY].empty: st.info("Set your inputs in the sidebar, then click **Run Analysis**.") return historical_df = st.session_state[HIST_KEY] forecast_df = st.session_state[FORECAST_KEY] if not historical_df.empty and "date" in historical_df.columns: historical_df["date"] = pd.to_datetime(historical_df["date"]) historical_df.sort_values("date", inplace=True) if not forecast_df.empty and "date" in forecast_df.columns: forecast_df["date"] = pd.to_datetime(forecast_df["date"]) forecast_df.sort_values("date", inplace=True) if "date" in historical_df.columns: historical_df = historical_df[historical_df["date"] >= cutoff_date] if "date" in forecast_df.columns: forecast_df = forecast_df[forecast_df["date"] >= cutoff_date] metrics = { "Revenue": { "historical": "revenue", "forecast": { "Low": "estimatedRevenueLow", "Avg": "estimatedRevenueAvg", "High": "estimatedRevenueHigh" } }, "EBITDA": { "historical": "ebitda", "forecast": { "Low": "estimatedEbitdaLow", "Avg": "estimatedEbitdaAvg", "High": "estimatedEbitdaHigh" } }, "EBIT": { "historical": "operatingIncome", "forecast": { "Low": "estimatedEbitLow", "Avg": "estimatedEbitAvg", "High": "estimatedEbitHigh" } }, "Net Income": { "historical": "netIncome", "forecast": { "Low": "estimatedNetIncomeLow", "Avg": "estimatedNetIncomeAvg", "High": "estimatedNetIncomeHigh" } }, "SG&A Expense": { "historical": "sellingGeneralAndAdministrativeExpenses", "forecast": { "Low": "estimatedSgaExpenseLow", "Avg": "estimatedSgaExpenseAvg", "High": "estimatedSgaExpenseHigh" } }, "EPS": { "historical": "eps", "forecast": { "Low": "estimatedEpsLow", "Avg": "estimatedEpsAvg", "High": "estimatedEpsHigh" } } } def create_plot(metric_name, hist_col, forecast_cols): fig = go.Figure() if hist_col in historical_df.columns and not historical_df.empty: bar_text = [format_value(val) for val in historical_df[hist_col]] fig.add_trace(go.Bar( x=historical_df["date"], y=historical_df[hist_col], text=bar_text, textposition="auto", name="Historical" )) if not forecast_df.empty: for label, col in forecast_cols.items(): if col in forecast_df.columns: fig.add_trace(go.Scatter( x=forecast_df["date"], y=forecast_df[col], mode="lines+markers", name=f"Forecast {label}" )) if metric_name == "EPS": analyst_field = "numberAnalystsEstimatedEps" else: analyst_field = "numberAnalystEstimatedRevenue" if analyst_field in forecast_df.columns and not forecast_df.empty: analysts_count = int(round(forecast_df[analyst_field].mean())) else: analysts_count = "N/A" title_text = f"{ticker} - {metric_name} | Analysts: {analysts_count}" fig.update_layout( title=title_text, barmode="stack", template="plotly_dark", paper_bgcolor="#0e1117", plot_bgcolor="#0e1117", xaxis=dict( title=xaxis_title, tickangle=45, tickformat=tickformat, dtick=dtick, showgrid=True, gridcolor="rgba(255, 255, 255, 0.1)" ), yaxis=dict( title=metric_name, showgrid=True, gridcolor="rgba(255, 255, 255, 0.1)" ), legend=dict(), margin=dict(l=40, r=40, t=80, b=80) ) return fig for metric, mapping in metrics.items(): with st.container(border=True): st.subheader(metric) st.write( f"This chart shows {metric} over the selected time periods. " f"Bars represent historical data and lines represent forecast ranges. " "Hover over markers for details." ) fig = create_plot(metric, mapping["historical"], mapping["forecast"]) st.plotly_chart(fig, use_container_width=True) with st.expander(f"View {metric} Data", expanded=False): hc = mapping["historical"] hist_disp = ( historical_df[["date", hc]].copy() if hc in historical_df.columns else pd.DataFrame() ) if not hist_disp.empty: hist_disp.rename(columns={hc: f"{metric}_Historical"}, inplace=True) forecast_disp = pd.DataFrame() if not forecast_df.empty: wanted_cols = ["date"] + list(mapping["forecast"].values()) existing_cols = [c for c in wanted_cols if c in forecast_df.columns] forecast_disp = forecast_df[existing_cols].copy() for fc_key, fc_val in mapping["forecast"].items(): if fc_val in forecast_disp.columns: forecast_disp.rename( columns={fc_val: f"{metric}_Forecast_{fc_key}"}, inplace=True ) if not hist_disp.empty and not forecast_disp.empty: merged_df = pd.merge(hist_disp, forecast_disp, on="date", how="outer") merged_df.sort_values("date", inplace=True) elif not hist_disp.empty: merged_df = hist_disp elif not forecast_disp.empty: merged_df = forecast_disp else: merged_df = pd.DataFrame() if merged_df.empty: st.write("No data found for this metric.") else: st.dataframe(merged_df.reset_index(drop=True)) if __name__ == "__main__": main() st.markdown( """ """, unsafe_allow_html=True )