File size: 2,967 Bytes
50e9fe0 |
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 83 84 85 86 87 88 89 |
"""Data fetching and processing utilities for the financial dashboard."""
import pandas as pd
from openbb import sdk
def load_stock_data(symbol: str) -> pd.DataFrame:
"""Load historical stock price data."""
df = sdk.equity.price.historical(symbol=symbol).to_dataframe()
return df
def load_company_profile(symbol: str):
"""Load company profile information."""
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
def load_income_statement(symbol: str) -> pd.DataFrame:
"""Load company income statement data."""
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["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 format_financial_value(value) -> str:
"""Format financial values with appropriate units."""
if pd.isna(value):
return "N/A"
if abs(value) >= 1e9:
return f"${value/1e9:.2f}B"
elif abs(value) >= 1e6:
return f"${value/1e6:.2f}M"
else:
return f"${value:.2f}"
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
|