87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
"""
|
|
kis_trader/utils/env.py — DB(env_config) 우선 + os.environ 폴백 환경변수 헬퍼
|
|
===========================================================================
|
|
기존 kis_scalping_ver2 / kis_short_ver3 에 흩어져 있던 `get_env_*` 를 통합.
|
|
- DB(env_config 최신 row) → os.environ → default 순으로 조회.
|
|
- 하드코딩 금지 원칙에 맞춰 전 모듈에서 이 함수들만 사용하도록 한다.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger("kis_trader.env")
|
|
|
|
# 순환 import 방지용 레이지 TradeDB 참조
|
|
_db_instance = None
|
|
|
|
|
|
def _get_db():
|
|
"""TradeDB 싱글톤. 최초 호출 시 MariaDB 연결."""
|
|
global _db_instance
|
|
if _db_instance is None:
|
|
try:
|
|
from database import TradeDB # 프로젝트 루트의 기존 모듈
|
|
_db_instance = TradeDB()
|
|
except Exception as e:
|
|
logger.warning("TradeDB 초기화 실패(%s) → env_config 조회 불가 → os.environ 폴백", e)
|
|
_db_instance = False # 실패 기록 (None과 구분)
|
|
return _db_instance or None
|
|
|
|
|
|
def set_db(db_obj) -> None:
|
|
"""외부에서 이미 만든 TradeDB 인스턴스를 재사용할 때 주입."""
|
|
global _db_instance
|
|
_db_instance = db_obj
|
|
|
|
|
|
def _strip_comment(val: Any) -> Any:
|
|
"""DB 컬럼에 `#` 이후 주석이 들어 있을 때 제거. 기존 규칙 유지."""
|
|
if isinstance(val, str) and "#" in val:
|
|
return val.split("#", 1)[0].strip()
|
|
return val
|
|
|
|
|
|
def get_env_from_db(key: str, default: str = "") -> str:
|
|
"""env_config 최신 row → 키 값을 문자열로 반환. 없으면 os.environ → default."""
|
|
db = _get_db()
|
|
if db is not None:
|
|
try:
|
|
row = db.get_latest_env()
|
|
if row and row.get("snapshot"):
|
|
v = row["snapshot"].get(key)
|
|
v = _strip_comment(v)
|
|
if v not in (None, ""):
|
|
return str(v)
|
|
except Exception as e:
|
|
logger.debug("env_config 조회 실패 (%s): %s", key, e)
|
|
# os.environ 폴백 (기존 호환)
|
|
return os.environ.get(key, str(default))
|
|
|
|
|
|
def get_env_float(key: str, default: float) -> float:
|
|
raw = get_env_from_db(key, str(default))
|
|
try:
|
|
return float(raw) if raw != "" else float(default)
|
|
except (ValueError, TypeError):
|
|
return float(default)
|
|
|
|
|
|
def get_env_int(key: str, default: int) -> int:
|
|
raw = get_env_from_db(key, str(default))
|
|
try:
|
|
# "1.0" 같은 값이 들어와도 int 캐스팅되도록 float 경유
|
|
return int(float(raw)) if raw != "" else int(default)
|
|
except (ValueError, TypeError):
|
|
return int(default)
|
|
|
|
|
|
def get_env_bool(key: str, default: bool = False) -> bool:
|
|
raw = str(get_env_from_db(key, str(default))).strip().lower()
|
|
if raw in ("true", "1", "yes", "y", "on"):
|
|
return True
|
|
if raw in ("false", "0", "no", "n", "off", ""):
|
|
return False
|
|
return bool(default)
|