import asyncio import queue import sys from loguru import logger import finestock from finestock import APIProvider from finestock.comm.api_interface import AuthenticationProvider, RealtimeProvider # 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: # Since refined queue is thread-safe standard queue, we can just poll it # In a real async app, you might use asyncio.Queue for await q.get() # But here we are injecting a standard queue as per interface. # To avoid blocking the loop, we use run_in_executor or just poll with sleep if not data_queue.empty(): data = data_queue.get_nowait() # Type hinting can be used here if data structure is known if isinstance(data, finestock.Price): type_str = "Index" if data.code == "001" else "Price/Trade" # Simple heuristic print(f"[{type_str}] Code: {data.code}, Time: {data.time}, Current: {data.close}, Vol: {data.volume}") elif isinstance(data, finestock.OrderBook): print(data) print(f"[OrderBook] Code: {data.code}, Buy: {data.total_buy}, Sell: {data.total_sell}") if data.buy: print(f" Best Buy: {data.buy[0].price} ({data.buy[0].qty})") if data.sell: print(f" Best Sell: {data.sell[0].price} ({data.sell[0].qty})") elif isinstance(data, finestock.Trade): print(f"[Order Status] Code: {data.code}, Flag: {data.trade_flag}, Order: {data.order_flag}, Qty: {data.qty}") else: print(f"Received 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() # 0. Select Provider # TARGET_PROVIDER = APIProvider.LSV TARGET_PROVIDER = APIProvider.KIWOOMV print(f"Selected Provider: {TARGET_PROVIDER.name}") # 1. Create API Instance api = finestock.create_api(TARGET_PROVIDER) # 2. Setup Authentication auth_api: AuthenticationProvider = api if isinstance(auth_api, AuthenticationProvider): if TARGET_PROVIDER == APIProvider.LSV: # LSV Keys APP_KEY = "PSgjLpp90XLhWexJYGL0P36oyzM3ne5kjhDW" APP_SECRET = "htA3h9kULFPD75vtHAt0JaXyYAjsbf0b" elif TARGET_PROVIDER == APIProvider.KIWOOM: # Kiwoom Keys (Test) APP_KEY = "RSvh1vtzHj8Z5F15Ee7_rJ5znHuAKOF6A8v90TEkyBk" APP_SECRET = "_7wuGnkD-Crf1TJKErJPYkQp7AVuh_gVOPgQ2Uyyh7k" elif TARGET_PROVIDER == APIProvider.KIWOOMV: APP_KEY = "5-7V0r97uwlzmPNFa_FnwFEZtnyhFAZ1qlnDXQH2rX4" APP_SECRET = "YOMvV7X1CwwST-_iFmRAPKXO7Wwr4zGGyTPX5gFcAqk" auth_api.set_oauth_info(APP_KEY, APP_SECRET) token = auth_api.oauth() print(f"Logged in with token: {token}") # 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 Prices # Subscribe to Realtime Prices, Orderbook, and Trades # Ex: Samsung Electronics (005930) logger.info("Subscribing to 005930 (Price, Orderbook, Trade)...") #await realtime_api.recv_price("005930") #완료 #await realtime_api.recv_orderbook("005930") #완료 #await realtime_api.recv_trade("005930") # Same as price (0B) logger.info("Subscribing to Order Status (00)...") await realtime_api.recv_order_status() # Subscribe to Index # Ex: KOSPI (001) logger.info("Subscribing to KOSPI Index (001)...") #await realtime_api.recv_index("001") #완료 # Start the data consumer task consumer_task = asyncio.create_task(consume_data(data_queue)) # Start the API run loop (This blocks until disconnected) logger.info("Starting API Event Loop...") await realtime_api.run() # Cleanup (if loop breaks) consumer_task.cancel() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logger.info("Terminated by user")