Coverage for brokers / broker_api_wrapper.py: 92%

1# user_api/broker_api_wrapper.py 

2 

3from brokers.korea_investment.korea_invest_client import KoreaInvestApiClient 

4from repositories.stock_code_repository import StockCodeRepository 

5from typing import Any, List 

6from common.types import ResCommonResponse, Exchange 

7from core.cache.cache_wrapper import cache_wrap_client 

8from core.retry_queue.api_request_queue import ApiRequestQueue 

9from core.retry_queue.client_with_retry_queue import retry_queue_wrap_client 

10 

11 

12class BrokerAPIWrapper: 

13 """ 

14 범용 사용자용 API Wrapper 클래스. 

15 증권사별 구현체를 내부적으로 호출하여, 일관된 방식의 인터페이스를 제공. 

16 """ 

17 

18 def __init__(self, broker: str = "korea_investment", env=None, logger=None, market_clock=None, 

19 cache_config=None, market_calendar_service=None): 

20 self._broker = broker 

21 self._logger = logger 

22 self._client = None 

23 self._stock_mapper = StockCodeRepository(logger=logger) 

24 self.env = env 

25 self._retry_queue: ApiRequestQueue | None = None 

26 

27 if broker == "korea_investment": 

28 if env is None: 

29 raise ValueError("KoreaInvest API를 사용하려면 env 인스턴스가 필요합니다.") 

30 

31 

32 self._client = KoreaInvestApiClient(env, logger, market_clock, market_calendar_service) 

33 # RetryQueue는 Cache 안쪽에 위치: 캐시 히트 시 Queue를 거치지 않고, 

34 # 캐시 miss 후 실제 API 호출 실패 시에만 KoreaInvestApiClient를 직접 재시도 

35 self._retry_queue = ApiRequestQueue(logger=logger) 

36 self._client = retry_queue_wrap_client(self._client, self._retry_queue) 

37 self._client = cache_wrap_client( 

38 

39 self._client, logger, market_clock, 

40 lambda: "PAPER" if env.is_paper_trading else "REAL", 

41 

42 config=cache_config, 

43 market_calendar_service=market_calendar_service 

44 ) 

45 

46 else: 

47 raise NotImplementedError(f"지원되지 않는 증권사: {broker}") 

48 

49 async def stop(self): 

50 """이벤트 루프 종료 전 대기 중인 재시도 태스크를 정리합니다.""" 

51 if self._retry_queue: 

52 await self._retry_queue.stop() 

53 

54 # --- StockCodeRepository delegation --- 

55 async def get_name_by_code(self, code: str) -> str: 

56 """종목코드로 종목명을 반환합니다.""" 

57 return self._stock_mapper.get_name_by_code(code) 

58 

59 async def get_code_by_name(self, name: str) -> str: 

60 """종목명으로 종목코드를 반환합니다.""" 

61 return self._stock_mapper.get_code_by_name(name) 

62 

63 async def get_all_stock_codes(self) -> Any: 

64 """StockCodeRepository를 통해 모든 종목의 코드와 이름을 포함하는 DataFrame을 반환합니다.""" 

65 if hasattr(self._stock_mapper, 'df'): 

66 return self._stock_mapper.df 

67 else: 

68 self._logger.error("StockCodeRepository가 초기화되지 않았거나 df 속성이 없습니다.") 

69 return None 

70 

71 async def get_all_stock_code_list(self) -> List[str]: 

72 """모든 종목 코드 리스트만 반환합니다.""" 

73 df = await self.get_all_stock_codes() 

74 if df is not None and '종목코드' in df.columns: 

75 return df['종목코드'].tolist() 

76 return [] 

77 

78 async def get_all_stock_name_list(self) -> List[str]: 

79 """모든 종목명 리스트만 반환합니다.""" 

80 df = await self.get_all_stock_codes() 

81 if df is not None and '종목명' in df.columns: 

82 return df['종목명'].tolist() 

83 return [] 

84 

85 # --- KoreaInvestApiClient / Quotations API delegation --- 

86 async def get_stock_info_by_code(self, stock_code: str, exchange: Exchange = Exchange.KRX) -> ResCommonResponse: 

87 """종목코드로 종목의 전체 정보를 가져옵니다 (KoreaInvestApiQuotations 위임).""" 

88 return await self._client.get_stock_info_by_code(stock_code, exchange=exchange) 

89 

90 async def get_current_price(self, code: str, exchange: Exchange = Exchange.KRX) -> ResCommonResponse: 

91 """현재가를 조회합니다 (KoreaInvestApiQuotations 위임).""" 

92 return await self._client.get_current_price(code, exchange=exchange) 

93 

94 async def get_stock_conclusion(self, code: str, exchange: Exchange = Exchange.KRX) -> ResCommonResponse: 

95 """주식 체결(체결강도) 정보를 조회합니다.""" 

96 return await self._client.get_stock_conclusion(code, exchange=exchange) 

97 

98 async def get_price_summary(self, code: str, exchange: Exchange = Exchange.KRX) -> ResCommonResponse: 

