Coverage for services / market_calendar_service.py: 92%
139 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-04 15:08 +0000
1import asyncio
2import logging
3from datetime import datetime, timedelta
4from typing import Optional
6from common.types import ResCommonResponse
7from core.performance_profiler import PerformanceProfiler
9class MarketCalendarService:
10 """
11 주식 시장의 개장일, 휴장일, 과거 최신 영업일 및 다음 개장 시간을 통합 관리하는 달력(Calendar) 매니저입니다.
12 """
13 def __init__(self, market_clock, logger=None, performance_profiler: Optional[PerformanceProfiler] = None):
14 self._market_clock = market_clock
15 self._logger = logger or logging.getLogger(__name__)
16 self._broker = None
17 self._pm = performance_profiler if performance_profiler else PerformanceProfiler(enabled=False)
19 # [과거/현재] get_latest_trading_date 캐시 변수 (기존 로직 유지)
20 self._cached_date = None
21 self._last_check_date = None
23 # [미래/달력] check_holiday API 기반 휴장일 캐시 변수 (신규 추가)
24 self._business_days_cache = {}
25 self._synced_months: set = set() # 동기화 완료된 월 집합 (단일 변수 → set으로 변경: 월 경계 재호출 방지)
27 def set_broker(self, broker):
28 self._broker = broker
30 # ==============================================================================
31 # 1. 과거/현재 기준 최신 영업일 조회 (기존 구현 완벽 유지 -> 기존 TC 통과 보장)
32 # ==============================================================================
33 async def get_latest_trading_date(self) -> Optional[str]:
34 """오늘을 포함하여 가장 최근에 장이 열렸던 영업일(YYYYMMDD)을 반환합니다."""
35 current_date = self._market_clock.get_current_kst_time().strftime("%Y%m%d")
37 # 캐시가 있고, 오늘 이미 확인했다면 캐시 반환
38 if self._cached_date and self._last_check_date == current_date:
39 return self._cached_date
41 if not self._broker:
42 self._logger.warning("MarketCalendarService: Broker is not set.")
43 return None
45 t_start = self._pm.start_timer()
46 try:
47 latest_date = await self._fetch_from_api()
48 if latest_date:
49 self._cached_date = latest_date
50 self._last_check_date = current_date
51 self._pm.log_timer("MarketCalendarService.get_latest_trading_date", t_start)
52 return latest_date
53 except Exception as e:
54 self._logger.error(f"최근 영업일 조회 실패: {e}")
55 self._pm.log_timer("MarketCalendarService.get_latest_trading_date [예외]", t_start)
56 return None
58 async def _fetch_from_api(self) -> Optional[str]:
59 """삼성전자(005930) 일봉 조회를 통해 가장 최근 영업일을 API에서 가져옵니다."""
60 if not self._broker: 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true
61 self._logger.warning("MarketCalendarService: Broker가 설정되지 않았습니다.")
62 return None
64 now = self._market_clock.get_current_kst_time()
65 end_dt = now.strftime("%Y%m%d")
66 # 7일 전부터 오늘까지 조회 (명절 연휴 등을 감안)
67 start_dt = (now - timedelta(days=7)).strftime("%Y%m%d")
69 try:
70 # 1. 내부 클래스에 직접 접근하지 않고, 상위 래퍼(BrokerAPIWrapper)의 메서드를 활용
71 # 2. 파라미터를 키워드 인자(kwargs)로 명시하여 안전하게 전달
72 resp = await self._broker.inquire_daily_itemchartprice(
73 stock_code="005930",
74 start_date=start_dt,
75 end_date=end_dt,
76 fid_period_div_code="D",
77 _skip_cache=True # 캐시 우회: cache_wrapper가 get_latest_trading_date를 호출하므로 무한 재귀 방지
78 )
80 # API 호출이 실패한 경우 원인 로깅
81 if resp.rt_cd != "0":
82 self._logger.error(f"일봉 조회 API 실패: {resp.msg1} (코드: {resp.rt_cd})")
83 return None
85 if resp.data:
86 first_item = resp.data[0]
87 # 3. 응답 데이터가 딕셔너리인지, 객체(Dataclass)인지 판별하여 안전하게 추출
88 if isinstance(first_item, dict): 88 ↛ 92line 88 didn't jump to line 92 because the condition on line 88 was always true
89 return first_item.get("stck_bsop_date")
90 else:
91 # 객체 형태인 경우 getattr 사용
92 return getattr(first_item, "stck_bsop_date", None)
94 return None
96 except Exception as e:
97 self._logger.error(f"최근 영업일 일봉 조회 중 예외 발생: {e}", exc_info=True)
98 return None
100 # ==============================================================================
101 # 2. 휴장일 판별 및 미래 개장일 계산 (chk-holiday API 활용 신규 로직)
102 # ==============================================================================
103 async def _sync_calendar_if_needed(self, target_date: Optional[datetime] = None):
104 """특정 날짜가 속한 '월'의 달력 데이터가 캐시에 없다면 API를 호출해 동기화합니다."""
105 if target_date is None:
106 target_date = self._market_clock.get_current_kst_time()
108 target_month = target_date.strftime("%Y%m")
110 # 이미 해당 월 데이터를 가져왔고, 해당 날짜가 캐시에 있다면 스킵
111 if target_month in self._synced_months and target_date.strftime("%Y%m%d") in self._business_days_cache:
112 return
114 if not self._broker: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true
115 self._logger.error("Broker가 설정되지 않아 휴장일 API를 호출할 수 없습니다.")
116 return
118 t_start = self._pm.start_timer()
119 # 한투 '국내휴장일조회' API 호출
120 target_date_str = target_date.strftime("%Y%m%d")
121 holiday_data: ResCommonResponse = await self._broker.check_holiday(target_date_str)
123 if holiday_data and holiday_data.rt_cd == "0" and holiday_data.data and "output" in holiday_data.data:
124 for day_info in holiday_data.data["output"]:
125 date_str = day_info["bass_dt"]
126 # 영업일이면서 거래일이어야 개장일
127 is_open = (day_info["bzdy_yn"] == "Y" and day_info["tr_day_yn"] == "Y")
128 self._business_days_cache[date_str] = is_open
129 self._synced_months.add(target_month)
130 else:
131 self._logger.warning(f"휴장일 API 동기화 실패 ({target_month}): {holiday_data.msg1 if holiday_data else 'No response'}")
133 self._pm.log_timer(f"MarketCalendarService._sync_calendar_if_needed({target_date_str})", t_start)
135 async def is_business_day(self, date_str: str = None) -> bool:
136 """특정 날짜(YYYYMMDD)가 공휴일/휴장일이 아닌 영업일인지 확인합니다."""
137 if not date_str: 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true
138 date_str = self._market_clock.get_current_kst_time().strftime("%Y%m%d")
140 target_date = datetime.strptime(date_str, "%Y%m%d")
142 # [최적화 1] 주말(토, 일)은 무조건 휴장일이므로 캐시/API 확인 스킵
143 if target_date.weekday() >= 5: 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true
144 return False
146 await self._sync_calendar_if_needed(target_date)
148 return self._business_days_cache.get(date_str, False)
150 async def is_market_open_now(self) -> bool:
151 """현재 시점이 휴일이 아니며, 장 운영 시간(09:00~15:40) 이내인지 확인합니다."""
152 # 장 운영 시간이 아니면 달력(API/캐시)을 확인할 필요도 없이 바로 False 반환 (성능 최적화)
153 if not self._market_clock.is_market_operating_hours():
154 return False
155 return await self.is_business_day()
157 async def get_next_open_day(self, current_date_str: str = None) -> str:
158 """기준일의 '다음 영업일(YYYYMMDD)'을 반환합니다 (연휴 완벽 스킵)."""
159 if not current_date_str:
160 current_date_str = self._market_clock.get_current_kst_time().strftime("%Y%m%d")
162 check_dt = datetime.strptime(current_date_str, "%Y%m%d") + timedelta(days=1)
164 # 최대 15일 탐색 (긴 명절 연휴 커버)
165 for _ in range(15):
166 # [최적화 2] 주말이면 달력 동기화 검사를 스킵하고 다음 날로 이동
167 if check_dt.weekday() >= 5:
168 check_dt += timedelta(days=1)
169 continue
171 await self._sync_calendar_if_needed(check_dt)
172 check_str = check_dt.strftime("%Y%m%d")
174 if self._business_days_cache.get(check_str) is True:
175 return check_str
177 check_dt += timedelta(days=1)
179 return current_date_str
181 async def get_next_open_time(self) -> datetime:
182 """다음 장이 열리는 정확한 '시간(datetime)'을 반환합니다."""
183 now = self._market_clock.get_current_kst_time()
184 today_str = now.strftime("%Y%m%d")
186 # 오늘이 영업일인데 아직 장 시작 전(09:00 이전)이라면 오늘이 개장일임
187 if await self.is_business_day(today_str) and now < self._market_clock.get_market_open_time():
188 next_open_str = today_str
189 else:
190 # 이미 장이 끝났거나 장 중이라면, 혹은 휴일이라면 다음 영업일을 찾음
191 next_open_str = await self.get_next_open_day(today_str)
193 open_time_str = self._market_clock.market_open_time_str
194 open_hour, open_minute = map(int, open_time_str.split(":"))
195 next_open_date = datetime.strptime(next_open_str, "%Y%m%d")
197 return self._market_clock.market_timezone.localize(
198 datetime(next_open_date.year, next_open_date.month, next_open_date.day, open_hour, open_minute)
199 )
201 async def wait_until_next_open(self):
202 """다음 개장 시간까지 스케줄러를 비동기적으로 대기시킵니다."""
203 now = self._market_clock.get_current_kst_time()
204 next_open = await self.get_next_open_time()
206 seconds_left = max(0.0, (next_open - now).total_seconds())
207 if seconds_left > 0: 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true
208 self._logger.info(f"다음 개장시간({next_open.strftime('%Y-%m-%d %H:%M:%S')})까지 {seconds_left:.1f}초 대기합니다. 💤")
209 await asyncio.sleep(seconds_left)
211 async def get_latest_market_close_time(self) -> Optional[datetime]:
212 """
213 가장 최근에 장이 마감된 정확한 '시간(datetime)'을 반환합니다.
214 (예: 월요일 오전 10시라면 -> 지난주 금요일 15:40 반환)
215 """
216 now = self._market_clock.get_current_kst_time()
217 today_str = now.strftime("%Y%m%d")
219 # 1. 오늘이 영업일이고, 현재 시간이 이미 오늘 장 마감(15:40) 이후라면? -> 오늘 15:40
220 if await self.is_business_day(today_str) and now >= self._market_clock.get_market_close_time():
221 latest_close_str = today_str
222 else:
223 # 2. 휴장일이거나, 아직 오늘 장이 안 끝났다면(장전/장중) -> 과거로 거슬러 올라감
224 check_dt = now - timedelta(days=1)
226 # 최대 15일 전까지 거슬러 올라가며 가장 최근 영업일을 찾음
227 for _ in range(15):
228 check_str = check_dt.strftime("%Y%m%d")
229 if await self.is_business_day(check_str):
230 latest_close_str = check_str
231 break
232 check_dt -= timedelta(days=1)
233 else:
234 self._logger.error("최근 15일 내에 영업일이 없습니다. (시스템 오류 의심)")
235 return None
237 # 찾아낸 영업일 문자열(latest_close_str)과 MarketClock의 마감 시간(15:40)을 결합
238 close_time_str = self._market_clock.market_close_time_str
239 close_hour, close_minute = map(int, close_time_str.split(":"))
240 close_date = datetime.strptime(latest_close_str, "%Y%m%d")
242 return self._market_clock.market_timezone.localize(
243 datetime(close_date.year, close_date.month, close_date.day, close_hour, close_minute)
244 )