""" 설정값 영속 저장. SQLite key-value 테이블 한 개로 단순화. - 텔레그램 토큰 / chat_id - 모니터링 심볼 - 알림 시간축 목록 - 쿨다운 / STOP_LOSS_PCT 등 운영 파라미터 Streamlit rerun 안전: 모듈 최상단의 connection 캐시는 모듈 캐싱 (sys.modules) 으로 process lifetime 동안 보존. 멀티 스레드 (alert thread) 도 같은 DB 파일을 읽으므로 SQLite 의 thread-safe 모드 (`check_same_thread=False`) 로 연다. """ import os import sqlite3 import threading from typing import Any, Optional, List DB_PATH = os.environ.get("SETTINGS_DB_PATH", "/app/data/settings.db") _lock = threading.RLock() _conn: Optional[sqlite3.Connection] = None def _get_conn() -> sqlite3.Connection: global _conn if _conn is None: os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) _conn = sqlite3.connect(DB_PATH, check_same_thread=False, isolation_level=None) _conn.execute("PRAGMA journal_mode=WAL") # 글로벌 settings 테이블 (옛 호환 + 기본값 시드용) _conn.execute( """ CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT NOT NULL, updated_at TEXT NOT NULL DEFAULT (datetime('now')) ) """ ) # 사용자별 settings 테이블 _conn.execute( """ CREATE TABLE IF NOT EXISTS user_settings ( user_id INTEGER NOT NULL, key TEXT NOT NULL, value TEXT NOT NULL, updated_at TEXT NOT NULL DEFAULT (datetime('now')), PRIMARY KEY (user_id, key) ) """ ) _conn.execute("CREATE INDEX IF NOT EXISTS idx_us_user ON user_settings(user_id)") return _conn DEFAULTS = { # ── 알림 / 모니터링 ── "telegram_token": "", "telegram_chat_id": "", "alert_symbol": "BTCUSDT", "alert_timeframes": "5m,15m,30m,1h", "alert_cooldown_sec": "600", "stop_loss_pct": "0.0075", "alert_enabled": "1", "daily_report_enabled": "1", "polling_interval_sec": "30", # ── 신호 임계값 (RSI / body) ── "long_rsi_max": "75", "short_rsi_min": "25", "strong_long_rsi_max": "65", "strong_short_rsi_min": "35", "body_pct_min": "0.002", # 일반 신호 캔들 body 최소 (양/음봉) "reversal_body_pct": "0.003", # 추세 꺾임 body 최소 "reversal_vol_mult": "1.3", # 추세 꺾임 거래량 동반 배수 # ── 거래량 / 펀딩비 ── "vol_exhaustion_mult": "3.0", # exhaustion 판정 (vol > avg × N) "vol_net_mult": "2.0", # vol_long/short_signal: net > avg × N "oi_active_pct": "0.001", # OI 활성도 임계 (변동률 절대값) "fr_long_overheat": "0.005", # 롱 과열 (배너) "fr_short_caution": "-0.005", # 숏스퀴즈 경보 "fr_short_extreme": "-0.007", # 숏 주의 신호 임계 # ── 차트 / UI ── "candle_limit_desktop": "53", "candle_limit_mobile": "14", "forming_stable_polls": "2", # forming candle 안정성 — 연속 N polls True 요구 } def init_db_with_env_defaults(): """최초 기동 시 글로벌 settings 의 .env 값 시드 (옛 호환용).""" with _lock: conn = _get_conn() for k, default in DEFAULTS.items(): cur = conn.execute("SELECT value FROM settings WHERE key=?", (k,)) if cur.fetchone() is not None: continue seed = default env_map = { "telegram_token": "TELEGRAM_TOKEN", "telegram_chat_id": "TELEGRAM_CHAT_ID", } if k in env_map: seed = os.environ.get(env_map[k], default) or default conn.execute("INSERT INTO settings(key, value) VALUES (?, ?)", (k, seed),) def ensure_user_defaults(user_id: int): """사용자가 처음 로그인 / 생성될 때 호출. 글로벌 DEFAULTS 를 그 사용자 키로 복사.""" if not user_id: return with _lock: conn = _get_conn() for k, default in DEFAULTS.items(): cur = conn.execute("SELECT value FROM user_settings WHERE user_id=? AND key=?", (user_id, k)) if cur.fetchone() is None: conn.execute( "INSERT INTO user_settings(user_id, key, value) VALUES (?, ?, ?)", (user_id, k, default), ) def get(key: str, default: Any = None, user_id: Optional[int] = None) -> str: """user_id 가 있으면 그 사용자 값. 없으면 글로벌 (옛 호환). 사용자 값 없으면 DEFAULTS.""" with _lock: conn = _get_conn() if user_id: cur = conn.execute("SELECT value FROM user_settings WHERE user_id=? AND key=?", (user_id, key)) row = cur.fetchone() if row is not None: return row[0] cur = conn.execute("SELECT value FROM settings WHERE key=?", (key,)) row = cur.fetchone() if row is None: return DEFAULTS.get(key, default) if default is None else default return row[0] def get_int(key: str, default: int = 0, user_id: Optional[int] = None) -> int: try: return int(get(key, default, user_id)) except (TypeError, ValueError): return default def get_float(key: str, default: float = 0.0, user_id: Optional[int] = None) -> float: try: return float(get(key, default, user_id)) except (TypeError, ValueError): return default def get_bool(key: str, default: bool = False, user_id: Optional[int] = None) -> bool: v = get(key, "1" if default else "0", user_id) return str(v).strip().lower() in ("1", "true", "yes", "on") def get_list(key: str, default=None, sep: str = ",", user_id: Optional[int] = None) -> list: v = get(key, "", user_id) if not v: return list(default or []) return [s.strip() for s in v.split(sep) if s.strip()] def set_value(key: str, value: Any, user_id: Optional[int] = None): with _lock: conn = _get_conn() if user_id: conn.execute( """ INSERT INTO user_settings(user_id, key, value, updated_at) VALUES(?, ?, ?, datetime('now')) ON CONFLICT(user_id, key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at """, (user_id, key, str(value)), ) else: conn.execute( """ INSERT INTO settings(key, value, updated_at) VALUES(?, ?, datetime('now')) ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at """, (key, str(value)), ) def all_settings(user_id: Optional[int] = None) -> dict: """user_id 가 있으면 그 사용자의 설정 (없는 키는 글로벌 fallback + DEFAULTS). 없으면 글로벌 settings 만 반환 (옛 호환).""" with _lock: conn = _get_conn() if user_id: ensure_user_defaults(user_id) cur = conn.execute("SELECT key, value FROM user_settings WHERE user_id=? ORDER BY key", (user_id,)) return dict(cur.fetchall()) cur = conn.execute("SELECT key, value FROM settings ORDER BY key") return dict(cur.fetchall()) def list_user_ids_with_alerts_enabled() -> list: """모든 사용자 중 alert_enabled=1 인 user_id 목록. 알림 스레드용.""" with _lock: conn = _get_conn() cur = conn.execute( "SELECT user_id FROM user_settings WHERE key='alert_enabled' AND value='1'" ) return [r[0] for r in cur.fetchall()]