signal-platform/app/services/sector_map.py

"""Ticker → GICS sector → SPDR sector ETF mapping (research only).

Sector residual momentum residualizes 12-1 momentum against SPY and the name's
sector ETF. Labels are persisted under ``data/research/ticker_sector_map.json``
so research runs do not depend on live FMP calls.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

# Eleven SPDR sector ETFs. Auxiliary series only — never tradable book members.
SECTOR_ETFS: tuple[str, ...] = (
    "XLB",
    "XLC",
    "XLE",
    "XLF",
    "XLI",
    "XLK",
    "XLP",
    "XLRE",
    "XLU",
    "XLV",
    "XLY",
)

# GICS sector name (and common aliases) → SPDR ETF.
# Keys are lower-case for matching.
GICS_SECTOR_TO_ETF: dict[str, str] = {
    "materials": "XLB",
    "basic materials": "XLB",
    "communication services": "XLC",
    "communications": "XLC",
    "telecommunication services": "XLC",
    "energy": "XLE",
    "financials": "XLF",
    "financial services": "XLF",
    "financial": "XLF",
    "industrials": "XLI",
    "industrial goods": "XLI",
    "information technology": "XLK",
    "technology": "XLK",
    "consumer staples": "XLP",
    "consumer defensive": "XLP",
    "real estate": "XLRE",
    "utilities": "XLU",
    "health care": "XLV",
    "healthcare": "XLV",
    "consumer discretionary": "XLY",
    "consumer cyclical": "XLY",
}

DEFAULT_SECTOR_MAP_PATH = Path("data/research/ticker_sector_map.json")


def normalise_symbol(symbol: str) -> str:
    """Alpaca-style symbols: BRK.B / BRK/B → BRK-B."""
    s = str(symbol or "").strip().upper()
    s = s.replace(".", "-").replace("/", "-")
    return s


def sector_to_etf(sector: str | None) -> str | None:
    if not sector:
        return None
    return GICS_SECTOR_TO_ETF.get(str(sector).strip().lower())


def etf_for_symbol(symbol: str, symbol_to_sector: dict[str, str]) -> str | None:
    sector = symbol_to_sector.get(normalise_symbol(symbol))
    return sector_to_etf(sector)


def load_ticker_sector_map(path: Path | str | None = None) -> dict[str, str]:
    """Load ``{symbol: gics_sector}`` from JSON. Empty dict if missing."""
    p = Path(path) if path is not None else DEFAULT_SECTOR_MAP_PATH
    if not p.exists():
        return {}
    raw = json.loads(p.read_text(encoding="utf-8"))
    if not isinstance(raw, dict):
        return {}
    out: dict[str, str] = {}
    # Accept either flat map or {"map": {...}, "meta": ...}
    payload = raw.get("map") if "map" in raw and isinstance(raw.get("map"), dict) else raw
    if not isinstance(payload, dict):
        return {}
    for sym, sector in payload.items():
        if sym in ("meta", "schema_version", "map"):
            continue
        if sector is None:
            continue
        ns = normalise_symbol(str(sym))
        if ns:
            out[ns] = str(sector).strip()
    return out


def save_ticker_sector_map(
    mapping: dict[str, str],
    path: Path | str | None = None,
    *,
    meta: dict[str, Any] | None = None,
) -> Path:
    p = Path(path) if path is not None else DEFAULT_SECTOR_MAP_PATH
    p.parent.mkdir(parents=True, exist_ok=True)
    # Normalise keys on write.
    clean = {
        normalise_symbol(k): str(v).strip()
        for k, v in mapping.items()
        if k and v and normalise_symbol(k)
    }
    payload: dict[str, Any] = {
        "schema_version": 1,
        "map": clean,
        "meta": meta or {},
    }
    p.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return p


def coverage_stats(
    symbols: list[str], mapping: dict[str, str]
) -> dict[str, Any]:
    total = len(symbols)
    mapped = [s for s in symbols if normalise_symbol(s) in mapping]
    with_etf = [
        s
        for s in mapped
        if sector_to_etf(mapping[normalise_symbol(s)]) is not None
    ]
    missing = [s for s in symbols if normalise_symbol(s) not in mapping]
    by_sector: dict[str, int] = {}
    for s in mapped:
        sec = mapping[normalise_symbol(s)]
        by_sector[sec] = by_sector.get(sec, 0) + 1
    return {
        "universe": total,
        "mapped": len(mapped),
        "mapped_pct": round(100.0 * len(mapped) / total, 1) if total else 0.0,
        "with_etf": len(with_etf),
        "missing": missing,
        "by_sector": dict(sorted(by_sector.items(), key=lambda kv: (-kv[1], kv[0]))),
    }