""" 공유 유틸리티 함수 및 상수 이 모듈은 여러 파일에서 중복되는 함수와 상수를 한 곳에 정의하여 코드 중복을 제거합니다. 포함 항목: - _is_domestic(): 국내/해외 종목 판별 - CircuitBreakerState Enum: 회로차단기 상태 - Chart data extraction helpers: 차트 데이터 추출 유틸 - DomesticChartItem / OverseasChartItem: 차트 데이터 TypedDict """ from enum import Enum from typing import List, Dict, Any, Optional, TypedDict # ── 해외 거래소 코드 목록 ──────────────────────────────────────────────────────── OVERSEAS_MARKETS = {"NASD", "NYSE", "AMEX", "TSEX", "SEHK", "SHAA", "SZAA", "HASE", "VNSE"} class CircuitBreakerState(Enum): """회로차단기(Circuit Breaker) 상태. States: CLOSED : 정상 운영 (요청 허용) OPEN : 차단 (recovery_timeout 경과 전까지 요청 거부) HALF_OPEN: 복구 시도 (1개 요청 허용 → 성공 시 CLOSED, 실패 시 OPEN) """ CLOSED = "CLOSED" OPEN = "OPEN" HALF_OPEN = "HALF_OPEN" class DomesticChartItem(TypedDict, total=False): """한국 국내 주식 차트 데이터 아이템 (KIS API). Field descriptions: stck_clpr: 종가 (closing price) stck_hgpr: 고가 (high price) stck_lwpr: 저가 (low price) stck_oprc: 시가 (opening price) acml_vol: 누적거래량 (accumulated volume) stck_bsop_date: 영업일자 (business date) """ stck_clpr: str stck_hgpr: str stck_lwpr: str stck_oprc: str acml_vol: str stck_bsop_date: str class OverseasChartItem(TypedDict, total=False): """해외 주식 차트 데이터 아이템 (KIS API). Field descriptions: clos: 종가 (closing price) high: 고가 (high price) low: 저가 (low price) open: 시가 (opening price) v: 거래량 (volume) """ clos: str high: str low: str open: str v: str def _is_domestic(stock_code: str, market: str = None) -> bool: """국내 종목 여부 판별. market 컬럼이 있으면 우선 사용, 없으면 종목코드 패턴으로 폴백. Args: stock_code: 종목 코드 (예: "005930") market: 시장 코드 (예: "NASD", "KRX"). Optional. Returns: True if domestic (KRX), False if overseas. """ if market: return market not in OVERSEAS_MARKETS # 폴백: 6자리 숫자이면 국내(KRX) return stock_code.isdigit() and len(stock_code) == 6 def extract_closes(chart_data: List[Dict[str, Any]], is_domestic: bool = True) -> List[float]: """차트 데이터에서 종가(close price) 추출. KIS API는 최신순으로 반환하므로 필요시 역순 처리. Args: chart_data: KIS API 차트 데이터 (최신순 또는 시간순) is_domestic: True면 국내(stck_clpr), False면 해외(clos) 필드 사용 Returns: 종가 리스트 (float) """ if not chart_data: return [] key = "stck_clpr" if is_domestic else "clos" return [float(item.get(key, 0)) for item in chart_data] def extract_highs(chart_data: List[Dict[str, Any]], is_domestic: bool = True) -> List[float]: """차트 데이터에서 고가(high price) 추출. Args: chart_data: KIS API 차트 데이터 is_domestic: True면 국내(stck_hgpr), False면 해외(high) 필드 사용 Returns: 고가 리스트 (float) """ if not chart_data: return [] key = "stck_hgpr" if is_domestic else "high" return [float(item.get(key, 0)) for item in chart_data] def extract_lows(chart_data: List[Dict[str, Any]], is_domestic: bool = True) -> List[float]: """차트 데이터에서 저가(low price) 추출. Args: chart_data: KIS API 차트 데이터 is_domestic: True면 국내(stck_lwpr), False면 해외(low) 필드 사용 Returns: 저가 리스트 (float) """ if not chart_data: return [] key = "stck_lwpr" if is_domestic else "low" return [float(item.get(key, 0)) for item in chart_data] def extract_opens(chart_data: List[Dict[str, Any]], is_domestic: bool = True) -> List[float]: """차트 데이터에서 시가(open price) 추출. Args: chart_data: KIS API 차트 데이터 is_domestic: True면 국내(stck_oprc), False면 해외(open) 필드 사용 Returns: 시가 리스트 (float) """ if not chart_data: return [] key = "stck_oprc" if is_domestic else "open" return [float(item.get(key, 0)) for item in chart_data] def extract_volumes(chart_data: List[Dict[str, Any]], is_domestic: bool = True) -> List[float]: """차트 데이터에서 거래량(volume) 추출. Args: chart_data: KIS API 차트 데이터 is_domestic: True면 국내(acml_vol), False면 해외(v) 필드 사용 Returns: 거래량 리스트 (float) """ if not chart_data: return [] key = "acml_vol" if is_domestic else "v" return [float(item.get(key, 0)) for item in chart_data]