"""미국 주식 거래소 자동 감지 모듈 티커 심볼로 거래소(NASD/NYSE/AMEX) 자동 판별. - 주요 종목은 하드코딩 매핑 (빠른 조회) - 미확인 종목은 KIS API 조회 후 캐싱 - Redis 캐시 활용 (TTL 24시간) """ import logging from typing import Optional logger = logging.getLogger(__name__) # ── 주요 미국 종목 → 거래소 매핑 (빠른 O(1) 조회) ──────────────────────────── # ticker → exchange_code KNOWN_EXCHANGES: dict[str, str] = { # ── NASDAQ (NASD) ────────────────────────────────────────────────────────── "AAPL": "NASD", "MSFT": "NASD", "GOOGL": "NASD", "GOOG": "NASD", "AMZN": "NASD", "META": "NASD", "TSLA": "NASD", "NVDA": "NASD", "NFLX": "NASD", "AVGO": "NASD", "COST": "NASD", "PEP": "NASD", "ADBE": "NASD", "CSCO": "NASD", "INTC": "NASD", "AMD": "NASD", "PYPL": "NASD", "QCOM": "NASD", "SBUX": "NASD", "TMUS": "NASD", "ABNB": "NASD", "COIN": "NASD", "MARA": "NASD", "MSTR": "NASD", "PLTR": "NASD", "RIVN": "NASD", "LCID": "NASD", "SOFI": "NASD", "CMCSA": "NASD", "TXN": "NASD", "INTU": "NASD", "ISRG": "NASD", "BKNG": "NASD", "ADP": "NASD", "MDLZ": "NASD", "REGN": "NASD", "VRTX": "NASD", "GILD": "NASD", "LRCX": "NASD", "PANW": "NASD", "SNPS": "NASD", "CDNS": "NASD", "KLAC": "NASD", "MRVL": "NASD", "CRWD": "NASD", "DDOG": "NASD", "ZS": "NASD", "FTNT": "NASD", "WDAY": "NASD", "TEAM": "NASD", "MNST": "NASD", "KDP": "NASD", "BIIB": "NASD", "ILMN": "NASD", "MRNA": "NASD", "ENPH": "NASD", "FSLR": "NASD", "DXCM": "NASD", "TTWO": "NASD", "EA": "NASD", "ROKU": "NASD", "ZM": "NASD", "DOCU": "NASD", "OKTA": "NASD", "HOOD": "NASD", "RBLX": "NASD", "CPRT": "NASD", "ODFL": "NASD", "SMCI": "NASD", "ARM": "NASD", "MU": "NASD", "AMAT": "NASD", "ASML": "NASD", "MELI": "NASD", "PDD": "NASD", "JD": "NASD", "BIDU": "NASD", "NTES": "NASD", "LI": "NASD", "NIO": "NASD", "XPEV": "NASD", "SIRI": "NASD", "WBD": "NASD", "PARA": "NASD", "SQQQ": "NASD", "TQQQ": "NASD", "QLD": "NASD", # ── NYSE ─────────────────────────────────────────────────────────────────── "JPM": "NYSE", "V": "NYSE", "JNJ": "NYSE", "WMT": "NYSE", "PG": "NYSE", "UNH": "NYSE", "HD": "NYSE", "BAC": "NYSE", "XOM": "NYSE", "CVX": "NYSE", "KO": "NYSE", "DIS": "NYSE", "MCD": "NYSE", "NKE": "NYSE", "BA": "NYSE", "GS": "NYSE", "CAT": "NYSE", "IBM": "NYSE", "GE": "NYSE", "F": "NYSE", "GM": "NYSE", "T": "NYSE", "VZ": "NYSE", "PFE": "NYSE", "MRK": "NYSE", "ABT": "NYSE", "MMM": "NYSE", "CRM": "NYSE", "ORCL": "NYSE", "ACN": "NYSE", "LLY": "NYSE", "ABBV": "NYSE", "TMO": "NYSE", "DHR": "NYSE", "NEE": "NYSE", "LIN": "NYSE", "LOW": "NYSE", "UPS": "NYSE", "RTX": "NYSE", "HON": "NYSE", "SPGI": "NYSE", "BLK": "NYSE", "AXP": "NYSE", "DE": "NYSE", "SYK": "NYSE", "EL": "NYSE", "CI": "NYSE", "CB": "NYSE", "SO": "NYSE", "DUK": "NYSE", "BMY": "NYSE", "ZTS": "NYSE", "SCHW": "NYSE", "C": "NYSE", "MS": "NYSE", "WFC": "NYSE", "BK": "NYSE", "USB": "NYSE", "TFC": "NYSE", "PNC": "NYSE", "BABA": "NYSE", "TSM": "NYSE", "SHOP": "NYSE", "SQ": "NYSE", "SNAP": "NYSE", "UBER": "NYSE", "LYFT": "NYSE", "PINS": "NYSE", "SNOW": "NYSE", "NET": "NYSE", "PATH": "NYSE", "U": "NYSE", "SE": "NYSE", "GRAB": "NYSE", "NU": "NYSE", # ── AMEX ─────────────────────────────────────────────────────────────────── "SPY": "AMEX", "QQQ": "AMEX", "IWM": "AMEX", "XLF": "AMEX", "XLE": "AMEX", "GDX": "AMEX", "SLV": "AMEX", "UVXY": "AMEX", "DIA": "AMEX", "VXX": "AMEX", "ARKK": "AMEX", "SOXL": "AMEX", "SOXS": "AMEX", "SPXU": "AMEX", "TZA": "AMEX", "TNA": "AMEX", "NUGT": "AMEX", "DUST": "AMEX", "JNUG": "AMEX", "GDXJ": "AMEX", "XLP": "AMEX", "XLV": "AMEX", "XLI": "AMEX", "XLK": "AMEX", "XLU": "AMEX", "XLB": "AMEX", "XLRE": "AMEX", "XLC": "AMEX", "VTI": "AMEX", "VOO": "AMEX", "VEA": "AMEX", "VWO": "AMEX", "BND": "AMEX", "AGG": "AMEX", "LQD": "AMEX", "HYG": "AMEX", "USO": "AMEX", "GLD": "AMEX", "IAU": "AMEX", "EEM": "AMEX", "EFA": "AMEX", "IVV": "AMEX", "IEFA": "AMEX", } # 역방향 인덱스는 필요 없음 — ticker → exchange 단방향 조회만 사용 def is_us_stock(code: str) -> bool: """미국 주식 여부 판별. 한국 주식 코드는 6자리 숫자(예: '005930', '035720'). 그 외(영문 티커)는 미국 주식으로 간주. Args: code: 종목 코드 (예: 'AAPL', '005930') Returns: True if 미국 주식, False if 한국 주식 """ if not code: return False return not (code.isdigit() and len(code) == 6) def detect_exchange(ticker: str) -> str: """티커 심볼로 거래소 코드 반환. 조회 순서: 1. KNOWN_EXCHANGES 하드코딩 매핑 (O(1)) 2. Redis 캐시 'exchange:{ticker}' (비동기 아님 — 동기 호출용) 3. 기본값 'NASD' 반환 (나스닥이 가장 일반적) Args: ticker: 미국 주식 티커 (예: 'AAPL', 'JPM') Returns: 거래소 코드 ('NASD', 'NYSE', 'AMEX') """ ticker_upper = ticker.upper().strip() # 1. 하드코딩 매핑 확인 if ticker_upper in KNOWN_EXCHANGES: return KNOWN_EXCHANGES[ticker_upper] # 2. Redis 캐시 확인 (동기 — 이벤트 루프 밖에서 호출될 수 있음) try: from app.core.redis import get_redis client = get_redis() if client: # 동기 컨텍스트에서는 캐시 스킵 (async 함수에서는 detect_exchange_via_api 사용) pass except Exception: pass # 3. 기본값 (나스닥이 가장 일반적) logger.warning(f"[EXCHANGE_DETECT] 미등록 티커 '{ticker_upper}' → 기본값 NASD 반환") return "NASD" async def detect_exchange_async(ticker: str) -> str: """티커 심볼로 거래소 코드 반환 (비동기 버전, Redis 캐시 조회 포함). 조회 순서: 1. KNOWN_EXCHANGES 하드코딩 매핑 2. Redis 캐시 'exchange:{ticker}' 3. 기본값 'NASD' Args: ticker: 미국 주식 티커 Returns: 거래소 코드 ('NASD', 'NYSE', 'AMEX') """ ticker_upper = ticker.upper().strip() # 1. 하드코딩 매핑 if ticker_upper in KNOWN_EXCHANGES: return KNOWN_EXCHANGES[ticker_upper] # 2. Redis 캐시 try: from app.core.redis import get_json cached = await get_json(f"exchange:{ticker_upper}") if cached: return str(cached) except Exception: pass # 3. 기본값 logger.warning(f"[EXCHANGE_DETECT] 미등록 티커 '{ticker_upper}' → 기본값 NASD 반환") return "NASD" async def detect_exchange_via_api(ticker: str) -> str: """KIS API를 통해 거래소를 자동 감지하고 Redis에 캐싱. 각 거래소(NASD → NYSE → AMEX)에 순차적으로 현재가 조회를 시도하여 유효한 응답을 받은 거래소를 반환. Args: ticker: 미국 주식 티커 (예: 'RIVN') Returns: 감지된 거래소 코드. 모두 실패 시 'NASD' 기본값. """ from app.services.korea_investment_api import korea_investment_api from app.core.redis import setex_json ticker_upper = ticker.upper().strip() # 이미 알려진 종목이면 즉시 반환 if ticker_upper in KNOWN_EXCHANGES: return KNOWN_EXCHANGES[ticker_upper] exchanges_to_try = ["NASD", "NYSE", "AMEX"] for exchange in exchanges_to_try: try: result = await korea_investment_api.get_overseas_price(ticker_upper, exchange) # 유효한 가격이 반환되면 해당 거래소로 확정 price = float(result.get("last", 0) or result.get("stck_prpr", 0) or 0) if price > 0: logger.info( f"[EXCHANGE_DETECT] '{ticker_upper}' → {exchange} (API 확인, 가격: ${price})" ) # Redis 캐시 (24시간) await setex_json(f"exchange:{ticker_upper}", 86400, exchange) return exchange except Exception as e: logger.debug(f"[EXCHANGE_DETECT] '{ticker_upper}' {exchange} 조회 실패: {e}") continue # 모든 거래소에서 실패 → 기본값 NASD logger.warning(f"[EXCHANGE_DETECT] '{ticker_upper}' 모든 거래소 조회 실패 → NASD 기본값") await setex_json(f"exchange:{ticker_upper}", 86400, "NASD") return "NASD" def get_market_info(code: str) -> dict: """종목 코드로 시장 정보 반환. Args: code: 종목 코드 (예: 'AAPL', '005930') Returns: dict with keys: - market: 'US' 또는 'KR' - exchange: 'NASD', 'NYSE', 'AMEX', 'KRX' - currency: 'USD' 또는 'KRW' """ if not code: return {"market": "KR", "exchange": "KRX", "currency": "KRW"} if is_us_stock(code): exchange = detect_exchange(code) return { "market": "US", "exchange": exchange, "currency": "USD", } return { "market": "KR", "exchange": "KRX", "currency": "KRW", }