""" Market Data Module 시세·차트 조회 전담 모듈 (Redis 캐시 적용) """ from typing import Dict, Any, Optional, List import logging import asyncio from datetime import datetime, timedelta import pytz from app.trading.kis_client import kis_client, real_market_client logger = logging.getLogger(__name__) # 차트 캐시 TTL 상수 TTL_DAILY_CHART = 300 TTL_MINUTE_CHART = 60 TTL_MINUTE_FULLDAY = 60 TTL_OVERSEAS_CHART = 300 async def _cache_get(key: str) -> Optional[Any]: """ Redis 캐시에서 값 조회 (non-blocking) """ try: from app.core.redis import get_json return await get_json(key) except Exception: return None async def _cache_set(key: str, ttl: int, data: Any) -> None: """Redis 캐시에 값 저장 (non-blocking)""" try: from app.core.redis import setex_json await setex_json(key, ttl, data) except Exception: pass class MarketDataProvider: """시세 및 차트 데이터 제공자""" def __init__(self, client=None, real_client=None): self.client = client or kis_client self.real_client = real_client or real_market_client async def get_current_price(self, stock_code: str) -> Dict[str, Any]: """현재가 조회""" try: client = self.real_client or self.client result, _ = await client.request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-price", "FHKST01010100", params={ "FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": stock_code } ) return result except Exception as e: logger.error(f"❌ 현재가 조회 실패 {stock_code}: {e}") raise async def get_daily_chart( self, stock_code: str, period: str = "D", limit: int = 240 ) -> Dict[str, Any]: """일봉/주봉/월봉 조회 (Redis 캐시, 페이지네이션)""" cache_key = f"chart:daily:{stock_code}:{period}:{limit}" # 캐시 히트 cached = await _cache_get(cache_key) if cached is not None: logger.debug(f"📦 [캐시 히트] {cache_key}") return cached # KIS API 호출 all_output2 = [] params = { "FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": stock_code, "FID_PERIOD_DIV_CODE": period, "FID_ORG_ADJ_PRC": "0" } try: client = self.real_client or self.client result, headers = await client.request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-daily-price", "FHKST01010400", params=params ) data_key = 'output2' if isinstance(result.get('output2'), list) else 'output' if not result or result.get('rt_cd') != '0' or not result.get(data_key): return result all_output2.extend(result.get(data_key, [])) tr_cont = headers.get('tr_cont') # 연속 조회 while tr_cont in ('F', 'M') and len(all_output2) < limit: await asyncio.sleep(self.client.min_request_interval + 0.1) extra_headers = {'tr_cont': 'Y'} result_cont, headers_cont = await client.request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-daily-price", "FHKST01010400", params=params, extra_headers=extra_headers ) if result_cont and result_cont.get('rt_cd') == '0': cont_key = 'output2' if isinstance(result_cont.get('output2'), list) else 'output' all_output2.extend(result_cont.get(cont_key, [])) tr_cont = headers_cont.get('tr_cont') if headers_cont else None final_result = result final_result['output2'] = all_output2 logger.info(f"✅ 차트 데이터 {len(all_output2)}개 수집: {stock_code}") # 캐시 저장 await _cache_set(cache_key, TTL_DAILY_CHART, final_result) return final_result except Exception as e: logger.error(f"❌ 차트 조회 실패 {stock_code}: {e}") raise async def get_minute_chart( self, stock_code: str, ref_time: str = "", include_past: bool = True, ) -> Dict[str, Any]: """국내주식 분봉차트 (Redis 캐시)""" try: if not ref_time: kst = pytz.timezone("Asia/Seoul") ref_time = datetime.now(kst).strftime("%H%M%S") ref_bucket = ref_time[:4] # HHMM cache_key = f"chart:minute:{stock_code}:{ref_bucket}" # 캐시 히트 cached = await _cache_get(cache_key) if cached is not None: logger.debug(f"📦 [캐시 히트] {cache_key}") return cached # KIS API 호출 client = self.real_client or self.client params = { "FID_ETC_CLS_CODE": "", "FID_INPUT_ISCD": stock_code, "FID_INPUT_HOUR_1": ref_time, "FID_PW_DATA_INCU_YN": "Y" if include_past else "N", } result, _ = await client.request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice", "FHKST03010200", params=params, ) # 캐시 저장 await _cache_set(cache_key, TTL_MINUTE_CHART, result) return result except Exception as e: logger.error(f"❌ 분봉 차트 실패 {stock_code}: {e}") raise async def get_overseas_price(self, symbol: str, exchange_code: str = "NASD") -> Dict[str, Any]: """해외주식 현재가""" try: client = self.real_client or self.client result, _ = await client.request( "GET", "/uapi/overseas-price/v1/quotations/price", "HHDFS00000300", params={ "AUTH": "", "EXCD": exchange_code.upper(), "SYMB": symbol.upper() } ) return result except Exception as e: logger.error(f"❌ 해외주식 시세 실패 {symbol}: {e}") raise async def get_orderbook(self, stock_code: str) -> Dict[str, Any]: """호가 정보 조회 (10단계)""" try: client = self.real_client or self.client result, _ = await client.request( "GET", "/uapi/domestic-stock/v1/quotations/inquire-asking-price-exp-ccn", "FHKST01010200", params={ "FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": stock_code } ) return result except Exception as e: logger.error(f"❌ 호가 조회 실패 {stock_code}: {e}") raise async def get_volume_rank(self, market: str = "J") -> Dict[str, Any]: """거래량 순위 조회""" try: client = self.real_client or self.client result, _ = await client.request( "GET", "/uapi/domestic-stock/v1/quotations/volume-rank", "FHPST01710000", params={ "FID_COND_MRKT_DIV_CODE": market, "FID_COND_SCR_DIV_CODE": "20171", "FID_INPUT_ISCD": "0000", "FID_DIV_CLS_CODE": "0", "FID_BLNG_CLS_CODE": "0", "FID_TRGT_CLS_CODE": "111111111", "FID_TRGT_EXLS_CLS_CODE": "000000", "FID_INPUT_PRICE_1": "5000", "FID_INPUT_PRICE_2": "", "FID_VOL_CNT": "", "FID_INPUT_DATE_1": "" } ) return result except Exception as e: logger.error(f"❌ 거래량 순위 실패: {e}") raise # 싱글톤 market_data_provider = MarketDataProvider()