shuim/auto-stock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 차트 API에 Redis 캐시 레이어 추가 (일봉/분봉/해외)

Browse Source
This commit is contained in:
YoungWooSeo
2026-02-26 13:36:06 +09:00
parent 92a1c81ef1
commit 82346fcc46
1 changed files with 130 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

175
backend/app/services/korea_investment_api.py
View File

@@ -12,6 +12,12 @@ from app.core.config import settings
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# ── 차트 캐시 TTL 상수 ────────────────────────────────────────────────────────
TTL_DAILY_CHART = 300 # 일봉 캐시 5분 (장중 업데이트 주기 고려)
TTL_MINUTE_CHART = 60 # 분봉 캐시 1분 (실시간성 유지)
TTL_MINUTE_FULLDAY = 60 # 당일 전체 분봉 캐시 1분
TTL_OVERSEAS_CHART = 300 # 해외 일봉 캐시 5분
class APICallMonitor: class APICallMonitor:
"""API 호출 모니터링 및 통계""" """API 호출 모니터링 및 통계"""
@@ -66,6 +72,26 @@ class APICallMonitor:
return self.get_error_rate() > threshold return self.get_error_rate() > threshold
# ── Redis 캐시 헬퍼 (redis.py 의존성 없이 안전하게 호출) ──────────────────────
async def _cache_get(key: str) -> Optional[Any]:
"""Redis 캐시에서 값 조회. Redis 미연결 시 None 반환 (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 KoreaInvestmentAPI: class KoreaInvestmentAPI:
"""한국투자증권 API 클라이언트 (안전한 재시도 및 쓰로틀링 적용)""" """한국투자증권 API 클라이언트 (안전한 재시도 및 쓰로틀링 적용)"""
@@ -376,10 +402,22 @@ class KoreaInvestmentAPI:
raise raise
async def get_daily_chart(self, stock_code: str, period: str = "D", limit: int = 240) -> Dict[str, Any]: async def get_daily_chart(self, stock_code: str, period: str = "D", limit: int = 240) -> Dict[str, Any]:
"""일봉/주봉/월봉 조회 (페이지네이션 지원으로 최대 240개까지 조회)""" """일봉/주봉/월봉 조회 (Redis 캐시 적용, 페이지네이션 지원으로 최대 240개까지 조회)
캐시 키: chart:daily:{stock_code}:{period}:{limit}
TTL: 5분 (장중 업데이트 주기 고려)
"""
cache_key = f"chart:daily:{stock_code}:{period}:{limit}"
# ── 1. 캐시 히트 확인 ───────────────────────────────────────────────
cached = await _cache_get(cache_key)
if cached is not None:
logger.debug(f"📦 [캐시 히트] {cache_key}")
return cached
# ── 2. KIS API 호출 ─────────────────────────────────────────────────
all_output2 = [] all_output2 = []
# 1. 첫 요청
params = { params = {
"FID_COND_MRKT_DIV_CODE": "J", "FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": stock_code, "FID_INPUT_ISCD": stock_code,
@@ -394,8 +432,6 @@ class KoreaInvestmentAPI:
params=params params=params
) )
# FHKST01010400은 'output' (배열) 또는 'output2' (배열)로 데이터를 반환
# 실전/모의투자 환경에 따라 키 이름이 다를 수 있음
data_key = 'output2' if isinstance(result.get('output2'), list) and result.get('output2') else \ data_key = 'output2' if isinstance(result.get('output2'), list) and result.get('output2') else \
'output' if isinstance(result.get('output'), list) and result.get('output') else None 'output' if isinstance(result.get('output'), list) and result.get('output') else None
@@ -406,9 +442,9 @@ class KoreaInvestmentAPI:
all_output2.extend(result.get(data_key, [])) all_output2.extend(result.get(data_key, []))
tr_cont = headers.get('tr_cont') tr_cont = headers.get('tr_cont')
# 2. 연속 조회 (필요 시) # 연속 조회
while tr_cont in ('F', 'M') and len(all_output2) < limit: while tr_cont in ('F', 'M') and len(all_output2) < limit:
await asyncio.sleep(self.min_request_interval + 0.1) # Rate limit await asyncio.sleep(self.min_request_interval + 0.1)
extra_headers = {'tr_cont': 'Y'} extra_headers = {'tr_cont': 'Y'}
result_cont, headers_cont = await self._make_request( result_cont, headers_cont = await self._make_request(
@@ -425,12 +461,16 @@ class KoreaInvestmentAPI:
tr_cont = headers_cont.get('tr_cont') if headers_cont else None tr_cont = headers_cont.get('tr_cont') if headers_cont else None
# 3. 최종 결과 조합 final_result = result
final_result = result # 첫 페이지 응답을 기반으로 조합
final_result['output2'] = all_output2 final_result['output2'] = all_output2
final_result['msg1'] = f"OK (fetched {len(all_output2)} data points)" final_result['msg1'] = f"OK (fetched {len(all_output2)} data points)"
logger.info(f"✅ Fetched {len(all_output2)} chart data points for {stock_code}") logger.info(f"✅ Fetched {len(all_output2)} chart data points for {stock_code}")
# ── 3. 캐시 저장 ────────────────────────────────────────────────
await _cache_set(cache_key, TTL_DAILY_CHART, final_result)
logger.debug(f"💾 [캐시 저장] {cache_key} TTL={TTL_DAILY_CHART}s")
return final_result return final_result
except Exception as e: except Exception as e:
@@ -440,12 +480,8 @@ class KoreaInvestmentAPI:
async def get_overseas_price(self, symbol: str, exchange_code: str = "NASD") -> Dict[str, Any]: async def get_overseas_price(self, symbol: str, exchange_code: str = "NASD") -> Dict[str, Any]:
"""해외주식 현재가 조회 (NASD, NYSE, AMEX)""" """해외주식 현재가 조회 (NASD, NYSE, AMEX)"""
try: try:
# 거래소 코드 보정
exch_map = {"NASD": "NAS", "NYSE": "NYS", "AMEX": "AMS"} exch_map = {"NASD": "NAS", "NYSE": "NYS", "AMEX": "AMS"}
formatted_exchange = exch_map.get(exchange_code.upper(), exchange_code.upper()) formatted_exchange = exch_map.get(exchange_code.upper(), exchange_code.upper())
# 모의투자 여부에 따라 TR_ID가 다를 수 있는지 확인 필요하나,
# 시세 조회는 실전/모의 공통인 경우가 많음 (HHDFS00000300)
result, _ = await self._make_request( result, _ = await self._make_request(
"GET", "GET",
@@ -453,7 +489,7 @@ class KoreaInvestmentAPI:
"HHDFS00000300", "HHDFS00000300",
params={ params={
"AUTH": "", "AUTH": "",
"EXCD": exchange_code.upper(), # NASD, NYSE, AMEX "EXCD": exchange_code.upper(),
"SYMB": symbol.upper() "SYMB": symbol.upper()
} }
) )
@@ -468,10 +504,14 @@ class KoreaInvestmentAPI:
ref_time: str = "", ref_time: str = "",
include_past: bool = True, include_past: bool = True,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""국내주식 분봉차트 조회 (FHKST03010200) """국내주식 분봉차트 조회 (FHKST03010200, Redis 캐시 적용)
캐시 키: chart:minute:{stock_code}:{ref_time_bucket}
- ref_time_bucket: ref_time을 1분 단위로 버킷화 (HHMM)
TTL: 1분
KIS API: /uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice KIS API: /uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice
한 번 호출에 최대 30개 분봉 반환. 전체 장 데이터는 여러 번 호출 필요. 한 번 호출에 최대 30개 분봉 반환.
Args: Args:
stock_code: 종목코드 (e.g. "005930") stock_code: 종목코드 (e.g. "005930")
@@ -483,15 +523,15 @@ class KoreaInvestmentAPI:
"output1": { 현재가 요약 }, "output1": { 현재가 요약 },
"output2": [ "output2": [
{ {
"stck_bsop_date": "20260222", # 날짜 YYYYMMDD "stck_bsop_date": "20260222",
"stck_cntg_hour": "091500", # 시간 HHMMSS "stck_cntg_hour": "091500",
"stck_prpr": "71800", # 종가(현재가) "stck_prpr": "71800",
"stck_oprc": "71500", # 시가 "stck_oprc": "71500",
"stck_hgpr": "72000", # 고가 "stck_hgpr": "72000",
"stck_lwpr": "71400", # 저가 "stck_lwpr": "71400",
"cntg_vol": "12345", # 분봉 거래량 "cntg_vol": "12345",
"acml_vol": "987654", # 누적 거래량 "acml_vol": "987654",
"acml_tr_pbmn": "70896123456", # 누적 거래대금(원) "acml_tr_pbmn": "70896123456",
}, ... }, ...
] ]
} }
@@ -503,6 +543,17 @@ class KoreaInvestmentAPI:
kst = pytz.timezone("Asia/Seoul") kst = pytz.timezone("Asia/Seoul")
ref_time = datetime.now(kst).strftime("%H%M%S") ref_time = datetime.now(kst).strftime("%H%M%S")
# 캐시 키: 분 단위 버킷 (HHMM) — 같은 분 안에서 중복 호출 방지
ref_bucket = ref_time[:4] # HHMM
cache_key = f"chart:minute:{stock_code}:{ref_bucket}"
# ── 1. 캐시 히트 확인 ─────────────────────────────────────────
cached = await _cache_get(cache_key)
if cached is not None:
logger.debug(f"📦 [캐시 히트] {cache_key}")
return cached
# ── 2. KIS API 호출 ───────────────────────────────────────────
params = { params = {
"FID_ETC_CLS_CODE": "", "FID_ETC_CLS_CODE": "",
"FID_INPUT_ISCD": stock_code, "FID_INPUT_ISCD": stock_code,
@@ -515,6 +566,11 @@ class KoreaInvestmentAPI:
"FHKST03010200", "FHKST03010200",
params=params, params=params,
) )
# ── 3. 캐시 저장 ──────────────────────────────────────────────
await _cache_set(cache_key, TTL_MINUTE_CHART, result)
logger.debug(f"💾 [캐시 저장] {cache_key} TTL={TTL_MINUTE_CHART}s")
return result return result
except Exception as e: except Exception as e:
logger.error(f"Error fetching minute chart for {stock_code}: {e}") logger.error(f"Error fetching minute chart for {stock_code}: {e}")
@@ -525,11 +581,14 @@ class KoreaInvestmentAPI:
stock_code: str, stock_code: str,
market_open: str = "090000", market_open: str = "090000",
) -> list: ) -> list:
"""당일 전체 분봉 데이터 수집 (여러 번 호출 집계) """당일 전체 분봉 데이터 수집 (Redis 캐시 적용, 여러 번 호출 집계)
캐시 키: chart:minute_fullday:{stock_code}:{YYYYMMDD_HHMM}
- HHMM: 현재 시각의 분 단위 버킷 → 1분마다 갱신
TTL: 1분
09:00부터 현재까지의 모든 1분봉을 순서대로 반환. 09:00부터 현재까지의 모든 1분봉을 순서대로 반환.
한 번 호출에 30개씩 → 최대 ~13회 호출 필요 (390분/30). 한 번 호출에 30개씩 → 최대 ~13회 호출 필요 (390분/30).
시간 역순 데이터를 정순으로 변환해서 반환.
Returns: Returns:
List[dict] — output2 레코드 목록 (오래된 것 → 최신 순) List[dict] — output2 레코드 목록 (오래된 것 → 최신 순)
@@ -537,8 +596,20 @@ class KoreaInvestmentAPI:
from datetime import datetime from datetime import datetime
import pytz import pytz
kst = pytz.timezone("Asia/Seoul") kst = pytz.timezone("Asia/Seoul")
now_str = datetime.now(kst).strftime("%H%M%S") now_dt = datetime.now(kst)
today_str = now_dt.strftime("%Y%m%d")
now_bucket = now_dt.strftime("%H%M") # 1분 단위 버킷
cache_key = f"chart:minute_fullday:{stock_code}:{today_str}_{now_bucket}"
# ── 1. 캐시 히트 확인 ───────────────────────────────────────────────
cached = await _cache_get(cache_key)
if cached is not None:
logger.debug(f"📦 [캐시 히트] {cache_key} ({len(cached)}개 분봉)")
return cached
# ── 2. KIS API 다중 호출 집계 ────────────────────────────────────────
now_str = now_dt.strftime("%H%M%S")
all_candles: list = [] all_candles: list = []
ref = now_str ref = now_str
seen_times: set = set() seen_times: set = set()
@@ -562,7 +633,6 @@ class KoreaInvestmentAPI:
all_candles.append(c) all_candles.append(c)
new_added += 1 new_added += 1
# 가장 오래된 시간 확인
times = [c.get("stck_cntg_hour", "999999") for c in candles if c.get("stck_cntg_hour")] times = [c.get("stck_cntg_hour", "999999") for c in candles if c.get("stck_cntg_hour")]
oldest = min(times) if times else "000000" oldest = min(times) if times else "000000"
@@ -574,13 +644,30 @@ class KoreaInvestmentAPI:
# 오래된 것 → 최신 순으로 정렬 # 오래된 것 → 최신 순으로 정렬
all_candles.sort(key=lambda c: c.get("stck_cntg_hour", "")) all_candles.sort(key=lambda c: c.get("stck_cntg_hour", ""))
# ── 3. 캐시 저장 ────────────────────────────────────────────────────
if all_candles:
await _cache_set(cache_key, TTL_MINUTE_FULLDAY, all_candles)
logger.debug(f"💾 [캐시 저장] {cache_key} ({len(all_candles)}개 분봉) TTL={TTL_MINUTE_FULLDAY}s")
return all_candles return all_candles
async def get_overseas_daily_chart(self, symbol: str, exchange_code: str = "NASD", period: str = "D") -> Dict[str, Any]: async def get_overseas_daily_chart(self, symbol: str, exchange_code: str = "NASD", period: str = "D") -> Dict[str, Any]:
"""해외주식 차트 조회""" """해외주식 차트 조회 (Redis 캐시 적용)
캐시 키: chart:overseas:{exchange_code}:{symbol}:{period}
TTL: 5분
"""
cache_key = f"chart:overseas:{exchange_code.upper()}:{symbol.upper()}:{period}"
# ── 1. 캐시 히트 확인 ───────────────────────────────────────────────
cached = await _cache_get(cache_key)
if cached is not None:
logger.debug(f"📦 [캐시 히트] {cache_key}")
return cached
# ── 2. KIS API 호출 ─────────────────────────────────────────────────
try: try:
# 기간 코드: O(일), P(주), Q(월) - KIS 해외주식 문서 기준
# 또는 0(일), 1(주), 2(월)
gubn = "0" gubn = "0"
if period == "W": gubn = "1" if period == "W": gubn = "1"
elif period == "M": gubn = "2" elif period == "M": gubn = "2"
@@ -598,6 +685,11 @@ class KoreaInvestmentAPI:
"MODP": "1" "MODP": "1"
} }
) )
# ── 3. 캐시 저장 ────────────────────────────────────────────────
await _cache_set(cache_key, TTL_OVERSEAS_CHART, result)
logger.debug(f"💾 [캐시 저장] {cache_key} TTL={TTL_OVERSEAS_CHART}s")
return result return result
except Exception as e: except Exception as e:
logger.error(f"Error fetching overseas chart for {symbol}: {e}") logger.error(f"Error fetching overseas chart for {symbol}: {e}")
@@ -621,12 +713,10 @@ class KoreaInvestmentAPI:
# TR_ID 설정 (미국 매수/매도) # TR_ID 설정 (미국 매수/매도)
if order_type == "BUY": if order_type == "BUY":
tr_id = "VTTT1002U" if is_virtual else "TTTT1002U" # 미국 매수 tr_id = "VTTT1002U" if is_virtual else "TTTT1002U"
else: else:
tr_id = "VTTT1006U" if is_virtual else "TTTT1006U" # 미국 매도 tr_id = "VTTT1006U" if is_virtual else "TTTT1006U"
# 해외주식 주문 (미국)
# OVRS_EXCG_CD: NASD, NYSE, AMEX
result, _ = await self._make_request( result, _ = await self._make_request(
"POST", "POST",
"/uapi/overseas-stock/v1/trading/order", "/uapi/overseas-stock/v1/trading/order",
@@ -639,7 +729,7 @@ class KoreaInvestmentAPI:
"ORD_QTY": str(quantity), "ORD_QTY": str(quantity),
"OVRS_ORD_UNPR": str(price), "OVRS_ORD_UNPR": str(price),
"ORD_SVR_DVSN_CD": "0", "ORD_SVR_DVSN_CD": "0",
"ORD_DVSN": "00" # 00: 지정가 "ORD_DVSN": "00"
} }
) )
return result return result
@@ -693,7 +783,6 @@ class KoreaInvestmentAPI:
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""주문 실행""" """주문 실행"""
try: try:
# 환경 설정 확인
if not self.app_key or not self.app_secret or not self.account_number: if not self.app_key or not self.app_secret or not self.account_number:
error_msg = f"KIS API 설정 누락: app_key={bool(self.app_key)}, secret={bool(self.app_secret)}, account={bool(self.account_number)}" error_msg = f"KIS API 설정 누락: app_key={bool(self.app_key)}, secret={bool(self.app_secret)}, account={bool(self.account_number)}"
logger.error(error_msg) logger.error(error_msg)
@@ -709,13 +798,11 @@ class KoreaInvestmentAPI:
# 모의투자 환경 확인 # 모의투자 환경 확인
is_virtual = "vts" in self.base_url.lower() is_virtual = "vts" in self.base_url.lower()
# 매수/매도 구분 (모의투자는 V 접두사)
if order_type == "BUY": if order_type == "BUY":
tr_id = "VTTC0802U" if is_virtual else "TTTC0802U" # 현금 매수 tr_id = "VTTC0802U" if is_virtual else "TTTC0802U"
else: else:
tr_id = "VTTC0801U" if is_virtual else "TTTC0801U" # 현금 매도 tr_id = "VTTC0801U" if is_virtual else "TTTC0801U"
# 주문 유형 (00: 지정가, 01: 시장가)
order_dvsn = "01" if price is None else "00" order_dvsn = "01" if price is None else "00"
order_price = "0" if price is None else str(int(price)) order_price = "0" if price is None else str(int(price))
@@ -778,7 +865,7 @@ class KoreaInvestmentAPI:
async def search_stocks_by_condition( async def search_stocks_by_condition(
self, self,
market: str = "J", # J: 코스피, Q: 코스닥 market: str = "J",
condition_name: str = "", condition_name: str = "",
max_results: int = 100 max_results: int = 100
) -> Dict[str, Any]: ) -> Dict[str, Any]:
@@ -790,7 +877,7 @@ class KoreaInvestmentAPI:
"HHKST03900300", "HHKST03900300",
params={ params={
"FID_COND_MRKT_DIV_CODE": market, "FID_COND_MRKT_DIV_CODE": market,
"FID_COND_SCR_DIV_CODE": "20171", # 조건검색 "FID_COND_SCR_DIV_CODE": "20171",
"FID_INPUT_ISCD": "0000", "FID_INPUT_ISCD": "0000",
"FID_DIV_CLS_CODE": "0", "FID_DIV_CLS_CODE": "0",
"FID_BLNG_CLS_CODE": "0", "FID_BLNG_CLS_CODE": "0",
@@ -810,7 +897,6 @@ class KoreaInvestmentAPI:
async def get_stock_list(self, market: str = "J") -> Dict[str, Any]: async def get_stock_list(self, market: str = "J") -> Dict[str, Any]:
"""전체 종목 리스트 조회""" """전체 종목 리스트 조회"""
try: try:
# 거래대금 상위 종목 조회
result, _ = await self._make_request( result, _ = await self._make_request(
"GET", "GET",
"/uapi/domestic-stock/v1/quotations/volume-rank", "/uapi/domestic-stock/v1/quotations/volume-rank",
@@ -851,7 +937,6 @@ class KoreaInvestmentAPI:
} }
) )
# 디버깅용 로그
logger.info(f"📊 Orderbook raw response for {stock_code}:") logger.info(f"📊 Orderbook raw response for {stock_code}:")
logger.info(f" rt_cd: {result.get('rt_cd')}, msg1: {result.get('msg1')}") logger.info(f" rt_cd: {result.get('rt_cd')}, msg1: {result.get('msg1')}")
if result.get('output1'): if result.get('output1'):