| """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() |
|
|
| |
| 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 |
|
|
