summaryrefslogtreecommitdiff
path: root/backend/app
diff options
context:
space:
mode:
Diffstat (limited to 'backend/app')
-rw-r--r--backend/app/main.py7
-rw-r--r--backend/app/schemas.py21
-rw-r--r--backend/app/services/data_service.py176
3 files changed, 200 insertions, 4 deletions
diff --git a/backend/app/main.py b/backend/app/main.py
index fc98a5e..eb5fa40 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -9,7 +9,7 @@ from fastapi import FastAPI, HTTPException, Query, status
from fastapi.middleware.cors import CORSMiddleware
from app.db import watchlist
-from app.schemas import HistoryPoint, MarketIndex, SearchResult, TickerOverview, WatchlistResponse
+from app.schemas import FinancialsResponse, HistoryPoint, MarketIndex, SearchResult, TickerOverview, WatchlistResponse
from app.services import data_service
load_dotenv()
@@ -65,6 +65,11 @@ def ticker_history(symbol: str, period: str = Query(default="1y", pattern="^(1m|
return data_service.get_price_history(symbol, period=period)
+@app.get("/api/tickers/{symbol}/financials", response_model=FinancialsResponse)
+def ticker_financials(symbol: str, period: str = Query(default="annual", pattern="^(annual|quarterly)$")) -> dict:
+ return data_service.get_financials(symbol, period=period)
+
+
@app.get("/api/watchlist", response_model=WatchlistResponse)
def get_watchlist() -> dict:
items = []
diff --git a/backend/app/schemas.py b/backend/app/schemas.py
index 2ee6dac..86f586c 100644
--- a/backend/app/schemas.py
+++ b/backend/app/schemas.py
@@ -107,6 +107,27 @@ class HistoryPoint(BaseModel):
volume: float | None = None
+class FinancialRow(BaseModel):
+ label: str
+ indent: int = 0
+ is_total: bool = False
+ is_section: bool = False
+ is_margin: bool = False
+ values: list[float | None] = Field(default_factory=list)
+
+
+class FinancialStatement(BaseModel):
+ columns: list[str] = Field(default_factory=list)
+ rows: list[FinancialRow] = Field(default_factory=list)
+
+
+class FinancialsResponse(BaseModel):
+ period: Literal["annual", "quarterly"]
+ income: FinancialStatement
+ balance: FinancialStatement
+ cash_flow: FinancialStatement
+
+
class WatchlistItem(BaseModel):
symbol: str
created_at: str
diff --git a/backend/app/services/data_service.py b/backend/app/services/data_service.py
index da76bdb..f7188f7 100644
--- a/backend/app/services/data_service.py
+++ b/backend/app/services/data_service.py
@@ -19,6 +19,9 @@ HISTORY_CACHE = TTLCache(maxsize=256, ttl=300)
INTRADAY_CACHE = TTLCache(maxsize=128, ttl=60)
MARKET_CACHE = TTLCache(maxsize=8, ttl=300)
STATEMENT_CACHE = TTLCache(maxsize=256, ttl=3600)
+INCOME_CACHE = TTLCache(maxsize=256, ttl=3600)
+BALANCE_CACHE = TTLCache(maxsize=256, ttl=3600)
+CF_CACHE = TTLCache(maxsize=256, ttl=3600)
SHARES_CACHE = TTLCache(maxsize=256, ttl=3600)
RATIO_CACHE = TTLCache(maxsize=256, ttl=3600)
BETA_CACHE = TTLCache(maxsize=256, ttl=3600)
@@ -116,6 +119,173 @@ def _safe_ratio(num: float | None, den: float | None) -> float | None:
return num / den
+def _build_income(frame: pd.DataFrame, frame_q: pd.DataFrame, quarterly: bool) -> dict:
+ if frame is None or frame.empty:
+ return {"columns": [], "rows": []}
+ n = min(len(frame.columns), 8 if quarterly else 4)
+ col_labels = [_fmt_col(c, quarterly) for c in frame.columns[:n]]
+ if not quarterly:
+ col_labels.append("TTM")
+
+ def v(label: str) -> list[float | None]:
+ base = _row_vals(frame, label, n)
+ return base + ([_statement_ttm(frame_q, label)] if not quarterly else [])
+
+ def vm(*labels: str) -> list[float | None]:
+ base = _row_vals_multi(frame, n, *labels)
+ if not quarterly:
+ ttm = None
+ for lbl in labels:
+ ttm = _statement_ttm(frame_q, lbl)
+ if ttm is not None:
+ break
+ base = base + [ttm]
+ return base
+
+ rev = v("Total Revenue")
+ gross = v("Gross Profit")
+ net = v("Net Income")
+
+ return {
+ "columns": col_labels,
+ "rows": [
+ _fin_row("Total Revenue", 0, True, rev),
+ _fin_row("Cost of Revenue", 1, False, v("Cost Of Revenue")),
+ _fin_row("Gross Profit", 0, True, gross),
+ _fin_margin("gross margin", [_safe_ratio(g, r) for g, r in zip(gross, rev)]),
+ _fin_row("Operating Expenses", 1, False, v("Operating Expense")),
+ _fin_row("Operating Income", 0, True, v("Operating Income")),
+ _fin_row("EBITDA", 1, False, vm("EBITDA", "Normalized EBITDA")),
+ _fin_row("Interest Expense", 1, False, v("Interest Expense")),
+ _fin_row("Pretax Income", 0, False, v("Pretax Income")),
+ _fin_row("Tax Provision", 1, False, v("Tax Provision")),
+ _fin_row("Net Income", 0, True, net),
+ _fin_margin("net margin", [_safe_ratio(ni, r) for ni, r in zip(net, rev)]),
+ _fin_row("EPS Basic", 1, False, v("Basic EPS")),
+ ],
+ }
+
+
+def _build_balance(frame: pd.DataFrame, frame_q: pd.DataFrame, quarterly: bool) -> dict:
+ if frame is None or frame.empty:
+ return {"columns": [], "rows": []}
+ n = min(len(frame.columns), 8 if quarterly else 4)
+ col_labels = [_fmt_col(c, quarterly) for c in frame.columns[:n]]
+ if not quarterly:
+ col_labels.append("MRQ")
+
+ def v(*labels: str) -> list[float | None]:
+ base = _row_vals_multi(frame, n, *labels)
+ if not quarterly:
+ val = None
+ for lbl in labels:
+ val = _balance_value(frame_q, lbl)
+ if val is not None:
+ break
+ base = base + [val]
+ return base
+
+ return {
+ "columns": col_labels,
+ "rows": [
+ _fin_section("ASSETS"),
+ _fin_row("Current Assets", 0, True, v("Current Assets")),
+ _fin_row("Cash & Equivalents", 1, False, v("Cash And Cash Equivalents", "Cash Cash Equivalents And Short Term Investments")),
+ _fin_row("Short Term Investments", 1, False, v("Other Short Term Investments", "Short Term Investments")),
+ _fin_row("Receivables", 1, False, v("Receivables", "Net Receivables")),
+ _fin_row("Inventory", 1, False, v("Inventory")),
+ _fin_row("Total Assets", 0, True, v("Total Assets")),
+ _fin_section("LIABILITIES"),
+ _fin_row("Current Liabilities", 0, True, v("Current Liabilities")),
+ _fin_row("Accounts Payable", 1, False, v("Payables And Accrued Expenses", "Accounts Payable")),
+ _fin_row("Short Term Debt", 1, False, v("Current Debt", "Short Term Debt And Capital Lease Obligation")),
+ _fin_row("Long Term Debt", 1, False, v("Long Term Debt", "Long Term Debt And Capital Lease Obligation")),
+ _fin_row("Total Liabilities", 0, True, v("Total Liabilities Net Minority Interest", "Total Liabilities")),
+ _fin_section("EQUITY"),
+ _fin_row("Stockholders Equity", 0, True, v("Stockholders Equity", "Common Stock Equity")),
+ ],
+ }
+
+
+def _build_cash_flow(cf: pd.DataFrame, cf_q: pd.DataFrame, inc: pd.DataFrame, inc_q: pd.DataFrame, quarterly: bool) -> dict:
+ if cf is None or cf.empty:
+ return {"columns": [], "rows": []}
+ n = min(len(cf.columns), 8 if quarterly else 4)
+ col_labels = [_fmt_col(c, quarterly) for c in cf.columns[:n]]
+ if not quarterly:
+ col_labels.append("TTM")
+
+ def cv(*labels: str) -> list[float | None]:
+ base = _row_vals_multi(cf, n, *labels)
+ if not quarterly:
+ ttm = None
+ for lbl in labels:
+ ttm = _statement_ttm(cf_q, lbl)
+ if ttm is not None:
+ break
+ base = base + [ttm]
+ return base
+
+ def iv(*labels: str) -> list[float | None]:
+ base = _row_vals_multi(inc, n, *labels)
+ if not quarterly:
+ ttm = None
+ for lbl in labels:
+ ttm = _statement_ttm(inc_q, lbl)
+ if ttm is not None:
+ break
+ base = base + [ttm]
+ return base
+
+ op_cf = cv("Operating Cash Flow", "Cash Flow From Continuing Operating Activities")
+ capex = cv("Capital Expenditure")
+ # CapEx is negative in yfinance; FCF = Operating CF + CapEx
+ fcf = [a + b if a is not None and b is not None else None for a, b in zip(op_cf, capex)]
+ rev = iv("Total Revenue")
+
+ return {
+ "columns": col_labels,
+ "rows": [
+ _fin_section("OPERATING"),
+ _fin_row("Net Income", 1, False, iv("Net Income")),
+ _fin_row("D&A", 1, False, cv("Depreciation And Amortization", "Reconciled Depreciation")),
+ _fin_row("Changes in Working Capital", 1, False, cv("Change In Working Capital")),
+ _fin_row("Operating Cash Flow", 0, True, op_cf),
+ _fin_section("INVESTING"),
+ _fin_row("CapEx", 1, False, capex),
+ _fin_row("Free Cash Flow", 0, True, fcf),
+ _fin_margin("FCF margin", [_safe_ratio(f, r) for f, r in zip(fcf, rev)]),
+ _fin_row("Investing Cash Flow", 0, True, cv("Investing Cash Flow", "Cash Flow From Continuing Investing Activities")),
+ _fin_section("FINANCING"),
+ _fin_row("Dividends Paid", 1, False, cv("Cash Dividends Paid", "Common Stock Dividend Paid")),
+ _fin_row("Buybacks", 1, False, cv("Repurchase Of Capital Stock", "Common Stock Repurchase")),
+ _fin_row("Financing Cash Flow", 0, True, cv("Financing Cash Flow", "Cash Flow From Continuing Financing Activities")),
+ _fin_row("Net Change in Cash", 0, True, cv("Changes In Cash", "End Cash Position")),
+ ],
+ }
+
+
+@cached(FINANCIALS_CACHE)
+def get_financials(symbol: str, period: str = "annual") -> dict:
+ sym = normalize_symbol(symbol)
+ quarterly = period == "quarterly"
+
+ inc = get_income_statement(sym, quarterly=quarterly)
+ bal = get_balance_sheet(sym, quarterly=quarterly)
+ cf = get_cash_flow(sym, quarterly=quarterly)
+
+ inc_q = get_income_statement(sym, quarterly=True) if not quarterly else inc
+ bal_q = get_balance_sheet(sym, quarterly=True) if not quarterly else bal
+ cf_q = get_cash_flow(sym, quarterly=True) if not quarterly else cf
+
+ return {
+ "period": period,
+ "income": _build_income(inc, inc_q, quarterly),
+ "balance": _build_balance(bal, bal_q, quarterly),
+ "cash_flow": _build_cash_flow(cf, cf_q, inc, inc_q, quarterly),
+ }
+
+
def _balance_value(frame: pd.DataFrame, *labels: str) -> float | None:
if frame is None or frame.empty:
return None
@@ -281,7 +451,7 @@ def get_intraday_history(symbol: str, period: str, interval: str) -> list[dict[s
return []
-@cached(STATEMENT_CACHE)
+@cached(INCOME_CACHE)
def get_income_statement(symbol: str, quarterly: bool = False) -> pd.DataFrame:
try:
ticker = yf.Ticker(normalize_symbol(symbol))
@@ -291,7 +461,7 @@ def get_income_statement(symbol: str, quarterly: bool = False) -> pd.DataFrame:
return pd.DataFrame()
-@cached(STATEMENT_CACHE)
+@cached(BALANCE_CACHE)
def get_balance_sheet(symbol: str, quarterly: bool = False) -> pd.DataFrame:
try:
ticker = yf.Ticker(normalize_symbol(symbol))
@@ -301,7 +471,7 @@ def get_balance_sheet(symbol: str, quarterly: bool = False) -> pd.DataFrame:
return pd.DataFrame()
-@cached(STATEMENT_CACHE)
+@cached(CF_CACHE)
def get_cash_flow(symbol: str, quarterly: bool = False) -> pd.DataFrame:
try:
ticker = yf.Ticker(normalize_symbol(symbol))
generated by cgit v1.3-2-g0d8e (git 2.53.0) at 2026-06-05 03:47:02 +0000