auto-stock/backend/app/services/utils.py

"""
공유 유틸리티 함수 및 상수

이 모듈은 여러 파일에서 중복되는 함수와 상수를 한 곳에 정의하여
코드 중복을 제거합니다.

포함 항목:
- _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]