import streamlit as st import pandas as pd import plotly.graph_objects as go import plotly.express as px import requests import yfinance as yf from datetime import datetime, date import os st.set_page_config(layout="wide") FMP_API_KEY = os.getenv("FMP_API_KEY") # ------------------------------------------------------------------- # Initialize session state defaults # ------------------------------------------------------------------- if "valid_ticker" not in st.session_state: st.session_state["valid_ticker"] = None if "ticker" not in st.session_state: st.session_state["ticker"] = None if "hist" not in st.session_state: st.session_state["hist"] = None if "consensus" not in st.session_state: st.session_state["consensus"] = None if "df_targets" not in st.session_state: st.session_state["df_targets"] = None if "df_rss" not in st.session_state: st.session_state["df_rss"] = None # ------------------------------------------------------------------- # Column reordering helper: move specified columns to the end # ------------------------------------------------------------------- def move_columns_to_end(df, cols_to_move): existing = [col for col in cols_to_move if col in df.columns] fixed_order = [col for col in df.columns if col not in existing] + existing return df[fixed_order] # ------------------------------------------------------------------- # Cache functions # ------------------------------------------------------------------- @st.cache_data def fetch_yfinance_data(symbol, period="5y"): try: ticker_obj = yf.Ticker(symbol) hist = ticker_obj.history(period=period) if hist.empty: raise ValueError("No historical data found.") return hist except: st.error("Unable to fetch historical price data.") return None @st.cache_data def fetch_fmp_consensus(symbol): try: url = f"https://financialmodelingprep.com/api/v4/price-target-consensus?symbol={symbol}&apikey={FMP_API_KEY}" response = requests.get(url) data = response.json() if data and len(data) > 0: return data[0] else: raise ValueError("No consensus data returned.") except: st.error("Unable to fetch consensus data.") return None @st.cache_data def fetch_price_target_data(symbol): try: url = f"https://financialmodelingprep.com/api/v4/price-target?symbol={symbol}&apikey={FMP_API_KEY}" response = requests.get(url) data = response.json() if data: df = pd.DataFrame(data) df['publishedDate'] = pd.to_datetime(df['publishedDate']) return df else: raise ValueError("No price target data returned.") except: st.error("Unable to fetch price target data.") return None @st.cache_data def fetch_price_target_rss_feed(num_pages=5): try: all_data = [] for page in range(num_pages): url = f"https://financialmodelingprep.com/api/v4/price-target-rss-feed?page={page}&apikey={FMP_API_KEY}" response = requests.get(url) if response.status_code == 200: data = response.json() all_data.extend(data) if all_data: df = pd.DataFrame(all_data) df['publishedDate'] = pd.to_datetime(df['publishedDate']) return df else: raise ValueError("No live feed data returned.") except: st.error("Unable to fetch live feed data.") return None def is_valid_ticker(tkr): try: _ = yf.Ticker(tkr).info return True except: return False # ------------------------------------------------------------------- # Sidebar # ------------------------------------------------------------------- st.sidebar.title("Analysis Parameters") with st.sidebar.expander("Page Selection", expanded=True): page = st.radio( "Select a page", ["Price Targets by Ticker", "Price Target Live Feed"], help="Choose a view for detailed stock data or a live feed of recent targets." ) if page == "Price Targets by Ticker": with st.sidebar.expander("Analysis Inputs", expanded=True): ticker = st.text_input( "Ticker Symbol", value="AAPL", help="Enter a valid stock ticker symbol (e.g. AAPL)." ) run_analysis = st.sidebar.button("Run Analysis") else: run_analysis = st.sidebar.button("Run Analysis", help="Fetch the latest live feed data.") # ------------------------------------------------------------------- # Logic to store data in session state if Run Analysis is clicked # ------------------------------------------------------------------- if page == "Price Targets by Ticker": if run_analysis: if not is_valid_ticker(ticker): st.session_state["valid_ticker"] = False else: st.session_state["valid_ticker"] = True st.session_state["ticker"] = ticker st.session_state["hist"] = fetch_yfinance_data(ticker) st.session_state["consensus"] = fetch_fmp_consensus(ticker) st.session_state["df_targets"] = fetch_price_target_data(ticker) elif page == "Price Target Live Feed": if run_analysis: st.session_state["df_rss"] = fetch_price_target_rss_feed(num_pages=5) # ------------------------------------------------------------------- # Main Page Content # ------------------------------------------------------------------- if page == "Price Targets by Ticker": st.title("Analyst Price Targets") if st.session_state["valid_ticker"] is None: st.markdown("Enter a stock symbol and click **Run Analysis** to load the data.") elif st.session_state["valid_ticker"] is False: st.error("Invalid symbol. Please try again.") else: ticker = st.session_state["ticker"] hist = st.session_state["hist"] consensus = st.session_state["consensus"] df_targets = st.session_state["df_targets"] # Fixed bubble size multiplier bubble_multiplier = 1.2 # ----------------------------------------- # 12 Month Analyst Forecast Consensus # ----------------------------------------- if hist is not None and consensus is not None: st.markdown("### Analyst Forecast (12-Month)") st.write("This chart shows the stock's closing price history. " "It also shows projected targets for the next year, " "including high, low, median, and overall consensus.") def plot_price_data_with_targets(history_df, cons, symbol, forecast_months=12): last_date = history_df.index[-1] future_date = last_date + pd.DateOffset(months=forecast_months) last_close = history_df['Close'][-1] extended_future_date = future_date + pd.DateOffset(days=90) fig = go.Figure() fig.add_trace(go.Scatter( x=history_df.index, y=history_df['Close'], mode='lines', name='Close Price', line=dict(color='royalblue', width=2), hovertemplate='Date: %{x}
Price: %{y:.2f}' )) fig.add_trace(go.Scatter( x=[last_date], y=[last_close], mode='markers', marker=dict(color='white', size=12, symbol='circle'), name="Current Price", hovertemplate='Date: %{x}
Price: %{y:.2f}' )) annotations = [ dict( x=last_date, y=last_close, text=f"{round(last_close)}", font=dict(size=16, color='white'), showarrow=False, yshift=30 ) ] targets = [ ("Target High", cons["targetHigh"], "green"), ("Target Low", cons["targetLow"], "red"), ("Target Consensus", cons["targetConsensus"], "orange"), ("Target Median", cons["targetMedian"], "purple") ] for name, val, color in targets: val_rounded = round(val) fig.add_trace(go.Scatter( x=[last_date, future_date], y=[last_close, val_rounded], mode='lines', line=dict(dash='dash', color=color, width=2), name=name, hovertemplate=f"{name}: {val_rounded}" )) annotations.append( dict( x=future_date, y=val_rounded, text=f"{val_rounded}", showarrow=True, arrowhead=2, ax=20, ay=0, font=dict(color=color, size=20) ) ) fig.add_shape( type="line", x0=last_date, x1=last_date, y0=history_df['Close'].min(), y1=history_df['Close'].max(), line=dict(color="gray", dash="dot") ) fig.update_layout( template='plotly_dark', paper_bgcolor='#0e1117', plot_bgcolor='#0e1117', font=dict(color='white'), title=dict(text=f"{symbol} Price History & 12-Month Targets", font=dict(color='white')), legend=dict( x=0.01, y=0.99, bordercolor="white", borderwidth=1, font=dict(color='white') ), xaxis=dict( range=[history_df.index[0], extended_future_date], showgrid=True, gridcolor='gray', title=dict(text="Date", font=dict(color='white')), tickfont=dict(color='white') ), yaxis=dict( showgrid=True, gridcolor='gray', title=dict(text="Price", font=dict(color='white')), tickfont=dict(color='white') ), annotations=annotations, margin=dict(l=40, r=40, t=60, b=40) ) return fig fig_consensus = plot_price_data_with_targets(hist, consensus, ticker) st.plotly_chart(fig_consensus, use_container_width=True) # ----------------------------------------- # Price Target Evolution (Bubble Chart) # ----------------------------------------- st.markdown("### Analyst Price Target Changes Over Time") st.write("This chart shows how price targets have shifted. " "Bubble sizes represent the percentage change from the posted price.") if df_targets is not None: def plot_price_target_evolution(df): df['publishedDate'] = pd.to_datetime(df['publishedDate'], errors='coerce').dt.tz_localize(None) df['targetChange'] = df['priceTarget'] - df['priceWhenPosted'] df['direction'] = df['targetChange'].apply( lambda x: "Raised" if x > 0 else ("Lowered" if x < 0 else "No Change") ) df['percentChange'] = (df['targetChange'] / df['priceWhenPosted']) * 100 color_map = {"Raised": "green", "Lowered": "red", "No Change": "gray"} colors = df['direction'].map(color_map) bubble_sizes = abs(df['percentChange']) * bubble_multiplier df['date'] = df['publishedDate'].dt.date daily_median = df.groupby('date')['priceTarget'].median() daily_median.index = pd.to_datetime(daily_median.index) fig = go.Figure() # Price When Posted line+markers fig.add_trace(go.Scatter( x=df['publishedDate'], y=df['priceWhenPosted'], mode='lines+markers', name='Price When Posted', line=dict(color='royalblue', width=2, dash='dot'), marker=dict(size=8), hovertemplate='Date: %{x}
Price When Posted: %{y:.2f}' )) # Bubble markers for Price Target fig.add_trace(go.Scatter( x=df['publishedDate'], y=df['priceTarget'], mode='markers', name='Price Target', marker=dict( size=bubble_sizes, color=colors, opacity=0.7, line=dict(width=1, color='black') ), hovertemplate=( "%{customdata[0]}
" "Published: %{x}
" "Price Target: %{y:.2f}
" "Price When Posted: %{customdata[1]:.2f}
" "Target Change: %{customdata[2]:.2f}
" "Percent Change: %{customdata[3]:.2f}%
" "Bubble Scale: 2.0" "" ), customdata=df[['newsTitle', 'priceWhenPosted', 'targetChange', 'percentChange']].values )) # Median line if not daily_median.empty: fig.add_trace(go.Scatter( x=daily_median.index, y=daily_median.values, mode='lines', name='Median Price Target', line=dict(color='white', dash='dash', width=3, shape='hv'), hovertemplate='Date: %{x}
Median Price Target: %{y:.2f}' )) # Annotation for latest price if not df.empty: current_date = df['publishedDate'].max() current_price = df.loc[df['publishedDate'] == current_date, 'priceWhenPosted'].iloc[-1] fig.add_annotation( x=current_date, y=current_price, text=f"{round(current_price)}", showarrow=False, font=dict(size=16, color='white'), yshift=30 ) fig.update_layout( template='plotly_dark', paper_bgcolor='#0e1117', plot_bgcolor='#0e1117', font=dict(color='white'), title=dict(text=f"{ticker}: Posted Price, Price Targets & Daily Median", font=dict(color='white')), legend=dict( x=0.01, y=0.99, bordercolor="white", borderwidth=1, font=dict(color='white') ), xaxis=dict( showgrid=True, gridcolor='gray', title=dict(text="Published Date", font=dict(color='white')), tickfont=dict(color='white') ), yaxis=dict( showgrid=True, gridcolor='gray', title=dict(text="Price (USD)", font=dict(color='white')), tickfont=dict(color='white') ), margin=dict(l=40, r=40, t=60, b=40) ) return fig fig_evolution = plot_price_target_evolution(df_targets) st.plotly_chart(fig_evolution, use_container_width=True) st.markdown("### Detailed Historical Price Targets") st.write("This table lists recent price targets, news headlines, and links.") df_targets["MovementChart"] = df_targets.apply( lambda row: [row["priceWhenPosted"], row["priceTarget"]], axis=1 ) df_targets = move_columns_to_end( df_targets, ["newsTitle","newsURL","newsPublisher","newsBaseURL","url"] ) with st.expander("Detailed Data", expanded=False): st.dataframe( df_targets, column_config={ "MovementChart": st.column_config.LineChartColumn( "From Posted to Target", help="Line from priceWhenPosted to priceTarget", ) }, height=300 ) elif page == "Price Target Live Feed": st.title("Live Analyst Targets") if st.session_state["df_rss"] is None: st.markdown("Click **Run Analysis** to fetch the latest feed.") else: df_rss = st.session_state["df_rss"] if not df_rss.empty: st.markdown("### Latest Analyst Announcements") st.write("This chart shows a daily view of median percentage changes in targets for various symbols.") def plot_rss_feed(df): df['date'] = df['publishedDate'].dt.date df['targetChange'] = df['priceTarget'] - df['priceWhenPosted'] df['percentChange'] = (df['targetChange'] / df['priceWhenPosted']) * 100 grouped = df.groupby(['date', 'symbol']).agg({ 'percentChange': 'median', 'priceTarget': 'median', 'priceWhenPosted': 'median' }).reset_index() if grouped.empty: return None grouped['date'] = pd.to_datetime(grouped['date']) fig = px.scatter( grouped, x='date', y='symbol', size=grouped['percentChange'].abs(), color='percentChange', color_continuous_scale='RdYlGn', title='Daily Median Analyst % Change by Symbol', labels={'date': 'Date', 'symbol': 'Ticker', 'percentChange': '% Change'} ) unique_symbols = grouped['symbol'].nunique() fig.update_layout( template='plotly_dark', paper_bgcolor='#0e1117', plot_bgcolor='#0e1117', font=dict(color='white'), title=dict(text='Daily Median Analyst % Change by Symbol', font=dict(color='white')), xaxis=dict( showgrid=True, gridcolor='gray', title=dict(text="Date", font=dict(color='white')), tickfont=dict(color='white') ), yaxis=dict( showgrid=True, gridcolor='gray', title=dict(text="Ticker", font=dict(color='white')), tickfont=dict(color='white') ), height=(unique_symbols * 10), margin=dict(l=40, r=40, t=60, b=40) ) fig.update_traces( customdata=grouped[['symbol', 'percentChange', 'priceTarget', 'priceWhenPosted']].values, hovertemplate=( "%{customdata[0]}
" "Date: %{x}
" "Median % Change: %{customdata[1]:.2f}%
" "Median Target: %{customdata[2]:.2f}
" "Median Posted: %{customdata[3]:.2f}" ) ) return fig feed_fig = plot_rss_feed(df_rss) if feed_fig: st.plotly_chart(feed_fig, use_container_width=True) else: st.info("No grouped data to plot.") st.markdown("### Detailed Live Feed Data") st.write("This table lists recent announcements with their posted price and target.") df_rss["MovementChart"] = df_rss.apply( lambda row: [row["priceWhenPosted"], row["priceTarget"]], axis=1 ) df_rss = move_columns_to_end( df_rss, ["newsTitle","newsURL","newsPublisher","newsBaseURL","url"] ) with st.expander("Detailed Data", expanded=False): st.dataframe( df_rss, column_config={ "MovementChart": st.column_config.LineChartColumn( "From Posted to Target", help="Line from priceWhenPosted to priceTarget", ) }, height=300 ) else: st.info("No live feed data available.") # Hide default Streamlit style st.markdown( """ """, unsafe_allow_html=True )