import streamlit as st
import requests
import time
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
import json
import yfinance as yf
import plotly.graph_objects as go
import plotly.express as px
from pathlib import Path
import os
PLOTLY_AVAILABLE = True
st.set_page_config(
page_title="StockSense AI Agent",
page_icon="đ",
layout="wide",
initial_sidebar_state="expanded",
menu_items={
'Get Help': 'https://github.com/Spkap/StockSense-AI',
'Report a bug': 'https://github.com/Spkap/StockSense-AI/issues',
'About': "# StockSense ReAct Agent\nAI-powered autonomous stock analysis using ReAct pattern"
}
)
st.markdown("""
""", unsafe_allow_html=True)
def initialize_session_state():
if 'analysis_result' not in st.session_state:
st.session_state.analysis_result = None
if 'analysis_history' not in st.session_state:
st.session_state.analysis_history = []
if 'backend_status' not in st.session_state:
st.session_state.backend_status = None
if 'selected_ticker' not in st.session_state:
st.session_state.selected_ticker = ""
initialize_session_state()
@st.cache_data
def load_company_ticker_mapping():
"""Load company name to ticker mapping from CSV file."""
csv_path = Path("nasdaq_screener.csv")
if csv_path.exists():
try:
df = pd.read_csv(csv_path)
if 'Symbol' in df.columns and 'Name' in df.columns:
# Create a mapping from company name to ticker
mapping = dict(zip(df['Name'], df['Symbol']))
return mapping, df
except Exception as e:
st.warning(f"Error loading CSV file: {e}")
# Fallback to popular stocks if CSV not available
fallback_mapping = {
"Apple Inc.": "AAPL",
"Microsoft Corporation": "MSFT",
"Alphabet Inc.": "GOOGL",
"Amazon.com Inc.": "AMZN",
"Tesla Inc.": "TSLA",
"NVIDIA Corporation": "NVDA",
"Meta Platforms Inc.": "META",
"Netflix Inc.": "NFLX",
"Advanced Micro Devices Inc.": "AMD",
"Intel Corporation": "INTC"
}
return fallback_mapping, None
def create_candlestick_chart(price_data: list) -> go.Figure:
"""Create an interactive candlestick chart with moving averages."""
if not price_data:
return None
# Convert to DataFrame
df = pd.DataFrame(price_data)
df['Date'] = pd.to_datetime(df['Date'])
df = df.sort_values('Date')
# Calculate moving averages
df['SMA_20'] = df['Close'].rolling(window=20, min_periods=1).mean()
df['SMA_50'] = df['Close'].rolling(window=50, min_periods=1).mean()
# Create candlestick chart
fig = go.Figure()
# Add candlestick
fig.add_trace(go.Candlestick(
x=df['Date'],
open=df['Open'],
high=df['High'],
low=df['Low'],
close=df['Close'],
name="Price",
increasing_line_color='#26a69a',
decreasing_line_color='#ef5350'
))
# Add moving averages
fig.add_trace(go.Scatter(
x=df['Date'],
y=df['SMA_20'],
mode='lines',
name='SMA 20',
line=dict(color='orange', width=2)
))
fig.add_trace(go.Scatter(
x=df['Date'],
y=df['SMA_50'],
mode='lines',
name='SMA 50',
line=dict(color='blue', width=2)
))
# Update layout
fig.update_layout(
title="Stock Price with Moving Averages",
yaxis_title="Price ($)",
xaxis_title="Date",
template="plotly_white",
height=500,
showlegend=True,
xaxis_rangeslider_visible=False
)
return fig
BACKEND_URL = "http://127.0.0.1:8000"
def check_backend_status() -> bool:
"""Check if the backend is online."""
try:
response = requests.get(f"{BACKEND_URL}/health", timeout=5)
status = response.status_code == 200
st.session_state.backend_status = status
return status
except:
st.session_state.backend_status = False
return False
def create_styled_header():
st.markdown("""
đ StockSense AI Agent
AI-Powered Stock Analysis Using Reasoning & Action
""", unsafe_allow_html=True)
def display_hero_section():
create_styled_header()
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
if check_backend_status():
st.success("đĸ Backend Connected & Ready", icon="â
")
else:
st.error("đ´ Backend Connection Failed", icon="â")
st.markdown("
", unsafe_allow_html=True)
def display_ticker_input():
st.markdown("### Select Stock to Analyze")
# Load company mapping
company_mapping, df = load_company_ticker_mapping()
company_names = list(company_mapping.keys())
col1, col2 = st.columns([2, 1])
with col1:
st.markdown("**Select Company:**")
# Company selectbox
selected_company = st.selectbox(
"Choose a company",
options=[""] + company_names,
index=0,
help="Select a company from the dropdown",
label_visibility="collapsed"
)
if selected_company:
ticker = company_mapping[selected_company]
st.session_state.selected_ticker = ticker
st.info(f"Selected: **{selected_company}** â **{ticker}**")
with col2:
st.markdown("**Or enter manually:**")
manual_ticker = st.text_input(
"Stock Ticker",
value=st.session_state.selected_ticker if not selected_company else "",
placeholder="e.g., AAPL",
help="Enter any valid stock ticker symbol",
label_visibility="collapsed"
)
if manual_ticker and manual_ticker != st.session_state.selected_ticker:
st.session_state.selected_ticker = manual_ticker.upper().strip()
# Quick select buttons for popular stocks
st.markdown("**Quick Select Popular Stocks:**")
cols = st.columns(5)
popular_tickers = ["AAPL", "MSFT", "GOOGL", "AMZN", "TSLA"]
for i, ticker in enumerate(popular_tickers):
with cols[i]:
if st.button(ticker, key=f"quick_{ticker}", use_container_width=True):
st.session_state.selected_ticker = ticker
cols = st.columns(5)
more_tickers = ["NVDA", "META", "NFLX", "AMD", "INTC"]
for i, ticker in enumerate(more_tickers):
with cols[i]:
if st.button(ticker, key=f"quick_{ticker}", use_container_width=True):
st.session_state.selected_ticker = ticker
return st.session_state.selected_ticker
def validate_ticker(ticker: str) -> tuple[bool, str]:
"""Validate ticker input."""
if not ticker:
return False, "Please select or enter a stock ticker symbol"
if not ticker.replace('.', '').isalpha() or len(ticker) < 1 or len(ticker) > 10:
return False, "Please enter a valid ticker (1-10 letters, dots allowed)"
return True, ""
def trigger_analysis(ticker: str) -> Optional[Dict[str, Any]]:
"""
Triggers backend analysis, handles direct responses for cache hits,
and patiently polls for results for fresh analyses.
"""
# Helper function to process a successful result and update session state
def _handle_success(result_data: Dict[str, Any], ticker: str) -> Dict[str, Any]:
result_obj = {
'ticker': ticker,
'data': result_data,
'timestamp': datetime.now().isoformat(),
'success': True
}
st.session_state.analysis_result = result_obj
st.session_state.analysis_history.insert(0, result_obj)
# Keep history to a max of 10 items
if len(st.session_state.analysis_history) > 10:
st.session_state.analysis_history.pop()
return result_obj
try:
if not check_backend_status():
st.error("đ Backend server is offline. Please check its status on Render.")
return None
# --- Step 1: Trigger the analysis ---
with st.spinner("Sending analysis request to the AI agent..."):
response = requests.post(f"{BACKEND_URL}/analyze/{ticker}", timeout=60)
if response.status_code != 200:
st.error(f"â Analysis request failed: Status {response.status_code}")
try:
st.error(f"Details: {response.json().get('detail', response.text)}")
except json.JSONDecodeError:
st.error(f"Details: {response.text}")
return None
# --- Step 2: Handle the response from the trigger ---
result = response.json()
analysis_data = result.get('data', {})
source = analysis_data.get('source')
if source == 'cache':
st.success("â
Analysis retrieved instantly from cache!")
return _handle_success(analysis_data, ticker)
elif source == 'react_analysis':
st.success("â
Fresh analysis completed directly!")
return _handle_success(analysis_data, ticker)
# --- Step 3: If not a direct response, start patient polling ---
st.info("đ¤ Agent has started a new analysis. Waiting for results...")
max_attempts = 30
sleep_interval = 3
status_text = st.empty()
progress_bar = st.progress(0, text="Initializing...")
for attempt in range(1, max_attempts + 1):
progress_bar.progress(attempt / max_attempts, text=f"Polling for results... (Attempt {attempt}/{max_attempts})")
try:
results_response = requests.get(f"{BACKEND_URL}/results/{ticker}", timeout=10)
if results_response.status_code == 200:
analysis_data = results_response.json().get('data', {})
status_text.success("â
Results received!")
progress_bar.empty()
return _handle_success(analysis_data, ticker)
# If 404, it means the result is not ready yet. Just wait and continue.
elif results_response.status_code == 404:
time.sleep(sleep_interval)
# Handle other potential server errors during polling
else:
status_text.error(f"Backend returned an error (Status {results_response.status_code}). Aborting.")
progress_bar.empty()
return None
except requests.exceptions.RequestException:
# Ignore transient network errors and just retry
time.sleep(sleep_interval)
progress_bar.empty()
status_text.error(
f"Analysis for '{ticker}' did not complete within {max_attempts * sleep_interval} seconds. "
"The backend might be busy or an error occurred. Please check the backend logs on Render."
)
return None
except requests.exceptions.Timeout:
st.error("âąī¸ Initial request timed out. The backend may be experiencing a cold start. Please try again in a moment.")
return None
except requests.exceptions.ConnectionError:
st.error("đ Cannot connect to backend. Please ensure the server is running and the URL is correct.")
return None
except Exception as e:
st.error(f"â An unexpected error occurred: {str(e)}")
return None
def display_enhanced_analysis_results(data: Dict[str, Any], ticker: str):
"""Display enhanced analysis results with tabbed interface."""
# Create tabs
tab1, tab2, tab3 = st.tabs(["đ Analysis Dashboard", "đ° News & Sentiment", "âī¸ Agent Reasoning"])
with tab1:
st.markdown("### đ Analysis Dashboard")
# Display analysis summary
summary = (data.get('summary') or
data.get('analysis_summary') or
data.get('analysis') or
"Analysis completed successfully")
if summary and summary != "Analysis completed successfully":
st.markdown(f"""
đ Analysis Summary
Stock: {ticker}
{summary}
""", unsafe_allow_html=True)
else:
st.info("đ Detailed analysis summary not available for this request")
# Display interactive candlestick chart
price_data = data.get('price_data', [])
if not price_data:
# Try alternative key names
price_data = (data.get('historical_data') or
data.get('price_history') or
data.get('ohlcv_data') or [])
if price_data and len(price_data) > 0:
st.markdown("#### đ Interactive Price Chart with Moving Averages")
try:
fig = create_candlestick_chart(price_data)
if fig:
st.plotly_chart(fig, use_container_width=True)
st.success(f"đ Displaying {len(price_data)} days of price data with technical indicators")
else:
st.warning("Unable to create candlestick chart - trying fallback visualization")
# Fallback to simple line chart
try:
df = pd.DataFrame(price_data)
df['Date'] = pd.to_datetime(df['Date'])
df = df.sort_values('Date').set_index('Date')
st.line_chart(df[['Close']], use_container_width=True)
except Exception as fallback_error:
st.error(f"Chart creation failed: {str(fallback_error)}")
except Exception as e:
st.error(f"Error creating chart: {str(e)}")
# Show raw data for debugging
with st.expander("đ Debug: Raw Price Data"):
st.json(price_data[:3]) # Show first 3 records
else:
# Show more helpful message
available_keys = list(data.keys())
st.info(f"đ No historical price data available in this analysis. Available data: {', '.join(available_keys)}")
# Try to get real-time price data as fallback
if st.button("đ Fetch Current Price Data", key="fetch_price_fallback"):
with st.spinner("Fetching current market data..."):
try:
from stocksense.data_collectors import get_price_history
df = get_price_history(ticker, period="1mo")
if df is not None and not df.empty:
# Convert to our format
df_reset = df.reset_index()
df_reset['Date'] = df_reset['Date'].dt.strftime('%Y-%m-%d')
fallback_data = df_reset.to_dict(orient='records')
fig = create_candlestick_chart(fallback_data)
if fig:
st.plotly_chart(fig, use_container_width=True)
st.success("đ Showing current market data (not from AI analysis)")
except Exception as e:
st.error(f"Failed to fetch fallback data: {str(e)}")
# Display key metrics
display_key_metrics(ticker)
with tab2:
st.markdown("### đ° News & Sentiment Analysis")
# Display detailed sentiment report
sentiment_report_raw = data.get('sentiment_report')
if sentiment_report_raw:
st.markdown("""
đ Market Sentiment Report
""", unsafe_allow_html=True)
# Try to parse as JSON first
if isinstance(sentiment_report_raw, str):
try:
sentiment_data = json.loads(sentiment_report_raw)
if isinstance(sentiment_data, list):
# Display structured sentiment data
for i, item in enumerate(sentiment_data, 1):
headline = item.get('headline', 'N/A')
sentiment = item.get('sentiment', 'N/A')
justification = item.get('justification', 'N/A')
with st.expander(f"đ° Article {i}: {headline[:80]}..."):
st.markdown(f"**Headline:** {headline}")
st.markdown(f"**Sentiment:** {sentiment}")
st.markdown(f"**Analysis:** {justification}")
else:
st.markdown(sentiment_report_raw)
except json.JSONDecodeError:
st.markdown(sentiment_report_raw)
elif isinstance(sentiment_report_raw, list):
# Already structured data
for i, item in enumerate(sentiment_report_raw, 1):
headline = item.get('headline', 'N/A')
sentiment = item.get('sentiment', 'N/A')
justification = item.get('justification', 'N/A')
with st.expander(f"īŋŊ Article {i}: {headline[:80]}..."):
st.markdown(f"**Headline:** {headline}")
st.markdown(f"**Sentiment:** {sentiment}")
st.markdown(f"**Analysis:** {justification}")
else:
st.markdown(str(sentiment_report_raw))
else:
# Try to show any related sentiment data
sentiment_info = []
if data.get('headlines_count'):
sentiment_info.append(f"Analyzed {data['headlines_count']} headlines")
if data.get('sentiment'):
sentiment_info.append(f"Overall sentiment: {data['sentiment']}")
if sentiment_info:
st.info("Sentiment analysis completed. " + " | ".join(sentiment_info))
else:
st.info("No detailed sentiment analysis data available")
with tab3:
st.markdown("### âī¸ Agent Reasoning Process")
# Check if we have any reasoning data - be more specific about what constitutes "available"
reasoning_steps = data.get('reasoning_steps') or []
has_reasoning_steps = len(reasoning_steps) > 0
has_tools_used = data.get('tools_used') and len(data.get('tools_used', [])) > 0
has_iterations = data.get('iterations') and data.get('iterations', 0) > 0
has_source_info = data.get('source') == 'react_analysis' # Fresh analysis vs cached
# For cached data, don't show reasoning even if it has generic cache steps
is_cached = data.get('source') == 'cache'
has_meaningful_reasoning = has_reasoning_steps and not is_cached
if has_meaningful_reasoning or (has_tools_used and not is_cached) or (has_iterations and not is_cached):
st.markdown("""
đ§ ReAct Agent Decision Process
""", unsafe_allow_html=True)
# Show reasoning steps if available
if has_meaningful_reasoning:
reasoning_data = reasoning_steps # Use the cleaned version
st.markdown("**đ Agent Reasoning Steps:**")
if isinstance(reasoning_data, str):
st.markdown(reasoning_data)
elif isinstance(reasoning_data, list):
for i, step in enumerate(reasoning_data, 1):
st.markdown(f"**Step {i}:** {step}")
else:
st.markdown(str(reasoning_data))
# Show tools used if available (but not for cached data)
if has_tools_used and not is_cached:
st.markdown("**đ§ Tools Used:**")
tools = data['tools_used']
if isinstance(tools, list):
for tool in set(tools): # Remove duplicates
st.markdown(f"âĸ {tool}")
else:
st.markdown(str(tools))
# Show iteration count if available (but not for cached data)
if has_iterations and not is_cached:
st.markdown(f"**đ Analysis Iterations:** {data['iterations']}")
# Show analysis type
if has_source_info and not is_cached:
st.info(f"â
Fresh ReAct analysis completed with {data.get('iterations', 0)} reasoning iterations")
else:
# Determine why no reasoning data is available
source = data.get('source', 'unknown')
if source == 'cache':
st.info("đ This is a cached result from a previous analysis. Detailed reasoning steps are not available for cached results.")
elif not has_source_info:
st.warning("â ī¸ No detailed reasoning information available. This may be an incomplete analysis or cached result.")
else:
# This shouldn't happen for fresh analyses
st.error("â No reasoning data available despite being a fresh analysis. This may indicate an issue with the ReAct agent.")
# Display raw data for debugging
with st.expander("đ Raw Analysis Data (Debug)"):
st.json(data)
def display_visualizations(ticker: str):
"""Displays data visualizations for price trends and sentiment distribution."""
st.markdown("### đ Market Insights")
col1, col2 = st.columns(2)
with col1:
st.markdown("**đ 30-Day Price Trend**")
from stocksense.data_collectors import get_price_history
with st.spinner("Fetching real market data..."):
price_data = get_price_history(ticker, period="1mo")
if price_data is not None and not price_data.empty:
price_df = pd.DataFrame({'Price': price_data['Close']})
st.line_chart(price_df, use_container_width=True)
current_price = price_data['Close'].iloc[-1]
prev_price = price_data['Close'].iloc[-2] if len(price_data) > 1 else current_price
change_pct = ((current_price - prev_price) / prev_price) * 100 if prev_price != 0 else 0
trend_icon = "đ" if change_pct > 0 else "đ"
st.markdown(f"""
{trend_icon} Current Price (Real Data)
${current_price:.2f}
{change_pct:+.2f}% from previous day
Source: Yahoo Finance
""", unsafe_allow_html=True)
else:
st.error(f"â Unable to fetch real price data for {ticker}")
st.info("Please check if the ticker symbol is valid and try again.")
with col2:
st.markdown("**đ Sentiment Distribution**")
from stocksense.database import get_latest_analysis
try:
cached_result = get_latest_analysis(ticker)
if cached_result and cached_result.get('sentiment_report'):
sentiment_data = cached_result['sentiment_report']
if isinstance(sentiment_data, str) and sentiment_data.strip():
sentiment_counts = {'Positive': 0, 'Negative': 0, 'Neutral': 0}
text_lower = sentiment_data.lower()
lines = sentiment_data.split('\n')
for line in lines:
line_lower = line.lower()
if 'sentiment:' in line_lower or 'sentiment is' in line_lower:
if 'positive' in line_lower:
sentiment_counts['Positive'] += 1
elif 'negative' in line_lower:
sentiment_counts['Negative'] += 1
elif 'neutral' in line_lower:
sentiment_counts['Neutral'] += 1
if sum(sentiment_counts.values()) == 0:
positive_words = ['positive', 'bullish', 'optimistic', 'strong', 'good', 'gains', 'up', 'growth', 'beat', 'exceeds']
negative_words = ['negative', 'bearish', 'pessimistic', 'weak', 'bad', 'losses', 'down', 'decline', 'miss', 'disappoints']
neutral_words = ['neutral', 'mixed', 'stable', 'unchanged', 'moderate']
for word in positive_words:
sentiment_counts['Positive'] += text_lower.count(word)
for word in negative_words:
sentiment_counts['Negative'] += text_lower.count(word)
for word in neutral_words:
sentiment_counts['Neutral'] += text_lower.count(word)
max_count = max(sentiment_counts.values())
if max_count > 10:
for key in sentiment_counts:
sentiment_counts[key] = min(sentiment_counts[key], 10)
if sum(sentiment_counts.values()) == 0:
sentiment_counts['Neutral'] = 1
sentiment_df = pd.DataFrame(
list(sentiment_counts.items()),
columns=['Sentiment', 'Count']
)
if PLOTLY_AVAILABLE:
colors = {
'Positive': '#28a745',
'Negative': '#dc3545',
'Neutral': '#6c757d'
}
fig = go.Figure(data=[
go.Bar(
x=sentiment_df['Sentiment'],
y=sentiment_df['Count'],
marker_color=[colors.get(sent, '#6c757d') for sent in sentiment_df['Sentiment']],
text=sentiment_df['Count'],
textposition='auto',
)
])
fig.update_layout(
title="Sentiment Distribution",
xaxis_title="Sentiment",
yaxis_title="Count",
showlegend=False,
height=300,
margin=dict(l=20, r=20, t=40, b=20)
)
st.plotly_chart(fig, use_container_width=True)
else:
st.bar_chart(sentiment_df.set_index('Sentiment'), use_container_width=True)
total_headlines = sum(sentiment_counts.values())
dominant_sentiment = max(sentiment_counts, key=sentiment_counts.get) if total_headlines > 0 else "Neutral"
st.markdown(f"""
đ Sentiment Summary
Analysis Available: â
Dominant: {dominant_sentiment}
From real market data
""", unsafe_allow_html=True)
else:
st.info("No sentiment analysis available. Run a new analysis to see sentiment data.")
placeholder_df = pd.DataFrame({
'Sentiment': ['Positive', 'Negative', 'Neutral'],
'Count': [0, 0, 0]
})
st.bar_chart(placeholder_df.set_index('Sentiment'), use_container_width=True)
else:
st.info("No recent sentiment analysis available. Run a new analysis to see sentiment data.")
placeholder_df = pd.DataFrame({
'Sentiment': ['Positive', 'Negative', 'Neutral'],
'Count': [0, 0, 0]
})
st.bar_chart(placeholder_df.set_index('Sentiment'), use_container_width=True)
except Exception as e:
st.warning("Unable to load sentiment data. Run a new analysis to generate sentiment insights.")
placeholder_df = pd.DataFrame({
'Sentiment': ['Positive', 'Negative', 'Neutral'],
'Count': [0, 0, 0]
})
st.bar_chart(placeholder_df.set_index('Sentiment'), use_container_width=True)
def display_key_metrics(ticker: str):
"""Displays key financial metrics."""
st.markdown("### đ° Key Metrics")
from stocksense.data_collectors import get_price_history
with st.spinner("Loading real market metrics..."):
price_data = get_price_history(ticker, period="1mo")
if price_data is not None and not price_data.empty:
current_price = price_data['Close'].iloc[-1]
prev_price = price_data['Close'].iloc[-2] if len(price_data) > 1 else current_price
high_30d = price_data['High'].max()
low_30d = price_data['Low'].min()
returns = price_data['Close'].pct_change().dropna()
volatility = returns.std() * 100 if len(returns) > 1 else 0
col1, col2, col3, col4 = st.columns(4)
with col1:
change_pct = ((current_price - prev_price) / prev_price * 100) if prev_price != 0 else 0
st.metric(
"đĩ Current Price",
f"${current_price:.2f}",
f"{change_pct:+.2f}%"
)
with col2:
st.metric(
"đ 30D High",
f"${high_30d:.2f}",
help="Highest price in the last 30 days"
)
with col3:
st.metric(
"đ 30D Low",
f"${low_30d:.2f}",
help="Lowest price in the last 30 days"
)
with col4:
st.metric(
"đ Volatility",
f"{volatility:.1f}%",
help="Price volatility over the period"
)
st.caption("đ All metrics sourced from Yahoo Finance real-time data")
else:
st.error(f"â Unable to fetch real market data for {ticker}")
st.info("Please verify the ticker symbol and try again.")
def clear_database_cache() -> bool:
"""Clear all cached analysis results from the database."""
try:
import sqlite3
import os
# Get database path (same logic as in database.py)
current_file_dir = os.path.dirname(os.path.abspath(__file__))
db_path = os.path.join(current_file_dir, 'stocksense.db')
if os.path.exists(db_path):
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute('DELETE FROM analysis_cache')
rows_deleted = cursor.rowcount
conn.commit()
return True, rows_deleted
else:
return True, 0 # No database file, so "cleared"
except Exception as e:
return False, str(e)
def get_cache_stats() -> dict:
"""Get statistics about cached analysis results."""
try:
import sqlite3
import os
current_file_dir = os.path.dirname(os.path.abspath(__file__))
db_path = os.path.join(current_file_dir, 'stocksense.db')
if not os.path.exists(db_path):
return {"total_analyses": 0, "unique_tickers": 0, "db_size_mb": 0}
# Get file size
db_size_mb = round(os.path.getsize(db_path) / (1024 * 1024), 2)
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
# Check if table exists
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='analysis_cache'")
if not cursor.fetchone():
return {"total_analyses": 0, "unique_tickers": 0, "db_size_mb": db_size_mb}
# Get total analyses
cursor.execute('SELECT COUNT(*) FROM analysis_cache')
total_analyses = cursor.fetchone()[0]
# Get unique tickers
cursor.execute('SELECT COUNT(DISTINCT ticker) FROM analysis_cache')
unique_tickers = cursor.fetchone()[0]
return {
"total_analyses": total_analyses,
"unique_tickers": unique_tickers,
"db_size_mb": db_size_mb
}
except Exception as e:
return {"total_analyses": 0, "unique_tickers": 0, "db_size_mb": 0, "error": str(e)}
def get_cached_tickers() -> list:
"""Get list of all cached ticker symbols."""
try:
import sqlite3
import os
current_file_dir = os.path.dirname(os.path.abspath(__file__))
db_path = os.path.join(current_file_dir, 'stocksense.db')
if not os.path.exists(db_path):
return []
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
# Check if table exists
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='analysis_cache'")
if not cursor.fetchone():
return []
cursor.execute('''
SELECT DISTINCT ticker, MAX(timestamp) as latest_timestamp, COUNT(*) as analysis_count
FROM analysis_cache
GROUP BY ticker
ORDER BY latest_timestamp DESC
''')
results = cursor.fetchall()
return [{"ticker": result[0], "latest": result[1], "count": result[2]} for result in results]
except Exception as e:
return []
def display_analysis_history():
"""Displays analysis history in sidebar."""
if st.session_state.analysis_history:
st.markdown("### đ Recent Analyses")
for i, analysis in enumerate(st.session_state.analysis_history[:5]):
ticker = analysis['ticker']
timestamp = analysis['timestamp']
dt = datetime.fromisoformat(timestamp)
time_str = dt.strftime("%m/%d %H:%M")
if st.button(f"đ {ticker} - {time_str}", key=f"history_{i}"):
st.session_state.analysis_result = analysis
st.rerun()
def display_sidebar():
with st.sidebar:
st.markdown("### đ§ System Status")
status = check_backend_status()
if status:
st.markdown('đĸ Backend Online
', unsafe_allow_html=True)
else:
st.markdown('đ´ Backend Offline
', unsafe_allow_html=True)
st.warning("Start the FastAPI server to enable analysis")
display_analysis_history()
st.markdown("---")
st.markdown("### âšī¸ About ReAct Agent")
with st.expander("How it works"):
st.markdown("""
**ReAct Pattern (Reasoning + Action):**
1. đ§ **Reasons** about market conditions
2. đ§ **Acts** by selecting appropriate tools
3. đ **Observes** the results
4. đ **Adapts** strategy based on findings
5. â
**Concludes** with comprehensive analysis
**Features:**
- Autonomous decision making
- Dynamic tool selection
- Real-time sentiment analysis
- Market trend identification
""")
with st.expander("Data Sources"):
st.markdown("""
- đ° **NewsAPI**: Latest market news
- đ **Yahoo Finance**: Price data
- đ¤ **Google Gemini**: AI analysis
- đž **SQLite**: Result caching
""")
st.markdown("---")
# Cache Management Section
st.markdown("### đž Cache Management")
# Get cache statistics
cache_stats = get_cache_stats()
if cache_stats.get("error"):
st.error(f"â Cache error: {cache_stats['error']}")
else:
# Display cache statistics in a nice format
col1, col2 = st.columns(2)
with col1:
st.metric("đ Total Analyses", cache_stats["total_analyses"])
with col2:
st.metric("đ¯ Unique Stocks", cache_stats["unique_tickers"])
if cache_stats["db_size_mb"] > 0:
st.metric("đž Database Size", f"{cache_stats['db_size_mb']} MB")
# Clear cache section
if cache_stats["total_analyses"] > 0:
with st.expander("đī¸ Clear Cache", expanded=False):
st.markdown(f"""
**Current Cache Status:**
- {cache_stats['total_analyses']} cached analyses
- {cache_stats['unique_tickers']} different stocks
- {cache_stats['db_size_mb']} MB database size
""")
# Show cached tickers
cached_tickers = get_cached_tickers()
if cached_tickers:
st.markdown("**Cached Stocks:**")
ticker_display = ", ".join([f"`{item['ticker']}`" for item in cached_tickers[:10]])
if len(cached_tickers) > 10:
ticker_display += f" *(+{len(cached_tickers)-10} more)*"
st.markdown(ticker_display)
st.markdown("""
â ī¸ **Warning:** This will permanently delete all cached analysis results.
Fresh analyses will take longer but will use the latest data.
""")
# Confirmation checkbox
confirm_clear = st.checkbox("I understand this action cannot be undone", key="confirm_cache_clear")
# Clear button (only enabled when confirmed)
if st.button(
"đī¸ Clear All Cached Results",
type="secondary",
disabled=not confirm_clear,
help="Permanently delete all cached analysis results",
use_container_width=True
):
with st.spinner("Clearing cache..."):
success, result = clear_database_cache()
if success:
st.success(f"â
Successfully cleared {result} cached analyses!")
# Also clear session state
st.session_state.analysis_result = None
st.session_state.analysis_history = []
time.sleep(1) # Brief pause to show success message
st.rerun()
else:
st.error(f"â Failed to clear cache: {result}")
else:
st.info("đ No cached results to clear")
st.markdown("---")
if st.button("đī¸ Clear Session Data", help="Clear current analysis results and history from this session only"):
st.session_state.analysis_result = None
st.session_state.analysis_history = []
st.rerun()
def main():
"""Main function for the Streamlit application."""
display_hero_section()
display_sidebar()
main_container = st.container()
with main_container:
ticker = display_ticker_input()
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
is_valid, error_msg = validate_ticker(ticker)
if not is_valid and ticker:
st.error(f"â {error_msg}")
analyze_button = st.button(
"đ Analyze with ReAct Agent",
type="primary",
use_container_width=True,
disabled=not is_valid,
help="Trigger autonomous AI analysis using the ReAct pattern"
)
if analyze_button and is_valid:
result = trigger_analysis(ticker)
if result and result.get('success'):
st.success(f"â
Analysis completed for **{ticker}**!")
st.markdown("---")
if st.session_state.analysis_result:
result_data = st.session_state.analysis_result
ticker = result_data['ticker']
data = result_data['data']
col1, col2 = st.columns([4, 1])
with col1:
st.markdown(f"## đ Analysis Results: {ticker}")
with col2:
if st.button("đī¸ Clear", help="Clear current results"):
st.session_state.analysis_result = None
st.rerun()
st.markdown("
", unsafe_allow_html=True)
# Use the enhanced tabbed display
display_enhanced_analysis_results(data, ticker)
else:
st.markdown("""
đ Welcome to StockSense AI Agent
Select a stock ticker above to begin your AI-powered market analysis
Our ReAct agent will autonomously reason about market conditions
and select the best tools for comprehensive analysis
""", unsafe_allow_html=True)
if __name__ == "__main__":
main()