hoon/kis_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 한투 호가 2키 전용, LS RAM 합집합, 분봉 쓰레기면 다음 소스 통째

Browse Source 
메인 시세 41에 H0STASP0를 붙이는 폴백을 없앤다. MINIMAL과 무관하게
후보∪보유∪영구를 LS RAM에 붙인다. 그 분 틱이 없거나 봉끝 대비 늦으면
그 WS 봉을 버리고 2차·LS·REST 봉을 통째로 쓴다.

같은 파일에 직전 커밋 이후 쌓인 시세 폴백/ENV 키 정리도 포함된다.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Your Name
2026-08-19 22:09:03 +09:00
parent 6d2a706a48
commit d1dc274f0e
10 changed files with 2146 additions and 459 deletions
Download Patch File Download Diff File Expand all files Collapse all files

464
scripts/test_live_execution_validation.py
View File

@@ -2,7 +2,7 @@
"""
scripts/test_live_execution_validation.py
=========================================
실매 코어 스모크 (주문·REST 연타 없음).
실매 코어 스모크 (주문·REST 연타·실WS 연결 없음).
반드시 **DB에서 읽은 실제 ON/OFF** 를 먼저 찍고, 그 값에 맞춰 기대 동작을 검사한다.
가상 호가 주입은 「DB에서 필터 ON인 전략」에만 실매와 동일하게 적용한다.
@@ -12,10 +12,12 @@ scripts/test_live_execution_validation.py
· ws_candles source/channel 정규화 · 실매=Optuna 읽기쌍
· 갭보정 키움 REST만 (KIS REST 강제 무시)
· 영구구독 마스터 · LS DB 적재 가드 · 운영설정 UI 키
· WS close 전 구독해제(한투/LS/키움) · freeze ON · 호가 TTL 분리
· Optuna OHLC 폴백 강제 OFF · approval 공유 · SKIP_HTS DB 표기
실행:
실행 (중요 코드 수정 후 완료 보고 전 최소 1회):
python3 -u scripts/test_live_execution_validation.py
# 로그 예: logs/test_live_execution_validation_*.log
# 로그: logs/test_live_execution_validation_YYYYMMDD_HHMMSS.log
"""
from __future__ import annotations
@@ -23,8 +25,9 @@ import logging
import sys
import threading
import time
from datetime import datetime
from pathlib import Path
from typing import Dict, List
from typing import Any, List
from unittest.mock import MagicMock, patch
ROOT = Path(__file__).resolve().parents[1]
@@ -58,6 +61,53 @@ def _전략표시(sid: str) -> str:
return f"{전략한글.get(sid, sid)}[{sid}]"
def _소스필수(실패목록: List[str], obj: Any, needle: str, 라벨: str) -> None:
"""실소켓/REST 없이 소스 문자열만 검사."""
import inspect
try:
src = inspect.getsource(obj)
except (OSError, TypeError) as exc:
실패목록.append(f"{라벨} getsource 실패")
_실패(f"{라벨}: getsource 실패 ({exc})")
return
if needle not in src:
실패목록.append(f"{라벨} 누락")
_실패(f"{라벨}: `{needle[:72]}` 없음")
else:
_성공(라벨)
def _파일필수(실패목록: List[str], rel: str, needle: str, 라벨: str) -> None:
path = ROOT / rel
try:
text = path.read_text(encoding="utf-8")
except OSError as exc:
실패목록.append(f"{라벨} 파일없음")
_실패(f"{라벨}: {rel} 읽기 실패 ({exc})")
return
if needle not in text:
실패목록.append(f"{라벨} 누락")
_실패(f"{라벨}: `{needle[:72]}` 없음 ({rel})")
else:
_성공(라벨)
def _파일금지(실패목록: List[str], rel: str, needle: str, 라벨: str) -> None:
path = ROOT / rel
try:
text = path.read_text(encoding="utf-8")
except OSError as exc:
실패목록.append(f"{라벨} 파일없음")
_실패(f"{라벨}: {rel} 읽기 실패 ({exc})")
return
if needle in text:
실패목록.append(f"{라벨} 금지패턴")
_실패(f"{라벨}: `{needle[:72]}` 가 있으면 안 됨 ({rel})")
else:
_성공(라벨)
def _db원문(키: str) -> str:
from kis_trader.utils.env import get_env_from_db
@@ -407,7 +457,7 @@ def run_validation() -> bool:
허브2._get_ls_ws = MagicMock(return_value=None)
home2 = 허브2._subscribe_tick_prefer_or_spill("000660", "kiwoom")
라벨2 = 허브2.get_tick_feed_label("000660")
if home2 != "kis" or 라벨2 != "kis(spill)":
if home2 != "kis" or "kis" not in 라벨2 or "spill" not in 라벨2:
실패목록.append(f"키움→한투 spill 실패 {home2}/{라벨2}")
_실패(f"키움 한도 spill 실패: {home2}/{라벨2}")
else:
@@ -415,12 +465,66 @@ def run_validation() -> bool:
주문 = OrderManager(client=MagicMock(), db=MagicMock())
주문.tick_feed_label_provider = 허브2.get_tick_feed_label
주문.ob_feed_label_provider = getattr(허브2, "get_ob_feed_label", None)
mm줄 = 주문._tick_feed_mm_line("000660")
if "시세:" not in mm줄:
if "시세:" not in mm줄 or "호가:" not in mm줄:
실패목록.append(f"MM 시세줄 이상 {mm줄}")
_실패(f"매터모스트 시세 표기 이상: {mm줄}")
_실패(f"매터모스트 시세/호가 표기 이상: {mm줄}")
else:
_성공(f"체결 알림 시세 한 줄 예: {mm줄}")
_성공(f"체결 알림 시세·호가 한 줄 예: {mm줄}")
# 메인 2초 실패 → 2차 get_price (RAM 없음=실패, REST 없음)
허브_fb = WSManager(db=_가짜DB(), kis_client=MagicMock())
키움_stale = MagicMock()
키움_stale.get_price = MagicMock(return_value=None)
한투_ok = MagicMock()
한투_ok.get_price = MagicMock(return_value={"stck_prpr": "12345"})
허브_fb._kiwoom_ws = 키움_stale
허브_fb.ws_cache = 한투_ok
허브_fb._get_ls_ws = MagicMock(return_value=None)
px_fb = 허브_fb.get_price("005930")
lab_fb = 허브_fb.get_tick_feed_label("005930")
if not isinstance(px_fb, dict) or str(px_fb.get("stck_prpr") or "") != "12345":
실패목록.append(f"get_price 2차 폴백 실패 {px_fb}")
_실패(f"메인 None 인데 2차 현재가를 안 씀: {px_fb}")
elif "kis" not in str(lab_fb):
실패목록.append(f"2차 시세라벨 이상 {lab_fb}")
_실패(f"2차 폴백 라벨에 kis 없음: {lab_fb}")
else:
_성공(f"메인 실패→2차 시세 {px_fb.get('stck_prpr')} 라벨={lab_fb}")
허브_fb3 = WSManager(db=_가짜DB(), kis_client=MagicMock())
키움_n = MagicMock(); 키움_n.get_price = MagicMock(return_value=None)
한투_n = MagicMock(); 한투_n.get_price = MagicMock(return_value=None)
엘에스_ok = MagicMock(); 엘에스_ok.get_price = MagicMock(return_value={"stck_prpr": "777"})
허브_fb3._kiwoom_ws = 키움_n
허브_fb3.ws_cache = 한투_n
허브_fb3._get_ls_ws = MagicMock(return_value=엘에스_ok)
px3 = 허브_fb3.get_price("005930")
lab3 = 허브_fb3.get_tick_feed_label("005930")
if not isinstance(px3, dict) or str(px3.get("stck_prpr") or "") != "777":
실패목록.append(f"get_price 3차 폴백 실패 {px3}")
_실패(f"1·2차 None 인데 3차 LS 를 안 씀: {px3}")
elif "ls" not in str(lab3):
실패목록.append(f"3차 시세라벨 이상 {lab3}")
_실패(f"3차 폴백 라벨에 ls 없음: {lab3}")
else:
_성공(f"1·2차 실패→3차 LS {px3.get('stck_prpr')} 라벨={lab3}")
from kis_trader.engine.feed_fallback import merge_ticks_time_axis_fallback
merged_empty = merge_ticks_time_axis_fallback(
[
{"tick_time": "20260819120000", "source": "kiwoom", "price": 1, "_lag_sec": 9},
{"tick_time": "20260819120000", "source": "kis", "price": 2, "_lag_sec": 8},
],
main_src="kiwoom",
max_lag_sec=2.0,
)
if merged_empty:
실패목록.append(f"늦은 보조를 씀 {merged_empty}")
_실패("메인·보조 둘 다 2초 초과인데 틱을 남김")
else:
_성공("메인·보조 둘 다 늦으면 그 초 틱 없음(OHLC 메우지 않음)")
허브3 = WSManager(db=_가짜DB(), kis_client=MagicMock())
한투3 = MagicMock()
@@ -456,6 +560,12 @@ def run_validation() -> bool:
"PERM_LS_FILL_SLEEP_MIN",
"PERM_LS_FILL_SLEEP_MAX",
"WS_CANDLE_FREEZE_ON_CONFIRM",
"LIVE_FEED_FALLBACK_MAX_AGE_SEC",
"LS_FEED_FALLBACK_SUBSCRIBE",
"LS_GAP_FILL_CANDIDATES",
"LS_WS_TICK_SAVE",
"LS_WS_ALSO_HOGA",
"LS_WS_ORDERBOOK_SAVE",
):
if 키 not in ENV_CONFIG_KEYS:
실패목록.append(f"ENV 누락 {키}")
@@ -485,6 +595,13 @@ def run_validation() -> bool:
"PERMANENT_SUBSCRIBE_ENABLED",
"WS_GAP_FILL_KIS_FALLBACK",
"PERM_LS_FILL_BARS",
"LIVE_FEED_FALLBACK_MAX_AGE_SEC",
"LS_FEED_FALLBACK_SUBSCRIBE",
"LS_WS_TICK_SAVE",
"LS_WS_ALSO_HOGA",
"LS_WS_ORDERBOOK_SAVE",
"CANDLE_GARBAGE_FALLBACK",
"WS_ORDERBOOK_SAVE_KIS",
):
if need not in ui키들:
실패목록.append(f"UI 누락 {need}")
@@ -529,6 +646,9 @@ def run_validation() -> bool:
if pairs[0][1] != "ws" or ("kiwoom", "rest") not in pairs:
실패목록.append(f"읽기쌍 규칙 위반 {pairs}")
_실패(f"메인은 WS, 구멍은 kiwoom+rest 여야 함: {pairs}")
elif ("ls", "ws") not in pairs or len(pairs) < 5:
실패목록.append(f"읽기쌍 길이 {pairs}")
_실패(f"메인WS+2차WS+LS+rest+rollup 5쌍이어야 함: {pairs}")
else:
_성공("메인 channel=ws, 구멍에 kiwoom+rest 포함")
@@ -611,6 +731,136 @@ def run_validation() -> bool:
else:
_성공("영구구독→LS sync 가 마스터 OFF 시 빈 집합")
if not hasattr(WSManager, "_sync_feed_fallback_to_ls"):
실패목록.append("_sync_feed_fallback_to_ls 없음")
_실패("LS 3차 폴백 구독 owner=_feed_fallback 없음")
else:
fb_src = inspect.getsource(WSManager._sync_feed_fallback_to_ls)
if "LS_FEED_FALLBACK_SUBSCRIBE" not in fb_src:
실패목록.append("LS_FEED_FALLBACK_SUBSCRIBE 미적용")
_실패("_sync_feed_fallback_to_ls 가 LS_FEED_FALLBACK_SUBSCRIBE 를 안 봄")
else:
_성공("LS 3차 폴백 구독 _sync_feed_fallback_to_ls")
st_src = inspect.getsource(WSManager.sync_targets)
if "_sync_feed_fallback_to_ls" not in st_src or "_ls_ram_universe_codes" not in st_src:
실패목록.append("sync_targets LS RAM 합집합 없음")
_실패("sync_targets 가 _sync_feed_fallback_to_ls(_ls_ram_universe_codes) 를 안 부름")
else:
_성공("sync_targets 끝에서 LS RAM 합집합 구독 (MINIMAL OFF 경로)")
rec_src = inspect.getsource(WSManager._reconcile_split_subscriptions)
if "_sync_feed_fallback_to_ls" not in rec_src or "_ls_ram_universe_codes" not in rec_src:
실패목록.append("split reconcile LS RAM 합집합 없음")
_실패("_reconcile_split_subscriptions 가 LS RAM 합집합을 안 부름")
else:
_성공("split reconcile 도 LS RAM 합집합")
start_src = inspect.getsource(WSManager.start)
if "kis_ws_ob = self.ws_cache" in start_src:
실패목록.append("한투 호가 메인 폴백")
_실패("WSManager.start 가 kis_ws_ob=메인 폴백(ASP0 합산)을 함")
elif "kis_ws_ob = None" not in start_src:
실패목록.append("kis_ws_ob None 미설정")
_실패("2키 실패 시 kis_ws_ob=None 이 없음")
else:
_성공("한투 호가: 2키 전용, 메인 ASP0 폴백 없음")
try_ob = inspect.getsource(WSManager._try_subscribe_vendor_ob)
if "ob is self.ws_cache" not in try_ob:
실패목록.append("OB 구독 메인 거절 없음")
_실패("_try_subscribe_vendor_ob 가 메인=OB 를 거절하지 않음")
else:
_성공("KIS 호가 구독은 전용 세션만")
from unittest.mock import patch as _patch
from kis_trader.engine.feed_fallback import bar_is_garbage, live_bar_is_garbage
_ct = "202608191000"
with _patch(
"kis_trader.engine.feed_fallback.candle_garbage_fallback_enabled",
return_value=True,
), _patch(
"kis_trader.engine.feed_fallback.live_feed_fallback_max_age_sec",
return_value=3.0,
):
if not bar_is_garbage([], candle_time=_ct, tf_min=1, source="kiwoom"):
실패목록.append("쓰레기 0건 실패")
_실패("bar_is_garbage 틱 0건은 구멍이어야 함")
else:
_성공("분봉 쓰레기: 그 분 틱 0건 = 구멍")
stale_in = [{
"source": "kiwoom",
"tick_time": "20260819100000",
"tick_time_raw": "20260819100000",
}]
if not bar_is_garbage(stale_in, candle_time=_ct, tf_min=1, source="kiwoom"):
실패목록.append("쓰레기 stale 실패")
_실패("bar_is_garbage 전부 봉끝 대비 나이초과면 구멍이어야 함")
else:
_성공("분봉 쓰레기: 전부 봉끝 대비 나이초과 = 구멍")
fresh = [{
"source": "kiwoom",
"tick_time": "20260819100058",
"tick_time_raw": "20260819100058",
}]
if bar_is_garbage(fresh, candle_time=_ct, tf_min=1, source="kiwoom"):
실패목록.append("신선 봉을 쓰레기로")
_실패("봉끝 2초 전 틱은 유지여야 함")
else:
_성공("분봉 쓰레기: 봉끝 직전 틱이면 유지")
other = [{
"source": "kiwoom",
"tick_time": "20260819100000",
"tick_time_raw": "20260819100000",
}]
if live_bar_is_garbage(other, candle_time=_ct, tf_min=1, source="ls"):
실패목록.append("LS 봉 오판")
_실패("실매 링에 LS 틱 없으면 LS 봉을 버리면 안 됨")
else:
_성공("실매 링: 해당 소스 틱 없으면 봉 유지")
from kis_trader.ws.candle_series import pick_bar_by_read_pairs
kis_ok = [{
"source": "kis",
"tick_time": "20260819100058",
"tick_time_raw": "20260819100058",
}]
picked = pick_bar_by_read_pairs(
[
{"candle_time": _ct, "close": 1, "source": "kiwoom", "channel": "ws"},
{"candle_time": _ct, "close": 2, "source": "kis", "channel": "ws"},
],
(("kiwoom", "ws"), ("kis", "ws")),
ticks=stale_in + kis_ok, tf_min=1, missing_policy="hole", live_cover=False,
)
if not picked or int(picked.get("close") or 0) != 2:
실패목록.append("pick 2차 실패")
_실패(f"메인 쓰레기면 2차 봉 통째: {picked}")
else:
_성공("pick_bar: 메인 쓰레기 → 2차 봉 통째")
from kis_trader.ws.ls_ws import LSWebSocketPriceCache
ls_msg_src = inspect.getsource(LSWebSocketPriceCache._on_message)
if '_tick_recorder and get_env_bool("LS_WS_TICK_SAVE"' in ls_msg_src:
실패목록.append("LS 틱콜백이 TICK_SAVE에 묶임")
_실패("LS_WS_TICK_SAVE OFF 가 호가 틱동기 콜백까지 끔")
elif "_tick_recorder" not in ls_msg_src:
실패목록.append("LS _tick_recorder 없음")
_실패("LS _on_message 에 _tick_recorder 없음")
else:
_성공("LS 틱콜백은 TICK_SAVE와 분리(호가 적재 가능)")
gap_src = inspect.getsource(WSManager._enqueue_ls_gap_fill)
if "LS_GAP_FILL_CANDIDATES" not in gap_src:
실패목록.append("LS_GAP_FILL_CANDIDATES 가드 없음")
_실패("후보 LS 갭 REST 기본 OFF 가드 없음")
else:
_성공("후보 LS 갭 REST 는 LS_GAP_FILL_CANDIDATES 가드")
# 갭 집합에 perm 합집합 금지 (후보|보유만)
recon_src = inspect.getsource(WSManager._reconcile_split_subscriptions)
if "_gap_refill_codes = set(kis_want) | set(kw_want) | perm" in recon_src:
@@ -693,6 +943,161 @@ def run_validation() -> bool:
실패목록.append(f"스키마 검증 오류: {exc}")
_실패(f"ws_candles 스키마 검증 오류: {exc}")
# ── 12) WS close 전 구독해제 (실소켓 없음 · 소스만) ──────────────
print("\n[12단계] WS 종료·양보 전 구독해제 (한투 공문 정상 케이스)")
print(" 소켓만 close 금지. 실WS/approval REST 호출 없음.")
try:
from kis_trader.ws.kis_ws_overseas import KISOverseasWebSocket
from kis_trader.ws.kiwoom_ws import KiwoomWebSocketPriceCache
from kis_trader.ws.ls_ws import LSWebSocketPriceCache
_소스필수(
실패목록, KISWebSocketPriceCache.stop,
"_unsubscribe_all_before_close", "한투 국내 stop → 구독해제",
)
_소스필수(
실패목록, KISWebSocketPriceCache._force_close_socket,
"_unsubscribe_all_before_close", "한투 국내 장양보 → 구독해제",
)
_소스필수(
실패목록, KISOverseasWebSocket.stop,
"_unsubscribe_all_before_close", "한투 해외 stop → 구독해제",
)
_소스필수(
실패목록, KISOverseasWebSocket._force_close_socket,
"_unsubscribe_all_before_close", "한투 해외 장양보 → 구독해제",
)
_소스필수(
실패목록, LSWebSocketPriceCache.stop,
"_graceful_unreg_all", "LS stop → UNREG",
)
_소스필수(
실패목록, LSWebSocketPriceCache._force_close_socket,
"_graceful_unreg_all", "LS 장양보 → UNREG",
)
_소스필수(
실패목록, LSWebSocketPriceCache._send_raw,
"_force_close_socket", "LS send 실패/락타임아웃 → UNREG 후 close",
)
_소스필수(
실패목록, KiwoomWebSocketPriceCache.stop,
"_unsubscribe_all_before_close", "키움 stop → REMOVE",
)
_소스필수(
실패목록, WSManager.stop,
"kis_ws_ob.stop", "WSManager.stop → 호가전용 소켓도 stop",
)
_파일필수(
실패목록,
"kis_trader/network/kiwoom_condition_manager.py",
"CNSRCLR",
"키움 조건검색 stop → CNSRCLR",
)
except Exception as exc:
실패목록.append(f"구독해제 검증 오류: {exc}")
_실패(f"구독해제 검증 오류: {exc}")
import traceback
traceback.print_exc()
# ── 13) CRITICAL 잠금 (freeze / 호가TTL / OHLC / HTS skip / approval) ─
print("\n[13단계] CRITICAL 잠금 · 호가 TTL 분리 · Optuna OHLC 폴백 OFF")
try:
from kis_trader.engine.orderbook_filter import _fetch_snapshot
from kis_trader.ws.kis_ws import CandleAggregator
freeze_on = get_env_bool("WS_CANDLE_FREEZE_ON_CONFIRM", True)
print(
f" CANDLE_FREEZE={'ON' if freeze_on else 'OFF'} "
f"| DB {_db원문('WS_CANDLE_FREEZE_ON_CONFIRM')} "
f"(코드기본=True, OFF는 사용자 명시 승인 전 금지)"
)
if not freeze_on:
실패목록.append("WS_CANDLE_FREEZE_ON_CONFIRM=OFF")
_실패("봉 freeze OFF — 사용자 승인 없이 금지")
else:
_성공("봉 freeze ON (확정봉 OHLCV 유지)")
_파일금지(
실패목록,
"kis_trader/ws/kis_ws.py",
'get_env_bool("WS_CANDLE_FREEZE_ON_CONFIRM", False)',
"kis_ws freeze 코드기본 True",
)
_소스필수(
실패목록, CandleAggregator.merge_confirmed_bars,
"WS_CANDLE_FREEZE_ON_CONFIRM",
"갭보정 병합 경로에 freeze 키 존재",
)
_소스필수(
실패목록, _fetch_snapshot,
"WS_ORDERBOOK_FILTER_MAX_AGE_SEC",
"호가필터 판정나이 = FILTER_MAX_AGE",
)
_소스필수(
실패목록, _fetch_snapshot,
"WS_ORDERBOOK_TICK_MAX_AGE_SEC",
"호가 저장TTL = TICK_MAX_AGE (판정과 분리)",
)
print(
f" 호가나이 DB: FILTER_MAX_AGE={_db원문('WS_ORDERBOOK_FILTER_MAX_AGE_SEC')} "
f"| TICK_MAX_AGE={_db원문('WS_ORDERBOOK_TICK_MAX_AGE_SEC')}"
)
_파일필수(
실패목록,
"kis_trader/backtest/param_search_optuna.py",
'base_params["backtest_tick_fallback_ohlc"] = False',
"Optuna OHLC 폴백 강제 OFF",
)
for sid in 전략목록:
키 = f"{sid}_SKIP_HTS_SCAN_DUPES"
if 키 not in ENV_CONFIG_KEYS:
실패목록.append(f"ENV 누락 {키}")
_실패(f"ENV_CONFIG_KEYS 에 {키} 없음")
else:
on = get_env_bool(키, False)
print(f" {_전략표시(sid)} {키}={'ON' if on else 'OFF'} | DB {_db원문(키)}")
if on:
_주의(f"{키}=true — 운영 기본은 false. 사용자 지시 없이 바꾸지 말 것")
else:
_성공(f"{키}=false (운영 기본 유지)")
_파일필수(
실패목록,
"kis_trader/strategies/base.py",
"TICK_SELL_ASYNC_PLACE",
"틱매도 place 비동기(수신스레드 REST 금지)",
)
_파일필수(
실패목록,
"kis_trader/strategies/base.py",
"_tick_sell_worker_loop",
"틱매도 워커",
)
_파일필수(
실패목록,
"kis_trader/strategies/base.py",
"_sell_inflight",
"틱·루프 종목당 매도 1장",
)
if "TICK_SELL_ASYNC_PLACE" not in ENV_CONFIG_KEYS:
실패목록.append("ENV 누락 TICK_SELL_ASYNC_PLACE")
_실패("ENV_CONFIG_KEYS 에 TICK_SELL_ASYNC_PLACE 없음")
else:
_성공("TICK_SELL_ASYNC_PLACE env 등록")
_소스필수(
실패목록, KISWebSocketPriceCache._get_approval_key,
"KISApprovalManager.instance",
"한투 approval_key 공유 캐시",
)
except Exception as exc:
실패목록.append(f"CRITICAL 잠금 검증 오류: {exc}")
_실패(f"CRITICAL 잠금 검증 오류: {exc}")
import traceback
traceback.print_exc()
# ── 최종 ───────────────────────────────────────────────────────
print("\n" + "=" * 78)
if 실패목록:
@@ -702,10 +1107,49 @@ def run_validation() -> bool:
print("=" * 78)
return False
print("최종: 통과 (위 DB ON/OFF 표기가 실매 기준과 일치하는지 함께 확인)")
print("👑 [최종 판정] 완결!")
print("=" * 78)
return True
class _Tee:
def __init__(self, *files: Any) -> None:
self.files = files
def write(self, s: str) -> None:
for f in self.files:
f.write(s)
f.flush()
def flush(self) -> None:
for f in self.files:
f.flush()
def _setup_run_log() -> tuple:
log_dir = ROOT / "logs"
log_dir.mkdir(exist_ok=True)
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
path = log_dir / f"test_live_execution_validation_{ts}.log"
fp = path.open("w", encoding="utf-8")
sys.stdout = _Tee(sys.__stdout__, fp) # type: ignore[assignment]
sys.stderr = _Tee(sys.__stderr__, fp) # type: ignore[assignment]
print(f"로그 파일: {path}")
return path, fp
if __name__ == "__main__":
ok = run_validation()
sys.exit(0 if ok else 1)
log_path, log_fp = _setup_run_log()
code = 1
try:
ok = run_validation()
print(f"로그 파일: {log_path}")
code = 0 if ok else 1
finally:
sys.stdout = sys.__stdout__
sys.stderr = sys.__stderr__
try:
log_fp.close()
except Exception:
pass
sys.exit(code)