|
|
"""Data fetching and processing utilities for the financial dashboard.""" |
|
|
|
|
|
import pandas as pd |
|
|
from openbb import sdk |
|
|
import streamlit as st |
|
|
|
|
|
|
|
|
@st.cache_data(ttl=3600) |
|
|
def load_stock_data(symbol: str) -> pd.DataFrame: |
|
|
"""Load historical stock price data with caching.""" |
|
|
df = sdk.equity.price.historical(symbol=symbol).to_dataframe() |
|
|
return df |
|
|
|
|
|
|
|
|
@st.cache_data(ttl=86400) |
|
|
def load_company_profile(symbol: str): |
|
|
"""Load company profile information with caching.""" |
|
|
profile_response = sdk.equity.profile(symbol=symbol) |
|
|
profile_info = profile_response.results[0] if hasattr(profile_response, 'results') and profile_response.results else None |
|
|
return profile_info |
|
|
|
|
|
|
|
|
@st.cache_data(ttl=86400) |
|
|
def load_income_statement(symbol: str) -> pd.DataFrame: |
|
|
"""Load company income statement data with caching.""" |
|
|
income_stmt = sdk.equity.fundamental.income(symbol=symbol).to_dataframe() |
|
|
return income_stmt |
|
|
|
|
|
|
|
|
def calculate_technical_indicators(df: pd.DataFrame, period: int) -> pd.DataFrame: |
|
|
"""Calculate SMA, EMA, and RSI indicators.""" |
|
|
df = df.copy() |
|
|
df["SMA"] = df["close"].rolling(period).mean() |
|
|
df["EMA"] = df["close"].ewm(span=period, adjust=False).mean() |
|
|
|
|
|
|
|
|
delta = df["close"].diff() |
|
|
gain = delta.clip(lower=0) |
|
|
loss = -1 * delta.clip(upper=0) |
|
|
avg_gain = gain.rolling(period).mean() |
|
|
avg_loss = loss.rolling(period).mean() |
|
|
rs = avg_gain / avg_loss |
|
|
df["RSI"] = 100 - (100 / (1 + rs)) |
|
|
|
|
|
return df |
|
|
|
|
|
|
|
|
def get_price_metrics(df: pd.DataFrame) -> dict: |
|
|
"""Calculate key price metrics.""" |
|
|
current_price = df["close"].iloc[-1] |
|
|
prev_close = df["close"].iloc[-2] if len(df) > 1 else df["close"].iloc[0] |
|
|
price_change = current_price - prev_close |
|
|
price_change_pct = (price_change / prev_close) * 100 if prev_close != 0 else 0 |
|
|
|
|
|
return { |
|
|
"current_price": current_price, |
|
|
"price_change": price_change, |
|
|
"price_change_pct": price_change_pct, |
|
|
"high_52w": df['high'].max(), |
|
|
"low_52w": df['low'].min(), |
|
|
} |
|
|
|
|
|
|
|
|
def get_profitability_metrics(income_data: pd.Series) -> dict: |
|
|
"""Calculate profitability metrics from income statement.""" |
|
|
total_rev = income_data.get('total_revenue', 0) |
|
|
gross_prof = income_data.get('gross_profit', 0) |
|
|
net_inc = income_data.get('net_income', 0) |
|
|
operating_inc = income_data.get('operating_income', 0) |
|
|
|
|
|
metrics = {} |
|
|
|
|
|
if total_rev and total_rev > 0: |
|
|
metrics["gross_margin"] = (gross_prof / total_rev) * 100 if pd.notna(gross_prof) else 0 |
|
|
metrics["net_margin"] = (net_inc / total_rev) * 100 if pd.notna(net_inc) else 0 |
|
|
if operating_inc: |
|
|
metrics["operating_margin"] = (operating_inc / total_rev) * 100 |
|
|
else: |
|
|
metrics = {"gross_margin": 0, "net_margin": 0} |
|
|
|
|
|
return metrics |
|
|
|