99 """주어진 종목코드에 대해 시가/현재가/등락률(%) 요약 정보를 반환합니다 (KoreaInvestApiQuotations 위임).""" 

100 return await self._client.get_price_summary(code, exchange=exchange) 

101 

102 async def get_market_cap(self, code: str, exchange: Exchange = Exchange.KRX) -> ResCommonResponse: 

103 """종목코드로 시가총액을 반환합니다 (KoreaInvestApiQuotations 위임).""" 

104 return await self._client.get_market_cap(code, exchange=exchange) 

105 

106 async def get_top_market_cap_stocks_code(self, market_code: str, count: int = 30) -> ResCommonResponse: 

107 """시가총액 상위 종목 목록을 반환합니다 (KoreaInvestApiQuotations 위임).""" 

108 return await self._client.get_top_market_cap_stocks_code(market_code, count) 

109 

110 async def inquire_daily_itemchartprice(self, stock_code: str, start_date: str, end_date: str, 

111 fid_period_div_code: str = 'D', 

112 exchange: Exchange = Exchange.KRX, **kwargs) -> ResCommonResponse: 

113 """일별/분봉 주식 시세 차트 데이터를 조회합니다 (KoreaInvestApiQuotations 위임).""" 

114 return await self._client.inquire_daily_itemchartprice(stock_code, start_date=start_date, end_date=end_date, 

115 fid_period_div_code=fid_period_div_code, 

116 exchange=exchange, **kwargs) 

117 async def inquire_time_itemchartprice( 

118 self, *, stock_code: str, input_hour_1: str, 

119 pw_data_incu_yn: str = "Y", etc_cls_code: str = "0" 

120 ) -> ResCommonResponse: 

121 return await self._client.inquire_time_itemchartprice( 

122 stock_code=stock_code, 

123 input_hour_1=input_hour_1, 

124 pw_data_incu_yn=pw_data_incu_yn, 

125 etc_cls_code=etc_cls_code, 

126 ) 

127 

128 async def inquire_time_dailychartprice( 

129 self, *, stock_code: str, input_date_1: str, input_hour_1: str = "", 

130 pw_data_incu_yn: str = "Y", fake_tick_incu_yn: str = "" 

131 ) -> ResCommonResponse: 

132 return await self._client.inquire_time_dailychartprice( 

133 stock_code=stock_code, 

134 input_date_1=input_date_1, 

135 input_hour_1=input_hour_1, 

136 pw_data_incu_yn=pw_data_incu_yn, 

137 fake_tick_incu_yn=fake_tick_incu_yn, 

138 ) 

139 

140 async def get_asking_price(self, stock_code: str, exchange: Exchange = Exchange.KRX) -> ResCommonResponse: 

141 """ 

142 종목의 실시간 호가(매도/매수 잔량 포함) 정보를 조회합니다. 

143 """ 

144 return await self._client.get_asking_price(stock_code, exchange=exchange) 

145 

146 async def get_time_concluded_prices(self, stock_code: str, exchange: Exchange = Exchange.KRX) -> ResCommonResponse: 

147 """ 

148 종목의 시간대별 체결가/체결량 정보를 조회합니다. 

149 """ 

150 return await self._client.get_time_concluded_prices(stock_code, exchange=exchange) 

151 

152 # async def search_stocks_by_keyword(self, keyword: str) -> ResCommonResponse: 

153 # """ 

154 # 키워드로 종목을 검색합니다. 

155 # """ 

156 # return await self._client.search_stocks_by_keyword(keyword) 

157 

158 async def get_top_rise_fall_stocks(self, rise: bool = True) -> ResCommonResponse: 

159 """ 

160 상승률 또는 하락률 상위 종목을 조회합니다. 

161 

162 Args: 

163 rise (bool): True이면 상승률, False이면 하락률 상위를 조회합니다. 

164 """ 

165 return await self._client.get_top_rise_fall_stocks(rise) 

166 

167 async def get_top_volume_stocks(self) -> ResCommonResponse: 

168 """ 

169 거래량 상위 종목을 조회합니다. 

170 """ 

171 return await self._client.get_top_volume_stocks() 

172 

173 async def get_investor_trade_by_stock_daily(self, stock_code: str, date: str = None) -> ResCommonResponse: 

174 """종목별 투자자 매매동향(일별) 조회 (실전 전용)""" 

175 return await self._client.get_investor_trade_by_stock_daily(stock_code, date) 

176 

177 async def get_investor_trade_by_stock_daily_multi(self, stock_code: str, date: str = None, days: int = 3) -> ResCommonResponse: 

178 """종목별 투자자 매매동향(일별) 다중일 조회 (실전 전용) — output2[:days] 리스트 반환""" 

179 return await self._client.get_investor_trade_by_stock_daily_multi(stock_code, date, days) 

180 

181 async def get_program_trade_by_stock_daily(self, stock_code: str, date: str = None) -> ResCommonResponse: 

182 """종목별 프로그램매매추이(일별) 조회 (실전 전용)""" 

