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

refactor(phase1): Add market data module for price/chart queries

Browse Source
This commit is contained in:
YoungWooSeo
2026-03-06 10:45:21 +09:00
parent 0bfca909cc
commit ea4ebae992
1 changed files with 247 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

247
backend/app/trading/market_data.py Normal file
View File

@@ -0,0 +1,247 @@
"""
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()