| """Loads the exported JSON artifacts (space/data/, populated by sync_data.sh from |
| outputs/dashboard/) once at app startup; per-call detail files are lazy-loaded on |
| demand since there are 675 of them.""" |
|
|
| import json |
| from functools import lru_cache |
| from pathlib import Path |
|
|
| import pandas as pd |
|
|
| DATA_DIR = Path(__file__).parent / "data" |
|
|
|
|
| def load_sentiment_ladder() -> dict: |
| with open(DATA_DIR / "sentiment_ladder.json") as f: |
| return json.load(f) |
|
|
|
|
| def load_methodology() -> dict: |
| with open(DATA_DIR / "methodology.json") as f: |
| return json.load(f) |
|
|
|
|
| def load_calls_index() -> pd.DataFrame: |
| with open(DATA_DIR / "calls_index.json") as f: |
| rows = json.load(f) |
| df = pd.DataFrame(rows) |
| df["call_date"] = pd.to_datetime(df["call_date"]) |
| return df |
|
|
|
|
| @lru_cache(maxsize=256) |
| def load_call_detail(call_id: str) -> dict | None: |
| path = DATA_DIR / "calls" / f"{call_id}.json" |
| if not path.exists(): |
| return None |
| with open(path) as f: |
| return json.load(f) |
|
|
|
|
| def load_company_index() -> pd.DataFrame | None: |
| """Flat one-row-per-ticker index — used by the Company tab to populate its ticker |
| picker (outputs/dashboard/company_summary.json). Guarded like VOLATILITY_LADDER in |
| app.py — the Workspace/Company tabs must not crash if this artifact hasn't been |
| exported yet.""" |
| path = DATA_DIR / "company_summary.json" |
| if not path.exists(): |
| return None |
| with open(path) as f: |
| rows = json.load(f) |
| return pd.DataFrame(rows) |
|
|
|
|
| @lru_cache(maxsize=64) |
| def load_company_series(ticker: str) -> dict | None: |
| """One ticker's cross-quarter call series for the Company tab (outputs/dashboard/companies.json).""" |
| path = DATA_DIR / "companies.json" |
| if not path.exists(): |
| return None |
| with open(path) as f: |
| companies = json.load(f) |
| for company in companies: |
| if company["ticker"] == ticker: |
| return company |
| return None |
|
|
|
|
| @lru_cache(maxsize=1) |
| def load_ticker_names() -> dict[str, str]: |
| """ticker -> company_name (data/processed/ticker_names.csv, synced by sync_data.sh). |
| Falls back to an empty map (callers should fall back to the raw ticker) if missing.""" |
| path = DATA_DIR / "ticker_names.csv" |
| if not path.exists(): |
| return {} |
| df = pd.read_csv(path) |
| return dict(zip(df["ticker"], df["company_name"])) |
|
|
