phase 6: financial statements

atlas-terminal/server/core/cache.py

@@ -24,6 +24,7 @@ class CachedGateway(DataGateway):
         "quote": 30,
         "profile": 86_400,
         "fundamentals": 43_200,
+        "financials": 43_200,
         "segments": 604_800,
         "history": 300,
         "news": 300,
@@ -64,6 +65,9 @@ class CachedGateway(DataGateway):
     async def fundamentals(self, symbol: str, period: str = "annual") -> Fundamentals:
         return await self._cached("fundamentals", (symbol, period), lambda: self.inner.fundamentals(symbol, period))
 
+    async def financials(self, symbol: str, statement: str = "income", period: str = "annual") -> dict[str, Any]:
+        return await self._cached("financials", (symbol, statement, period), lambda: self.inner.financials(symbol, statement, period))
+
     async def history(self, symbol: str, range: str = "1y") -> OHLCV:
         return await self._cached("history", (symbol, range), lambda: self.inner.history(symbol, range))
 

atlas-terminal/server/core/chained_gateway.py

@@ -57,6 +57,9 @@ class ChainedGateway(DataGateway):
     async def fundamentals(self, symbol: str, period: str = "annual") -> Fundamentals:
         return await self._try(symbol, "fundamentals", lambda provider: provider.fundamentals(symbol, period))
 
+    async def financials(self, symbol: str, statement: str = "income", period: str = "annual") -> dict:
+        return await self._try(symbol, "financials", lambda provider: provider.financials(symbol, statement, period))
+
     async def history(self, symbol: str, range: str = "1y") -> OHLCV:
         return await self._try(symbol, "history", lambda provider: provider.history(symbol, range))
 

atlas-terminal/server/core/data_gateway.py

@@ -130,6 +130,8 @@ class DataGateway(Protocol):
 
     async def fundamentals(self, symbol: str, period: str = "annual") -> Fundamentals: ...
 
+    async def financials(self, symbol: str, statement: str = "income", period: str = "annual") -> dict[str, Any]: ...
+
     async def history(self, symbol: str, range: str = "1y") -> OHLCV: ...
 
     async def news(self, symbols: list[str], limit: int = 20) -> list[Article]: ...

atlas-terminal/server/core/providers/base.py

@@ -55,6 +55,9 @@ class BaseProvider:
     async def fundamentals(self, symbol: str, period: str = "annual") -> Fundamentals:
         raise ProviderNotImplemented(f"{self.name}.fundamentals")
 
+    async def financials(self, symbol: str, statement: str = "income", period: str = "annual") -> dict[str, Any]:
+        raise ProviderNotImplemented(f"{self.name}.financials")
+
     async def history(self, symbol: str, range: str = "1y") -> OHLCV:
         raise ProviderNotImplemented(f"{self.name}.history")
 

atlas-terminal/server/core/providers/fmp.py

@@ -59,3 +59,39 @@ class FMPProvider(BaseProvider):
 
     async def segments(self, symbol: str) -> list[Segment]:
         raise ProviderNotImplemented("FMP segments parser is planned for Phase 1.4")
+
+    async def financials(self, symbol: str, statement: str = "income", period: str = "annual") -> dict[str, object]:
+        normalized = symbol.strip().upper()
+        statement_key = statement.strip().lower()
+        path_map = {
+            "income": "income-statement",
+            "balance": "balance-sheet-statement",
+            "cashflow": "cash-flow-statement",
+            "cash_flow": "cash-flow-statement",
+        }
+        path = path_map.get(statement_key)
+        if not path:
+            raise ProviderError(f"unsupported statement: {statement}")
+        period_key = "quarter" if period.strip().lower().startswith("q") else "annual"
+        data = await self._get_json(f"/{path}/{normalized}", {"period": period_key, "limit": 5})
+        rows = data if isinstance(data, list) else []
+        if not rows:
+            raise ProviderError("missing financial statement rows")
+        periods = [str(row.get("date") or row.get("calendarYear") or idx) for idx, row in enumerate(rows)]
+        line_items: dict[str, list[float | int | None]] = {}
+        for row in rows:
+            if not isinstance(row, dict):
+                continue
+            for key, value in row.items():
+                if key in {"date", "symbol", "reportedCurrency", "cik", "fillingDate", "acceptedDate", "calendarYear", "period", "link", "finalLink"}:
+                    continue
+                if isinstance(value, (int, float)) or value is None:
+                    line_items.setdefault(key, []).append(value)
+        return {
+            "ticker": normalized,
+            "statement": statement_key,
+            "period": period_key,
+            "source": self.name,
+            "periods": periods,
+            "line_items": line_items,
+        }

atlas-terminal/server/core/providers/yfinance.py

@@ -106,6 +106,49 @@ class YFinanceProvider(BaseProvider):
 
         return await self._to_thread(fetch)
 
+    async def financials(self, symbol: str, statement: str = "income", period: str = "annual") -> dict[str, Any]:
+        def fetch() -> dict[str, Any]:
+            import pandas as pd
+
+            normalized = symbol.strip().upper()
+            ticker = self._ticker(normalized)
+            statement_key = statement.strip().lower()
+            quarterly = period.strip().lower().startswith("q")
+            if statement_key == "income":
+                df = ticker.quarterly_income_stmt if quarterly else ticker.income_stmt
+            elif statement_key == "balance":
+                df = ticker.quarterly_balance_sheet if quarterly else ticker.balance_sheet
+            elif statement_key in {"cashflow", "cash_flow"}:
+                df = ticker.quarterly_cashflow if quarterly else ticker.cashflow
+            else:
+                raise ProviderError(f"unsupported statement: {statement}")
+            if df is None or not isinstance(df, pd.DataFrame) or df.empty:
+                raise ProviderError("missing financial statement dataframe")
+            sliced = df.iloc[:, :5]
+            periods = [str(col)[:10] for col in sliced.columns]
+            line_items: dict[str, list[float | None]] = {}
+            for idx, row in sliced.iterrows():
+                values: list[float | None] = []
+                for value in row.tolist():
+                    try:
+                        if pd.isna(value):
+                            values.append(None)
+                        else:
+                            values.append(float(value))
+                    except (TypeError, ValueError):
+                        values.append(None)
+                line_items[str(idx).replace(" ", "")] = values
+            return {
+                "ticker": normalized,
+                "statement": statement_key,
+                "period": "quarter" if quarterly else "annual",
+                "source": self.name,
+                "periods": periods,
+                "line_items": line_items,
+            }
+
+        return await self._to_thread(fetch)
+
     async def history(self, symbol: str, range: str = "1y") -> OHLCV:
         def fetch() -> OHLCV:
             hist = self._ticker(symbol).history(period=range)