gahusb
7ea1a21487
Atomic mv of root V1 assets (main_server.py + modules/ + data/ + tests/ + entry scripts + docs + logs) into signal_v1/ subdirectory. load_dotenv() updated to load web-ai/.env explicitly via Path. Adds web-ai/CLAUDE.md (workspace guide) and web-ai/start.bat (signal_v1 entry wrapper). Prepares for signal_v2/ Phase 2. Tests: signal_v1/tests/unit baseline preserved (no regression). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
214 lines
8.8 KiB
Python
214 lines
8.8 KiB
Python
"""
|
|
KRX (한국거래소) 시장 캘린더
|
|
장 운영: 평일 09:00~15:30 KST (공휴일 제외)
|
|
|
|
우선순위:
|
|
1. exchange_calendars 라이브러리 (pip install exchange-calendars) → 음력 자동 계산
|
|
2. 하드코딩 폴백 (2024~2026 공휴일 내장)
|
|
"""
|
|
import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
KST = ZoneInfo("Asia/Seoul")
|
|
MARKET_OPEN = datetime.time(9, 0)
|
|
MARKET_CLOSE = datetime.time(15, 30)
|
|
|
|
# ── KRX 공휴일 하드코딩 (exchange_calendars 미설치 시 폴백) ──────────────────
|
|
# 출처: KRX 공식 휴장일 공고 (2024~2026)
|
|
STATIC_HOLIDAYS: frozenset[datetime.date] = frozenset({
|
|
# 2024
|
|
datetime.date(2024, 1, 1), # 신정
|
|
datetime.date(2024, 2, 9), # 설날 연휴
|
|
datetime.date(2024, 2, 12), # 대체공휴일
|
|
datetime.date(2024, 3, 1), # 삼일절
|
|
datetime.date(2024, 4, 10), # 국회의원선거
|
|
datetime.date(2024, 5, 5), # 어린이날
|
|
datetime.date(2024, 5, 6), # 대체공휴일
|
|
datetime.date(2024, 5, 15), # 부처님오신날
|
|
datetime.date(2024, 6, 6), # 현충일
|
|
datetime.date(2024, 8, 15), # 광복절
|
|
datetime.date(2024, 9, 16), # 추석 연휴
|
|
datetime.date(2024, 9, 17), # 추석
|
|
datetime.date(2024, 9, 18), # 추석 연휴
|
|
datetime.date(2024, 10, 3), # 개천절
|
|
datetime.date(2024, 10, 9), # 한글날
|
|
datetime.date(2024, 12, 25), # 성탄절
|
|
datetime.date(2024, 12, 31), # 연말 휴장
|
|
# 2025
|
|
datetime.date(2025, 1, 1), # 신정
|
|
datetime.date(2025, 1, 28), # 설날 연휴
|
|
datetime.date(2025, 1, 29), # 설날
|
|
datetime.date(2025, 1, 30), # 설날 연휴
|
|
datetime.date(2025, 3, 1), # 삼일절
|
|
datetime.date(2025, 3, 3), # 대체공휴일
|
|
datetime.date(2025, 5, 5), # 어린이날
|
|
datetime.date(2025, 5, 6), # 대체공휴일
|
|
datetime.date(2025, 6, 6), # 현충일
|
|
datetime.date(2025, 8, 15), # 광복절
|
|
datetime.date(2025, 10, 2), # 대체공휴일
|
|
datetime.date(2025, 10, 3), # 개천절
|
|
datetime.date(2025, 10, 6), # 추석 연휴
|
|
datetime.date(2025, 10, 7), # 추석
|
|
datetime.date(2025, 10, 8), # 추석 연휴
|
|
datetime.date(2025, 10, 9), # 한글날
|
|
datetime.date(2025, 12, 25), # 성탄절
|
|
datetime.date(2025, 12, 31), # 연말 휴장
|
|
# 2026
|
|
datetime.date(2026, 1, 1), # 신정
|
|
datetime.date(2026, 2, 16), # 설날 연휴
|
|
datetime.date(2026, 2, 17), # 설날
|
|
datetime.date(2026, 2, 18), # 설날 연휴
|
|
datetime.date(2026, 3, 1), # 삼일절
|
|
datetime.date(2026, 3, 2), # 대체공휴일
|
|
datetime.date(2026, 5, 5), # 어린이날
|
|
datetime.date(2026, 5, 24), # 부처님오신날
|
|
datetime.date(2026, 6, 6), # 현충일
|
|
datetime.date(2026, 8, 14), # 대체공휴일
|
|
datetime.date(2026, 8, 15), # 광복절
|
|
datetime.date(2026, 9, 24), # 추석 연휴
|
|
datetime.date(2026, 9, 25), # 추석
|
|
datetime.date(2026, 10, 3), # 개천절
|
|
datetime.date(2026, 10, 9), # 한글날
|
|
datetime.date(2026, 12, 25), # 성탄절
|
|
datetime.date(2026, 12, 31), # 연말 휴장
|
|
})
|
|
|
|
|
|
class KRXCalendar:
|
|
"""
|
|
KRX 시장 캘린더
|
|
|
|
>>> cal = KRXCalendar()
|
|
>>> cal.is_trading_day(datetime.date(2026, 1, 1)) # 신정
|
|
False
|
|
>>> cal.is_trading_day(datetime.date(2026, 1, 2)) # 평일
|
|
True
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._ec_cal = None
|
|
try:
|
|
import exchange_calendars as ec
|
|
self._ec_cal = ec.get_calendar("XKRX")
|
|
print("[KRXCalendar] exchange_calendars 로드 성공 (정확한 음력 공휴일 사용)")
|
|
except ImportError:
|
|
print("[KRXCalendar] exchange_calendars 미설치 → 하드코딩 폴백 (pip install exchange-calendars 권장)")
|
|
except Exception as e:
|
|
print(f"[KRXCalendar] exchange_calendars 로드 실패: {e} → 폴백 사용")
|
|
|
|
# ── 날짜 판별 ──────────────────────────────────────────────────────────────
|
|
|
|
def is_trading_day(self, date: datetime.date | None = None) -> bool:
|
|
"""주어진 날짜가 KRX 거래일인지 확인 (기본: 오늘 KST)"""
|
|
if date is None:
|
|
date = datetime.datetime.now(KST).date()
|
|
if date.weekday() >= 5: # 토(5), 일(6)
|
|
return False
|
|
if self._ec_cal:
|
|
try:
|
|
return self._ec_cal.is_session(date.isoformat())
|
|
except Exception:
|
|
pass
|
|
return date not in STATIC_HOLIDAYS
|
|
|
|
def now_kst(self) -> datetime.datetime:
|
|
"""현재 KST 시각"""
|
|
return datetime.datetime.now(KST)
|
|
|
|
def is_market_open(self) -> bool:
|
|
"""현재 KST 기준 장 중 여부 (09:00 ≤ time < 15:30)"""
|
|
now = self.now_kst()
|
|
if not self.is_trading_day(now.date()):
|
|
return False
|
|
return MARKET_OPEN <= now.time() < MARKET_CLOSE
|
|
|
|
def is_pre_market(self) -> bool:
|
|
"""장 시작 전 (당일 거래일이고 09:00 이전)"""
|
|
now = self.now_kst()
|
|
return self.is_trading_day(now.date()) and now.time() < MARKET_OPEN
|
|
|
|
def is_post_market(self) -> bool:
|
|
"""장 마감 후 (당일 거래일이고 15:30 이후)"""
|
|
now = self.now_kst()
|
|
return self.is_trading_day(now.date()) and now.time() >= MARKET_CLOSE
|
|
|
|
# ── 다음 장 시각 계산 ──────────────────────────────────────────────────────
|
|
|
|
def next_trading_open(self) -> datetime.datetime:
|
|
"""
|
|
다음 장 시작 시각 (KST)
|
|
- 오늘이 거래일이고 아직 09:00 이전 → 오늘 09:00 반환
|
|
- 그 외 → 다음 거래일 09:00 반환
|
|
"""
|
|
now = self.now_kst()
|
|
date = now.date()
|
|
if self.is_trading_day(date) and now.time() < MARKET_OPEN:
|
|
return datetime.datetime.combine(date, MARKET_OPEN, tzinfo=KST)
|
|
# 다음 거래일 탐색 (최대 14일)
|
|
next_date = date + datetime.timedelta(days=1)
|
|
for _ in range(14):
|
|
if self.is_trading_day(next_date):
|
|
return datetime.datetime.combine(next_date, MARKET_OPEN, tzinfo=KST)
|
|
next_date += datetime.timedelta(days=1)
|
|
raise RuntimeError("14일 이내에 거래일을 찾지 못했습니다.")
|
|
|
|
def today_close(self) -> datetime.datetime | None:
|
|
"""오늘 장 종료 시각. 오늘이 거래일이 아니면 None."""
|
|
now = self.now_kst()
|
|
if not self.is_trading_day(now.date()):
|
|
return None
|
|
return datetime.datetime.combine(now.date(), MARKET_CLOSE, tzinfo=KST)
|
|
|
|
# ── 잔여 시간 계산 ──────────────────────────────────────────────────────────
|
|
|
|
def seconds_to_open(self) -> float:
|
|
"""장 시작까지 남은 초 (이미 장 중이거나 장 마감 후면 0)"""
|
|
if self.is_market_open():
|
|
return 0.0
|
|
try:
|
|
return max(0.0, (self.next_trading_open() - self.now_kst()).total_seconds())
|
|
except RuntimeError:
|
|
return 0.0
|
|
|
|
def seconds_to_close(self) -> float:
|
|
"""장 종료까지 남은 초 (장 외 시간이면 0)"""
|
|
now = self.now_kst()
|
|
if not self.is_trading_day(now.date()):
|
|
return 0.0
|
|
close_dt = datetime.datetime.combine(now.date(), MARKET_CLOSE, tzinfo=KST)
|
|
return max(0.0, (close_dt - now).total_seconds())
|
|
|
|
def minutes_to_close(self) -> float:
|
|
return self.seconds_to_close() / 60
|
|
|
|
# ── 상태 요약 ──────────────────────────────────────────────────────────────
|
|
|
|
def status_summary(self) -> str:
|
|
"""현재 시장 상태 요약 문자열 (로그/알림용)"""
|
|
now = self.now_kst()
|
|
today = now.date()
|
|
if not self.is_trading_day(today):
|
|
try:
|
|
nxt = self.next_trading_open()
|
|
return f"휴장 | 다음 거래일: {nxt.strftime('%m/%d(%a) %H:%M')}"
|
|
except Exception:
|
|
return "휴장"
|
|
if self.is_market_open():
|
|
mins = int(self.minutes_to_close())
|
|
return f"장 중 | 마감까지 {mins}분"
|
|
if now.time() < MARKET_OPEN:
|
|
secs = self.seconds_to_open()
|
|
return f"장 시작 전 | 개장까지 {int(secs / 60)}분"
|
|
return "장 마감"
|
|
|
|
|
|
# 싱글톤 (프로세스 내 공유)
|
|
_calendar: KRXCalendar | None = None
|
|
|
|
|
|
def get_calendar() -> KRXCalendar:
|
|
global _calendar
|
|
if _calendar is None:
|
|
_calendar = KRXCalendar()
|
|
return _calendar
|