import gradio as gr
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import plotly.graph_objects as go
import plotly.express as px
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')
# Hugging Face imports with error handling
try:
from transformers import pipeline
from sentence_transformers import SentenceTransformer
import torch
HF_AVAILABLE = True
except ImportError:
HF_AVAILABLE = False
print("Hugging Face libraries not available. Running in lightweight mode.")
class FinancialSpacesAgent:
"""Optimized Financial Agent for Hugging Face Spaces"""
def __init__(self):
print("đ Initializing Financial Analysis Agent for Spaces...")
self.models_loaded = False
self.company_cache = {}
self.load_models_optimized()
def load_models_optimized(self):
"""Load models optimized for Spaces memory constraints"""
try:
if not HF_AVAILABLE:
self.models_loaded = False
print("Running in lightweight mode (no AI models)")
return
# Load lightweight models
print("đĻ Loading AI models...")
# 1. Lightweight sentiment model
try:
self.sentiment_model = pipeline(
"text-classification",
model="ProsusAI/finbert",
device_map="auto" if torch.cuda.is_available() else "cpu",
max_length=128,
truncation=True
)
print("â
Financial sentiment model loaded")
except:
print("â ī¸ Could not load FinBERT, using smaller model")
self.sentiment_model = pipeline(
"sentiment-analysis",
model="distilbert-base-uncased-finetuned-sst-2-english",
device_map="auto" if torch.cuda.is_available() else "cpu"
)
# 2. Mini summarizer (skip to save memory)
self.summarizer = None
print("â ī¸ Summarization model skipped to save memory")
# 3. Lightweight embeddings
try:
self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
print("â
Embeddings model loaded")
except:
self.embedder = None
# 4. Text generation (skip to save memory)
self.text_generator = None
self.models_loaded = True
print("đ All models loaded successfully!")
except Exception as e:
print(f"â ī¸ Model loading failed: {e}")
print("Running in data-only mode")
self.models_loaded = False
def fetch_company_data(self, ticker):
"""Fetch and cache company data"""
cache_key = f"{ticker}_{datetime.now().strftime('%Y%m%d')}"
if cache_key in self.company_cache:
return self.company_cache[cache_key]
try:
print(f"đĨ Fetching data for {ticker}...")
stock = yf.Ticker(ticker)
# Basic info with timeout protection
info = stock.info
# Get essential financials
data = {
'ticker': ticker,
'info': info,
'balance_sheet': stock.balance_sheet,
'income_statement': stock.financials,
'cash_flow': stock.cashflow,
'history_1y': stock.history(period="1y"),
'history_5y': stock.history(period="5y"),
'dividends': stock.dividends,
'news': getattr(stock, 'news', [])[:5] # Limit news to save memory
}
# Calculate key metrics
data['metrics'] = self.calculate_key_metrics(data)
# Cache for today
self.company_cache[cache_key] = data
return data
except Exception as e:
print(f"Error fetching data for {ticker}: {e}")
return {'error': str(e), 'ticker': ticker}
def calculate_key_metrics(self, data):
"""Calculate essential financial metrics"""
metrics = {}
try:
info = data.get('info', {})
inc = data.get('income_statement', pd.DataFrame())
bal = data.get('balance_sheet', pd.DataFrame())
if not inc.empty and not bal.empty:
# Latest period data
if len(inc.columns) > 0:
latest_inc = inc.iloc[:, 0]
if len(bal.columns) > 0:
latest_bal = bal.iloc[:, 0]
# Profitability
if 'Total Revenue' in latest_inc.index and latest_inc['Total Revenue'] != 0:
if 'Net Income' in latest_inc.index:
metrics['net_margin'] = (latest_inc['Net Income'] / latest_inc['Total Revenue'] * 100)
# ROE
if 'Total Stockholder Equity' in latest_bal.index and latest_bal['Total Stockholder Equity'] != 0:
if 'Net Income' in latest_inc.index:
metrics['roe'] = (latest_inc['Net Income'] / latest_bal['Total Stockholder Equity'] * 100)
# Liquidity
if 'Total Current Assets' in latest_bal.index and 'Total Current Liabilities' in latest_bal.index:
if latest_bal['Total Current Liabilities'] != 0:
metrics['current_ratio'] = (latest_bal['Total Current Assets'] / latest_bal['Total Current Liabilities'])
# Leverage
if 'Total Liab' in latest_bal.index and 'Total Stockholder Equity' in latest_bal.index:
if latest_bal['Total Stockholder Equity'] != 0:
metrics['debt_to_equity'] = (latest_bal['Total Liab'] / latest_bal['Total Stockholder Equity'])
# Market metrics
if info:
metrics['pe_ratio'] = info.get('trailingPE', 'N/A')
metrics['pb_ratio'] = info.get('priceToBook', 'N/A')
metrics['dividend_yield'] = info.get('dividendYield', 0) * 100 if info.get('dividendYield') else 0
metrics['market_cap'] = info.get('marketCap', 'N/A')
metrics['beta'] = info.get('beta', 'N/A')
except Exception as e:
print(f"Error calculating metrics: {e}")
return metrics
def create_dashboard(self, ticker, data):
"""Create interactive Plotly dashboard"""
try:
metrics = data.get('metrics', {})
history = data.get('history_1y', pd.DataFrame())
# Create subplots
from plotly.subplots import make_subplots
fig = make_subplots(
rows=2, cols=2,
subplot_titles=(
'đ Price Trend',
'đ° Key Metrics',
'đ Financial Health',
'đ¯ Recommendation'
),
specs=[
[{"type": "scatter"}, {"type": "bar"}],
[{"type": "indicator"}, {"type": "indicator"}]
],
vertical_spacing=0.2,
horizontal_spacing=0.15
)
# 1. Price Trend
if not history.empty:
fig.add_trace(
go.Scatter(
x=history.index,
y=history['Close'],
mode='lines',
name='Price',
line=dict(color='#2E86AB', width=2),
fill='tozeroy',
fillcolor='rgba(46, 134, 171, 0.1)'
),
row=1, col=1
)
# Add moving average
if len(history) > 20:
ma20 = history['Close'].rolling(window=20).mean()
fig.add_trace(
go.Scatter(
x=history.index,
y=ma20,
mode='lines',
name='MA20',
line=dict(color='#F18F01', width=1, dash='dash')
),
row=1, col=1
)
fig.update_xaxes(title_text="Date", row=1, col=1)
fig.update_yaxes(title_text="Price ($)", row=1, col=1)
# 2. Key Metrics Bar Chart
if metrics:
metric_keys = ['roe', 'net_margin', 'current_ratio', 'debt_to_equity']
metric_labels = ['ROE %', 'Net Margin %', 'Current Ratio', 'Debt/Equity']
metric_values = []
metric_colors = []
for key in metric_keys:
val = metrics.get(key, 0)
if isinstance(val, (int, float)):
metric_values.append(val)
# Color coding
if key == 'roe':
metric_colors.append('#2E86AB' if val > 15 else '#A23B72' if val > 8 else '#C73E1D')
elif key == 'net_margin':
metric_colors.append('#2E86AB' if val > 10 else '#A23B72' if val > 5 else '#C73E1D')
elif key == 'current_ratio':
metric_colors.append('#2E86AB' if val > 1.5 else '#A23B72' if val > 1 else '#C73E1D')
elif key == 'debt_to_equity':
metric_colors.append('#2E86AB' if val < 1 else '#A23B72' if val < 2 else '#C73E1D')
else:
metric_values.append(0)
metric_colors.append('#CCCCCC')
fig.add_trace(
go.Bar(
x=metric_labels,
y=metric_values,
marker_color=metric_colors,
text=[f'{v:.1f}' if isinstance(v, (int, float)) else 'N/A' for v in metric_values],
textposition='auto',
name='Metrics'
),
row=1, col=2
)
fig.update_xaxes(title_text="Metrics", row=1, col=2, tickangle=45)
fig.update_yaxes(title_text="Value", row=1, col=2)
# 3. Financial Health Gauge
health_score = self.calculate_health_score(metrics)
fig.add_trace(
go.Indicator(
mode="gauge+number",
value=health_score,
title={'text': "Financial Health"},
gauge={
'axis': {'range': [None, 10]},
'bar': {'color': "darkblue"},
'steps': [
{'range': [0, 3], 'color': "red"},
{'range': [3, 7], 'color': "yellow"},
{'range': [7, 10], 'color': "green"}
],
'threshold': {
'line': {'color': "black", 'width': 4},
'thickness': 0.75,
'value': health_score
}
}
),
row=2, col=1
)
# 4. Recommendation Indicator
recommendation = self.generate_recommendation(metrics)
color_map = {
'Strong Buy': 'green',
'Buy': 'lightgreen',
'Hold': 'yellow',
'Sell': 'orange',
'Strong Sell': 'red'
}
fig.add_trace(
go.Indicator(
mode="number",
value=health_score,
title={'text': f"Recommendation: {recommendation}"},
number={'font': {'color': color_map.get(recommendation, 'black'), 'size': 24}},
domain={'row': 2, 'column': 2}
),
row=2, col=2
)
# Update layout
fig.update_layout(
height=700,
showlegend=True,
title_text=f"đ Financial Dashboard - {ticker}",
title_font_size=20,
template="plotly_white"
)
return fig
except Exception as e:
print(f"Error creating dashboard: {e}")
# Return simple figure
fig = go.Figure()
fig.add_annotation(
text="Dashboard generation failed
Try a different ticker",
xref="paper", yref="paper",
x=0.5, y=0.5,
showarrow=False,
font_size=16,
align="center"
)
fig.update_layout(
title=f"Error for {ticker}",
height=400
)
return fig
def calculate_health_score(self, metrics):
"""Calculate financial health score (1-10)"""
score = 5.0 # Start at neutral
try:
# ROE contribution
roe = metrics.get('roe', 0)
if isinstance(roe, (int, float)):
if roe > 20:
score += 2
elif roe > 10:
score += 1
elif roe < 5:
score -= 1
# Net Margin contribution
margin = metrics.get('net_margin', 0)
if isinstance(margin, (int, float)):
if margin > 15:
score += 1.5
elif margin > 5:
score += 0.5
elif margin < 0:
score -= 1.5
# Liquidity contribution
current = metrics.get('current_ratio', 0)
if isinstance(current, (int, float)):
if current > 2:
score += 1
elif current < 1:
score -= 1
# Leverage contribution
debt_eq = metrics.get('debt_to_equity', 0)
if isinstance(debt_eq, (int, float)):
if debt_eq > 2:
score -= 1.5
elif debt_eq < 0.5:
score += 0.5
# Cap score between 1 and 10
score = max(1, min(10, score))
except:
pass
return round(score, 1)
def generate_recommendation(self, metrics):
"""Generate investment recommendation"""
score = self.calculate_health_score(metrics)
if score >= 8:
return "Strong Buy"
elif score >= 6:
return "Buy"
elif score >= 4:
return "Hold"
elif score >= 2:
return "Sell"
else:
return "Strong Sell"
def generate_ai_analysis(self, ticker, data):
"""Generate AI analysis of company"""
try:
metrics = data.get('metrics', {})
# Create analysis
analysis = f"""
## đ¤ AI Analysis for {ticker}
### đ Key Metrics Analysis:
"""
# Add metric analysis
metric_analysis = {
'roe': "Return on Equity measures how effectively management is using shareholders' funds.",
'net_margin': "Net Profit Margin shows what percentage of revenue is actual profit.",
'current_ratio': "Current Ratio indicates short-term liquidity and ability to pay debts.",
'debt_to_equity': "Debt-to-Equity ratio shows the company's financial leverage."
}
for key, desc in metric_analysis.items():
val = metrics.get(key, 'N/A')
if isinstance(val, (int, float)):
analysis += f"**{key.replace('_', ' ').title()}**: {val:.1f} - {desc}\n\n"
# Add overall assessment
health_score = self.calculate_health_score(metrics)
recommendation = self.generate_recommendation(metrics)
analysis += f"""
### đĨ Financial Health Score: **{health_score}/10**
### đ¯ Recommendation: **{recommendation}**
### đ Market Outlook:
"""
# Add simple outlook based on metrics
if health_score >= 7:
analysis += "The company shows strong financial fundamentals with good profitability and manageable leverage."
elif health_score >= 5:
analysis += "The company has average financial health with room for improvement in some areas."
else:
analysis += "The company shows concerning financial metrics that warrant caution."
analysis += "\n\n*Note: This analysis is AI-generated for educational purposes only.*"
return analysis
except Exception as e:
return f"Error in AI analysis: {str(e)}"
def create_metrics_table(self, metrics):
"""Create a nice HTML table for metrics"""
if not metrics:
return "
No metrics available
" table_html = """| Metric | Value | Assessment |
|---|---|---|
| {key.replace('_', ' ').title()} | {display_value} | {assessment} |