diff --git a/kis_trader/backtest/optuna_breakout.py b/kis_trader/backtest/optuna_breakout.py index 5fcafa5..b91e014 100644 --- a/kis_trader/backtest/optuna_breakout.py +++ b/kis_trader/backtest/optuna_breakout.py @@ -33,6 +33,7 @@ from kis_trader.backtest.tail_param_search import _results_dir_for_write from kis_trader.strategies.breakout import breakout_backtest_wants_tick_replay, breakout_entry_mode from kis_trader.engine.indicator_cache import attach_indicator_caches_to_params from kis_trader.backtest.breakout_tick_loader import load_breakout_ticks_by_code +from kis_trader.utils.env import get_env_bool logger = logging.getLogger("param_search_optuna") @@ -63,6 +64,7 @@ class BreakoutSearchContext: start_key: str end_key: str cache_holder: Dict[str, Any] = field(default_factory=dict) + shared_tick_store: Any = None # ws_ticks 공유메모리 핸들 (종료 시 unlink) def prepare_breakout_search_context( @@ -159,6 +161,25 @@ def prepare_breakout_search_context( finally: _tick_db.close() + # ── ws_ticks 공유메모리 (Optuna, opt-in) — dict→numpy 컬럼 shared_memory 로 RAM 절감 ── + # 끄려면 OPTUNA_PARAM_SEARCH_SHARED_TICKS=0. numpy/shm 미지원·빌드 실패 시 자동 폴백. + shared_tick_store = None + if get_env_bool("OPTUNA_PARAM_SEARCH_SHARED_TICKS", True) and ticks_by_code: + from kis_trader.backtest.shared_ticks import build_shared_ticks_view + _view, shared_tick_store = build_shared_ticks_view(ticks_by_code, enabled=True) + if shared_tick_store is not None: + import atexit as _atexit + _atexit.register(shared_tick_store.unlink) # 크래시 시 /dev/shm 누수 방지 + logger.info("📦 ws_ticks 공유메모리 ON (Optuna) — dict 사본 제거, RAM 절감") + ticks_by_code = _view + import gc as _gc + _gc.collect() + try: + import ctypes as _ctypes + _ctypes.CDLL("libc.so.6").malloc_trim(0) + except Exception: + pass + grid = grids[mode] _ob_axes = ("max_spread_pct", "min_bid_ask_ratio", "ask_wall_max_qty") _ob_sweeping = any(len(set(grid.get(k) or [])) > 1 for k in _ob_axes) @@ -238,6 +259,7 @@ def prepare_breakout_search_context( start_key=start_key, end_key=end_key, cache_holder=cache_holder, + shared_tick_store=shared_tick_store, ) @@ -309,7 +331,17 @@ def run_breakout_optuna( logger.info("🔬 Optuna BREAKOUT | study=%s | trials=%d", study_name, n_trials) t0 = time.time() - study.optimize(objective, n_trials=n_trials, n_jobs=n_jobs, show_progress_bar=show_progress) + try: + study.optimize(objective, n_trials=n_trials, n_jobs=n_jobs, show_progress_bar=show_progress) + finally: + # 탐색 종료(또는 예외) 시 공유메모리 즉시 해제 (atexit 는 크래시 대비 이중 안전장치). + _store = getattr(ctx, "shared_tick_store", None) + if _store is not None: + try: + _store.unlink() + except Exception: + pass + ctx.shared_tick_store = None elapsed = time.time() - t0 passing: List[Dict[str, Any]] = [] diff --git a/kis_trader/backtest/optuna_momentum.py b/kis_trader/backtest/optuna_momentum.py index c01c205..08dfce6 100644 --- a/kis_trader/backtest/optuna_momentum.py +++ b/kis_trader/backtest/optuna_momentum.py @@ -33,7 +33,7 @@ from kis_trader.backtest.param_search_momentum import ( from kis_trader.backtest.tail_param_search import _results_dir_for_write from kis_trader.engine import momentum_engine as me from kis_trader.engine.indicator_cache import attach_indicator_caches_to_params -from kis_trader.utils.env import get_env_float +from kis_trader.utils.env import get_env_bool, get_env_float logger = logging.getLogger("param_search_optuna") @@ -63,6 +63,7 @@ class MomentumSearchContext: start_key: str end_key: str cache_holder: Dict[str, Any] = field(default_factory=dict) + shared_tick_store: Any = None # ws_ticks 공유메모리 핸들 (종료 시 unlink) def prepare_momentum_search_context( @@ -209,6 +210,25 @@ def prepare_momentum_search_context( finally: _snap_db.close() + # ── ws_ticks 공유메모리 (Optuna, opt-in) — dict→numpy 컬럼 shared_memory 로 RAM 절감 ── + # 끄려면 OPTUNA_PARAM_SEARCH_SHARED_TICKS=0. numpy/shm 미지원·빌드 실패 시 자동 폴백. + shared_tick_store = None + if get_env_bool("OPTUNA_PARAM_SEARCH_SHARED_TICKS", True) and ticks_by_code: + from kis_trader.backtest.shared_ticks import build_shared_ticks_view + _view, shared_tick_store = build_shared_ticks_view(ticks_by_code, enabled=True) + if shared_tick_store is not None: + import atexit as _atexit + _atexit.register(shared_tick_store.unlink) # 크래시 시 /dev/shm 누수 방지 + logger.info("📦 ws_ticks 공유메모리 ON (Optuna) — dict 사본 제거, RAM 절감") + ticks_by_code = _view + import gc as _gc + _gc.collect() + try: + import ctypes as _ctypes + _ctypes.CDLL("libc.so.6").malloc_trim(0) + except Exception: + pass + cache_holder: Dict[str, Any] = {} attach_indicator_caches_to_params(cache_holder, codes_candles) @@ -234,6 +254,7 @@ def prepare_momentum_search_context( start_key=start_key, end_key=end_key, cache_holder=cache_holder, + shared_tick_store=shared_tick_store, ) @@ -322,7 +343,17 @@ def run_momentum_optuna( study_name, n_trials, sort_by, ) t0 = time.time() - study.optimize(objective, n_trials=n_trials, n_jobs=n_jobs, show_progress_bar=show_progress) + try: + study.optimize(objective, n_trials=n_trials, n_jobs=n_jobs, show_progress_bar=show_progress) + finally: + # 탐색 종료(또는 예외) 시 공유메모리 즉시 해제 (atexit 는 크래시 대비 이중 안전장치). + _store = getattr(ctx, "shared_tick_store", None) + if _store is not None: + try: + _store.unlink() + except Exception: + pass + ctx.shared_tick_store = None elapsed = time.time() - t0 passing: List[Dict[str, Any]] = [] diff --git a/kis_trader/backtest/param_search_optuna.py b/kis_trader/backtest/param_search_optuna.py index b0eff03..c2e8323 100644 --- a/kis_trader/backtest/param_search_optuna.py +++ b/kis_trader/backtest/param_search_optuna.py @@ -81,7 +81,7 @@ from kis_trader.backtest.tail_param_search import ( ) from kis_trader.engine import tail_engine as te from kis_trader.engine.indicator_cache import attach_indicator_caches_to_params -from kis_trader.utils.env import get_env_from_db, get_env_int +from kis_trader.utils.env import get_env_bool, get_env_from_db, get_env_int logging.basicConfig(level=logging.INFO, format="%(message)s") logger = logging.getLogger("param_search_optuna") @@ -130,6 +130,7 @@ class TailSearchContext: grid_keys: List[str] ob_filter_on: bool cache_holder: Dict[str, Any] = field(default_factory=dict) + shared_tick_store: Any = None # ws_ticks 공유메모리 핸들 (종료 시 unlink) def prepare_tail_search_context( @@ -267,6 +268,25 @@ def prepare_tail_search_context( elif tail_backtest_wants_tick_replay(_tick_probe): logger.warning("⚠️ ws_ticks 없음 — OHLC 폴백 (WS_TICK_SAVE_ENABLED 후 재탐색)") + # ── ws_ticks 공유메모리 (Optuna, opt-in) — dict→numpy 컬럼 shared_memory 로 RAM 절감 ── + # 끄려면 OPTUNA_PARAM_SEARCH_SHARED_TICKS=0. numpy/shm 미지원·빌드 실패 시 자동 폴백. + shared_tick_store = None + if get_env_bool("OPTUNA_PARAM_SEARCH_SHARED_TICKS", True) and ticks_by_code: + from kis_trader.backtest.shared_ticks import build_shared_ticks_view + _view, shared_tick_store = build_shared_ticks_view(ticks_by_code, enabled=True) + if shared_tick_store is not None: + import atexit as _atexit + _atexit.register(shared_tick_store.unlink) # 크래시 시 /dev/shm 누수 방지 + logger.info("📦 ws_ticks 공유메모리 ON (Optuna) — dict 사본 제거, RAM 절감") + ticks_by_code = _view + import gc as _gc + _gc.collect() + try: + import ctypes as _ctypes + _ctypes.CDLL("libc.so.6").malloc_trim(0) + except Exception: + pass + from kis_trader.backtest.tail_param_search import _tail_grids pre_grid = _tail_grids(mode) _ob_axes = ("max_spread_pct", "min_bid_ask_ratio") @@ -343,6 +363,7 @@ def prepare_tail_search_context( grid_keys=tail_grid_axis_keys(mode), ob_filter_on=ob_filter_on, cache_holder=cache_holder, + shared_tick_store=shared_tick_store, ) finally: db.close() @@ -424,12 +445,22 @@ def run_tail_optuna( study_name, n_trials, sampler_name, storage_url, n_jobs, ) t0 = time.time() - study.optimize( - objective, - n_trials=n_trials, - n_jobs=n_jobs, - show_progress_bar=show_progress, - ) + try: + study.optimize( + objective, + n_trials=n_trials, + n_jobs=n_jobs, + show_progress_bar=show_progress, + ) + finally: + # 탐색 종료(또는 예외) 시 공유메모리 즉시 해제 (atexit 는 크래시 대비 이중 안전장치). + _store = getattr(ctx, "shared_tick_store", None) + if _store is not None: + try: + _store.unlink() + except Exception: + pass + ctx.shared_tick_store = None elapsed = time.time() - t0 logger.info("✅ Optuna 완료 | %.1f초 | 완료 trial %d", elapsed, len(study.trials)) diff --git a/kis_trader/backtest/shared_ticks.py b/kis_trader/backtest/shared_ticks.py index c8a0225..8b48d28 100644 --- a/kis_trader/backtest/shared_ticks.py +++ b/kis_trader/backtest/shared_ticks.py @@ -346,9 +346,29 @@ class SharedTickStore: self._shms = {} def attach_mapping(self) -> SharedTicksMapping: - """같은 프로세스(부모)에서도 검증용으로 매핑 뷰를 얻는다.""" + """같은 프로세스(부모)에서도 검증용으로 매핑 뷰를 얻는다(이름으로 재-attach).""" return SharedTicksMapping(self._meta) + def local_mapping(self) -> SharedTicksMapping: + """같은 프로세스(Optuna 단일프로세스/스레드)용 — 부모가 이미 연 버퍼를 그대로 재사용. + + cross-process 용 ``attach_mapping`` 은 워커에서 ``SharedMemory(name=...)`` 로 다시 + attach 하지만, 같은 프로세스에서 그렇게 하면 resource_tracker 재등록/해제가 엇갈려 + 종료 시 무해하지만 시끄러운 ``KeyError`` 노이즈가 난다. 여기서는 새 attach 없이 + 생성자가 보관 중인 shm 버퍼로 numpy 배열을 직접 얹어(추가 attach 0회) 그 노이즈를 + 원천 제거한다. 배열을 미리 연결하므로 스레드(n_jobs>1) 초기화 레이스도 없다. + """ + m = SharedTicksMapping(self._meta) + total = int(self._meta["total"]) + src_dtype = self._meta["src_dtype"] + m._price = np.ndarray((total,), dtype=_DT_PRICE, buffer=self._shms["price"].buf) + m._volume = np.ndarray((total,), dtype=_DT_VOLUME, buffer=self._shms["volume"].buf) + m._tick_time = np.ndarray((total,), dtype=_DT_TICKTIME, buffer=self._shms["tick_time"].buf) + m._source = np.ndarray((total,), dtype=src_dtype, buffer=self._shms["source"].buf) + m._epoch = np.ndarray((total,), dtype=_DT_EPOCH, buffer=self._shms["epoch"].buf) + m._attached = True # 배열 이미 연결 → _ensure() no-op (재-attach 안 함) + return m + def build_shared_ticks( ticks_by_code: Optional[Dict[str, Dict[str, List[Dict[str, Any]]]]], @@ -446,9 +466,40 @@ def build_shared_ticks( return SharedTickStore(shms, meta) +def build_shared_ticks_view( + ticks_by_code: Optional[Dict[str, Dict[str, List[Dict[str, Any]]]]], + *, + enabled: bool = True, +) -> Tuple[Any, Optional["SharedTickStore"]]: + """단일 프로세스(Optuna 등)용 — dict 를 컬럼 공유메모리 '뷰' 로 치환해 RAM 절감. + + [Grid 와 차이] + Grid 파라서치는 워커 '프로세스' 간 사본 제거가 목적이라 descriptor 를 워커로 + 넘겨 attach 시킨다. Optuna 는 단일 프로세스(또는 n_jobs 스레드)라 사본 문제는 + 없지만, dict-of-dict(파이썬 객체 오버헤드 큼)을 numpy 컬럼(shared_memory)으로 + 바꾸면 **메모리 사용량**이 크게 준다 → RAM 이 적은 WSL 에서도 대용량 틱으로 + Optuna 를 돌릴 수 있다. + + 반환 ``(view, store)``: + · 성공: ``(SharedTicksMapping, SharedTickStore)`` — 종료 시 ``store.unlink()`` 필수 + · 미지원/빈데이터/off/실패: ``(원본 ticks_by_code, None)`` — 호출부가 dict 경로 그대로 사용 + + ``SharedTicksMapping`` 은 원본 dict 와 bit-identical(Grid E2E 검증) 이라 evaluate 무변경. + """ + if not enabled or not ticks_by_code or not shared_ticks_available(): + return ticks_by_code, None + store = build_shared_ticks(ticks_by_code) + if store is None: + return ticks_by_code, None + # 단일 프로세스용 로컬 매핑 — 부모 버퍼 재사용(추가 attach 0회) → 종료 noise 없음 + 스레드 안전. + view = store.local_mapping() + return view, store + + __all__ = [ "shared_ticks_available", "build_shared_ticks", + "build_shared_ticks_view", "SharedTickStore", "SharedTicksMapping", "SharedBucketMapping",