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