Coverage for strategies / backtest_data_provider.py: 97%

1# strategies/backtest_data_provider.py 

2 

3import logging 

4from core.market_clock import MarketClock 

5from services.stock_query_service import StockQueryService 

6 

7 

8class BacktestDataProvider: 

9 """ 

10 백테스트 전략 실행에 필요한 주가 데이터를 제공하는 클래스입니다. 

11 모의(mock) 데이터 조회 및 실제 과거 데이터 조회 로직을 포함합니다. 

12 """ 

13 

14 def __init__(self, stock_query_service: StockQueryService, market_clock: MarketClock, logger=None): 

15 self.stock_query_service = stock_query_service 

16 self._market_clock = market_clock 

17 self._logger = logger if logger else logging.getLogger(__name__) 

18 

19 async def mock_price_lookup(self, stock_code: str) -> int: 

20 """ 

21 백테스트용으로 주가 상승을 가정한 모의 가격 제공 

22 (실제로는 DB, CSV, 또는 API를 통해 특정 시점 데이터를 받아야 함) 

23 """ 

24 try: 

25 resp = await self.stock_query_service.get_current_price(stock_code) 

26 if not resp or resp.rt_cd != "0" or not resp.data: 26 ↛ 27line 26 didn't jump to line 27 because the condition on line 26 was never true

27 return 0 

28 output = resp.data.get("output", {}) 

29 current = int(output.get("stck_prpr", 0)) 

30 return int(current * 1.05) 

31 except Exception as e: 

32 self._logger.warning(f"[백테스트] {stock_code} 모의 가격 조회 실패: {e}") 

33 return 0 

34 

35 async def realistic_price_lookup(self, stock_code: str, base_summary: dict, minutes_after: int) -> int: 

36 """ 

37 백테스트용으로, 실제 과거 분봉 데이터를 기반으로 N분 후의 가격을 조회합니다. 

38 

39 :param stock_code: 종목코드 

40 :param base_summary: 초기 등락률이 감지된 시점의 가격 요약 정보 

41 :param minutes_after: 몇 분 후의 가격을 조회할지 

42 :return: N분 후의 실제 종가 

43 """ 

44 try: 

45 backtest_date = self._market_clock.get_current_kst_time().strftime('%Y%m%d') 

46 

47 # StockQueryService를 통해 분봉 데이터 조회 (시간 오름차순 정렬됨) 

48 chart_data = await self.stock_query_service.get_day_intraday_minutes_list( 

49 stock_code, date_ymd=backtest_date 

50 ) 

51 

52 if not isinstance(chart_data, list) or not chart_data: 

53 self._logger.warning(f"[백테스트] {stock_code}의 분봉 데이터가 없거나 형식이 올바르지 않습니다.") 

54 return base_summary.get("current", 0) 

55 

56 base_price = base_summary.get("current", 0) 

57 base_index = -1 

58 

59 for i, candle in enumerate(chart_data): 

60 # get_day_intraday_minutes_list는 stck_prpr(현재가)를 포함 

61 if int(candle.get('stck_prpr', 0)) == base_price: 

62 base_index = i 

63 break 

64 

65 if base_index == -1: 

66 self._logger.warning(f"[백테스트] {stock_code}의 기준 시점 분봉({base_price})을 찾지 못했습니다.") 

67 return base_price 

68 

69 # 데이터가 시간 오름차순(과거->미래)이므로 인덱스를 더함 

70 after_index = base_index + minutes_after 

71 

72 if after_index >= len(chart_data): 

73 after_index = len(chart_data) - 1 

74 

75 if after_index < 0: 

76 after_index = 0 

77 

78 after_price = int(chart_data[after_index].get('stck_prpr', 0)) 

79 

80 self._logger.info(f"[백테스트] {stock_code} | 기준가: {base_price} | {minutes_after}분 후 가격: {after_price}") 

81 return after_price 

82 

83 except Exception as e: 

84 self._logger.exception(f"[백테스트] {stock_code} 가격 조회 중 오류 발생: {e}") 

85 return base_summary.get("current", 0)