286 lines
13 KiB
Python
286 lines
13 KiB
Python
"""Tests for us2kr.market_time.MarketClock (DST, holidays, early close, weekends, KRX)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
from freezegun import freeze_time
|
|
|
|
from us2kr.market_time import MarketClock
|
|
|
|
ET = ZoneInfo("America/New_York")
|
|
KST = ZoneInfo("Asia/Seoul")
|
|
|
|
|
|
def kst(y: int, m: int, d: int, hh: int, mm: int = 0) -> datetime:
|
|
return datetime(y, m, d, hh, mm, tzinfo=KST)
|
|
|
|
|
|
def et(y: int, m: int, d: int, hh: int, mm: int = 0) -> datetime:
|
|
return datetime(y, m, d, hh, mm, tzinfo=ET)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def clock() -> MarketClock:
|
|
return MarketClock(run_offset_minutes=75, anchor=date(2026, 8, 25))
|
|
|
|
|
|
# --------------------------------------------------------------------------- close / briefing
|
|
|
|
|
|
def test_edt_day_close_and_briefing(clock: MarketClock) -> None:
|
|
session = date(2026, 6, 15)
|
|
close = clock.us_session_close(session)
|
|
assert close.tzinfo is not None and close.tzinfo.key == "America/New_York"
|
|
assert close == et(2026, 6, 15, 16, 0)
|
|
assert close.utcoffset() == timedelta(hours=-4) # EDT
|
|
assert close.astimezone(KST) == kst(2026, 6, 16, 5, 0)
|
|
assert clock.briefing_time_for(session) == kst(2026, 6, 16, 6, 15)
|
|
assert clock.briefing_time_for(session).tzinfo.key == "Asia/Seoul"
|
|
assert clock.krx_date_for(session) == date(2026, 6, 16)
|
|
|
|
|
|
def test_est_day_close_and_briefing(clock: MarketClock) -> None:
|
|
session = date(2026, 1, 15)
|
|
close = clock.us_session_close(session)
|
|
assert close == et(2026, 1, 15, 16, 0)
|
|
assert close.utcoffset() == timedelta(hours=-5) # EST
|
|
assert close.astimezone(KST) == kst(2026, 1, 16, 6, 0)
|
|
assert clock.briefing_time_for(session) == kst(2026, 1, 16, 7, 15)
|
|
assert clock.krx_date_for(session) == date(2026, 1, 16)
|
|
|
|
|
|
def test_custom_offset() -> None:
|
|
clock = MarketClock(run_offset_minutes=30, anchor=date(2026, 8, 25))
|
|
assert clock.briefing_time_for(date(2026, 6, 15)) == kst(2026, 6, 16, 5, 30)
|
|
|
|
|
|
def test_negative_offset_rejected() -> None:
|
|
with pytest.raises(ValueError):
|
|
MarketClock(run_offset_minutes=-1, anchor=date(2026, 8, 25))
|
|
|
|
|
|
# --------------------------------------------------------------------------- holidays / early close
|
|
|
|
|
|
@pytest.mark.parametrize("holiday", [date(2026, 7, 3), date(2026, 11, 26), date(2026, 1, 1)])
|
|
def test_us_holiday(clock: MarketClock, holiday: date) -> None:
|
|
assert clock.is_us_session(holiday) is False
|
|
with pytest.raises(ValueError):
|
|
clock.us_session_close(holiday)
|
|
with pytest.raises(ValueError):
|
|
clock.briefing_time_for(holiday)
|
|
|
|
|
|
def test_previous_session_skips_holiday_and_weekend(clock: MarketClock) -> None:
|
|
# 07-03 (Fri) observed Independence Day, 07-04/05 weekend -> Monday's previous session is Thu 07-02
|
|
assert clock.previous_us_session(date(2026, 7, 6)) == date(2026, 7, 2)
|
|
assert clock.previous_us_session(date(2026, 7, 3)) == date(2026, 7, 2)
|
|
# strictly before: a session date returns the one before it
|
|
assert clock.previous_us_session(date(2026, 7, 2)) == date(2026, 7, 1)
|
|
assert clock.next_us_session(date(2026, 7, 2)) == date(2026, 7, 6)
|
|
|
|
|
|
def test_early_close_day(clock: MarketClock) -> None:
|
|
session = date(2026, 11, 27) # day after Thanksgiving: 13:00 ET close
|
|
assert clock.is_us_session(session)
|
|
close = clock.us_session_close(session)
|
|
assert close == et(2026, 11, 27, 13, 0)
|
|
assert close.astimezone(KST) == kst(2026, 11, 28, 3, 0)
|
|
assert clock.briefing_time_for(session) == kst(2026, 11, 28, 4, 15)
|
|
assert clock.krx_date_for(session) == date(2026, 11, 28)
|
|
|
|
|
|
# --------------------------------------------------------------------------- weekend
|
|
|
|
|
|
def test_weekend_not_session(clock: MarketClock) -> None:
|
|
assert clock.is_us_session(date(2026, 8, 22)) is False # Saturday
|
|
assert clock.is_us_session(date(2026, 8, 23)) is False # Sunday
|
|
assert clock.is_krx_session(date(2026, 8, 22)) is False
|
|
assert clock.is_us_session(date(2026, 8, 21)) is True
|
|
# Sunday afternoon -> Friday is the latest completed session
|
|
assert clock.latest_completed_session(kst(2026, 8, 23, 15, 0)) == date(2026, 8, 21)
|
|
|
|
|
|
# --------------------------------------------------------------------------- DST transitions
|
|
|
|
|
|
def test_dst_spring_forward_2026_03_08(clock: MarketClock) -> None:
|
|
# Friday before the switch is still EST (close 06:00 KST -> briefing 07:15)
|
|
assert clock.briefing_time_for(date(2026, 3, 6)) == kst(2026, 3, 7, 7, 15)
|
|
# Monday after the switch is EDT (close 05:00 KST -> briefing 06:15)
|
|
assert clock.briefing_time_for(date(2026, 3, 9)) == kst(2026, 3, 10, 6, 15)
|
|
assert clock.is_us_session(date(2026, 3, 8)) is False
|
|
# asking on the switch day itself resolves to Monday's session
|
|
session, run_at = clock.next_briefing(kst(2026, 3, 8, 12, 0))
|
|
assert (session, run_at) == (date(2026, 3, 9), kst(2026, 3, 10, 6, 15))
|
|
|
|
|
|
def test_dst_fall_back_2026_11_01(clock: MarketClock) -> None:
|
|
assert clock.briefing_time_for(date(2026, 10, 30)) == kst(2026, 10, 31, 6, 15) # EDT
|
|
assert clock.briefing_time_for(date(2026, 11, 2)) == kst(2026, 11, 3, 7, 15) # EST
|
|
assert clock.is_us_session(date(2026, 11, 1)) is False
|
|
session, run_at = clock.next_briefing(kst(2026, 11, 1, 12, 0))
|
|
assert (session, run_at) == (date(2026, 11, 2), kst(2026, 11, 3, 7, 15))
|
|
|
|
|
|
# --------------------------------------------------------------------------- next_briefing
|
|
|
|
|
|
def test_next_briefing_across_weekend(clock: MarketClock) -> None:
|
|
# Friday 16:30 ET (= Sat 05:30 KST): Friday's briefing (Sat 06:15 KST = 17:15 ET) is still ahead
|
|
session, run_at = clock.next_briefing(et(2026, 8, 21, 16, 30))
|
|
assert session == date(2026, 8, 21)
|
|
assert run_at == kst(2026, 8, 22, 6, 15)
|
|
assert run_at.tzinfo.key == "Asia/Seoul"
|
|
# right after it fires, the next one is Monday's session -> Tuesday KST
|
|
session2, run_at2 = clock.next_briefing(run_at) # boundary: equal is NOT strictly after
|
|
assert (session2, run_at2) == (date(2026, 8, 24), kst(2026, 8, 25, 6, 15))
|
|
session3, run_at3 = clock.next_briefing(run_at + timedelta(minutes=1))
|
|
assert (session3, run_at3) == (date(2026, 8, 24), kst(2026, 8, 25, 6, 15))
|
|
|
|
|
|
def test_next_briefing_before_close_same_day(clock: MarketClock) -> None:
|
|
session, run_at = clock.next_briefing(et(2026, 8, 24, 10, 0))
|
|
assert (session, run_at) == (date(2026, 8, 24), kst(2026, 8, 25, 6, 15))
|
|
|
|
|
|
def test_next_briefing_over_holiday(clock: MarketClock) -> None:
|
|
# Thu 07-02 briefing fired (Fri 07-03 06:15 KST); 07-03 holiday + weekend -> Mon 07-06 session
|
|
session, run_at = clock.next_briefing(kst(2026, 7, 3, 7, 0))
|
|
assert (session, run_at) == (date(2026, 7, 6), kst(2026, 7, 7, 6, 15))
|
|
|
|
|
|
@freeze_time("2026-08-24 21:00:00") # UTC = 17:00 ET Monday = 06:00 KST Tuesday
|
|
def test_next_briefing_with_frozen_now(clock: MarketClock) -> None:
|
|
now = datetime.now(tz=KST)
|
|
assert now == kst(2026, 8, 25, 6, 0)
|
|
session, run_at = clock.next_briefing(now)
|
|
assert (session, run_at) == (date(2026, 8, 24), kst(2026, 8, 25, 6, 15))
|
|
|
|
|
|
# --------------------------------------------------------------------------- latest_completed_session
|
|
|
|
|
|
def test_latest_completed_session_around_close(clock: MarketClock) -> None:
|
|
assert clock.latest_completed_session(et(2026, 8, 24, 15, 59)) == date(2026, 8, 21)
|
|
assert clock.latest_completed_session(et(2026, 8, 24, 16, 0)) == date(2026, 8, 24) # inclusive
|
|
assert clock.latest_completed_session(et(2026, 8, 24, 16, 1)) == date(2026, 8, 24)
|
|
# early-close day: 13:30 ET is already after the 13:00 close
|
|
assert clock.latest_completed_session(et(2026, 11, 27, 13, 30)) == date(2026, 11, 27)
|
|
assert clock.latest_completed_session(et(2026, 11, 27, 12, 30)) == date(2026, 11, 25)
|
|
|
|
|
|
def test_latest_completed_session_accepts_any_tz(clock: MarketClock) -> None:
|
|
# 06:00 KST Tuesday == 17:00 ET Monday
|
|
assert clock.latest_completed_session(kst(2026, 8, 25, 6, 0)) == date(2026, 8, 24)
|
|
assert clock.latest_completed_session(kst(2026, 8, 25, 4, 0)) == date(2026, 8, 21)
|
|
|
|
|
|
def test_naive_datetime_rejected(clock: MarketClock) -> None:
|
|
naive = datetime(2026, 8, 25, 6, 0) # noqa: DTZ001 - deliberately naive
|
|
with pytest.raises(ValueError):
|
|
clock.latest_completed_session(naive)
|
|
with pytest.raises(ValueError):
|
|
clock.next_briefing(naive)
|
|
with pytest.raises(ValueError):
|
|
clock.catchup_candidate(naive, 3)
|
|
|
|
|
|
# --------------------------------------------------------------------------- catch-up
|
|
|
|
|
|
def test_catchup_candidate_inside_window(clock: MarketClock) -> None:
|
|
# Monday 08-24 briefing fires 06:15 KST Tuesday; daemon restarts at 07:00 -> catch up
|
|
assert clock.catchup_candidate(kst(2026, 8, 25, 7, 0), window_hours=3) == date(2026, 8, 24)
|
|
assert clock.catchup_candidate(kst(2026, 8, 25, 6, 15), window_hours=3) == date(2026, 8, 24) # inclusive
|
|
|
|
|
|
def test_catchup_candidate_outside_window(clock: MarketClock) -> None:
|
|
assert clock.catchup_candidate(kst(2026, 8, 25, 9, 16), window_hours=3) is None # 3h01m later
|
|
assert clock.catchup_candidate(kst(2026, 8, 25, 15, 0), window_hours=3) is None
|
|
# between close (05:00) and briefing (06:15): nothing was missed yet
|
|
assert clock.catchup_candidate(kst(2026, 8, 25, 5, 30), window_hours=3) is None
|
|
# weekend afternoon: Friday's briefing is far in the past
|
|
assert clock.catchup_candidate(kst(2026, 8, 23, 15, 0), window_hours=3) is None
|
|
|
|
|
|
def test_catchup_candidate_wide_window_reaches_previous_session(clock: MarketClock) -> None:
|
|
# 05:30 KST Tuesday: Monday's briefing not yet due, Friday's (Sat 06:15) is ~71.25h ago
|
|
assert clock.catchup_candidate(kst(2026, 8, 25, 5, 30), window_hours=72) == date(2026, 8, 21)
|
|
|
|
|
|
# --------------------------------------------------------------------------- KRX
|
|
|
|
|
|
def test_krx_chuseok_holiday(clock: MarketClock) -> None:
|
|
# 추석 2026: 09-24(목), 09-25(금) KRX closed while XNYS trades
|
|
assert clock.is_krx_session(date(2026, 9, 23)) is True
|
|
assert clock.is_krx_session(date(2026, 9, 24)) is False
|
|
assert clock.is_krx_session(date(2026, 9, 25)) is False
|
|
assert clock.is_us_session(date(2026, 9, 24)) is True
|
|
# briefing for the Wed 09-23 US session lands on the Korean holiday
|
|
assert clock.krx_date_for(date(2026, 9, 23)) == date(2026, 9, 24)
|
|
assert clock.is_krx_session(clock.krx_date_for(date(2026, 9, 23))) is False
|
|
# a normal day
|
|
assert clock.krx_date_for(date(2026, 8, 24)) == date(2026, 8, 25)
|
|
assert clock.is_krx_session(date(2026, 8, 25)) is True
|
|
|
|
|
|
def test_krx_only_holiday_does_not_affect_us_schedule(clock: MarketClock) -> None:
|
|
# 2026-10-09 한글날 (Fri): KRX closed, US open -> briefing still computed
|
|
assert clock.is_krx_session(date(2026, 10, 9)) is False
|
|
assert clock.is_us_session(date(2026, 10, 9)) is True
|
|
assert clock.briefing_time_for(date(2026, 10, 8)) == kst(2026, 10, 9, 6, 15)
|
|
|
|
|
|
def test_next_krx_session(clock: MarketClock) -> None:
|
|
assert clock.next_krx_session(date(2026, 8, 22)) == date(2026, 8, 24) # Sat -> Mon
|
|
assert clock.next_krx_session(date(2026, 8, 25)) == date(2026, 8, 25) # session day returns itself
|
|
assert clock.next_krx_session(date(2026, 9, 24)) == date(2026, 9, 28) # 추석 -> following Monday
|
|
assert clock.next_krx_session(date(2026, 10, 9)) == date(2026, 10, 12) # 한글날(Fri) -> Mon
|
|
assert clock.next_krx_session(date(2026, 12, 25)) == date(2026, 12, 28)
|
|
|
|
|
|
def test_krx_known_adhoc_closures(clock: MarketClock) -> None:
|
|
# exchange_calendars 4.13.2 marks 2026-06-03 (지방선거) as a session; KRX closes on election days.
|
|
assert clock.krx_date_for(date(2026, 6, 2)) == date(2026, 6, 3)
|
|
assert clock.is_krx_session(date(2026, 6, 3)) is False
|
|
assert clock.is_krx_session(date(2026, 6, 4)) is True
|
|
assert clock.next_krx_session(date(2026, 6, 3)) == date(2026, 6, 4)
|
|
assert clock.is_krx_session(date(2025, 10, 10)) is False # 임시공휴일
|
|
# past election days are already in the library data
|
|
assert clock.is_krx_session(date(2025, 6, 3)) is False
|
|
|
|
|
|
def test_krx_extra_holidays_override() -> None:
|
|
c = MarketClock(anchor=date(2026, 8, 25), extra_krx_holidays=[date(2026, 8, 20), date(2026, 8, 21)])
|
|
assert c.is_krx_session(date(2026, 8, 20)) is False
|
|
assert c.next_krx_session(date(2026, 8, 20)) == date(2026, 8, 24) # Thu/Fri closed -> Mon
|
|
assert MarketClock(anchor=date(2026, 8, 25)).is_krx_session(date(2026, 8, 20)) is True
|
|
# override days do not change the US-driven schedule
|
|
assert c.briefing_time_for(date(2026, 8, 19)) == kst(2026, 8, 20, 6, 15)
|
|
|
|
|
|
# --------------------------------------------------------------------------- calendar window
|
|
|
|
|
|
def test_out_of_window_date_triggers_rebuild() -> None:
|
|
narrow = MarketClock(anchor=date(2026, 8, 25), years_back=0, years_ahead=0)
|
|
assert narrow.is_us_session(date(2026, 6, 15)) is True
|
|
# 2028-01-03 is outside the 2026-only window: must rebuild instead of raising
|
|
assert narrow.is_us_session(date(2028, 1, 3)) is True
|
|
assert narrow.is_us_session(date(2028, 1, 1)) is False
|
|
assert narrow.us_session_close(date(2028, 1, 3)) == et(2028, 1, 3, 16, 0)
|
|
|
|
|
|
def test_default_anchor_is_today() -> None:
|
|
clock = MarketClock()
|
|
today = datetime.now(tz=KST).date()
|
|
# today's date is always resolvable without a rebuild
|
|
assert isinstance(clock.is_us_session(today), bool)
|
|
assert isinstance(clock.next_briefing(datetime.now(tz=KST))[1], datetime)
|