finestock/example_async_nh.py

import asyncio
import os
import queue
import sys
from loguru import logger

import finestock
from finestock import APIProvider
from finestock.comm.api_interface import AuthenticationProvider, RealtimeProvider

try:
    # 선택 의존성: 설치돼 있으면 .env 파일을 자동으로 읽어 os.environ에 채워준다.
    # pip install python-dotenv
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    pass

# Configure logging
logger.remove()
logger.add(sys.stdout, level="DEBUG")


async def consume_data(data_queue: queue.Queue):
    """
    Background task to consume data from the queue asynchronously.
    """
    logger.info("Data consumer started...")
    while True:
        try:
            if not data_queue.empty():
                data = data_queue.get_nowait()
                print(f"Received Realtime Data: {data}")
            else:
                await asyncio.sleep(0.1)  # Yield control
        except Exception as e:
            logger.error(f"Consumer Error: {e}")
            await asyncio.sleep(1)


async def main():
    finestock.print_version_info()

    # 1. Create API Instance
    # 모의투자로 테스트하려면 APIProvider.NHV를 사용한다. 단, 접근토큰발급(oauth())은
    # NH·NHV 모두 항상 운영 도메인에서만 발급된다(모의투자 계좌로 거래하더라도 토큰은 동일).
    api = finestock.create_api(APIProvider.NH)

    # 2. Setup Authentication
    # 실키/시크릿은 절대 코드에 하드코딩하지 말 것. 환경변수로 주입한다.
    # 예 (PowerShell): $env:APP_KEY="..."; $env:APP_SECRET="..."
    auth_api: AuthenticationProvider = api
    if isinstance(auth_api, AuthenticationProvider):
        app_key = os.environ.get("APP_KEY", "YOUR_APP_KEY")
        app_secret = os.environ.get("APP_SECRET", "YOUR_APP_SECRET")
        auth_api.set_oauth_info(app_key, app_secret)
        auth_api.oauth()
        logger.info("Logged in.")

    # 3. Setup Realtime Data (Async)
    realtime_api: RealtimeProvider = api
    if isinstance(realtime_api, RealtimeProvider):
        # Create Queue for data
        data_queue = queue.Queue()
        realtime_api.set_data_queue(data_queue)

        # Connect to WebSocket
        logger.info("Connecting to WebSocket...")
        await realtime_api.connect()

        # Subscribe to Realtime Price (체결가) — 삼성전자(005930)
        logger.info("Subscribing to 005930 price...")
        await realtime_api.recv_price("005930")

        # Subscribe to Realtime Orderbook (호가) — 삼성전자(005930)
        logger.info("Subscribing to 005930 orderbook...")
        await realtime_api.recv_orderbook("005930")

        # 참고: NH krstock API에는 지수 실시간 채널이 없어 recv_index()는 stub이다
        # (호출하면 "not supported" 로그만 남기고 아무것도 구독하지 않는다).

        # Start the data consumer task
        consumer_task = asyncio.create_task(consume_data(data_queue))

        # Start the API run loop
        logger.info("Starting API Event Loop...")

        # Run for 30 seconds then exit for verification
        try:
            await asyncio.wait_for(realtime_api.run(), timeout=30)
        except asyncio.TimeoutError:
            logger.info("Test finished (timeout)")

        # Cleanup
        consumer_task.cancel()
        await realtime_api.disconnect()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("Terminated by user")