Coverage for core / market_clock.py: 100%

1# core/market_clock.py 

2import time 

3from typing import Optional 

4import pytz 

5import logging 

6import asyncio 

7from datetime import datetime, timedelta, date, time as dt_time 

8 

9class MarketClock: 

10 """ 

11 주식 거래와 관련된 '시간(시계)'을 관리하는 클래스입니다. 

12 순수 시간대 계산 및 포맷 변환, KST 타임존 처리를 담당하며, 

13 공휴일 및 실제 영업일 판단은 MarketCalendarService에서 수행해야 합니다. 

14 """ 

15 

16 def __init__(self, market_open_time="09:00", market_close_time="15:40", timezone="Asia/Seoul", logger=None): 

17 self.market_open_time_str = market_open_time 

18 self.market_close_time_str = market_close_time 

19 self.timezone_name = timezone 

20 self.logger = logger if logger else logging.getLogger(__name__) 

21 

22 # [최적화 1] 시간 문자열을 한 번만 파싱하여 time 객체로 캐싱 

23 open_h, open_m = map(int, self.market_open_time_str.split(':')) 

24 self._open_time_obj = dt_time(open_h, open_m) 

25 close_h, close_m = map(int, self.market_close_time_str.split(':')) 

26 self._close_time_obj = dt_time(close_h, close_m) 

27 

28 try: 

29 self.market_timezone = pytz.timezone(self.timezone_name) 

30 except pytz.UnknownTimeZoneError: 

31 self.logger.error(f"알 수 없는 시간대: {self.timezone_name}. 'Asia/Seoul'로 기본 설정합니다.") 

32 self.timezone_name = "Asia/Seoul" 

33 self.market_timezone = pytz.timezone(self.timezone_name) 

34 

35 def get_current_kst_time(self): 

36 """현재 한국 시간(KST)을 timezone-aware datetime 객체로 반환합니다.""" 

37 return datetime.now(self.market_timezone) 

38 

39 def get_current_kst_date_str(self): 

40 """현재 KST 기준 날짜를 YYYYMMDD 포맷으로 반환합니다.""" 

41 return self.get_current_kst_time().strftime("%Y%m%d") 

42 

43 def is_market_operating_hours(self, now=None) -> bool: 

44 """ 

45 단순히 현재 '시간'이 시장 운영 시간(예: 09:00 ~ 15:40) 내에 있는지 확인합니다. 

46 (주의: 공휴일, 임시휴일 등 '영업일' 여부는 MarketCalendarService에서 판단해야 합니다.) 

47 """ 

48 now = now or self.get_current_kst_time() 

49 

50 # 주말(토, 일)은 기본적으로 1차 제외 

51 if now.weekday() >= 5: 

52 return False 

53 

54 # [최적화 2] 무거운 datetime 조합과 타임존 연산 없이 순수 시간(time) 객체만으로 비교 

55 return self._open_time_obj <= now.time() <= self._close_time_obj 

56 

57 def get_market_open_time(self, target_dt: Optional[datetime] = None) -> datetime: 

58 """오늘 날짜 또는 지정된 날짜 기준 시장 개장 시간(09:00) 반환""" 

59 now = target_dt or self.get_current_kst_time() 

60 return self.market_timezone.localize(datetime( 

61 now.year, now.month, now.day, 

62 hour=self._open_time_obj.hour, 

63 minute=self._open_time_obj.minute, 

64 second=0, microsecond=0 

65 )) 

66 

67 def get_market_close_time(self, target_dt: Optional[datetime] = None) -> datetime: 

68 """오늘 날짜 또는 지정된 날짜 기준 시장 마감 시간(15:40) 반환""" 

69 now = target_dt or self.get_current_kst_time() 

70 return self.market_timezone.localize(datetime( 

71 now.year, now.month, now.day, 

72 hour=self._close_time_obj.hour, 

73 minute=self._close_time_obj.minute, 

74 second=0, microsecond=0 

75 )) 

76 

77 def get_seconds_until_market_close(self, now=None) -> float: 

78 """ 

79 현재 시간 또는 지정된 시간부터 해당 날짜의 장 마감(15:40)까지 남은 초(seconds)를 계산합니다. 

80 (장 마감 후 계산 시 음수가 반환될 수 있습니다.) 

81 """ 

82 now = now or self.get_current_kst_time() 

83 close_time = self.get_market_close_time(target_dt=now) 

84 diff = (close_time - now).total_seconds() 

85 return diff 

86 

87 def get_sleep_seconds_until_market_close(self, now=None) -> float: 

88 """ 

89 현재 시간부터 오늘 장 마감(15:40)까지 대기해야 할 남은 초를 반환합니다. 

90 이미 마감 시간을 지났다면 0.0을 반환합니다. 

91 """ 

92 diff = self.get_seconds_until_market_close(now) 

93 return max(0.0, diff) 

94 

95 def sleep(self, seconds): 

96 """지정된 시간(초)만큼 프로그램을 일시 중지합니다 (동기).""" 

97 if seconds > 0: 

98 self.logger.info(f"{seconds:.2f}초 동안 대기합니다 (동기).") 

99 time.sleep(seconds) 

100 

101 async def async_sleep(self, seconds): 

102 """지정된 시간(초)만큼 비동기적으로 프로그램을 일시 중지합니다.""" 

103 if seconds > 0: 

104 self.logger.info(f"{seconds:.2f}초 동안 대기합니다 (비동기).") 

105 await asyncio.sleep(seconds) 

106 

107 def to_yyyymmdd(self, val) -> str: 

108 """여러 타입을 YYYYMMDD 문자열로 안전 변환""" 

109 if val is None: 

110 return self.get_current_kst_date_str() 

111 if isinstance(val, str): 

112 return val 

113 if isinstance(val, (datetime, date)): 

114 return val.strftime("%Y%m%d") 

115 if callable(val): 

116 return self.to_yyyymmdd(val()) 

117 return str(val) 

118 

119 def to_hhmmss(self, t: str | int) -> str: 

120 """ 

121 다양한 입력(YYYYMMDDHH, YYYYMMDDHHMM, HH, HHMM 등)을 안전하게 HHMMSS로 정규화. 

122 규칙: 

123 - 긴 포맷은 뒤 6자리만 취함 

124 - HH만 오면 HH0000, HHMM이면 HHMM00 

125 - 애매한 길이(1/3/5자)는 왼쪽 0 패딩 

126 """ 

127 if t is None: 

128 t = self.get_current_kst_time() 

129 

130 s = ''.join(ch for ch in str(t).strip() if ch.isdigit()) 

131 

132 if len(s) == 2: # HH 

133 return s + "0000" 

134 if len(s) == 4: # HHMM 

135 return s + "00" 

136 

137 if len(s) >= 6: 

138 return s[-6:] 

139 return s.rjust(6, "0") 

140 

141 def dec_minute(self, hhmmss: str, minutes: int = 1) -> str: 

142 """HHMMSS 포맷의 문자열 시간에서 특정 분(minute)을 뺀 시간을 반환합니다.""" 

143 hh = int(hhmmss[0:2]) 

144 mm = int(hhmmss[2:4]) 

145 ss = int(hhmmss[4:6]) 

146 dt = datetime(2000, 1, 1, hh, mm, ss) - timedelta(minutes=minutes) 

147 return dt.strftime("%H%M%S")