whdwo798
a64a3f017b
- main.py: sleep 0.05/0.1 → 1.1초 (KIS rate limit 준수) - main.py: 전일 날짜 계산 수정 (월요일→금요일), 인라인 주석 env 파싱, 장 중 재시작 즉시 루프 진입 - strategy/volatility_breakout.py: has_prev_data() 추가, 중복 수집 skip - db/repository.py, order_executor.py: UPDATE ORDER BY → 서브쿼리 수정 (SQLite 호환) - kis_client.py: get_balance TR ID VTTC8001R → VTTC8434R - test_connection.py: API 호출 간 sleep 추가 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
"""
|
|
test_kis_connection.py
|
|
KIS API 연결 테스트 스크립트
|
|
실행: python test_kis_connection.py
|
|
|
|
테스트 항목:
|
|
1. 토큰 발급
|
|
2. 삼성전자 현재가 조회
|
|
3. 잔고 조회
|
|
4. 거래량 순위 조회
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import json
|
|
import sys
|
|
from dotenv import load_dotenv
|
|
|
|
# .env 로드
|
|
load_dotenv()
|
|
|
|
# 프로젝트 경로 추가
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from app.execution.kis_client import KISClient
|
|
|
|
async def test_connection():
|
|
print("=" * 55)
|
|
print(" KIS API 연결 테스트")
|
|
mode = "모의투자" if os.getenv("KIS_MOCK","true") == "true" else "실거래"
|
|
print(f" 현재 모드: {mode}")
|
|
print("=" * 55)
|
|
|
|
client = KISClient()
|
|
|
|
# ── 1. 토큰 발급 ──
|
|
print("\n[1] 액세스 토큰 발급...")
|
|
try:
|
|
token = await client.get_access_token()
|
|
print(f" ✅ 성공: {token[:20]}...")
|
|
except Exception as e:
|
|
print(f" ❌ 실패: {e}")
|
|
print(" → .env의 KIS 키를 확인해주세요")
|
|
return
|
|
|
|
await asyncio.sleep(0.5)
|
|
|
|
# ── 2. 삼성전자 현재가 ──
|
|
print("\n[2] 삼성전자(005930) 현재가 조회...")
|
|
try:
|
|
price = await client.get_price("005930")
|
|
print(f" ✅ 현재가: {price['current']:,}원")
|
|
print(f" 시가: {price['open']:,} | 고가: {price['high']:,} | 저가: {price['low']:,}")
|
|
print(f" 등락률: {price['change_pct']:+.2f}%")
|
|
except Exception as e:
|
|
print(f" ❌ 실패: {e}")
|
|
|
|
await asyncio.sleep(0.5)
|
|
|
|
# ── 3. 잔고 조회 ──
|
|
print("\n[3] 계좌 잔고 조회...")
|
|
try:
|
|
balance = await client.get_balance()
|
|
print(f" ✅ 예수금: {balance['cash']:,}원")
|
|
print(f" 보유 종목: {balance['total_cnt']}개")
|
|
for h in balance['holdings']:
|
|
print(f" └ {h['name']}({h['ticker']}) "
|
|
f"{h['qty']}주 / 평균가 {h['avg_price']:,}원 / "
|
|
f"{h['pnl_pct']:+.2f}%")
|
|
except Exception as e:
|
|
print(f" ❌ 실패: {e}")
|
|
|
|
await asyncio.sleep(0.5)
|
|
|
|
# ── 4. 거래량 순위 ──
|
|
print("\n[4] 거래량 순위 상위 5종목...")
|
|
try:
|
|
rank = await client.get_volume_rank(top_n=5)
|
|
for r in rank:
|
|
print(f" {r['rank']}위 {r['name']}({r['ticker']}) "
|
|
f"거래량 {r['volume']:,} / {r['change_pct']:+.2f}%")
|
|
except Exception as e:
|
|
print(f" ❌ 실패: {e}")
|
|
|
|
print("\n" + "=" * 55)
|
|
print(" 테스트 완료")
|
|
print("=" * 55)
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_connection())
|