File size: 2,812 Bytes
f7323a3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 |
"""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()
# Calculate RSI
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
|