You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
"""Proxy trade symbol helpers for event-driven strategies."""
|
|
from __future__ import annotations
|
|
|
|
SECTOR_ETF_BY_SECTOR: dict[str, str] = {
|
|
"Technology": "XLK",
|
|
"Health Care": "XLV",
|
|
"Healthcare": "XLV",
|
|
"Financials": "XLF",
|
|
"Consumer Discretionary": "XLY",
|
|
"Industrials": "XLI",
|
|
"Energy": "XLE",
|
|
"Utilities": "XLU",
|
|
"Real Estate": "XLRE",
|
|
"Materials": "XLB",
|
|
"Communication Services": "XLC",
|
|
"Consumer Staples": "XLP",
|
|
}
|
|
|
|
LEADER_PEER_CANDIDATES_BY_SYMBOL: dict[str, tuple[str, ...]] = {
|
|
"AVGO": ("NVDA", "AMD", "QCOM", "MRVL"),
|
|
"SNOW": ("CRM", "NOW", "ORCL", "MDB"),
|
|
"URI": ("CAT", "DE", "PWR", "PH"),
|
|
"ISRG": ("ABBV", "MRK", "AMGN", "JNJ", "LLY", "BSX"),
|
|
"LLY": ("ABBV", "MRK", "JNJ", "AMGN"),
|
|
"CCI": ("AMT", "SBAC", "EQIX", "DLR"),
|
|
"AXON": ("TDY", "LHX", "GD", "RTX"),
|
|
"CACI": ("LDOS", "BAH", "SAIC", "LHX"),
|
|
"AEIS": ("ENTG", "KLAC", "AMAT", "LRCX"),
|
|
"NVDA": ("AVGO", "AMD", "QCOM", "MRVL"),
|
|
"AMD": ("NVDA", "AVGO", "QCOM", "MRVL"),
|
|
"MSFT": ("ORCL", "CRM", "NOW", "ADBE"),
|
|
"META": ("GOOGL", "NFLX", "TMUS", "DIS"),
|
|
"AMZN": ("HD", "TJX", "BKNG", "LOW"),
|
|
}
|
|
|
|
SECTOR_STOCK_FALLBACKS: dict[str, tuple[str, ...]] = {
|
|
"Technology": ("MSFT", "NVDA", "AVGO", "ORCL", "CRM", "ADBE", "AMD", "INTU"),
|
|
"Health Care": ("LLY", "ABBV", "MRK", "JNJ", "AMGN", "ISRG", "BSX"),
|
|
"Healthcare": ("LLY", "ABBV", "MRK", "JNJ", "AMGN", "ISRG", "BSX"),
|
|
"Financials": ("JPM", "GS", "MS", "BLK", "SPGI", "ICE"),
|
|
"Consumer Discretionary": ("AMZN", "HD", "TJX", "LOW", "BKNG", "MCD"),
|
|
"Industrials": ("CAT", "DE", "HON", "GE", "PWR", "RTX", "LHX"),
|
|
"Energy": ("XOM", "CVX", "COP", "SLB", "EOG"),
|
|
"Utilities": ("NEE", "DUK", "SO", "CEG"),
|
|
"Real Estate": ("AMT", "EQIX", "PLD", "DLR", "SBAC", "CCI"),
|
|
"Materials": ("LIN", "APD", "NEM", "SHW"),
|
|
"Communication Services": ("GOOGL", "META", "NFLX", "TMUS", "DIS"),
|
|
"Consumer Staples": ("COST", "PG", "KO", "PM", "WMT"),
|
|
}
|
|
|
|
|
|
def sector_etf_for_sector(sector: str | None) -> str | None:
|
|
if not sector:
|
|
return None
|
|
normalized = str(sector).strip()
|
|
if not normalized:
|
|
return None
|
|
return SECTOR_ETF_BY_SECTOR.get(normalized)
|
|
|
|
|
|
def peer_candidates_for_symbol(
|
|
source_symbol: str | None,
|
|
sector: str | None,
|
|
) -> list[str]:
|
|
source = (source_symbol or "").strip().upper()
|
|
ordered: list[str] = []
|
|
if source:
|
|
ordered.extend(LEADER_PEER_CANDIDATES_BY_SYMBOL.get(source, ()))
|
|
if sector:
|
|
normalized_sector = str(sector).strip()
|
|
if normalized_sector:
|
|
ordered.extend(SECTOR_STOCK_FALLBACKS.get(normalized_sector, ()))
|
|
|
|
seen: set[str] = set()
|
|
result: list[str] = []
|
|
for symbol in ordered:
|
|
candidate = str(symbol).strip().upper()
|
|
if not candidate or candidate == source or candidate in seen:
|
|
continue
|
|
seen.add(candidate)
|
|
result.append(candidate)
|
|
return result
|