183 return await self._client.get_program_trade_by_stock_daily(stock_code, date) 

184 # 

185 # async def get_stock_news(self, stock_code: str) -> ResCommonResponse: 

186 # """ 

187 # 특정 종목의 뉴스를 조회합니다. 

188 # """ 

189 # return await self._client.get_stock_news(stock_code) 

190 

191 async def get_multi_price(self, stock_codes: list[str]) -> ResCommonResponse: 

192 """복수종목 현재가를 조회합니다 (최대 30종목, KoreaInvestApiQuotations 위임).""" 

193 return await self._client.get_multi_price(stock_codes) 

194 

195 async def get_etf_info(self, etf_code: str) -> ResCommonResponse: 

196 """ 

197 특정 ETF의 상세 정보를 조회합니다. 

198 """ 

199 return await self._client.get_etf_info(etf_code) 

200 

201 async def get_financial_ratio(self, stock_code: str) -> ResCommonResponse: 

202 """기업 재무비율을 조회합니다 (영업이익 증가율 등).""" 

203 return await self._client.get_financial_ratio(stock_code) 

204 

205 async def check_holiday(self, date: str) -> ResCommonResponse: 

206 """국내 휴장일 조회""" 

207 return await self._client.check_holiday(date) 

208 

209 # --- KoreaInvestApiClient / Account API delegation --- 

210 async def get_account_balance(self, exchange: Exchange = Exchange.KRX) -> ResCommonResponse: 

211 """계좌 잔고를 조회합니다 (KoreaInvestApiAccount 위임).""" 

212 return await self._client.get_account_balance(exchange=exchange) 

213 

214 # --- KoreaInvestApiClient / Trading API delegation --- 

215 async def place_stock_order(self, stock_code, order_price, order_qty, is_buy: bool, 

216 exchange: Exchange = Exchange.KRX) -> ResCommonResponse: 

217 """범용 주식 주문을 실행합니다 (KoreaInvestApiTrading 위임).""" 

218 return await self._client.place_stock_order(stock_code, order_price, order_qty, is_buy, exchange=exchange) 

219 

220 # --- KoreaInvestApiClient / WebSocket API delegation --- 

221 def is_websocket_receive_alive(self) -> bool: 

222 """웹소켓 수신 태스크가 살아있는지 확인.""" 

223 return self._client.is_websocket_receive_alive() 

224 

225 async def connect_websocket(self, on_message_callback=None) -> Any: # 실제 반환 값에 따라 타입 변경 

226 """웹소켓 연결을 시작합니다 (KoreaInvestWebSocketAPI 위임).""" 

227 return await self._client.connect_websocket(on_message_callback) 

228 

229 async def disconnect_websocket(self) -> Any: # 실제 반환 값에 따라 타입 변경 

230 """웹소켓 연결을 종료합니다 (KoreaInvestWebSocketAPI 위임).""" 

231 return await self._client.disconnect_websocket() 

232 

233 async def subscribe_realtime_price(self, stock_code: str) -> Any: # 실제 반환 값에 따라 타입 변경 

234 """실시간 체결 데이터 구독합니다 (KoreaInvestWebSocketAPI 위임).""" 

235 return await self._client.subscribe_realtime_price(stock_code) 

236 

237 async def unsubscribe_realtime_price(self, stock_code: str) -> Any: # 실제 반환 값에 따라 타입 변경 

238 """실시간 체결 데이터 구독 해지합니다 (KoreaInvestWebSocketAPI 위임).""" 

239 return await self._client.unsubscribe_realtime_price(stock_code) 

240 

241 async def subscribe_unified_price(self, stock_code: str) -> bool: 

242 """실시간 통합 체결가(H0UNCNT0) 구독합니다 (KRX+NXT 통합).""" 

243 return await self._client.subscribe_unified_price(stock_code) 

244 

245 async def unsubscribe_unified_price(self, stock_code: str) -> bool: 

246 """실시간 통합 체결가(H0UNCNT0) 구독 해지합니다.""" 

247 return await self._client.unsubscribe_unified_price(stock_code) 

248 

249 async def subscribe_realtime_quote(self, stock_code: str) -> Any: # 실제 반환 값에 따라 타입 변경 

250 """실시간 호가 데이터 구독합니다 (KoreaInvestWebSocketAPI 위임).""" 

251 return await self._client.subscribe_realtime_quote(stock_code) 

252 

253 async def unsubscribe_realtime_quote(self, stock_code: str) -> Any: # 실제 반환 값에 따라 타입 변경 

254 """실시간 호가 데이터 구독 해지합니다 (KoreaInvestWebSocketAPI 위임).""" 

255 return await self._client.unsubscribe_realtime_quote(stock_code) 

256 

257 async def subscribe_program_trading(self, stock_code: str): 

258 return await self._client.subscribe_program_trading(stock_code) 

259 

260 async def unsubscribe_program_trading(self, stock_code: str): 

261 return await self._client.unsubscribe_program_trading(stock_code)