"""Credit Peer Curve — computes average OAS term structures across peer universes.""" from __future__ import annotations import logging import time from collections import Counter from datetime import date, datetime from typing import Optional logger = logging.getLogger(__name__) # ── Rating hierarchy ────────────────────────────────────────────────────────── RATING_HIERARCHY = ["AAA", "AA", "A", "BBB", "BB", "B", "CCC", "CC", "D"] _MOODY_TO_BUCKET: dict[str, str] = { "Aaa": "AAA", "Aa1": "AA", "Aa2": "AA", "Aa3": "AA", "A1": "A", "A2": "A", "A3": "A", "Baa1": "BBB", "Baa2": "BBB", "Baa3": "BBB", "Ba1": "BB", "Ba2": "BB", "Ba3": "BB", "B1": "B", "B2": "B", "B3": "B", "Caa1": "CCC", "Caa2": "CCC", "Caa3": "CCC", "Ca": "CC", "C": "CC", "D": "D", "DDD": "D", "DD": "D", "WD": "D", } _SP_TO_BUCKET: dict[str, str] = { "AAA": "AAA", "AA+": "AA", "AA": "AA", "AA-": "AA", "A+": "A", "A": "A", "A-": "A", "BBB+": "BBB", "BBB": "BBB", "BBB-": "BBB", "BB+": "BB", "BB": "BB", "BB-": "BB", "B+": "B", "B": "B", "B-": "B", "CCC+": "CCC", "CCC": "CCC", "CCC-": "CCC", "CC": "CC", "C": "CC", "D": "D", "SD": "D", } # ── Maturity buckets ────────────────────────────────────────────────────────── MATURITY_BUCKETS = ["<1yr", "2yr", "5yr", "10yr", "20yr", "30yr+"] UNIVERSE_KEYS = [ "issuer", "rating_peers", "sector", "sub_sector", "rating_sector", "rating_sub_sector", "lower_rating_sub_sector", "upper_rating_sub_sector", ] def _rating_bucket(moodys: Optional[str], sp: Optional[str] = None) -> Optional[str]: """Moody's first; fall back to S&P if Moody's absent.""" m = _MOODY_TO_BUCKET.get((moodys or "").strip()) if m: return m return _SP_TO_BUCKET.get((sp or "").strip()) or None def _lower_rating_bucket(bucket: str) -> Optional[str]: if bucket not in RATING_HIERARCHY: return None idx = RATING_HIERARCHY.index(bucket) if idx + 1 >= len(RATING_HIERARCHY): return None return RATING_HIERARCHY[idx + 1] def _upper_rating_bucket(bucket: str) -> Optional[str]: if bucket not in RATING_HIERARCHY: return None idx = RATING_HIERARCHY.index(bucket) if idx - 1 < 0: return None return RATING_HIERARCHY[idx - 1] def _years_to_maturity(maturity_str: Optional[str]) -> Optional[float]: if not maturity_str: return None try: mat = datetime.strptime(maturity_str[:10], "%Y-%m-%d").date() delta = (mat - date.today()).days / 365.25 return delta if delta > 0 else None except (ValueError, TypeError): return None def _maturity_bucket(years: float) -> str: if years <= 1: return "<1yr" if years <= 3: return "2yr" if years <= 7: return "5yr" if years <= 15: return "10yr" if years <= 25: return "20yr" return "30yr+" def _best_spread(bond: dict) -> Optional[float]: for key in ("oas", "z_spread", "asset_swap_spread"): v = bond.get(key) if v is not None: return v return None def _avg_spread(bonds: list[dict]) -> Optional[float]: spreads = [s for b in bonds if (s := _best_spread(b)) is not None] return round(sum(spreads) / len(spreads), 1) if spreads else None def _dominant(bonds: list[dict], field: str) -> Optional[str]: vals = [v for b in bonds if (v := b.get(field))] return Counter(vals).most_common(1)[0][0] if vals else None def _genealogy_prefix(genealogy: Optional[str], depth: int) -> Optional[str]: """Extract first `depth` code segments from a TRBC genealogy path. depth=3 → Industry Group (sector level, e.g. 'B:180\\B:181\\B:182') depth=4 → Industry (sub-sector level, e.g. 'B:180\\B:181\\B:182\\B:183') """ if not genealogy: return None parts = genealogy.split("\\") return "\\".join(parts[:depth]) if len(parts) >= depth else None # ── Service ─────────────────────────────────────────────────────────────────── class PeerCurveService: _cache: dict[str, tuple[float, dict]] = {} _in_progress: set[str] = set() _CACHE_TTL = 1800 # 30 minutes def __init__(self, app_key: str, username: str, password: str): self._app_key = app_key self._username = username self._password = password self._client = None def _get_client(self): if self._client is None: from app.providers.lseg_api import LsegApiClient self._client = LsegApiClient( app_key=self._app_key, username=self._username, password=self._password, ) return self._client def _background_compute(self, issuers: list[str], display_name: str, cache_key: str) -> None: try: result = self._compute(issuers, display_name) self._cache[cache_key] = (time.time(), result) logger.info("PeerCurve: background build complete for '%s'", display_name) except Exception as exc: logger.exception("PeerCurve: background build failed for '%s'", display_name) self._cache[cache_key] = (time.time(), {"__error__": str(exc)}) finally: self._in_progress.discard(cache_key) def build(self, issuers: list[str], display_name: str) -> Optional[dict]: """Return cached result immediately, or None if still computing (starts background build).""" import threading cache_key = "|".join(sorted(i.strip().lower() for i in issuers)) if cache_key in self._cache: ts, result = self._cache[cache_key] if time.time() - ts < self._CACHE_TTL: logger.info("PeerCurve: cache hit for '%s'", display_name) return result del self._cache[cache_key] if cache_key not in self._in_progress: self._in_progress.add(cache_key) threading.Thread( target=self._background_compute, args=(issuers, display_name, cache_key), daemon=True ).start() logger.info("PeerCurve: started background build for '%s'", display_name) return None def _compute(self, issuers: list[str], display_name: str) -> dict: client = self._get_client() # ── Step 1: issuer bonds (fully enriched) ──────────────────────────── logger.info("PeerCurve: fetching issuer bonds for '%s'", display_name) issuer_bonds = client.fetch_bonds_for_issuers(issuers) if not issuer_bonds: raise ValueError(f"No bonds found for: {', '.join(issuers)}") issuer_rics: set[str] = {b["ric"] for b in issuer_bonds if b.get("ric")} # ── Step 2: GICS via equity lookup + TRBC from bond market data ───── rating_bucket = _dominant( [{**b, "_rb": _rating_bucket(b.get("rating_moodys"), b.get("rating_sp"))} for b in issuer_bonds], "_rb", ) trbc_ig_dom = _dominant(issuer_bonds, "trbc_ig") trbc_ind_dom = _dominant(issuer_bonds, "trbc_industry") # One equity search per issuer to get proper GICS names gics = client.lookup_issuer_gics(issuers[0]) logger.info( "PeerCurve: issuer=%s gics=%s trbc_ig=%s trbc_industry=%s rating=%s", display_name, gics, trbc_ig_dom, trbc_ind_dom, rating_bucket, ) # ── Step 3: search for sector-level peers via TRBC genealogy filter ─ # Use depth=2 (one level above the company's Industry Group) so the sector # universe is meaningfully broader than the sub_sector. depth=3 produces # ≈22 unique issuers for Apple; depth=2 gives ≈141 issuers. genealogy = _dominant(issuer_bonds, "genealogy") sector_prefix = _genealogy_prefix(genealogy, depth=2) # broad sector pool logger.info("PeerCurve: genealogy=%s sector_prefix(d2)=%s", genealogy, sector_prefix) if sector_prefix: peer_refs = client.search_peers_by_sector(sector_prefix, top=300) else: fallback_query = (trbc_ig_dom or trbc_ind_dom or "").split(" & ")[0].strip() logger.warning("PeerCurve: no genealogy, text fallback query='%s'", fallback_query) peer_refs = client.search_bonds_by_query(fallback_query, top=300) # ── Step 3b: cross-sector rating peers (separate universe for rating_peers) ─ # rating_peers is meant to compare the issuer against other bonds of the SAME rating # regardless of sector — a Aaa issuer should be compared to all Aaa corporates, # not just Aaa bonds within the same sector (which may be zero). rating_refs: list[dict] = [] if rating_bucket: rating_refs = client.search_peers_by_rating(rating_bucket, top=300) logger.info("PeerCurve: cross-sector rating '%s' → %d refs", rating_bucket, len(rating_refs)) # ── Dedup helpers ───────────────────────────────────────────────────── _today = date.today().isoformat() def _valid_refs(refs: list[dict], exclude_rics: set[str]) -> list[dict]: seen: set[str] = set() out = [] for r in refs: ric = r.get("ric") if not ric or ric in exclude_rics or ric in seen: continue if not r.get("maturity_date") or r["maturity_date"][:10] < _today: continue seen.add(ric) out.append(r) return out sector_refs = _valid_refs(peer_refs, issuer_rics) clean_rating_refs = _valid_refs(rating_refs, issuer_rics) # Refs needed for market data: union of both sets sector_ric_set = {r["ric"] for r in sector_refs} extra_rating_refs = [r for r in clean_rating_refs if r["ric"] not in sector_ric_set] all_peer_refs = sector_refs + extra_rating_refs logger.info( "PeerCurve: %d sector refs, %d rating refs (%d new), %d total for market data", len(sector_refs), len(clean_rating_refs), len(extra_rating_refs), len(all_peer_refs), ) # ── Step 4: fetch market data for all peer refs in one batch ───────── peer_market: dict[str, dict] = {} if all_peer_refs: all_peer_rics = [r["ric"] for r in all_peer_refs] peer_market = client.fetch_market_data(all_peer_rics) # ── Step 5: build enriched peer bond lists ─────────────────────────── def _enrich(ref: dict) -> dict: ric = ref["ric"] m = peer_market.get(ric, {}) return { **ref, "price": ref.get("price") or m.get("bid_price"), "ytm": m.get("ytm"), "ytw": m.get("ytw"), "z_spread": m.get("z_spread"), "oas": m.get("oas"), "asset_swap_spread": m.get("asset_swap_spread"), "country": m.get("country"), "trbc_ig": m.get("trbc_ig"), "trbc_industry": m.get("trbc_industry"), "duration": m.get("duration"), "_is_issuer": False, } peer_bonds = [_enrich(r) for r in sector_refs] rating_peer_bonds = [_enrich(r) for r in clean_rating_refs] # Mark issuer bonds for b in issuer_bonds: b["_is_issuer"] = True # ── Step 6: combine + enrich with derived fields ───────────────────── # all_bonds: issuer + sector peers — used for sector/sub_sector universes # rating_pool: issuer + cross-sector rating peers — used for rating_peers universe all_bonds = issuer_bonds + peer_bonds def _enrich_derived(b: dict) -> None: b["rating_bucket"] = _rating_bucket(b.get("rating_moodys"), b.get("rating_sp")) yrs = _years_to_maturity(b.get("maturity_date")) b["years_to_maturity"] = yrs b["maturity_bucket"] = _maturity_bucket(yrs) if yrs is not None else None for b in all_bonds: _enrich_derived(b) for b in rating_peer_bonds: _enrich_derived(b) # Drop peer bonds with no maturity bucket (perpetuals, already-matured) before = len(all_bonds) all_bonds = [b for b in all_bonds if b.get("_is_issuer") or b.get("maturity_bucket") is not None] rating_peer_bonds = [b for b in rating_peer_bonds if b.get("maturity_bucket") is not None] dropped = before - len(all_bonds) if dropped: logger.info("PeerCurve: dropped %d sector peer bonds with no valid maturity", dropped) logger.info( "PeerCurve: %d sector bonds (%d issuer + %d sector peers), %d cross-sector rating peers", len(all_bonds), sum(1 for b in all_bonds if b.get("_is_issuer")), sum(1 for b in all_bonds if not b.get("_is_issuer")), len(rating_peer_bonds), ) # ── Step 6b: determine display sector/sub_sector (GICS) and matching fields ─ # Display uses GICS names from the equity lookup. # Universe matching uses TRBC (available on all bonds) — trbc_ig / trbc_industry. # # Hierarchy rules (user requirement): # Utilities / Materials / Energy → display Sector + Industry (skip IG) # Industrials → display Industry Group + Industry (skip Sector) # Everything else → display Sector + Industry Group _USE_INDUSTRY = {"Utilities", "Materials", "Energy"} _SKIP_SECTOR = {"Industrials"} gics_sector = gics.get("sector", "") gics_ig = gics.get("industry_group", "") gics_ind = gics.get("industry", "") if gics_sector in _USE_INDUSTRY: sector = gics_sector # e.g. "Utilities" sub_sector = gics_ind # e.g. "Electric Utilities" sector_label, sub_sector_label = "Sector", "Industry" # Match peers by TRBC industry (aligns with GICS industry for these sectors) match_sector = trbc_ig_dom match_sub = trbc_ind_dom elif gics_sector in _SKIP_SECTOR: sector = gics_ig # e.g. "Capital Goods" sub_sector = gics_ind # e.g. "Aerospace & Defense" sector_label, sub_sector_label = "Industry Group", "Industry" match_sector = trbc_ig_dom match_sub = trbc_ind_dom else: sector = gics_sector or trbc_ig_dom # fall back to TRBC if GICS lookup failed sub_sector = gics_ig or trbc_ig_dom sector_label, sub_sector_label = "Sector", "Industry Group" match_sector = trbc_ig_dom match_sub = trbc_ind_dom # TRBC Industry is narrower than IG → distinct universe # Tag each bond with matching fields for universe filtering for b in all_bonds: b["eff_sector"] = b.get("trbc_ig") b["eff_sub_sector"] = b.get("trbc_industry") logger.info( "PeerCurve: GICS display sector=%s (%s) sub=%s (%s) match_sector=%s match_sub=%s", sector, sector_label, sub_sector, sub_sector_label, match_sector, match_sub, ) # ── Step 7: build peer universes ───────────────────────────────────── peer_buckets = [rating_bucket] if rating_bucket else [] lower_bucket = _lower_rating_bucket(rating_bucket) if rating_bucket else None upper_bucket = _upper_rating_bucket(rating_bucket) if rating_bucket else None def _smatch(b: dict) -> bool: return b.get("eff_sector") == match_sector if match_sector else False def _ssmatch(b: dict) -> bool: return b.get("eff_sub_sector") == match_sub if match_sub else False # rating_peers uses the cross-sector rating pool so it compares against # all same-rated corporates, not just same-rated bonds within the same sector. issuer_only = [b for b in all_bonds if b.get("_is_issuer")] rating_pool = issuer_only + rating_peer_bonds universes: dict[str, list[dict]] = { "issuer": issuer_only, "rating_peers": ( [b for b in rating_pool if b.get("rating_bucket") in peer_buckets] if peer_buckets else [] ), "sector": [b for b in all_bonds if _smatch(b)], "sub_sector": [b for b in all_bonds if _ssmatch(b)] if match_sub else [], "rating_sector": ( [b for b in all_bonds if b.get("rating_bucket") in peer_buckets and _smatch(b)] if peer_buckets else [] ), "rating_sub_sector": ( [b for b in all_bonds if b.get("rating_bucket") in peer_buckets and _ssmatch(b)] if peer_buckets and match_sub else [] ), "lower_rating_sub_sector": ( [b for b in all_bonds if b.get("rating_bucket") == lower_bucket and _ssmatch(b)] if lower_bucket and match_sub else [] ), "upper_rating_sub_sector": ( [b for b in all_bonds if b.get("rating_bucket") == upper_bucket and _ssmatch(b)] if upper_bucket and match_sub else [] ), } # ── Step 8: compute OAS table ───────────────────────────────────────── table: dict[str, dict] = {} for bucket in MATURITY_BUCKETS: row = {} for key, bonds in universes.items(): bucket_bonds = [b for b in bonds if b.get("maturity_bucket") == bucket] row[key] = _avg_spread(bucket_bonds) table[bucket] = row all_row = {key: _avg_spread(bonds) for key, bonds in universes.items()} table["All"] = all_row # ── Step 9: determine spread source ────────────────────────────────── all_issuer = universes["issuer"] oas_count = sum(1 for b in all_issuer if b.get("oas") is not None) z_count = sum(1 for b in all_issuer if b.get("z_spread") is not None) n = len(all_issuer) or 1 if oas_count / n > 0.5: spread_source = "oas" elif z_count / n > 0.5: spread_source = "z_spread" else: spread_source = "asset_swap_spread" return { "issuer": display_name, "sector": sector, "sub_sector": sub_sector, "sector_label": sector_label, "sub_sector_label": sub_sector_label, "rating_bucket": rating_bucket, "rating_peer_buckets": peer_buckets, "lower_rating_bucket": lower_bucket, "upper_rating_bucket": upper_bucket, "table": table, "bond_counts": {k: len(v) for k, v in universes.items()}, "chart_data": self._chart_rows(universes), "spread_source": spread_source, } def _chart_rows(self, universes: dict[str, list[dict]]) -> list[dict]: rows = [] for bucket in MATURITY_BUCKETS: row: dict = {"bucket": bucket} for key, bonds in universes.items(): bb = [b for b in bonds if b.get("maturity_bucket") == bucket] row[key] = _avg_spread(bb) rows.append(row) return rows