""" kis_trader/strategies/dbband_strategy.py — 더블 볼린저 라이브 전략 ================================================================ - **유니버스**: ``dbband_stock_config`` 등록 종목만 매수 스캔. - **파라미터**: 종목별 ``dbband_stock_config`` (없으면 env ``DBBAND_*`` 폴백). - **분봉(tf)**: 종목별 ``tf_min`` → WS/백테, 없으면 env ``DBBAND_TIMEFRAME``. - 신호·청산: ``dbband_engine`` 단일 소스. """ from __future__ import annotations import random from datetime import datetime as dt from typing import Dict, List, Optional try: from ..engine import dbband_engine as bbe except ImportError: bbe = None from ..utils.env import get_env_bool, get_env_int from .base import BaseStrategy from .dbband_stock_cfg import ( effective_dbband_tf_for_code, ensure_dbband_stock_config_table, fetch_latest_dbband_stock_config_by_code, load_dbband_engine_cfg, ) class DbBandStrategy(BaseStrategy): strategy_id = "DBBAND" loop_min_sleep = 1.5 loop_max_sleep = 2.5 def __init__(self, **kwargs): super().__init__(**kwargs) self.candle_tf = 15 self._env_tf_default = 15 self._dbband_cfg: Dict = {} self._holding_row_by_code: Dict[str, Dict] = {} self.min_price = 1000.0 self._fired_entry_key: Dict[str, str] = {} self.reload_config() def reload_config(self) -> None: if bbe is not None: try: self._dbband_cfg = bbe.get_dbband_defaults_from_db(self.db) except Exception as e: self.logger.debug("dbband defaults 조회 실패: %s", e) self._dbband_cfg = {} self.candle_tf = int(self._dbband_cfg.get("timeframe") or get_env_int("DBBAND_TIMEFRAME", 15)) self._env_tf_default = int(self.candle_tf) self.min_price = float(self._dbband_cfg.get("min_price") or 1000.0) self.slot_money = int(float(self._dbband_cfg.get("slot_money") or get_env_int("DBBAND_SLOT_MONEY", 3_000_000))) try: ensure_dbband_stock_config_table(self.db) self._holding_row_by_code = fetch_latest_dbband_stock_config_by_code(self.db) except Exception as e: self.logger.warning("dbband_stock_config 로드 실패: %s", e) self._holding_row_by_code = {} def _candidate_filter(self, candidate: Dict) -> bool: code = str(candidate.get("code") or "").strip() if not code: return False if self._holding_row_by_code: return code in self._holding_row_by_code return True def _merged_cfg(self, code: str) -> Dict: base = dict(self._dbband_cfg or {}) row = self._holding_row_by_code.get(code) if row: return dict(row) return load_dbband_engine_cfg(self.db, code, base) def _candle_tf_for_code(self, code: str) -> int: return effective_dbband_tf_for_code(self.db, code, self._env_tf_default) def check_buy(self, code: str, name: str) -> Optional[Dict]: if bbe is None: self.logger.warning("dbband_engine 미탑재 → 매수 스킵") return None if self._holding_row_by_code and code not in self._holding_row_by_code: return None try: if get_env_bool("FORCE_BUY_TEST", False): return self._force_buy_test(code, name) merged = self._merged_cfg(code) tf_code = self._candle_tf_for_code(code) min_len = max( int(merged.get("trend_ma_period") or 200) + 10, get_env_int("DBBAND_LIVE_MIN_CANDLES", 210), ) candles_raw = self.ws.get_candles(code, tf_code, n=min_len + 10) if len(candles_raw) < min_len: return None candles = [self._norm_candle(c) for c in candles_raw] today = dt.now().strftime("%Y%m%d") last_exit_dt = None if code in self.recently_sold: try: last_exit_dt = dt.fromtimestamp(self.recently_sold[code]) if last_exit_dt.strftime("%Y%m%d") != today: last_exit_dt = None except Exception: pass try: today_trades = self.db.get_trades_by_date(today) daily_cnt = len([ t for t in today_trades if t.get("code") == code and str(t.get("strategy", "")).startswith("DBBAND") ]) except Exception: daily_cnt = 0 state = {"last_exit_dt": last_exit_dt, "daily_cnt": daily_cnt} reject, msg, sig = bbe.check_buy_signal_live(candles, merged, state) if reject: self.logger.info("🔍 [DBBAND탈락] %s %s: %s — %s", name, code, reject, msg) return None if not sig: return None ent_key = str(sig.get("entry_bar_key") or "") if self._fired_entry_key.get(code) == ent_key: return None self._fired_entry_key[code] = ent_key entry_px = float(sig.get("entry_price") or 0) if entry_px <= 0: return None slot = int(float(merged.get("slot_money") or self.slot_money)) qty, rej = self._resolve_buy_qty_live(entry_px, invest_cap=float(slot)) if rej: self.logger.info( "🔍 [탈락-%s] %s %s: %s", rej, name, code, rej, ) return None return { "code": code, "name": name, "price": entry_px, "qty": qty, "stop_price": float(sig.get("stop_price") or 0), "target_price": float(sig.get("target_price") or 0), "side": str(sig.get("side") or "long"), "dbband_entry_bar_key": ent_key, "signal_candle_key": sig.get("signal_candle_key"), "entry_mode": sig.get("entry_mode"), } except Exception as e: self.logger.error("DBBAND check_buy 오류 %s: %s", code, e) return None def check_sell_signals(self) -> List[Dict]: if not self.holdings or bbe is None: return [] signals: List[Dict] = [] base_n = get_env_int("DBBAND_CANDLE_FETCH_N", 80) for code, holding in list(self.holdings.items()): try: name = holding.get("name", code) buy_price = float(holding.get("buy_price", 0)) qty = int(holding.get("qty", 0)) if qty <= 0 or buy_price <= 0: continue merged = self._merged_cfg(code) tf_code = self._candle_tf_for_code(code) candles_raw = self.ws.get_candles(code, tf_code, n=base_n) if not candles_raw: continue candles = [self._norm_candle(c) for c in candles_raw] last = candles[-1] position = { "entry_price": buy_price, "entry_time": str(holding.get("buy_time", ""))[:12], "stop": float(holding.get("stop_price") or buy_price * 0.98), "target": float(holding.get("target_price") or buy_price * 1.03), "max_price": float(holding.get("max_price") or buy_price), "min_price": float(holding.get("min_price") or buy_price), "qty": qty, "side": str(holding.get("side") or "long"), } closes = [float(c["close"]) for c in candles] period = int(merged.get("bb_period") or 20) inner_std = float(merged.get("bb_inner_std") or 2.0) _, i_upper, i_lower = bbe.compute_bb_series(closes, period, inner_std) sell = bbe.check_sell_signal_live( position, last, merged, inner_upper=i_upper[-1] if i_upper else None, inner_lower=i_lower[-1] if i_lower else None, is_eod=False, ) if sell: reason, exit_px = sell signals.append({ "code": code, "name": name, "qty": qty, "price": exit_px, "reason": reason, }) if position.get("max_price"): holding["max_price"] = position["max_price"] except Exception as e: self.logger.error("DBBAND 매도체크 %s: %s", code, e) return signals def run_loop_body(self) -> None: random.sleep(random.uniform(self.loop_min_sleep, self.loop_max_sleep)) super().run_loop_body()