161 lines
4.8 KiB
Python
Executable File
161 lines
4.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
API 모니터링 테스트 스크립트
|
|
|
|
KIS API의 재시도 및 쓰로틀링 기능을 테스트합니다.
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# 프로젝트 루트 추가
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from app.services.korea_investment_api import KoreaInvestmentAPI
|
|
import logging
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def test_throttling():
|
|
"""쓰로틀링 테스트: 빠른 연속 호출"""
|
|
logger.info("=" * 60)
|
|
logger.info("쓰로틀링 테스트 시작: 5개 종목 연속 조회")
|
|
logger.info("=" * 60)
|
|
|
|
kis_api = KoreaInvestmentAPI()
|
|
|
|
test_stocks = [
|
|
"005930", # 삼성전자
|
|
"000660", # SK하이닉스
|
|
"035420", # 네이버
|
|
"051910", # LG화학
|
|
"005380", # 현대차
|
|
]
|
|
|
|
import time
|
|
start = time.time()
|
|
|
|
for stock_code in test_stocks:
|
|
logger.info(f"📊 Fetching {stock_code}...")
|
|
try:
|
|
result = await kis_api.get_current_price(stock_code)
|
|
if result and result.get("output"):
|
|
price = result["output"].get("stck_prpr", "N/A")
|
|
logger.info(f"✅ {stock_code}: {price}원")
|
|
except Exception as e:
|
|
logger.error(f"❌ {stock_code}: {e}")
|
|
|
|
elapsed = time.time() - start
|
|
logger.info(f"⏱️ Total time: {elapsed:.2f}초")
|
|
logger.info(f"📈 Expected: ~{len(test_stocks) * kis_api.min_request_interval:.1f}초 (쓰로틀링 적용)")
|
|
|
|
# 통계 출력
|
|
stats = kis_api.get_monitor_stats()
|
|
logger.info("=" * 60)
|
|
logger.info("📡 API 통계:")
|
|
logger.info(f" - 총 호출: {stats['total_calls']}건")
|
|
logger.info(f" - 오류: {stats['total_errors']}건")
|
|
logger.info(f" - 재시도: {stats['total_retries']}건")
|
|
logger.info(f" - 오류율: {stats['error_rate']:.1%}")
|
|
logger.info(f" - 평균 응답: {stats['avg_duration']:.2f}초")
|
|
logger.info("=" * 60)
|
|
|
|
|
|
async def test_retry():
|
|
"""재시도 테스트: 잘못된 종목 코드로 테스트"""
|
|
logger.info("=" * 60)
|
|
logger.info("재시도 테스트 시작: 오류 케이스")
|
|
logger.info("=" * 60)
|
|
|
|
kis_api = KoreaInvestmentAPI()
|
|
|
|
# 잘못된 종목 코드
|
|
invalid_code = "000000"
|
|
|
|
try:
|
|
logger.info(f"📊 Fetching invalid stock: {invalid_code}")
|
|
result = await kis_api.get_current_price(invalid_code)
|
|
logger.info(f"✅ Result: {result}")
|
|
except Exception as e:
|
|
logger.error(f"❌ Expected error: {e}")
|
|
|
|
# 통계 출력
|
|
stats = kis_api.get_monitor_stats()
|
|
logger.info("=" * 60)
|
|
logger.info("📡 API 통계:")
|
|
logger.info(f" - 총 호출: {stats['total_calls']}건")
|
|
logger.info(f" - 오류: {stats['total_errors']}건")
|
|
logger.info(f" - 재시도: {stats['total_retries']}건")
|
|
logger.info(f" - 오류율: {stats['error_rate']:.1%}")
|
|
logger.info("=" * 60)
|
|
|
|
|
|
async def test_concurrent():
|
|
"""동시 요청 테스트: 세마포어 확인"""
|
|
logger.info("=" * 60)
|
|
logger.info(f"동시 요청 테스트 시작 (최대 동시 요청: 2건)")
|
|
logger.info("=" * 60)
|
|
|
|
kis_api = KoreaInvestmentAPI()
|
|
|
|
test_stocks = [
|
|
"005930", # 삼성전자
|
|
"000660", # SK하이닉스
|
|
"035420", # 네이버
|
|
"051910", # LG화학
|
|
]
|
|
|
|
import time
|
|
start = time.time()
|
|
|
|
# 병렬 실행
|
|
tasks = [kis_api.get_current_price(code) for code in test_stocks]
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
elapsed = time.time() - start
|
|
|
|
success = sum(1 for r in results if not isinstance(r, Exception))
|
|
logger.info(f"✅ 성공: {success}/{len(test_stocks)}")
|
|
logger.info(f"⏱️ Total time: {elapsed:.2f}초")
|
|
|
|
# 통계 출력
|
|
stats = kis_api.get_monitor_stats()
|
|
logger.info("=" * 60)
|
|
logger.info("📡 API 통계:")
|
|
logger.info(f" - 총 호출: {stats['total_calls']}건")
|
|
logger.info(f" - 오류: {stats['total_errors']}건")
|
|
logger.info(f" - 재시도: {stats['total_retries']}건")
|
|
logger.info(f" - 오류율: {stats['error_rate']:.1%}")
|
|
logger.info(f" - 평균 응답: {stats['avg_duration']:.2f}초")
|
|
logger.info("=" * 60)
|
|
|
|
|
|
async def main():
|
|
"""메인 테스트 실행"""
|
|
logger.info("🚀 KIS API 안전 운영 테스트 시작")
|
|
logger.info("")
|
|
|
|
# 1. 쓰로틀링 테스트
|
|
await test_throttling()
|
|
await asyncio.sleep(2)
|
|
|
|
# 2. 재시도 테스트
|
|
await test_retry()
|
|
await asyncio.sleep(2)
|
|
|
|
# 3. 동시 요청 테스트
|
|
await test_concurrent()
|
|
|
|
logger.info("")
|
|
logger.info("✅ 모든 테스트 완료")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|