"""
주식 거래소의 개장/폐장 시간을 확인하는 함수를 제공합니다.
"""
from cachetools.func import ttl_cache
from dataclasses import dataclass
from enum import Enum
from pyqqq.datatypes import Exchange
from pyqqq.utils.api_client import send_request, raise_for_status
from typing import Optional, Union
import datetime
import pyqqq.config as c
import requests
import pandas_market_calendars as mcal
KRX_AFTER_MARKET_START_DATE = datetime.date(2026, 9, 14)
""" KRX 애프터마켓 도입일. 이 날짜부터 KRX 운영 시간이 20:00 까지 연장됩니다. """
KRX_DEFAULT_OPEN_TIME = datetime.time(9, 0, 0)
""" KRX 기본 개장 시간 """
KRX_DEFAULT_CLOSE_TIME = datetime.time(15, 30, 0)
""" 애프터마켓 도입 이전의 KRX 기본 폐장 시간 """
KRX_AFTER_MARKET_CLOSE_TIME = datetime.time(20, 0, 0)
""" 애프터마켓 도입 이후의 KRX 기본 폐장 시간 """
NXT_PRE_MARKET_OPEN_TIME = datetime.time(8, 0, 0)
""" NXT 프리마켓 개장 시간 """
NXT_PRE_MARKET_CLOSE_TIME = datetime.time(8, 50, 0)
""" NXT 프리마켓 종료 시간 """
NXT_DEFAULT_OPEN_TIME = datetime.time(9, 0, 30)
""" NXT 기본 정규장 개장 시간 """
NXT_REGULAR_CLOSE_TIME = datetime.time(15, 20, 0)
""" NXT 기본 정규장 마감 시간 """
NXT_DEFAULT_CLOSE_TIME = datetime.time(20, 0, 0)
""" NXT 기본 폐장 시간 (애프터마켓 종료) """
SCHEDULE_REASON_HOLIDAY = "holiday"
""" 휴장 """
SCHEDULE_REASON_OPENING_DAY = "opening_day"
""" 개장일. 개장 시간만 1시간 연기되고 마감 시간은 평소와 같습니다. """
SCHEDULE_REASON_CSAT_DAY = "csat_day"
""" 수능일. 운영 시간이 1시간씩 순연됩니다. 애프터마켓은 개시만 연기되고 종료 시간은 같습니다. """
SCHEDULE_REASON_IRREGULAR = "irregular_schedule"
""" 사유가 구분되지 않은 조정. 스케줄의 개장/폐장 시간을 그대로 사용합니다. """
_DELAYED_OPEN_REASONS = (SCHEDULE_REASON_OPENING_DAY, SCHEDULE_REASON_CSAT_DAY)
""" 개장이 지연되는 조정 사유 """
_KRX_AFTER_MARKET_OPEN_DELAY = datetime.timedelta(minutes=30)
""" KRX 정규장 마감 이후 애프터마켓 개장까지의 간격 """
_NXT_REGULAR_DURATION = datetime.timedelta(hours=6, minutes=19, seconds=30)
""" NXT 정규장 운영 시간 (09:00:30 ~ 15:20) """
_NXT_AFTER_MARKET_OPEN_DELAY = datetime.timedelta(minutes=20)
""" NXT 정규장 마감 이후 애프터마켓 개장까지의 간격 """
[docs]
class MarketSessionType(Enum):
"""시장 세션 종류입니다."""
PRE_MARKET = "pre_market"
""" 프리마켓 (NXT) """
REGULAR = "regular"
""" 정규장. KRX는 종가 단일가까지 포함합니다. """
AFTER_MARKET = "after_market"
""" 애프터마켓. KRX는 2026년 9월 14일 부터 운영합니다. """
[docs]
@dataclass
class MarketSession:
"""
시장 세션의 운영 시간을 담고 있는 클래스입니다.
"""
type: MarketSessionType
""" 세션 종류 """
start_time: datetime.time
""" 세션 시작 시간 """
end_time: datetime.time
""" 세션 종료 시간 """
[docs]
def contains(self, time: datetime.time) -> bool:
"""주어진 시각이 세션 운영 시간에 포함되는지 확인합니다."""
return self.start_time <= time <= self.end_time
[docs]
@dataclass
class MarketSchedule:
"""
시장 운영 정보를 담고 있는 클래스입니다.
"""
full_day_closed: bool
""" 운영 여부 """
exchange: str = "KRX"
""" 시장 코드 """
open_time: Optional[datetime.time] = None
""" 개장 시간 """
close_time: Optional[datetime.time] = None
""" 폐장 시간 """
reason: Optional[str] = None
""" 장 운영이 중단되거나 시간이 조정된 이유 """
sessions: Optional[list[MarketSession]] = None
""" 세션별 운영 시간 목록. 휴장일이면 빈 목록입니다. """
[docs]
def __init__(self, **kwargs):
for k, v in kwargs.items():
if k == "_id":
continue
elif k in ["open_time", "close_time"] and type(v) is int:
date = datetime.datetime.fromtimestamp(int(v / 1000))
setattr(self, k, date.time())
else:
setattr(self, k, v)
[docs]
def get_session(self, session_type: MarketSessionType) -> Optional[MarketSession]:
"""
주어진 종류의 세션을 반환합니다.
Args:
session_type (MarketSessionType): 세션 종류
Returns:
Optional[MarketSession]: 해당 세션. 운영하지 않으면 None
"""
for session in self.sessions or []:
if session.type == session_type:
return session
return None
[docs]
def has_session(self, session_type: MarketSessionType) -> bool:
"""
주어진 종류의 세션을 운영하는지 확인합니다.
Args:
session_type (MarketSessionType): 세션 종류
Returns:
bool: 운영 여부
"""
return self.get_session(session_type) is not None
[docs]
def session_at(self, time: datetime.time) -> Optional[MarketSession]:
"""
주어진 시각에 운영 중인 세션을 반환합니다.
Args:
time (datetime.time): 시각
Returns:
Optional[MarketSession]: 운영 중인 세션. 없으면 None
"""
for session in self.sessions or []:
if session.contains(time):
return session
return None
[docs]
def is_full_day_closed(
now: Optional[Union[datetime.datetime, datetime.date]] = None,
exchange: Union[str, Exchange] = "KRX",
) -> bool:
"""
주식 거래소가 휴장일인지 확인합니다.
Args:
now (Union[datetime.datetime, datetime.date]): 현재 시각. 기본값: 현재 시각
exchange (Union[str, Exchange]): 거래소 이름. 기본값: KRX
Returns:
bool: 휴장일 여부
"""
if now is None:
now = datetime.datetime.now()
elif isinstance(now, datetime.date) and not isinstance(now, datetime.datetime):
now = datetime.datetime.combine(now, datetime.time.min)
exchange = _validate_exchange(exchange)
if exchange == Exchange.NXT and now.date() < datetime.date(2025, 3, 4):
raise ValueError("NXT 거래소는 2025년 3월 4일 부터 운영되었습니다. 이전 날짜는 지원하지 않습니다.")
return get_market_schedule(now.date(), exchange).full_day_closed
[docs]
def is_before_opening(
now: Optional[datetime.datetime] = None,
exchange: Union[str, Exchange] = "KRX",
) -> bool:
"""
주식 거래소가 아직 개장 전인지 확인합니다.
Args:
now (datetime.datetime): 현재 시각. 기본값: 현재 시각
exchange (Union[str, Exchange]): 거래소 이름. 기본값: KRX
Returns:
bool: 개장 전 여부
"""
if now is None:
now = datetime.datetime.now()
exchange = _validate_exchange(exchange)
if exchange == Exchange.NXT and now.date() < datetime.date(2025, 3, 4):
raise ValueError("NXT 거래소는 2025년 3월 4일 부터 운영되었습니다. 이전 날짜는 지원하지 않습니다.")
schedule = get_market_schedule(now.date(), exchange)
if schedule is None:
return None
return schedule.full_day_closed or now.time() < schedule.open_time
[docs]
def is_after_closing(
now: Optional[datetime.datetime] = None,
exchange: Union[str, Exchange] = "KRX",
) -> bool:
"""
주식 거래소가 이미 폐장 후인지 확인합니다.
Args:
now (datetime.datetime): 현재 시각. 기본값: 현재 시각
exchange (Union[str, Exchange]): 거래소 이름. 기본값: KRX
Returns:
bool: 폐장 후 여부
"""
if now is None:
now = datetime.datetime.now()
exchange = _validate_exchange(exchange)
if exchange == Exchange.NXT and now.date() < datetime.date(2025, 3, 4):
raise ValueError("NXT 거래소는 2025년 3월 4일 부터 운영되었습니다. 이전 날짜는 지원하지 않습니다.")
schedule = get_market_schedule(now.date(), exchange)
if schedule is None:
return None
return schedule.full_day_closed or now.time() > schedule.close_time
[docs]
def is_trading_time(
now: Optional[datetime.datetime] = None,
exchange: Union[str, Exchange] = "KRX",
) -> bool:
"""
주식 거래소가 거래 시간인지 확인합니다.
Args:
now (datetime.datetime): 현재 시각. 기본값: 현재 시각
exchange (Union[str, Exchange]): 거래소 이름. 기본값: KRX
Returns:
bool: 거래 시간 여부
"""
if now is None:
now = datetime.datetime.now()
exchange = _validate_exchange(exchange)
if exchange == Exchange.NXT and now.date() < datetime.date(2025, 3, 4):
raise ValueError("NXT 거래소는 2025년 3월 4일 부터 운영되었습니다. 이전 날짜는 지원하지 않습니다.")
schedule = get_market_schedule(now.date(), exchange)
if schedule.full_day_closed:
return False
if now.time() < schedule.open_time or now.time() > schedule.close_time:
return False
return True
[docs]
def get_market_schedule(
date: datetime.date,
exchange: Union[str, Exchange] = "KRX",
) -> MarketSchedule:
"""
주식 거래소의 개장/폐장 시간을 확인합니다.
2018년 1월 1일 이후의 데이터만 조회 가능합니다.
KRX는 2026년 9월 14일 부터 애프터마켓이 도입되어 폐장 시간이 20:00 으로 연장됩니다.
Args:
date (datetime.date): 날짜. datetime.datetime 이 전달되면 날짜 부분만 사용합니다.
exchange (Union[str, Exchange]): 거래소 이름. 기본값: KRX
Returns:
MarketSchedule: 거래소 개장/폐장 정보
"""
exchange = _validate_exchange(exchange)
if isinstance(date, datetime.datetime):
date = date.date()
if exchange == Exchange.NYSE:
return _get_nyse_schedule(date)
elif exchange == Exchange.NXT:
return _get_nxt_schedule(date)
else:
return _get_krx_schedule(date)
@ttl_cache(maxsize=10, ttl=3600)
def _get_nyse_schedule(date: datetime.date) -> MarketSchedule:
"""NYSE 시장 스케줄을 조회합니다."""
cal = mcal.get_calendar("NYSE")
schedules = cal.schedule(date, date, tz=cal.tz)
if schedules.empty:
return MarketSchedule(exchange="NYSE", full_day_closed=True, open_time=None, close_time=None, reason="holiday", sessions=[])
row = schedules.iloc[0]
open_time = row["market_open"].time()
close_time = row["market_close"].time()
sessions = [MarketSession(MarketSessionType.REGULAR, open_time, close_time)]
return MarketSchedule(exchange="NYSE", full_day_closed=False, open_time=open_time, close_time=close_time, sessions=sessions)
def _get_krx_schedule(date: datetime.date) -> MarketSchedule:
"""KRX 시장 스케줄을 조회합니다."""
regular_schedule = _get_krx_regular_schedule(date)
sessions = _build_krx_sessions(date, regular_schedule)
schedule = _apply_krx_after_market(regular_schedule, date)
schedule.sessions = sessions
return schedule
def _get_krx_regular_schedule(date: datetime.date) -> MarketSchedule:
"""애프터마켓을 제외한 KRX 정규장 스케줄을 조회합니다."""
full_day_closed = date.weekday() in [5, 6]
if not full_day_closed:
schedule = _fetch_market_scheldue(date, "KRX")
if schedule is not None:
return MarketSchedule(**schedule.json())
else:
open_time = KRX_DEFAULT_OPEN_TIME
close_time = KRX_DEFAULT_CLOSE_TIME
reason = None
else:
open_time = None
close_time = None
reason = "holiday"
return MarketSchedule(exchange="KRX", full_day_closed=full_day_closed, open_time=open_time, close_time=close_time, reason=reason)
def _shift_time(time: datetime.time, delta: datetime.timedelta) -> datetime.time:
"""시각에 시간 간격을 더합니다."""
return (datetime.datetime.combine(datetime.date.min, time) + delta).time()
def _build_krx_sessions(date: datetime.date, schedule: MarketSchedule) -> list[MarketSession]:
"""KRX 세션 목록을 만듭니다.
애프터마켓은 정규장 마감 30분 후에 개장해서 20:00 에 종료하며,
애프터마켓 도입 이전에는 정규장만 운영됩니다.
"""
if schedule.full_day_closed or schedule.open_time is None:
return []
regular_close_time = _get_krx_regular_close_time(schedule)
if regular_close_time is None:
return []
sessions = [MarketSession(MarketSessionType.REGULAR, schedule.open_time, regular_close_time)]
if date >= KRX_AFTER_MARKET_START_DATE:
open_time = _shift_time(regular_close_time, _KRX_AFTER_MARKET_OPEN_DELAY)
if open_time < KRX_AFTER_MARKET_CLOSE_TIME:
sessions.append(MarketSession(MarketSessionType.AFTER_MARKET, open_time, KRX_AFTER_MARKET_CLOSE_TIME))
return sessions
def _get_krx_regular_close_time(schedule: MarketSchedule) -> Optional[datetime.time]:
"""KRX 정규장 마감 시간을 조정 사유에 따라 산출합니다.
개장일은 개장 시간만 연기되므로 마감 시간이 평소와 같고,
수능일은 개장이 연기된 만큼 마감 시간도 순연됩니다.
"""
if schedule.reason == SCHEDULE_REASON_OPENING_DAY:
return KRX_DEFAULT_CLOSE_TIME
elif schedule.reason == SCHEDULE_REASON_CSAT_DAY:
return _shift_time(KRX_DEFAULT_CLOSE_TIME, _get_open_delay(schedule.open_time, KRX_DEFAULT_OPEN_TIME))
else:
return schedule.close_time
def _get_open_delay(open_time: datetime.time, default_open_time: datetime.time) -> datetime.timedelta:
"""기본 개장 시간 대비 얼마나 연기되었는지 계산합니다."""
base_date = datetime.date.min
return datetime.datetime.combine(base_date, open_time) - datetime.datetime.combine(base_date, default_open_time)
def _build_nxt_sessions(date: datetime.date, schedule: MarketSchedule) -> list[MarketSession]:
"""NXT 세션 목록을 만듭니다.
개장이 지연된 날에는 프리마켓을 운영하지 않으며, 스케줄의 개장 시간이 정규장 개장 시간이 됩니다.
개장일은 개장 시간만 연기되므로 정규장 마감 시간이 평소와 같고, 수능일은 마감 시간도 순연됩니다.
애프터마켓은 정규장 마감 20분 후에 개장해서 스케줄의 폐장 시간에 종료합니다.
"""
if schedule.full_day_closed or schedule.open_time is None:
return []
sessions = []
is_open_delayed = schedule.reason in _DELAYED_OPEN_REASONS or schedule.open_time > NXT_PRE_MARKET_OPEN_TIME
if is_open_delayed:
regular_open_time = schedule.open_time
else:
sessions.append(MarketSession(MarketSessionType.PRE_MARKET, NXT_PRE_MARKET_OPEN_TIME, NXT_PRE_MARKET_CLOSE_TIME))
regular_open_time = NXT_DEFAULT_OPEN_TIME
if schedule.reason == SCHEDULE_REASON_OPENING_DAY:
regular_close_time = NXT_REGULAR_CLOSE_TIME
else:
regular_close_time = _shift_time(regular_open_time, _NXT_REGULAR_DURATION)
sessions.append(MarketSession(MarketSessionType.REGULAR, regular_open_time, regular_close_time))
after_market_open_time = _shift_time(regular_close_time, _NXT_AFTER_MARKET_OPEN_DELAY)
after_market_close_time = schedule.close_time or NXT_DEFAULT_CLOSE_TIME
if after_market_open_time < after_market_close_time:
sessions.append(MarketSession(MarketSessionType.AFTER_MARKET, after_market_open_time, after_market_close_time))
return sessions
def _apply_krx_after_market(schedule: MarketSchedule, date: datetime.date) -> MarketSchedule:
"""애프터마켓 도입 이후의 스케줄이라면 폐장 시간을 20:00 까지 연장합니다.
이미 20:00 이후로 조정된 스케줄은 그대로 유지합니다.
"""
if date < KRX_AFTER_MARKET_START_DATE:
return schedule
if schedule.full_day_closed or schedule.close_time is None:
return schedule
if schedule.close_time < KRX_AFTER_MARKET_CLOSE_TIME:
schedule.close_time = KRX_AFTER_MARKET_CLOSE_TIME
return schedule
def _get_nxt_schedule(date: datetime.date) -> MarketSchedule:
"""NXT 시장 스케줄을 조회합니다."""
if date < datetime.date(2025, 3, 4):
raise ValueError("NXT 거래소는 2025년 3월 4일 부터 운영되었습니다. 이전 날짜는 지원하지 않습니다.")
schedule = _fetch_market_scheldue(date, "NXT")
if schedule is not None:
schedule = MarketSchedule(**schedule.json())
schedule.sessions = _build_nxt_sessions(date, schedule)
return schedule
schedule = _get_krx_schedule(date)
schedule.exchange = "NXT"
if not schedule.full_day_closed:
schedule = MarketSchedule(
exchange="NXT",
full_day_closed=False,
open_time=NXT_PRE_MARKET_OPEN_TIME,
close_time=NXT_DEFAULT_CLOSE_TIME,
reason=schedule.reason,
)
schedule.sessions = _build_nxt_sessions(date, schedule)
return schedule
else:
schedule.sessions = []
return schedule
[docs]
def get_last_trading_day(date: Optional[datetime.date] = None, exchange: Union[str, Exchange] = "KRX") -> datetime.date:
"""
주어진 날짜의 이전 거래일을 반환합니다.
Args:
date (datetime.date): 날짜. 기본값: 오늘
exchange (Union[str, Exchange]): 거래소 이름. 기본값: KRX
Returns:
datetime.date: 이전 거래일
"""
if date is None:
date = datetime.date.today()
exchange = _validate_exchange(exchange)
if exchange == Exchange.NYSE:
cal = mcal.get_calendar("NYSE")
start_date = date - datetime.timedelta(days=10)
end_date = date - datetime.timedelta(days=1)
schedules = cal.schedule(start_date, end_date)
return schedules.index[-1].to_pydatetime().date()
else:
while True:
date -= datetime.timedelta(days=1)
schedule = get_market_schedule(date, exchange)
if not schedule.full_day_closed:
return date
[docs]
def get_next_trading_day(date: datetime.date = None, exchange: Union[str, Exchange] = "KRX") -> datetime.date:
"""
주어진 날짜의 다음 거래일을 반환합니다.
Args:
date (datetime.date): 날짜. 기본값: 오늘
exchange (Union[str, Exchange]): 거래소 이름. 기본값: KRX
Returns:
datetime.date: 다음 거래일
"""
if date is None:
date = datetime.date.today()
exchange = _validate_exchange(exchange)
if exchange == Exchange.NXT and date < datetime.date(2025, 3, 4):
raise ValueError("NXT 거래소는 2025년 3월 4일 부터 운영되었습니다. 이전 날짜는 지원하지 않습니다.")
if exchange == Exchange.NYSE:
cal = mcal.get_calendar("NYSE")
start_date = date + datetime.timedelta(days=1)
end_date = date + datetime.timedelta(days=10)
schedules = cal.schedule(start_date, end_date)
return schedules.index[0].to_pydatetime().date()
else:
while True:
date += datetime.timedelta(days=1)
schedule = get_market_schedule(date, exchange)
if not schedule.full_day_closed:
return date
[docs]
def get_trading_day_with_offset(from_date: Optional[datetime.date] = None, offset_days: int = 0, exchange: Union[str, Exchange] = "KRX") -> datetime.date:
"""
주어진 날짜로부터 주어진 오프셋만큼의 거래일을 반환합니다.
Args:
from_date (datetime.date): 날짜. 기본값: 오늘
offset_days (int): 오프셋. 양수면 이후, 음수면 이전 거래일
exchange (Union[str, Exchange]): 거래소 이름. 기본값: KRX
Returns:
datetime.date: 거래일
"""
if from_date is None:
from_date = datetime.date.today()
exchange = _validate_exchange(exchange)
if exchange == Exchange.NXT and from_date < datetime.date(2025, 3, 4):
raise ValueError("NXT 거래소는 2025년 3월 4일 부터 운영되었습니다. 이전 날짜는 지원하지 않습니다.")
if exchange == Exchange.NYSE:
cal = mcal.get_calendar("NYSE")
if offset_days >= 0:
start_date = from_date
end_date = from_date + datetime.timedelta(days=max(10, offset_days * 3))
schedules = cal.schedule(start_date, end_date)
return schedules.index[offset_days].to_pydatetime().date()
else:
start_date = from_date + datetime.timedelta(days=-max(10, abs(offset_days) * 3))
end_date = from_date
schedules = cal.schedule(start_date, end_date)
return schedules.index[offset_days - 1].to_pydatetime().date()
else:
url = f"{c.PYQQQ_API_URL}/domestic-stock/market-schedules/KRX/trading-day"
params = {
"fromDate": from_date.strftime("%Y%m%d"),
"offset": offset_days,
}
r = send_request("GET", url, params=params)
raise_for_status(r)
result = r.json()
date = result["date"]
offset_date = datetime.datetime.strptime(date, "%Y%m%d").date()
if exchange == Exchange.NXT and offset_date < datetime.date(2025, 3, 4):
raise ValueError("NXT 거래소는 2025년 3월 4일 부터 운영되었습니다. 이전 날짜는 지원하지 않습니다.")
return offset_date
@ttl_cache(maxsize=1000, ttl=3600)
def _fetch_market_scheldue(date: datetime.date, exchange: str) -> requests.Response | None:
url = f"{c.PYQQQ_API_URL}/domestic-stock/market-schedules/{exchange}"
params = {"date": date}
r = send_request("GET", url, params=params)
if r.status_code == 404:
return None
else:
raise_for_status(r)
return r
def _validate_exchange(exchange: Union[str, Exchange]) -> Exchange:
if isinstance(exchange, str):
assert exchange in [e.value for e in Exchange], "지원하지 않는 거래소 코드입니다."
exchange = Exchange(exchange) # 안전하게 변환
else:
assert exchange in Exchange, "지원하지 않는 거래소 코드입니다."
return exchange