File size: 2,379 Bytes
2b21f2d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """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"]))
|