summaryrefslogtreecommitdiff
path: root/backend/app/services
diff options
context:
space:
mode:
Diffstat (limited to 'backend/app/services')
-rw-r--r--backend/app/services/data_service.py42
1 files changed, 42 insertions, 0 deletions
diff --git a/backend/app/services/data_service.py b/backend/app/services/data_service.py
index 95650ad..da76bdb 100644
--- a/backend/app/services/data_service.py
+++ b/backend/app/services/data_service.py
@@ -23,6 +23,7 @@ SHARES_CACHE = TTLCache(maxsize=256, ttl=3600)
RATIO_CACHE = TTLCache(maxsize=256, ttl=3600)
BETA_CACHE = TTLCache(maxsize=256, ttl=3600)
SHORT_CACHE = TTLCache(maxsize=256, ttl=3600)
+FINANCIALS_CACHE = TTLCache(maxsize=128, ttl=3600)
PERIODS = {"1m", "3m", "6m", "1y", "2y", "5y"}
YF_PERIOD_MAP = {"1m": "1mo", "3m": "3mo", "6m": "6mo", "1y": "1y", "2y": "2y", "5y": "5y"}
@@ -74,6 +75,47 @@ def _cap_ratio(value: float | None, lower: float, upper: float) -> float | None:
return value
+def _fmt_col(ts: Any, quarterly: bool) -> str:
+ t = pd.Timestamp(ts)
+ if quarterly:
+ q = (t.month - 1) // 3 + 1
+ return f"Q{q} {t.year}"
+ return f"FY {t.year}"
+
+
+def _row_vals(frame: pd.DataFrame, label: str, n: int) -> list[float | None]:
+ if frame is None or frame.empty or label not in frame.index:
+ return [None] * n
+ series = pd.to_numeric(frame.loc[label], errors="coerce")
+ return [_safe_float(series.iloc[i]) if i < len(series) else None for i in range(n)]
+
+
+def _row_vals_multi(frame: pd.DataFrame, n: int, *labels: str) -> list[float | None]:
+ for label in labels:
+ vals = _row_vals(frame, label, n)
+ if any(v is not None for v in vals):
+ return vals
+ return [None] * n
+
+
+def _fin_row(label: str, indent: int, is_total: bool, values: list[float | None]) -> dict:
+ return {"label": label, "indent": indent, "is_total": is_total, "is_section": False, "is_margin": False, "values": values}
+
+
+def _fin_section(label: str) -> dict:
+ return {"label": label, "indent": 0, "is_total": False, "is_section": True, "is_margin": False, "values": []}
+
+
+def _fin_margin(label: str, values: list[float | None]) -> dict:
+ return {"label": label, "indent": 1, "is_total": False, "is_section": False, "is_margin": True, "values": values}
+
+
+def _safe_ratio(num: float | None, den: float | None) -> float | None:
+ if num is None or den is None or den == 0:
+ return None
+ return num / den
+
+
def _balance_value(frame: pd.DataFrame, *labels: str) -> float | None:
if frame is None or frame.empty:
return None