Files
kis_trader/database.py
Your Name 1d69c217e2 fix(정합성): 틱 lag wall-clock 정합 + 3벤더 DB 저장 스위치 통일
- feed_fallback.bar_is_garbage: 봉끝 기준 → 각 틱의 recv_ts wall-clock 기준으로 정정
  실매 RAM 3초컷과 동일 논리 → 유동성 낮은 종목 부당 스킵 해소
- candle_garbage_fallback_enabled: 기본 True 복원 (wall-clock 정정 후 안전)
- param_search_optuna·run_tail_backtest_cli: CANDLE_GARBAGE_FALLBACK·BACKTEST_USE_RUST
  강제 os.environ 세팅 제거 → DB env·CLI 플래그로만 관리 (UI 존중)
- WS_TICK_DB_SAVE_LAG_CUT_ENABLED 신설 (bool, 기본 false, 3벤더 공통)
  OFF=키움/KIS/LS 모든 틱 lag 무관 전부 저장 (벤더 통계·재현·백테 정합)
  ON=lag > LIVE_FEED_FALLBACK_MAX_AGE_SEC 이면 미저장 (미래 A안)
- KIWOOM_TICK_LIVE_MAX_LAG_SEC 완전 폐기 → 위 스위치로 통일
- kiwoom_ws: _skip_persist 로직 새 스위치로 교체
- kis_ws·ls_ws: _skip_persist_kis/_ls 신규 (벤더별 상이했던 정책 통일)
- docs/정합성.md §9 신설 (문제·결정·시나리오·향후 A안 전환법)
- docs/rust_engine_parity_port_plan.md (신규 설계)

실매 스모크: logs/test_live_execution_validation_20260906_191845.log
  → 최종: 통과 · 👑 완결
브라우저 검증: http://192.168.0.149:5050/#liveconfig → 새 스위치 노출, JS 오류 없음

영향: 실매(DB 저장 정책 통일, RAM 컷 변경 없음) + 백테/Optuna(실매 정합)

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-06 19:23:01 +09:00

5984 lines
273 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
트레이딩 봇 데이터베이스 관리 모듈
- MariaDB 기반 (pymysql) — SQLite 에서 이전
- 활성 트레이딩 관리 (active_trades)
- 매매 히스토리 관리 (trade_history)
MariaDB 접속 정보 우선순위:
1) 환경변수 DB_HOST / DB_PORT / DB_USER / DB_PASS / DB_NAME
2) 아래 DEFAULT 상수 (기본값)
"""
import json
import os
import datetime
import logging
import threading
from typing import Any, Dict, List, Optional, Tuple
try:
import pymysql
import pymysql.cursors
_PYMYSQL_AVAILABLE = True
except ImportError:
_PYMYSQL_AVAILABLE = False
logger = logging.getLogger("TradeDB")
# 인증/계좌/ID 전용 키 — env_auth_config 컬럼 테이블로 분리 저장
ENV_AUTH_KEYS = (
"KIS_APP_KEY_REAL", "KIS_APP_SECRET_REAL",
"KIS_APP_KEY_MOCK", "KIS_APP_SECRET_MOCK",
"KIS_APP_KEY_OB_REAL", "KIS_APP_SECRET_OB_REAL",
"KIS_APP_KEY_OB_MOCK", "KIS_APP_SECRET_OB_MOCK",
"KIS_ACCOUNT_NO_REAL", "KIS_ACCOUNT_CODE_REAL",
"KIS_ACCOUNT_NO_MOCK", "KIS_ACCOUNT_CODE_MOCK",
"KIS_APP_KEY", "KIS_APP_SECRET", "KIS_ACCOUNT_NO", "KIS_ACCOUNT_CODE",
"KIS_HTS_ID",
"KIWOOM_APP_KEY_REAL", "KIWOOM_APP_SECRET_REAL",
"KIWOOM_APP_KEY_MOCK", "KIWOOM_APP_SECRET_MOCK",
"KIWOOM_APP_KEY", "KIWOOM_APP_SECRET",
"LS_APP_KEY_REAL", "LS_APP_SECRET_REAL",
"LS_APP_KEY_MOCK", "LS_APP_SECRET_MOCK",
)
# 웹·알람이 아직 쓰는 구모듈은 legacy/ (remove/ 와 별도 — remove 삭제해도 유지).
try:
from kis_trader.utils.legacy_root import ensure_legacy_root as _ensure_legacy_root
_ensure_legacy_root()
except Exception:
pass
# ── MariaDB 기본 접속 정보 (환경변수로 재정의 가능) ─────────────────
_DB_HOST = os.environ.get("DB_HOST", "192.168.0.141")
_DB_PORT = int(os.environ.get("DB_PORT", "3306"))
_DB_USER = os.environ.get("DB_USER", "jae")
_DB_PASS = os.environ.get("DB_PASS", "1234")
_DB_NAME = os.environ.get("DB_NAME", "kis_quant_db")
# ══════════════════════════════════════════════════════════════════════
# SQLite 호환 래퍼 — 기존 conn.execute() / with conn: 패턴 유지
# ══════════════════════════════════════════════════════════════════════
class _MariaDBConn:
"""
pymysql 연결을 sqlite3 인터페이스처럼 감싸는 래퍼.
- conn.execute(sql, params) → DictCursor 반환 (row['col'] 접근)
- with conn: ... → 자동커밋(autocommit=True) 이므로 no-op
- conn.commit() → 명시적 커밋 (autocommit=True라 호환)
- ? 플레이스홀더 → %s 자동 변환
- SQLite-only DDL 키워드 자동 변환
"""
# SQL 텍스트 자동 변환 규칙 (SQLite → MySQL)
_REPLACE_PAIRS = [
# DDL 키워드
("INTEGER PRIMARY KEY AUTOINCREMENT", "INT NOT NULL AUTO_INCREMENT PRIMARY KEY"),
("INTEGER PRIMARY KEY", "INT NOT NULL PRIMARY KEY"),
("AUTOINCREMENT", "AUTO_INCREMENT"),
# DML — SQLite 전용 구문
("INSERT OR REPLACE INTO", "REPLACE INTO"),
("last_insert_rowid()", "LAST_INSERT_ID()"),
# ON CONFLICT 처리: 단순 패턴 제거 후 ON DUPLICATE KEY UPDATE 로 수동 변환
# (복잡한 케이스는 각 메서드에서 직접 처리)
]
def __init__(self):
self._lock = threading.Lock()
self._conn = None
self._connect()
def _connect(self):
"""pymysql 연결 (재연결 포함)."""
if not _PYMYSQL_AVAILABLE:
raise ImportError(
"pymysql 미설치. 설치: pip install pymysql\n"
"또는: pip install PyMySQL"
)
self._conn = pymysql.connect(
host=_DB_HOST, port=_DB_PORT,
user=_DB_USER, password=_DB_PASS,
database=_DB_NAME,
charset="utf8mb4",
autocommit=True,
cursorclass=pymysql.cursors.DictCursor,
connect_timeout=10,
read_timeout=30,
write_timeout=30,
)
logger.debug("✅ MariaDB 연결 완료 (%s:%s/%s)", _DB_HOST, _DB_PORT, _DB_NAME)
def _ensure_connected(self):
"""연결 끊김 시 자동 재접속."""
try:
self._conn.ping(reconnect=True)
except Exception:
try:
self._connect()
except Exception as e:
logger.error("❌ MariaDB 재접속 실패: %s", e)
raise
@staticmethod
def _translate_sql(sql: str) -> str:
"""SQLite 전용 SQL 구문을 MySQL 호환으로 변환."""
for old, new in _MariaDBConn._REPLACE_PAIRS:
sql = sql.replace(old, new)
# ? → %s (파라미터 플레이스홀더)
# 단, 이미 %s 가 있는 경우 중복 변환 방지
if "?" in sql:
sql = sql.replace("?", "%s")
return sql
def execute(self, sql: str, params=None):
"""
SQL 실행. sqlite3.Connection.execute() 와 동일한 인터페이스.
반환값: DictCursor (fetchone/fetchall/lastrowid 사용 가능)
"""
sql = self._translate_sql(sql)
with self._lock:
self._ensure_connected()
cur = self._conn.cursor()
cur.execute(sql, params or ())
return cur
def executemany_tx(self, sql: str, seq_of_params) -> int:
"""executemany 를 autocommit OFF 한 트랜잭션으로 실행. 반환=행수."""
sql = self._translate_sql(sql)
rows = list(seq_of_params or [])
if not rows:
return 0
with self._lock:
self._ensure_connected()
self._conn.autocommit(False)
try:
cur = self._conn.cursor()
cur.executemany(sql, rows)
self._conn.commit()
return int(cur.rowcount if cur.rowcount is not None else len(rows))
except Exception:
try:
self._conn.rollback()
except Exception:
pass
raise
finally:
try:
self._conn.autocommit(True)
except Exception:
pass
def execute_long(self, sql: str, params=None, timeout_sec: int = 600):
"""DDL 등 긴 쿼리 — 클라이언트 read/write timeout 만 잠시 올림."""
sql = self._translate_sql(sql)
with self._lock:
self._ensure_connected()
old_r = getattr(self._conn, "_read_timeout", 30)
old_w = getattr(self._conn, "_write_timeout", 30)
self._conn._read_timeout = int(timeout_sec)
self._conn._write_timeout = int(timeout_sec)
try:
cur = self._conn.cursor()
cur.execute(sql, params or ())
return cur
finally:
try:
self._conn._read_timeout = old_r
self._conn._write_timeout = old_w
except Exception:
pass
def __enter__(self):
"""with conn: 패턴 호환 (autocommit=True 이므로 실질적으로 no-op)."""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
pass
def commit(self):
"""명시적 커밋 (autocommit=True 환경에서 호환성 유지용)."""
try:
with self._lock:
self._conn.commit()
except Exception:
pass
@property
def row_factory(self):
return None # DictCursor 가 이미 dict 반환, 호환용 stub
@row_factory.setter
def row_factory(self, _):
pass # sqlite3 호환용 stub (무시)
def close(self):
try:
self._conn.close()
except Exception:
pass
# ── information_schema 기반 컬럼 목록 조회 (PRAGMA 대체) ──────
def get_columns(self, table_name: str) -> list:
"""
PRAGMA table_info() 대체.
반환: 컬럼명 문자열 리스트
"""
try:
cur = self.execute(
"SELECT COLUMN_NAME FROM information_schema.COLUMNS "
"WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s "
"ORDER BY ORDINAL_POSITION",
(_DB_NAME, table_name),
)
return [row["COLUMN_NAME"] for row in cur.fetchall()]
except Exception as e:
logger.debug("get_columns(%s) 실패: %s", table_name, e)
return []
# ML 학습용 진입 시점 피처 컬럼 (active_trades / trade_history 공통)
ML_ENTRY_FEATURE_COLUMNS = (
"rsi", "volume_ratio", "tail_length_pct", "ma5_gap_pct", "ma20_gap_pct",
"foreign_net_buy", "institution_net_buy", "market_hour",
)
# env_config 테이블 컬럼 (키 하나당 컬럼 하나, 추가/삭제 시 여기와 CREATE TABLE만 수정)
ENV_CONFIG_KEYS = (
"STOP_LOSS_PCT", "SHOULDER_CUT_PCT", "SHOULDER_MIN_HIGH_PCT", "SHOULDER_MIN_NET_PCT", "STOP_ATR_MULTIPLIER_TAIL", "TARGET_ATR_MULTIPLIER_TAIL",
"TAIL_ATR_SL_MIN_PCT", "TAIL_ATR_SL_MAX_PCT", "TAIL_ATR_TP_MIN_PCT", "TAIL_ATR_TP_MAX_PCT",
"MAX_POSITION_PCT", "USE_SLOT_CAP", "SLOT_CAP_PCT", "MAX_STOCKS",
"USE_KELLY", "RISK_PCT_PER_TRADE", "MIN_POSITION_AMOUNT",
"USE_RISK_CHECK", "DAILY_STOP_LOSS_PCT", "CONSECUTIVE_LOSS_LIMIT",
# 일일 익절 목표 — 전략별만 ({SID}_DAILY_PROFIT_* → config_*).
# 글로벌 DAILY_PROFIT_TARGET_ENABLED 폐기(헷갈림) — 금액/%/수동중단·알림·트레일 규칙만 마스터 유지.
"DAILY_PROFIT_TARGET_KRW",
"DAILY_PROFIT_TARGET_PCT",
"DAILY_PROFIT_TARGET_BUDGET_KRW",
"DAILY_PROFIT_HALT_NEW_BUYS",
"DAILY_PROFIT_NOTIFY_MM",
# 일일 누적손익 트레일 익절 (고점 대비 되돌림 시 신규매수 중단) — 마스터·총합 규칙
"DAILY_PROFIT_MODE",
"DAILY_PROFIT_TRAIL_TIERS",
"DAILY_PROFIT_TRAIL_DROP_PCT",
"DAILY_PROFIT_TRAIL_ARM_KRW",
"DAILY_PROFIT_TRAIL_ARM_PCT",
"USE_BAN_SYSTEM", "BAN_HOURS", "USE_STOCK_FILTER", "RSI_OVERHEAT_THRESHOLD",
"MIN_RECOVERY_RATIO", "MAX_RECOVERY_RATIO",
"USE_TWAP", "TWAP_MIN_SPLIT", "TWAP_MAX_SPLIT", "TWAP_MIN_DELAY", "TWAP_MAX_DELAY",
"USE_ML_SIGNAL", "ML_MIN_PROBABILITY", "USE_NEWS_ANALYSIS", "NEWS_ANALYSIS_HOUR", "NEWS_MAX_COUNT",
"USE_QUICK_PROFIT_PROTECTION", "HIGH_PRICE_CHASE_THRESHOLD", "MAX_DAILY_CHANGE_PCT",
"MA20_MAX_ABOVE_PCT", "VOLUME_AVG_MULTIPLIER",
# 스캘핑 TRIGGER 거래량배수 (param_search vol_mult / 웹 저장). overflow 말고 env_config 정식 키.
"VOL_MULTIPLIER",
"CANDLE_OPEN_PRICE_BUFFER",
"MIN_CANDLE_LEN_TAIL", "MIN_PRICE_TAIL", "TAIL_SCORE_BASE", "TAIL_SCORE_RATIO_MULT",
"TAIL_RATIO_MIN", "TAIL_PCT_MIN", # 꼬리 비율/하락률 최소 (AI·봇 공통 기준)
"TAIL_SKIP_HTS_SCAN_DUPES", "TAIL_USE_INTRADAY_DROP", "TAIL_USE_MA20_FILTER",
"TAIL_USE_RSI_FILTER", "TAIL_USE_DAILY_RANGE_FILTER", "TAIL_USE_HIGH_CHASE_FILTER",
"TAIL_BAR_CHG_MIN_PCT", "TAIL_BAR_CHG_MAX_PCT", "TAIL_RSI_THRESHOLD",
"TAIL_PATTERN_HAMMER", "TAIL_PATTERN_PIN", "TAIL_PATTERN_ENGULFING",
"TAIL_PATTERN_PIERCING", "TAIL_PATTERN_HARAMI", "TAIL_PATTERN_DOJI",
"TAIL_PATTERN_MORNING_STAR",
"TAIL_PIN_CLOSE_UPPER_RATIO", "TAIL_PIN_MAX_UPPER_TAIL_RATIO",
"TAIL_ENGULF_MIN_BODY_RATIO", "TAIL_PIERCING_PENETRATE_RATIO",
"TAIL_HARAMI_MAX_BODY_RATIO", "TAIL_DOJI_BODY_MAX_RATIO",
"TAIL_MORNING_STAR_BODY_MAX_RATIO",
# 꼬리잡기 TAIL_* 통일 키 (레거시 MIN_DROP_RATE·STOP_LOSS_PCT 등 대체)
"TAIL_MIN_DROP_RATE", "TAIL_MIN_RECOVERY_RATIO", "TAIL_MAX_RECOVERY_3M",
"TAIL_STOP_LOSS_PCT", "TAIL_TAKE_PROFIT_PCT",
"TAIL_SHOULDER_MIN_HIGH_PCT", "TAIL_SHOULDER_CUT_PCT",
"TAIL_RATCHET_TIERS", "TAIL_MAX_HOLD_BARS",
"TAIL_TRAIL_PCT", "TAIL_TRAIL_ARM_PCT",
"TAIL_BACKTEST_USE_TICK_DB", "TAIL_BACKTEST_TICK_FALLBACK_OHLC",
"TAIL_PARAM_SEARCH_MIN_START",
"TAIL_STOP_ATR_MULT", "TAIL_TARGET_ATR_MULT",
# 신호봉 거래량 폭증 필터 (param_search tail_vol_mult · 0=OFF)
"TAIL_VOL_MULT", "TAIL_VOL_WIN",
"TAIL_MIN_PRICE", "TAIL_MAX_DAILY_CHG", "TAIL_MA20_MAX_ABOVE_PCT",
"TAIL_MAX_LOSS_KRW", "TAIL_MIN_DROP_FOR_LOSS_CUT",
"TAIL_RSI_PERIOD", "TAIL_COOLDOWN_SEC", "TAIL_MAX_DAILY",
"TAIL_SYMBOL_DAILY_LOSS_LIMIT_KRW", "TAIL_SYMBOL_DAILY_LOSS_LIMIT_PCT",
"TAIL_REENTRY_MIN_EDGE_KRW", "TAIL_REENTRY_REQUIRE_NONNEG",
"TAIL_HIGH_CHASE_THR", "TAIL_TIME_START", "TAIL_TIME_END",
"TAIL_MAX_STOCKS", "TAIL_TOTAL_BUDGET_KRW", "TAIL_SLOT_MONEY",
"TAIL_ENTRY_MODE", "TAIL_LIMIT_ATR_MULT", "TAIL_LIMIT_ANCHOR",
"TAIL_LIMIT_VALID_BARS", "TAIL_LIMIT_FILL_SLIP_PCT",
"TAIL_MIN_INVEST_RATIO_OF_SLOT",
"TAIL_PARAM_SEARCH_ENTRY_MODE",
"OPTUNA_MODE_POOL",
"OPTUNA_MODE_TOP_N",
"OPTUNA_MODE_BAND_DECAY_IQR",
"OPTUNA_MODE_REFINE_PHASE2_TRIALS",
"OPTUNA_MODE_REFINE_BAND_EXPAND_IQR",
"TAIL_GRID_FAST_MAX_DAILY_CHG", "TAIL_GRID_COARSE_MAX_DAILY_CHG",
"TAIL_GRID_FAST_SYMBOL_LOSS_PCT", "TAIL_GRID_FAST_SYMBOL_LOSS_KRW",
"TAIL_GRID_FAST_REENTRY_MIN_EDGE",
"TAIL_GRID_FAST_LIMIT_ATR_MULT", "TAIL_GRID_COARSE_LIMIT_ATR_MULT",
# 더블 볼린저 DBBAND_* (config_dbband)
"DBBAND_BB_PERIOD", "DBBAND_BB_INNER_STD", "DBBAND_BB_OUTER_STD",
"DBBAND_TREND_MA_PERIOD", "DBBAND_USE_TREND_FILTER", "DBBAND_SIDE_MODE",
"DBBAND_ENTRY_VALID_BARS", "DBBAND_ENTRY_MODE",
"DBBAND_STOP_MODE", "DBBAND_STOP_BUFFER_PCT", "DBBAND_STOP_LOSS_PCT",
"DBBAND_TP_MODE", "DBBAND_TAKE_PROFIT_PCT", "DBBAND_RR_RATIO",
"DBBAND_EXIT_MODE",
"DBBAND_SHOULDER_MIN_HIGH_PCT", "DBBAND_SHOULDER_CUT_PCT",
"DBBAND_TRAIL_PCT", "DBBAND_TRAIL_ARM_PCT",
"DBBAND_TIME_START", "DBBAND_TIME_END", "DBBAND_COOLDOWN_SEC",
"DBBAND_MAX_DAILY", "DBBAND_MIN_PRICE",
"DBBAND_SLOT_MONEY", "DBBAND_MAX_STOCKS", "DBBAND_TOTAL_BUDGET_KRW",
"DBBAND_MAX_BUY_AMOUNT", "DBBAND_MAX_HOLD_BARS", "DBBAND_TIMEFRAME",
"DBBAND_MIN_INVEST_RATIO_OF_SLOT", "DBBAND_FORCE_EOD_EXIT",
"DBBAND_LIVE_MIN_CANDLES", "DBBAND_LIVE_SIGNAL_LOOKBACK_BARS",
"DBBAND_UNIVERSE_SOURCE", "DBBAND_CANDLE_FETCH_N",
"STRATEGY_DBBAND_ENABLED", "KIS_DBBAND_MM_CHANNEL",
"RANK_DBBAND_SORT", "RANK_DBBAND_LIMIT",
"CONDITION_DBBAND_NAME", "CONDITION_DBBAND_SEQ",
"DBBAND_GRID_BB_PERIOD0", "DBBAND_GRID_BB_PERIOD1",
"DBBAND_GRID_INNER_STD0", "DBBAND_GRID_INNER_STD1",
"DBBAND_GRID_OUTER_STD0", "DBBAND_GRID_OUTER_STD1",
"DBBAND_GRID_TREND_MA0", "DBBAND_GRID_TREND_MA1",
"DBBAND_GRID_SL0", "DBBAND_GRID_SL1",
"DBBAND_GRID_TP0", "DBBAND_GRID_TP1",
"DBBAND_GRID_RR0", "DBBAND_GRID_RR1",
"INTRADAY_INVESTOR_NET_BUY_THRESHOLD", "SIZE_CLASS_LARGE_MIN", "SIZE_CLASS_MID_MIN",
# 단타 스캔 후보 점수 (전부 env/DB)
"SCAN_INVESTOR_NET_STRONG", "SCAN_INVESTOR_SCORE_STRONG", "SCAN_INVESTOR_SCORE_WEAK",
"SCAN_INVESTOR_BONUS_STRONG", "SCAN_INVESTOR_BONUS_WEAK",
"SCAN_VOLUME_BONUS_MIN", "SCAN_VOLUME_BONUS_POINT",
"SCAN_EXEC_STRENGTH_HIGH", "SCAN_EXEC_STRENGTH_MID", "SCAN_EXEC_BONUS_HIGH", "SCAN_EXEC_BONUS_MID",
"SCAN_SCORE_DROP_WEIGHT", "SCAN_SCORE_RECOVERY_WEIGHT",
"MIN_SCORE_DISPLAY", "MM_TOP_N",
"SLOT_MONEY_DEFAULT", "SLOT_BASE_AMOUNT_CAP",
"SIZE_CLASS_SMALL_RATIO", "SIZE_CLASS_MID_RATIO",
# 개미털기/유니버스 필터 추가 키
"SCAN_MIN_PREV_DAY_PCT", "UPDATE_UNIVERSE_MIN_CANDIDATES",
"UPDATE_UNIVERSE_MIN_SCORE", "UPDATE_UNIVERSE_FALLBACK_TOP_N", "UPDATE_UNIVERSE_TOP_LOG", "UPDATE_UNIVERSE_TOP_N",
"SCAN_INTERVAL_SEC", # 유니버스 스캔 주기(초), kiwoom_universe_scanner 전용
"USE_RANDOM_SPLIT", "FORCE_MARKET_OPEN", "FORCE_BUY_TEST", "FORCE_BUY_TEST_CODE", "TOTAL_DEPOSIT",
# POP/LOCK·금액 손절 관련 추가 키
"ROUND_TRIP_COST_PCT", "POP_NET_PCT", "LOCK_NET_PCT", "MAX_LOSS_PER_TRADE_KRW",
# 금액손실컷 발동 최소 하락률(%): 이 하락률 미만이면 슬리피지/흔들림으로 간주하고 금액손실컷 미발동 (꼬리잡기)
"MIN_DROP_PCT_FOR_LOSS_CUT",
# 한투 API 관련 키 추가 (실전/모의 계좌 분리)
"KIS_APP_KEY_REAL", "KIS_APP_SECRET_REAL",
"KIS_APP_KEY_MOCK", "KIS_APP_SECRET_MOCK",
"KIS_ACCOUNT_NO_REAL", "KIS_ACCOUNT_CODE_REAL", # 실전 계좌 (KIS_MOCK=false 시 사용)
"KIS_ACCOUNT_NO_MOCK", "KIS_ACCOUNT_CODE_MOCK", # 모의 계좌 (KIS_MOCK=true 시 사용)
"KIS_MOCK",
# 단타 봇 전용 키
"TAKE_PROFIT_PCT", "MIN_DROP_RATE", "MIN_RECOVERY_RATIO_SHORT",
# 단타 매도 로직 (env/DB에서 수치 로드)
"SCALP_ATR_UP_MULT", "SCALP_ATR_DOWN_MULT", "SCALP_ATR_DROP_MULT",
"QUICK_PROFIT_PROTECT_HOURS", "QUICK_PROFIT_MAX_RATIO", "QUICK_PROFIT_CURRENT_MIN",
"MIN_HOLD_EARLY_TAKE_PCT", "MIN_HOLD_HIGH_PCT", "MIN_HOLD_DROP_FROM_HIGH",
"POST_HOLD_TAKE_PCT", "POST_HOLD_DROP_FROM_HIGH",
# 늘림목 봇 전용 키
"MAX_PER", "MAX_PEG", "MIN_GROWTH_PCT", "DCA_INTERVALS", "DCA_AMOUNTS",
# Mattermost 및 AI 리포트 관련 키
"MM_SERVER_URL", "MM_BOT_TOKEN_", "MATTERMOST_CHANNEL", "GEMINI_API_KEY",
"AI_JOURNAL_LINES", "ANTHROPIC_API_KEY", "CLAUDE_MODEL_ID", "CLAUDE_MAX_TOKENS",
"MM_BUTLER_CHANNEL",
# true=체결 알림을 전략 채널 + MATTERMOST_CHANNEL(통합) 양쪽 발송
"MM_DUAL_CHANNEL_ENABLED",
# OpenRouter API 연동용 키
"OPENROUTER_API_KEY", "OPENROUTER_MODEL_ID",
# Open DART — 폐기. API키도 env_config 물리컬럼에서 제거 (row size 확보).
# 전략 키(DART_* / STRATEGY_DART_*) 는 아래 목록·config_dart 에만 잔존(미사용).
"STRATEGY_DART_ENABLED", "KIS_DART_MM_CHANNEL",
# SCAN 감시
"DART_SCAN_ENABLED", # 폴링·DB·MM (구 DART_ENABLED)
"DART_ENABLED", # 하위호환 alias → SCAN
"DART_POLL_SEC", "DART_LOOKBACK_DAYS", "DART_MM_CHANNEL",
"DART_SAVE_CANDIDATES", "DART_REPORT_NAME_KEYWORDS",
"DART_SKIP_CORRECTION", "DART_LISTED_ONLY", "DART_LIST_MAX_PAGES",
# SCAN 품질필터 — 건설/재개발 제외 · 매출대비% · 수출·테마
"DART_QUALITY_FILTER_ENABLED",
"DART_EXCLUDE_CORP_KEYWORDS",
"DART_EXCLUDE_CONTENT_KEYWORDS",
"DART_MIN_SALES_PCT",
"DART_SKIP_IF_SALES_PCT_UNKNOWN",
"DART_THEME_KEYWORDS",
"DART_REQUIRE_THEME",
"DART_FETCH_DOCUMENT",
# 임시 구독(영구구독 아님)
"DART_SUBSCRIBE_ENABLED", # 기본 false — 워치리스트 적재
"DART_WATCH_TTL_HOURS", # 워치 유지 시간. 기본 24
"DART_WATCH_MAX", # 동시 워치 상한. 기본 15
# 실매 TRIGGER (기본 OFF)
"DART_TRADE_ENABLED",
"DART_SLOT_MONEY", "DART_MAX_STOCKS", "DART_TOTAL_BUDGET_KRW",
"DART_MAX_BUY_AMOUNT", "DART_MIN_INVEST_RATIO_OF_SLOT",
"DART_CAND_LIMIT",
"DART_TIME_START", "DART_TIME_END",
"DART_RSI_PERIOD", "DART_RSI_OVERSOLD", "DART_RSI_RECLAIM",
"DART_STOP_LOSS_PCT", "DART_TAKE_PROFIT_PCT",
"DART_TRAIL_PCT", "DART_TRAIL_ARM_PCT",
"DART_MAX_HOLD_BARS", "DART_EVENT_WINDOW_BARS",
"DART_MIN_PRICE", "DART_VOL_MULT", "DART_VOL_WINDOW",
"DART_MAX_LOSS_PER_TRADE_KRW",
"DART_FORCE_EOD_EXIT",
"DART_BACKTEST_USE_TICK_EXIT", "DART_BACKTEST_TICK_FALLBACK_OHLC",
# 봇별 Mattermost 채널 — 전략별 config_* (KIS_LONG 은 홀딩봇 전용 → env_config)
"KIS_LONG_MM_CHANNEL",
# 롱 위시리스트/뉴스 리포트 전용 키
"LONG_DAILY_LOOKBACK_DAYS", "LONG_MA_SHORT_DAYS", "LONG_MA_LONG_DAYS",
"LONG_REPORT_AM_HOUR", "LONG_REPORT_AM_MIN",
"LONG_REPORT_PM_HOUR", "LONG_REPORT_PM_MIN",
"LONG_NEWS_ENABLED", "LONG_NEWS_INTERVAL_MIN",
"LONG_NEWS_ACTIVE_START_HOUR", "LONG_NEWS_ACTIVE_END_HOUR",
"LONG_ANALYSIS_DELAY_MIN_SEC", "LONG_ANALYSIS_DELAY_MAX_SEC",
# RSI 기간: 스윙=14(기본), 단타=5, 스캘핑=3 권장 (DB에서 실시간 변경 가능)
"RSI_PERIOD",
# KIS WebSocket 관련 키
"KIS_WS_URL_REAL", "KIS_WS_URL_MOCK", "KIS_WS_MOCK_ENABLED",
# KIS WS 재연결·구독 (키움 CNSRREQ_GAP 과 동일 — 구독 연속 폭주 시 세션 즉시 종료 방지)
"KIS_WS_SUBSCRIBE_GAP_MIN_SEC", "KIS_WS_SUBSCRIBE_GAP_MAX_SEC",
"KIS_WS_RECONNECT_SESSION_WAIT_SEC", "KIS_WS_RECONNECT_REFRESH_KEY",
"WS_RECONNECT_DELAY_SECS",
"KIS_WS_APPROVAL_MIN_REISSUE_SEC", "KIS_WS_APPROVAL_MAX_VALID_SEC",
# invalid approval 연속 시 6h 가드 우회 REST 재발급 (죽은 키 루프 방지)
"KIS_WS_INVALID_APPROVAL_REISSUE_AFTER", "KIS_WS_INVALID_APPROVAL_BYPASS_6H",
"KIS_WS_EMERGENCY_REISSUE_MIN_SEC",
"ENV_MERGED_CACHE_TTL_SEC",
"KIS_WS_INSTANT_DROP_SEC", "KIS_WS_INSTANT_DROP_MAX",
"KIS_WS_INSTANT_DROP_COOLDOWN_SEC",
# 국내↔해외 WS approval 세션 시간분할 (1키=1세션). HHMM, 여유 버퍼 포함.
"KIS_WS_KR_HOLD_START_HM", "KIS_WS_KR_HOLD_END_HM",
"KIS_WS_US_HOLD_START_HM", "KIS_WS_US_HOLD_END_HM",
"KIS_WS_SESSION_GUARD_SEC",
# 재진입 쿨다운: 매도 후 같은 종목 재매수를 N초 동안 차단 (반복매매 루프 방지)
"REENTRY_COOLDOWN_SEC",
# 꼬리잡기/단타 매수 허용 시간대 (HHMM 정수, 930=09:30, 1500=15:00) — 백테스트·실매 공통
"TIME_START", "TIME_END",
# 매도 실패 백오프: 영업일 아님·장외 시간 오류 시 N초 동안 재시도 금지 (API 낭비·차단 방지)
"SELL_FAILURE_BACKOFF_SEC",
# 실잔고 inquire-balance 캐시 TTL(초) — 매도 루프 내 N종목 1회 조회 공유
"BROKER_HOLDINGS_CACHE_TTL_SEC",
# 유령잔고(DB·메모리만 존재) 정리 후 동일 (전략,종목) 재시도 쿨다운(초)
"GHOST_POSITION_COOLDOWN_SEC",
# ── 스캘핑봇(kis_scalping_ver1) 전용 키 ──────────────────────────────
# RSI 과매도 임계값: 이 값 이하면 "과매도 → 되돌림 가능" 후보로 판단
"SCALP_RSI_OVERSOLD",
"SCALP_RSI_PERIOD",
# RSI 과매수 임계값: 이 값 이상이면 신규 진입 금지 (고점 추격 방지)
"SCALP_RSI_OVERBOUGHT",
# 스캘핑봇이 사용할 봉 단위 (분): 1 또는 3 권장
"SCALP_CANDLE_TIMEFRAME",
# 장 시작 후 몇 분 뒤부터 매매 허용 (장 시작 직후 변동성 회피)
"SCALP_MARKET_OPEN_WAIT_MIN",
# WS 재접속 후 갭 보정에 사용할 REST 분봉 조회 캔들 수
"SCALP_GAP_FILL_LIMIT",
# 확정 봉 기준 최소 거래량: 이 값 미만인 봉은 유동성 부족으로 무시
"SCALP_MIN_VOLUME",
# 스캘핑·꼬리 MM alias → config_scalp / config_short
"KIS_SCALP_MM_CHANNEL",
"KIS_SHORT_MM_CHANNEL",
# 시스템 알림(시작·종료·장시작·장마감) 단일 MM alias (기본 default=키스명령봇)
"KIS_SYSTEM_MM_CHANNEL",
# 시작·종료 알림 추가 발송 MM alias (기본 stock — 재시작/종료 짝 확인용)
"KIS_LIFECYCLE_MM_CHANNEL",
# journalctl 오류감시 → Mattermost (scripts/kis_error_watch_mm.py)
"ERROR_WATCH_ENABLED",
"ERROR_WATCH_UNIT",
"ERROR_WATCH_MM_CHANNEL",
"ERROR_WATCH_COOLDOWN_SEC",
"ERROR_WATCH_CONTEXT_LINES",
"ERROR_WATCH_TRACEBACK_EXTRA_LINES",
"ERROR_WATCH_HEALTH_CHECK_SEC",
"ERROR_WATCH_GLOBAL_MIN_GAP_SEC",
"ERROR_WATCH_MATCH_REGEX",
"ERROR_WATCH_IGNORE_REGEX",
"ERROR_WATCH_MM_JITTER",
"ERROR_WATCH_STARTUP_NOTIFY",
# API 거절/예외 JSONL 백로그 (나중에 고칠 목록) + 영구형 거절 주문 쿨다운
"KIS_API_REJECT_LOG_ENABLED",
"KIS_API_REJECT_LOG_PATH",
"KIS_API_REJECT_LOG_DEDUP_SEC",
"KIS_API_REJECT_ORDER_COOLDOWN_ENABLED",
"KIS_API_REJECT_ORDER_COOLDOWN_SEC",
"KIS_API_REJECT_COOLDOWN_MSG_CD",
"KIS_API_REJECT_COOLDOWN_MSG_SUBSTR",
"KIS_API_REJECT_COOLDOWN_HTTP_MIN",
# 미등록 보유분 일괄 시장가매도 시 종목 간 간격(초) — 429 방지
"BULK_SELL_INTERVAL_SEC",
# 수동매수 보호 종목코드(쉼표구분) — 일괄매도에서 무조건 제외
"MANUAL_HOLD_CODES",
# 장마감 후 봇 고아(active_trades 미기록) 자동 복구 ON/OFF
"ORPHAN_RECONCILE_ENABLED",
# 봇 기동 직후 실잔고 1회 → active_trades 맞춤 (매도 없음, 기본 ON)
"ORPHAN_RECONCILE_ON_STARTUP",
# Pre-EOD 고아복구 — 가장 이른 EOD N분 (EOD 청산 전 1회)
"ORPHAN_RECONCILE_PRE_EOD_ENABLED",
"ORPHAN_RECONCILE_PRE_EOD_LEAD_MIN",
# Pre/Post EOD 고아복구와 동일 잔고조회에서 DB有·브로커0 유령 삭제
"GHOST_PURGE_ON_RECONCILE",
# 당일 청산(장마감청산 등) 히스토리 있으면 고아복구로 active 재등록 금지
"ORPHAN_SKIP_AFTER_SELL_HISTORY",
# 재등록 금지 sell_reason 키워드(쉼표). 비우면 매수 이후 모든 청산 히스토리 금지
"ORPHAN_SKIP_SELL_REASONS",
# active 에 남아 있는데 매수시각 이후 청산 히스토리가 있으면 DB 행 정리
"ORPHAN_PURGE_ACTIVE_AFTER_SELL_HISTORY",
# insert 실패(주문DB중복) 시 브로커 체결 조회 복구
"DUPLICATE_ORDER_FILL_RECOVERY_ENABLED",
"DUPLICATE_ORDER_RECOVERY_WAIT_SEC",
# 장중 잔고 qty 드리프트 감시 (기본 OFF — REST 주기 폴링)
"INTRADAY_HOLDINGS_DRIFT_ENABLED",
"INTRADAY_HOLDINGS_DRIFT_INTERVAL_SEC",
"INTRADAY_HOLDINGS_DRIFT_AUTO_RECOVER",
# ws_candles 자동 정리 보존 일수
"SCALP_CANDLE_KEEP_DAYS",
# ws_ticks 실시간 체결 틱 (RAM 링버퍼 + 배치 INSERT, C안)
"WS_TICK_SAVE_ENABLED",
"WS_TICK_SAVE_US_ENABLED",
"WS_TICK_BUFFER_MAX_PER_CODE",
"WS_TICK_DB_BATCH_SIZE",
"WS_TICK_DB_FLUSH_SEC",
"WS_TICK_WRITE_QUEUE_MAX",
"WS_TICK_KEEP_DAYS",
"WS_TICK_CLEANUP_CHUNK",
"WS_TICK_CLEANUP_MAX_LOOPS",
# 백테/Optuna ws_ticks 벌크 로드 (일별 청크) — TradeDB 기본 read_timeout(30s) 우회
"WS_TICK_LOAD_READ_TIMEOUT",
"WS_TICK_LOAD_MAX_RETRIES",
"WS_TICK_RECORD_SCOPE",
"WS_TICK_DEFAULT_MARKET",
"WS_TICK_DEFAULT_EXCHANGE",
"WS_TICK_DEFAULT_CURRENCY",
# 영구구독 기준봉(분) — 1분이면 15/60 롤업. UI·저장 통일.
"PERM_SUB_BASE_TF_MIN",
# 영구구독 탭 현재가 자동새로고침 초 (웹 UI)
"PERM_SUB_UI_REFRESH_SEC",
# 영구구독 목표가 MM 알람 (permanent 채널)
"PERM_ALERT_ENABLED",
"PERM_ALERT_POLL_SEC",
"KIS_PERM_SUB_MM_CHANNEL",
# 해외 탭 표시용 USD→KRW (Frankfurter 등). 매매 체결환율 아님.
"FX_USD_KRW_API_BASE",
"FX_USD_KRW_CACHE_DIR",
"FX_USD_KRW_TIMEOUT_SEC",
"FX_USD_KRW_MIN_INTERVAL_SEC",
"FX_USD_KRW_MAX_RETRIES",
"FX_USD_KRW_LOOKBACK_DAYS",
"FX_USD_KRW_FALLBACK",
"FX_USD_KRW_RANGE_MAX_DAYS",
"FX_USD_KRW_HTTP_LOOKBACK",
# 외부 환율 HTTP (Frankfurter). 기본 OFF — 캐시/폴백만으로 $↔원 표시
"FX_USD_KRW_HTTP_ENABLED",
# 백테 틱재생: 실매 후보=키움 정합. 분봉에 kiwoom 있으면 kis(과거 volume 오염) 제외
"WS_TICK_PREFER_KIWOOM",
# 후보/보유 이탈 후 틱 구독·저장 유지 (키움 한도 내. 영구 KR은 LS, 한투 41과 분리)
"WS_TICK_GRACE_SEC",
"WS_TICK_GRACE_HEADROOM",
# ws_orderbook / ws_program — TRIGGER 호가·프로그램 스냅샷 (백테 재현)
"WS_ORDERBOOK_SAVE_ENABLED",
"WS_PROGRAM_SAVE_ENABLED",
"WS_TRIGGER_SNAPSHOT_DB_INTERVAL_SEC",
"WS_TRIGGER_SNAPSHOT_DB_BATCH_SIZE",
"WS_TRIGGER_SNAPSHOT_DB_FLUSH_SEC",
"WS_TRIGGER_SNAPSHOT_WRITE_QUEUE_MAX",
"WS_ORDERBOOK_KEEP_DAYS",
"WS_PROGRAM_KEEP_DAYS",
# TRIGGER 필터 판정 시점 스냅샷 (실매 RAM → DB, 백테 재생)
"WS_TRIGGER_EVAL_SAVE_ENABLED",
"WS_ORDERBOOK_COLLECT_ENABLED",
# 진입 호가필터 RAM 나이. 0=마지막 호가로 검사(저장 TTL 과 분리)
"WS_ORDERBOOK_FILTER_MAX_AGE_SEC",
# 매수/조회 get_price() 기본 나이. 0=마지막 RAM(체결 없어도 유지). 양수=그 초 넘으면 None.
"WS_PRICE_MAX_AGE_SEC",
# 실매·옵투나 읽기 폴백 나이(초). 메인 체결시각 vs 지금. 넘기면 그 벤더 실패→2차→3차.
"LIVE_FEED_FALLBACK_MAX_AGE_SEC",
# ws_ticks DB 저장 시 lag 컷 스위치 (2026-09-06 신설, 3벤더 공통 · docs/정합성.md §9)
# false(기본): DB 는 모든 틱 저장 (벤더 통계·재현·백테 정합용)
# true: lag > LIVE_FEED_FALLBACK_MAX_AGE_SEC 인 틱은 미저장 (미래 A안 전환용)
"WS_TICK_DB_SAVE_LAG_CUT_ENABLED",
# 수집통계 탭: 종목별 recv_ts 공백(끊김) 소프트/하드/상한(초)
"FEED_STATS_DISCONNECT_SOFT_SEC",
"FEED_STATS_DISCONNECT_HARD_SEC",
"FEED_STATS_DISCONNECT_CAP_SEC",
# 실매 틱/호가 1차 벤더 (LS는 1차 아님 · 구독 spill 3차). 호가 kis=2키 OB 세션.
"LIVE_TICK_PROVIDER",
"LIVE_OB_PROVIDER",
# 벤더별 호가 DB 적재. KIS=2키(H0STASP0)만. 메인 시세 41 합산 금지.
"WS_ORDERBOOK_SAVE_KIS",
"WS_ORDERBOOK_SAVE_KIWOOM",
"TRIGGER_FEED_DETAIL_LOG",
# 옵투나/백테 피드 추적 (틱·호가 벤더 비율 + TRIGGER 샘플 상한)
"BT_FEED_DETAIL_LOG",
"BT_FEED_DETAIL_LOG_MAX",
"CANDLE_GARBAGE_FALLBACK",
# LS에 메인 틱/호가 유니버스(후보보유grace) 전체 미러. 기본 OFF — spill/permanent만.
"LS_FEED_FALLBACK_SUBSCRIBE",
# hold reopen 시 RAM 유지: 영구 + (기본) 보유. OFF=영구만.
"LS_WS_HOLD_RAM_KEEP_HOLDINGS",
# 후보 LS 갭 REST(t8412). 기본 OFF — 구독만. 영구 갭은 LS_GAP_FILL_ENABLED.
"LS_GAP_FILL_CANDIDATES",
# 필터 ON + 호가 사진 한 장도 없으면 거절(기본 true). 만료 Fail-Open 과 별개.
"WS_ORDERBOOK_FILTER_REJECT_IF_EMPTY",
# 실매 매도: 마지막 WS 체결가 유지. 죽은선 감지 시에만 REST 1회(쿨다운).
"SELL_WS_STALE_REST_SEC",
"SELL_WS_STALE_REST_COOLDOWN_SEC",
"SELL_WS_LAST_RAM_MAX_AGE_SEC",
"WS_PROGRAM_COLLECT_ENABLED",
"WS_TRIGGER_EVAL_DB_BATCH_SIZE",
"WS_TRIGGER_EVAL_DB_FLUSH_SEC",
"WS_TRIGGER_EVAL_WRITE_QUEUE_MAX",
"BACKTEST_USE_TRIGGER_EVAL_SNAPSHOT",
"BACKTEST_USE_TRIGGER_EVAL_SNAPSHOT",
"MOMENTUM_BACKTEST_USE_TRIGGER_EVAL_SNAPSHOT",
"BREAKOUT_BACKTEST_USE_TRIGGER_EVAL_SNAPSHOT",
"TAIL_BACKTEST_USE_TRIGGER_EVAL_SNAPSHOT",
"BACKTEST_USE_TRIGGER_SNAPSHOT_DB",
"MOMENTUM_BACKTEST_USE_TRIGGER_SNAPSHOT_DB",
"MOMENTUM_BACKTEST_USE_TICK_EXIT",
"MOMENTUM_BACKTEST_USE_TICK_ENTRY",
"MOMENTUM_BACKTEST_TICK_FALLBACK_OHLC",
"MOMENTUM_BACKTEST_WALLCLOCK_LAST_PRICE",
"BREAKOUT_BACKTEST_USE_TICK_DB",
"BREAKOUT_BACKTEST_USE_TICK_EXIT",
"BREAKOUT_BACKTEST_TICK_FALLBACK_OHLC",
"BREAKOUT_BACKTEST_POLL_MS",
"BREAKOUT_BACKTEST_SELL_SLIP_PCT",
# 돌파 백테 전일 봉 웜업(실매 WS 버퍼 정합) · 유니버스 EXIT 디바운스
"BREAKOUT_BACKTEST_CANDLE_WARMUP_BARS",
"BREAKOUT_BACKTEST_UNIVERSE_SCAN_AT",
"BREAKOUT_UNIVERSE_EXIT_DEBOUNCE_SEC",
# 백테 유니버스 이력 소스: kiwoom(target_candidates_history) | ls(ls_candidates_history)
"BACKTEST_UNIVERSE_HISTORY_SOURCE",
"BACKTEST_LS_UNIVERSE_SESSION_ONLY",
# 실매 유니버스슬롯 정합 이력 테이블 (비우면 UNIVERSE_SOURCE 로 자동: ls_condition→ls)
"LIVE_UNIVERSE_HISTORY_SOURCE",
"SCALP_LIVE_UNIVERSE_HISTORY_SOURCE",
"BREAKOUT_LIVE_UNIVERSE_HISTORY_SOURCE",
"SHORT_LIVE_UNIVERSE_HISTORY_SOURCE",
"MOMENTUM_LIVE_UNIVERSE_HISTORY_SOURCE",
"TAIL_BACKTEST_USE_TICK_EXIT",
"TAIL_BACKTEST_USE_SPILL_FALLBACK",
"TAIL_BACKTEST_POLL_MS",
"TAIL_BACKTEST_SELL_SLIP_PCT",
"SCALP_BACKTEST_USE_TICK_EXIT",
"SCALP_BACKTEST_USE_TICK_ENTRY",
"SCALP_BACKTEST_TICK_FALLBACK_OHLC",
"SCALP_BACKTEST_POLL_MS",
"SCALP_BACKTEST_SELL_SLIP_PCT",
"SCALP_BACKTEST_CANDLE_WARMUP_BARS",
"SCALP_UNIVERSE_EXIT_DEBOUNCE_SEC",
"RANGE_BREAK_BACKTEST_USE_TICK_EXIT",
"RANGE_BREAK_BACKTEST_TICK_FALLBACK_OHLC",
"RANGE_BREAK_BACKTEST_POLL_MS",
"RANGE_BREAK_BACKTEST_SELL_SLIP_PCT",
"DBBAND_BACKTEST_USE_TICK_EXIT",
"DBBAND_BACKTEST_TICK_FALLBACK_OHLC",
"DBBAND_BACKTEST_POLL_MS",
"DBBAND_BACKTEST_SELL_SLIP_PCT",
"MOMENTUM_BACKTEST_POLL_MS",
"MOMENTUM_BACKTEST_SELL_SLIP_PCT",
"MOMENTUM_BACKTEST_BUY_SLIP_PCT",
"MOMENTUM_BACKTEST_UNIVERSE_STRICT",
"MOMENTUM_BACKTEST_UNIVERSE_STRICT_LAG_MIN",
"MOMENTUM_BACKTEST_LIVE_SCAN_QUEUE",
"MOMENTUM_BACKTEST_SCAN_SEC",
"MOMENTUM_BACKTEST_UNIVERSE_SCAN_AT",
"MOMENTUM_BACKTEST_CANDLE_WARMUP_BARS",
# 백테 REST 전일봉 웜업: 1차→실패시만 사다리 증량(최대 4단) · DB 캐시 공유
"MOMENTUM_BACKTEST_REST_WARMUP",
"MOMENTUM_BACKTEST_REST_WARMUP_BARS",
"MOMENTUM_BACKTEST_REST_WARMUP_BARS_RETRY",
"MOMENTUM_BACKTEST_REST_WARMUP_BARS_3",
"MOMENTUM_BACKTEST_REST_WARMUP_BARS_4",
"MOMENTUM_BACKTEST_REST_WARMUP_DB_CACHE",
"MOMENTUM_BACKTEST_REST_MAX_CODES",
"MOMENTUM_BACKTEST_REST_SLEEP_SEC",
"MOMENTUM_BACKTEST_REST_KIWOOM_MOCK",
"MOMENTUM_PREV_DAY_OPEN_HM_MAX",
"MOMENTUM_UNIVERSE_EXIT_DEBOUNCE_SEC",
"MOMENTUM_LIVE_ALIGN_USE_FORMING_BAR",
"MOMENTUM_COOLDOWN_USE_CANDLE_FLOOR",
"MOMENTUM_COOLDOWN_ENGINE_ONLY",
"MOMENTUM_BACKTEST_SKIP_PRE_SUBSCRIBE",
"BREAKOUT_BACKTEST_USE_TRIGGER_SNAPSHOT_DB",
"TAIL_BACKTEST_USE_TRIGGER_SNAPSHOT_DB",
# 휩쏘(TRIGGER) 필터 — 공통 + 전략별
"WHIPSAW_FILTER_ENABLED",
"WHIPSAW_BACKTEST_OHLC_FALLBACK",
"WHIPSAW_MODE",
"WHIPSAW_SUBBAR_SEC",
"WHIPSAW_LOOKBACK_SEC",
"WHIPSAW_DIP_PCT",
"WHIPSAW_RECOVERY_TOL_PCT",
"WHIPSAW_MIN_SUBBARS",
"WHIPSAW_TICK_BUFFER_LIMIT",
"MOMENTUM_WHIPSAW_FILTER_ENABLED",
"MOMENTUM_WHIPSAW_MODE",
"MOMENTUM_WHIPSAW_SUBBAR_SEC",
"MOMENTUM_WHIPSAW_LOOKBACK_SEC",
"MOMENTUM_WHIPSAW_DIP_PCT",
"MOMENTUM_WHIPSAW_RECOVERY_TOL_PCT",
# 스캘핑 휩쏘 — Optuna 추천·실매 TRIGGER (꼬리/돌파는 DB 적용 skip, 스캘은 필수)
"SCALP_WHIPSAW_FILTER_ENABLED",
"SCALP_WHIPSAW_MODE",
"SCALP_WHIPSAW_SUBBAR_SEC",
"SCALP_WHIPSAW_LOOKBACK_SEC",
"SCALP_WHIPSAW_DIP_PCT",
"SCALP_WHIPSAW_RECOVERY_TOL_PCT",
"BREAKOUT_WHIPSAW_FILTER_ENABLED",
"BREAKOUT_WHIPSAW_MODE",
"BREAKOUT_WHIPSAW_SUBBAR_SEC",
"BREAKOUT_WHIPSAW_LOOKBACK_SEC",
"BREAKOUT_WHIPSAW_DIP_PCT",
"BREAKOUT_WHIPSAW_RECOVERY_TOL_PCT",
"TAIL_WHIPSAW_FILTER_ENABLED",
"TAIL_WHIPSAW_MODE",
"TAIL_WHIPSAW_SUBBAR_SEC",
"TAIL_WHIPSAW_LOOKBACK_SEC",
"TAIL_WHIPSAW_DIP_PCT",
"TAIL_WHIPSAW_RECOVERY_TOL_PCT",
"TAIL_WHIPSAW_MIN_SUBBARS",
"TAIL_WHIPSAW_TICK_BUFFER_LIMIT",
# 스캘핑 전용 손절 % (꼬리잡기 STOP_LOSS_PCT와 분리, 기본 1.5%)
# 1분봉 초단타: 손절이 넓으면(-4%) 자금이 묶여 회전율 0 → 타이트하게
"SCALP_STOP_LOSS_PCT",
# 스캘핑 전용 익절 % (꼬리잡기 TAKE_PROFIT_PCT와 분리, 기본 1.5%)
# 1분봉에서 +5% 익절은 도달 불가 → 박리다매 전략으로 1.5%씩 수익 적립
"SCALP_TAKE_PROFIT_PCT",
# 스캘핑 익절 상한·어깨컷 (꼬리와 동일 1순위 청산)
"SCALP_TP_MAX_PCT",
"SCALP_SHOULDER_MIN_HIGH_PCT", "SCALP_SHOULDER_CUT_PCT",
# 스캘핑 낙폭 필터 % (꼬리잡기 MIN_DROP_RATE와 분리, 기본 1.5%)
# 3% 기준은 1분봉 소형주에서 너무 엄격 → 1.5%로 완화해 타점 빈도 증가
"SCALP_MIN_DROP_RATE",
# 봉부족 감지 시 재갭보정 최소 간격(초): 같은 종목 중복 REST 호출 방지 (기본 30초)
"SCALP_GAP_RETRY_SEC",
# 스캘핑 전용 재진입 쿨다운(초): 매도 후 같은 종목 N초 동안 재매수 차단 (기본 600=10분)
# 백테스트와 동일 파라미터로 맞추려면 DB에 값 저장 후 봇/백테스트 모두 이 값 사용
"SCALP_REQUIRE_REVERSAL_CANDLE", # true=직전 음봉+현재 양봉 필수, false=RSI 과매도만으로 진입 허용
"SCALP_COOLDOWN_SEC",
# 최소 보유 초: 진입 후 N초 미만이면 청산 검사 스킵 (0=OFF). Optuna/Grid min_hold_sec 와 동일 키.
"SCALP_MIN_HOLD_SEC",
# 트레일링 발동 최소 수익률(%): 고점이 매수가 대비 이 이상 올라야 트레일링 활성화
# 0.5%면 수수료(~0.21%) 뺀 나머지만 이익 → 1.5 이상 권장
"SCALP_TRAIL_TRIGGER_PCT",
# 트레일링/본절사수 후 최소 순이익(%): 수수료+세금 위에 이 값만큼 추가 마진 확보
# 0이면 본절(수수료 이후 0원), 0.2면 최소 +0.2% 순이익 보장
"SCALP_MIN_PROFIT_PCT",
# 스캘핑 전용 방어로직 (꼬리잡기와 값 분리: HIGH_PRICE_CHASE_THRESHOLD / MIN_PRICE_TAIL 등과 별도)
"SCALP_HIGH_PRICE_CHASE_THRESHOLD", # 고점추격 방지: 당일고가 대비 비율 이하면 진입 허용 (0.96 등)
"SCALP_MAX_DAILY_CHANGE_PCT", # 급등주 필터: 당일 등락률 이 값 초과 시 진입 금지 (%)
"SCALP_MIN_PRICE", # 최소 가격(원): 이 값 미만 종목 진입 금지
"SCALP_MAX_LOSS_PER_TRADE_KRW", # 1회 최대 손실(원): 종목당 손실 상한
"SCALP_MIN_DROP_PCT_FOR_LOSS_CUT", # 금액손실컷 발동 최소 하락률(%): 이 미만은 흔들림으로 미발동
"SCALP_USE_DEFENSE_FILTERS", # true/false: 스캘핑 진입 방어 필터(낙폭/고점추격/급등/최소가격) ON/OFF
# kiwoom scalp_re SCAN 사용 시 HTS B/C/F/D 와 겹치는 TRIGGER(낙폭·RSI·되돌림) 생략. 비우면 UNIVERSE_SOURCE=condition 일 때 자동 true
"SCALP_SKIP_HTS_SCAN_DUPES",
# HTS C(MACD+Stoch 골든크로스) — TRIGGER 전용. true면 RSI reversal 대신 MACD 진입.
"SCALP_USE_MACD_CROSS",
"SCALP_MACD_FAST", "SCALP_MACD_SLOW", "SCALP_MACD_SIGNAL",
"SCALP_STOCH_K_PERIOD", "SCALP_STOCH_D_PERIOD", "SCALP_STOCH_SLOW",
# Optuna/Grid 탐색 축 (CSV, env_config — 웹·봇 기본값과 별도)
"SCALP_GRID_TRIGGER_RSI_OVERSOLD", "SCALP_GRID_TRIGGER_DROP_RATE",
"SCALP_GRID_TRIGGER_HIGH_CHASE_THR", "SCALP_GRID_TRIGGER_VOL_MULT",
"SCALP_GRID_TRIGGER_REQUIRE_REVERSAL",
"SCALP_GRID_SKIP_HTS_SCAN_DUPES",
"SCALP_GRID_EXIT_SL_PCT", "SCALP_GRID_EXIT_TP_PCT", "SCALP_GRID_EXIT_TP_MAX_PCT",
"SCALP_GRID_EXIT_SHOULDER_MIN_HIGH_PCT", "SCALP_GRID_EXIT_SHOULDER_CUT_PCT",
"SCALP_GRID_EXIT_MIN_HOLD_SEC",
"SCALP_SCORE_MDD_FLOOR",
# ── 단타봇(kis_short_ver2) 전용 키 ───────────────────────────────────
# 켈리 공식 사용 여부 (true=켈리 적용, false=고정 비중)
"USE_KELLY_FORMULA",
# 켈리 공식 적용 배수 (0.25 = Full Kelly의 25%, 과도한 베팅 방지)
"KELLY_MULTIPLIER",
# 시장가 IOC 주문 사용 여부 — 매수 (실전: true=IOC13, false=일반01 · 모의는 코드에서 01 고정)
"USE_MARKET_IOC",
# 시장가 IOC 주문 사용 여부 — 매도 (실전: true=IOC13, false=일반01 · 모의는 코드에서 01 고정)
"USE_MARKET_IOC_SELL",
# 체결 확인 엄격 모드: true=모의도 fill 없으면 가정 체결 금지 (실전 훈련)
"STRICT_FILL_VERIFY",
# 주문 직후 체결 조회 대기(초) — 시장가
"ORDER_FILL_WAIT_SEC",
# 지정가 주문 직후 체결 조회 대기(초)
"LIMIT_ORDER_FILL_WAIT_SEC",
# heartbeat 미체결 재조회 간격(초)
"PENDING_FILL_POLL_INTERVAL_SEC",
# 미체결 재조회 시 get_execution 1회 대기(초)
"PENDING_FILL_POLL_SEC",
# (레거시) 매수 미체결 최대 대기 — PENDING_BUY_MAX_AGE_SEC 미설정 시 fallback
"PENDING_FILL_MAX_AGE_SEC",
# 매수 미체결 최대 대기(초) — 초과 시 취소·다음 신호 대기
"PENDING_BUY_MAX_AGE_SEC",
# 익절·지정가 매도 미체결 최대 대기(초)
"PENDING_SELL_MAX_AGE_SEC",
# 손절·장마감·긴급 매도 미체결 최대 대기(초) — 짧게, 만료 시 시장가 재주문
"PENDING_SELL_STOP_MAX_AGE_SEC",
# 손절 등 긴급 매도 만료 취소 후 즉시 시장가 재주문
"SELL_PENDING_REORDER_ON_EXPIRE",
# 동일 전략·종목 미체결 매수 있으면 재주문 스킵 (체결 API 지연 시 주문 폭주 방지)
"BUY_DEDUP_PENDING",
# 매수 부분체결 후 잔량 자동 취소 (지정가·비-IOC 시장가)
"AUTO_CANCEL_PARTIAL_BUY_REMAINDER",
# heartbeat poll_pending_fills: 당일 체결 1 REST 일괄 조회 (실매 전용)
"PENDING_POLL_BATCH_FETCH",
# 일별주문체결조회 TR (기본 실전 TTTC0081R / 모의 VTTC0081R — 레거시 *8001R 금지)
"DAILY_CCLD_TR_ID",
"DAILY_CCLD_MAX_PAGES",
"DAILY_CCLD_PAGE_GAP_SEC",
"DAILY_CCLD_EXCG_ID",
# 매수가능조회(inquire-psbl-order) — 주문가능현금 전용
"PSBL_ORDER_REF_CODE",
"PSBL_ORDER_CMA_ICLD",
"PSBL_ORDER_OVRS_ICLD",
# 익절지정가 취소실패+잔고0 → 정상 매도확정 / PENDING매도 시 유령금지
"SELL_LIMIT_CANCEL_BEFORE_MARKET_RETRY",
"SELL_LIMIT_RECHECK_WAIT_SEC",
"SELL_LIMIT_CANCEL_FAIL_BALANCE_CONFIRM",
"GHOST_PURGE_BLOCK_WHILE_PENDING_SELL",
"GHOST_PURGE_RECORD_HISTORY",
# 유령정리 전 정정취소가능주문조회 (매도가능 0 ≠ 보유 0)
"INQUIRE_PSBL_RVSECNCL_BEFORE_GHOST",
"PSBL_RVSECNCL_CACHE_TTL_SEC",
"PSBL_RVSECNCL_MAX_PAGES",
"PSBL_RVSECNCL_PAGE_GAP_SEC",
"PSBL_RVSECNCL_TR_ID",
"PSBL_RVSECNCL_INQR_DVSN_1",
"PSBL_RVSECNCL_INQR_DVSN_2",
"CANCELABLE_CCLD_MAX_PAGES",
"SELL_LOCKED_ENQUEUE_COOLDOWN_SEC",
"CANCELABLE_RECONCILE_ENABLED",
"CANCELABLE_RECONCILE_CANCEL_RETRY",
# 잔고 연속조회 최대 페이지 (1p=실전50/모의20종목) — 보유 많을 때 누락 방지
"BALANCE_MAX_PAGES",
# WebSocket 실시간 가격 캐시 유효기간(초): 이 시간 이상 지나면 REST 재조회
"KIS_PRICE_CACHE_TTL_SEC",
# WS 재접속 후 갭 보정에 사용할 REST 분봉 조회 캔들 수 (단타봇)
"SHORT_GAP_FILL_LIMIT",
# 매수 직후 최소 보유 시간(초): 이 기간 내 매도 신호 무시 (API 잔고 반영 지연 대응)
"MIN_HOLD_AFTER_BUY_SEC",
# 최소 보유 시간(시간): 이 기간 이전에는 손절 외 매도 금지 (꼬리잡기 전략 충분히 대기)
"MIN_HOLD_HOURS",
# 3개월 최대 회복 비율: 전고점 대비 이 비율 이상 회복한 종목은 추격 매수 제외
"MAX_RECOVERY_RATIO_3M",
# 유니버스 상위 N개 후보 경량 체크 (매수 후보 1차 필터)
"CANDIDATE_LIST_TOP_N_LIGHT",
# 매수 신호 체크 시 사용할 유니버스 최대 종목 수 (과부하 방지)
"SCAN_UNIVERSE_MAX_CODES",
# 꼬리 캔들 패턴 인식 시 과거 몇 봉까지 확인할지 (lookback)
"TAIL_CANDLE_LOOKBACK",
# ── 레거시 단일 계좌 키 (KIS_APP_KEY_REAL/MOCK 이전 버전 호환용) ──────
"KIS_APP_KEY", "KIS_APP_SECRET", "KIS_ACCOUNT_NO", "KIS_ACCOUNT_CODE",
# ── mm_butler 전용 키 ────────────────────────────────────────────────
# Gemini AI 모델 ID (gemini-2.5-flash, gemini-1.5-pro 등)
"GEMINI_MODEL_ID",
# MM 원격 명령 폴링 주기(초): 너무 짧으면 API 과부하
"MM_BUTLER_POLL_SEC",
# AI 소스 텍스트 최대 길이(문자): 초과 시 잘라서 전송 (토큰 비용 관리)
"AI_SOURCE_MAX_CHARS",
# ── 수수료·거래세 (전략 공통) ──────────────────────────────────────────
# FEE_RATE_PCT : 위탁수수료율 (매수/매도 각각, 기본 0.015%)
# SELL_TAX_RATE_PCT: 증권거래세율 (매도 시만 부과, 기본 0.18% ← 2025 코스피/코스닥 공통)
"FEE_RATE_PCT", "SELL_TAX_RATE_PCT",
# ── 키움증권 REST API 키 (60분봉 과거 데이터 수집 전용) ────────────────
# 실전/모의 분리 (KIS_MOCK 값에 따라 자동 선택)
# KIWOOM_APP_KEY_REAL : 키움증권 실전 앱키
# KIWOOM_APP_SECRET_REAL: 키움증권 실전 시크릿
# KIWOOM_APP_KEY_MOCK : 키움증권 모의 앱키 (mockapi.kiwoom.com)
# KIWOOM_APP_SECRET_MOCK: 키움증권 모의 시크릿
# KIWOOM_APP_KEY / KIWOOM_APP_SECRET: 레거시 (단일 키 호환용)
"KIWOOM_APP_KEY_REAL", "KIWOOM_APP_SECRET_REAL",
"KIWOOM_APP_KEY_MOCK", "KIWOOM_APP_SECRET_MOCK",
"KIWOOM_APP_KEY", "KIWOOM_APP_SECRET",
# ── LS증권 Open API 키 (시세/주문 연동용 — 값 수동 입력) ─────────────
# LS_APP_KEY_REAL / LS_APP_SECRET_REAL : 실전
# LS_APP_KEY_MOCK / LS_APP_SECRET_MOCK : 모의투자 (별도 키 페어)
# LS_USER_ID : HTS/OpenAPI 로그인 ID (t1866 서버저장조건 목록 조회용, 최대 8자)
"LS_APP_KEY_REAL", "LS_APP_SECRET_REAL",
"LS_APP_KEY_MOCK", "LS_APP_SECRET_MOCK",
"LS_USER_ID",
# ── WebSocket 영구 구독 종목 (시장 방향 필터용) ─────────────────────
# KOSPI/KOSDAQ 지수 ETF는 매매 후보와 무관하게 항상 구독 유지.
# 60분봉 RSI 로 상승장/하락장 체크 → 스캘핑/꼬리잡기 진입 방향 결정.
# 기본값: KODEX200(069500), KODEX KOSDAQ150(229200)
# 콤마 구분 코드: "069500,229200"
"PERMANENT_WS_CODES",
# 영구구독 마스터 — false면 테이블 행 유지, KR LS / US 해외 WS 구독만 안 함
"PERMANENT_SUBSCRIBE_ENABLED",
# 영구구독 「확정봉 가져오기」 — 키움 ka10080 → ls_ws_candles 구멍만
"PERM_LS_FILL_BARS", "PERM_LS_FILL_SLEEP_MIN", "PERM_LS_FILL_SLEEP_MAX",
# ── 시장 방향 필터 (상승장에서만 롱 진입) ──────────────────────────
# USE_MARKET_REGIME_FILTER: true=활성, false=비활성 (기본 false)
# MARKET_REGIME_MIN_RSI : ETF 60분봉 RSI 이 값 이상이어야 롱 진입 허용 (기본 48)
"USE_MARKET_REGIME_FILTER", "MARKET_REGIME_MIN_RSI",
# ── 테마 과열 필터 (테마 전체가 과열이면 신규 진입 억제) ───────────
# USE_THEME_HEAT_FILTER : true=활성 (기본 false)
# THEME_HEAT_RSI_MAX : 테마 평균 RSI 이 값 초과면 진입 차단 (기본 72)
"USE_THEME_HEAT_FILTER", "THEME_HEAT_RSI_MAX",
# ── kis_trader 통합봇 전용 키 (전략 스위치 / 유니버스 / 랭킹) ────────
# 전략 on/off → config_{전략} (classify_config_key)
"STRATEGY_SCALP_ENABLED", "STRATEGY_SHORT_ENABLED", "STRATEGY_BREAKOUT_ENABLED",
"STRATEGY_RANGE_BREAK_ENABLED",
"STRATEGY_UPDOW_ENABLED",
# 중복매매 정책: allow(기본)=전략별 ODNO·active_trades 분리 / block=동일종목 1전략만
"STRATEGY_SAME_CODE_POLICY",
# 가용 예수금 캐시 (kv_store account.*) — 부족 시에만 매수 qty 축소
"ORDER_CASH_PCT", "ORDER_CASH_FEE_BUFFER", "ORDER_CASH_DIVIDE_BY_MAX_STOCKS",
"ACCOUNT_CASH_PERSIST_SEC",
# 라이브 진입 타이밍 = 백테 (신호봉 확정 → 다음 봉 시가)
"SCALP_LIVE_BACKTEST_ALIGN", "SCALP_LIVE_SIGNAL_LOOKBACK_BARS",
"SCALP_LIVE_ALIGN_USE_FORMING_BAR",
"SCALP_BACKTEST_USE_TICK_ENTRY", "SCALP_BACKTEST_TICK_FALLBACK_OHLC",
# 실매 진입 계산가: RAM 틱 첫가 → 없으면 분봉 시가 (백테 align_entry_price_from_ticks 정합)
"LIVE_ALIGN_ENTRY_FROM_TICK_BUFFER", "LIVE_ENTRY_TICK_BUFFER_LIMIT",
# 중분 편입 시 해당 진입봉 시가 매수 보류(다음 분) — 실매·백테·Optuna 공통
"ALIGN_DEFER_MID_BAR_ENROLL",
"BACKTEST_EXIT_CHECKS_PER_BAR",
"SHORT_LIVE_BACKTEST_ALIGN", "SHORT_LIVE_SIGNAL_LOOKBACK_BARS",
"MOMENTUM_LIVE_BACKTEST_ALIGN", "MOMENTUM_LIVE_SIGNAL_LOOKBACK_BARS",
"MOMENTUM_LIVE_ALIGN_USE_FORMING_BAR",
"MOMENTUM_COOLDOWN_USE_CANDLE_FLOOR", "MOMENTUM_COOLDOWN_ENGINE_ONLY",
"MOMENTUM_BACKTEST_SKIP_PRE_SUBSCRIBE",
# MOMENTUM E조건(전일시가) — 실매매 최소 1분봉 수·갭보정 REST 조회량 (약 2영업일)
"MOMENTUM_LIVE_MIN_CANDLES", "MOMENTUM_GAP_FILL_LIMIT",
"MOMENTUM_PREV_DAY_OPEN_HM_MAX",
"BREAKOUT_LIVE_BACKTEST_ALIGN", "BREAKOUT_LIVE_SIGNAL_LOOKBACK_BARS",
"RANGE_BREAK_LIVE_BACKTEST_ALIGN", "RANGE_BREAK_LIVE_SIGNAL_LOOKBACK_BARS",
"BREAKOUT_ENTRY_MODE",
"BREAKOUT_INTRABAR_SLIPPAGE_PCT",
# 유니버스 소스: ranking(기본) | condition
"SCALP_UNIVERSE_SOURCE", "SHORT_UNIVERSE_SOURCE", "BREAKOUT_UNIVERSE_SOURCE",
"RANGE_BREAK_UNIVERSE_SOURCE",
# 랭킹 정렬/크기 — sort: volume|trading_value|strength|fluct_up|fluct_down
"RANK_SCALP_SORT", "RANK_SCALP_LIMIT",
"RANK_SHORT_SORT", "RANK_SHORT_LIMIT",
"RANK_BREAKOUT_SORT", "RANK_BREAKOUT_LIMIT",
"RANK_RANGE_BREAK_SORT", "RANK_RANGE_BREAK_LIMIT",
# 랭킹 폴링 주기(초) — 기본 10초 (모의 TPS 5건/s 대비 안전, 시간당 720회)
"RANKING_POLL_INTERVAL_SEC",
# 조건검색 파라미터
"KIS_HTS_ID", "CONDITION_POLL_INTERVAL_SEC",
# 조건검색 EXIT 유예 시간 — 한 번 빠진 종목을 N초간 universe 에 keep
# (단발성 EXIT/RE-ENTER 회전으로 WS 구독·캔들 데이터가 휘발되는 사이클 차단)
# 0 = 비활성(기존 동작), 권장 60
"CONDITION_EXIT_GRACE_SEC",
# 매도 후 보유구간 1분봉 REST 백필 (백테 봉구멍·슬롯 좀비 방지)
"POST_SELL_CANDLE_BACKFILL",
"POST_SELL_CANDLE_ROLLUP_3M",
"POST_SELL_CANDLE_MAX_BARS",
"POST_SELL_CANDLE_SLEEP_MIN_SEC",
"POST_SELL_CANDLE_SLEEP_MAX_SEC",
"POST_SELL_CANDLE_FORCE_MOCK",
"CONDITION_SCALP_NAME", "CONDITION_SCALP_SEQ",
"CONDITION_SHORT_NAME", "CONDITION_SHORT_SEQ",
"CONDITION_BREAKOUT_NAME", "CONDITION_BREAKOUT_SEQ",
"CONDITION_RANGE_BREAK_NAME", "CONDITION_RANGE_BREAK_SEQ",
# 키움 조건식 seq (선택). UNIVERSE_SOURCE=kiwoom_condition 일 때 CNSRLST name→seq 자동 해결.
"CONDITION_SCALP_KIWOOM_SEQ",
"CONDITION_SHORT_KIWOOM_SEQ",
"CONDITION_BREAKOUT_KIWOOM_SEQ",
"CONDITION_MOMENTUM_KIWOOM_SEQ",
# 키움 전용 이름 — KIS CONDITION_{SID}_NAME 과 다를 때만 설정 (예: SCALP scalp vs scalp_re)
"CONDITION_SCALP_KIWOOM_NAME",
"CONDITION_SHORT_KIWOOM_NAME",
"CONDITION_BREAKOUT_KIWOOM_NAME",
"CONDITION_MOMENTUM_KIWOOM_NAME",
# LS 서버저장조건 전용 이름 (비우면 KIWOOM/CONDITION_NAME 동명)
"CONDITION_SCALP_LS_NAME",
"CONDITION_SHORT_LS_NAME",
"CONDITION_BREAKOUT_LS_NAME",
"CONDITION_MOMENTUM_LS_NAME",
"LS_CONDITION_MANAGER_ENABLED",
"LS_CONDITION_FLUSH_SEC",
# LS HTS 조건 CRUD 후 재시작 없이 이름→index/AFR 재부착
"LS_CONDITION_REMAP_SEC",
"LS_CONDITION_SNAPSHOT_REFRESH_SEC",
"LS_CONDITION_TR_GAP_SEC",
"LS_CONDITION_MISSING_WARN_SEC",
# LS /oauth2/token — expires_in 재사용·최소 재발급 간격
"LS_TOKEN_EXPIRES_IN_DEFAULT",
"LS_TOKEN_REFRESH_MARGIN_SEC",
"LS_TOKEN_MIN_REISSUE_SEC",
# 유니버스 히스토리(백테스트용) — 기본 true, 배치 INSERT 로 DB 부담 최소화
"UNIVERSE_HISTORY_SAVE",
# WS 갭보정 파라미터 (KIS_FALLBACK 은 기본 false — 모의서버 500 폭탄 회피)
"WS_GAP_FILL_OFF_HOURS", "WS_GAP_FILL_LIMIT", "WS_CANDLE_RAM_BUFFER", "WS_TIMEFRAMES",
"WS_GAP_FILL_KIS_FALLBACK",
# 저유동 종목 봉 강제확정 — 다음 체결 틱이 없어도 봉주기 경과 시 확정 처리
# (2026-07-08 원티드랩 14분 무거래 → 신호 인식 지연 사례 대응, false로 즉시 롤백 가능)
"WS_CANDLE_FORCE_CONFIRM_ENABLED",
"WS_CANDLE_FORCE_CONFIRM_GRACE_SEC",
"WS_CANDLE_STALE_CHECK_INTERVAL_SEC",
"WS_GAP_FILL_MAX_RETRIES", "WS_GAP_FILL_RETRY_DELAY_SEC",
# 갭보정 실패 후 전략 force 재큐 차단(초) · 빈응답 전역 회로차단
"WS_GAP_FILL_GIVE_UP_SEC",
"WS_GAP_FILL_EMPTY_CIRCUIT_WINDOW_SEC",
"WS_GAP_FILL_EMPTY_CIRCUIT_MAX",
"WS_GAP_FILL_EMPTY_CIRCUIT_PAUSE_SEC",
"WS_GAP_FILL_EMPTY_LOG_SEC",
"WS_GAP_FILL_ABORT_TF_ON_EMPTY",
"KIWOOM_KA10080_MAX_INFLIGHT",
"KIWOOM_KA10080_RATE_RETRIES",
"KIWOOM_KA10080_RATE_RETRY_BASE_SEC",
"KIWOOM_KA10080_PAGE_SLEEP_SEC",
"WS_GAP_BULK_REFILL_DEBOUNCE_SEC",
"WS_GAP_FILL_PRIORITY_TFS",
"WS_GAP_FILL_TF_SLEEP_MIN_SEC", "WS_GAP_FILL_TF_SLEEP_MAX_SEC",
"WS_GAP_FILL_CODE_SLEEP_MIN_SEC", "WS_GAP_FILL_CODE_SLEEP_MAX_SEC",
"WS_GAP_FILL_PHASE_PAUSE_SEC",
"WS_GAP_FILL_WORKERS",
# 갭보정 REST가 진행 중(미완성) 당일 분봉을 confirmed 에 넣지 않음 (장초 직전봉% 왜곡 방지)
# 2026-07-16 이노테나/한울: 미완성 09:00 이 confirmed → prevChg≈몸통% 로 실매만 통과
"WS_GAP_FILL_SKIP_INCOMPLETE_BUCKET",
# 확정봉(is_confirmed=1) OHLCV 동결 — 갭보정/백필이 volume 등을 키워 실매↔백테 정합 깨는 것 방지
# docs/정합성.md · 기본 true · false 로 즉시 롤백
"WS_CANDLE_FREEZE_ON_CONFIRM",
"WS_CANDLE_FREEZE_DB_LOOKUP_CHUNK",
# 1M REST → RAM 3M 롤업 (꼬리 웜업·3M REST 생략)
"WS_GAP_ROLLUP_3M_FROM_1M",
"WS_GAP_FILL_CANDIDATE_MODE",
# 꼬리 백테: 1M→3M 구멍 보강 / 유니버스 scan_at
"TAIL_BT_SYNTH_3M_FROM_1M",
"TAIL_BACKTEST_UNIVERSE_SCAN_AT",
"TAIL_BT_INCLUSION_LOOKBACK",
# 꼬리 백테 전일 웜업 (실매 get_candles n=50 / SHORT_GAP_FILL_LIMIT 150 정합)
"TAIL_BACKTEST_CANDLE_WARMUP_BARS",
"TAIL_BACKTEST_REST_WARMUP",
"TAIL_BACKTEST_REST_WARMUP_BARS",
"TAIL_BACKTEST_REST_SLEEP_SEC",
"TAIL_BACKTEST_REST_MAX_CODES",
"TAIL_UNIVERSE_EXIT_DEBOUNCE_SEC",
"KIWOOM_CNSRREQ_GAP_MIN_SEC", "KIWOOM_CNSRREQ_GAP_MAX_SEC",
"KIWOOM_CNSRREQ_MAX_RETRIES", "KIWOOM_CNSRREQ_RETRY_SEC",
# true(기본)=실매 소스가 ls_condition 이어도 키움 조건 → target_candidates_history 적재
"KIWOOM_CONDITION_DUAL_HISTORY",
# 키움 Bye 후 조건검색 유령등록(900003) 방지: REQ 전 CLR + settle
"KIWOOM_CNSRCLR_BEFORE_REQ",
"KIWOOM_CNSRCLR_GAP_MIN_SEC", "KIWOOM_CNSRCLR_GAP_MAX_SEC",
"KIWOOM_CNSRCLR_SETTLE_SEC",
# 종목당 최대 매수금액 하드캡 (손절%가 작으면 포지션 사이즈 폭주 방지)
"MAX_BUY_AMOUNT_PER_STOCK",
"SCALP_MAX_BUY_AMOUNT", "SHORT_MAX_BUY_AMOUNT", "TAIL_MAX_BUY_AMOUNT",
# KIS REST 안정화 파라미터
"KIS_MIN_INTERVAL_SEC", "KIS_REST_MAX_RETRIES",
"KIS_REST_BACKOFF_CAP_SEC", "KIS_REST_TIMEOUT_SEC",
# 도메인별 REST 최소 호출 간격 (한투 유량: 실전 18건/초, 모의 1건/초)
"KIS_MIN_INTERVAL_SEC_MOCK", "KIS_MIN_INTERVAL_SEC_REAL",
# 해외주식 주문/시세 (미국) — 표시·실주문 공통
"KIS_OVRS_DEFAULT_EXCG",
"KIS_OVRS_ORD_DVSN",
"KIS_OVRS_ORD_SVR_DVSN_CD",
"KIS_OVRS_ORDER_PATH",
"KIS_OVRS_US_BUY_TR_ID",
"KIS_OVRS_US_SELL_TR_ID",
"KIS_OVRS_PRICE_DECIMALS",
"KIS_OVRS_BUY_LIMIT_SLIPPAGE_PCT",
"KIS_OVRS_SELL_LIMIT_SLIPPAGE_PCT",
"KIS_OVRS_TEST_SYMBOL",
"KIS_OVRS_TEST_EXCHANGE",
# 스캔 루프 회전율 sleep (REST 유량과 별개 · WS 따라가기 속도 · 비우면 전략별 기본값)
"SCAN_REJECT_SLEEP_MIN", "SCAN_REJECT_SLEEP_MAX",
"SCAN_BUY_OK_SLEEP_MIN", "SCAN_BUY_OK_SLEEP_MAX",
"SCAN_BUY_FAIL_SLEEP_MIN", "SCAN_BUY_FAIL_SLEEP_MAX",
"STRATEGY_LOOP_SLEEP_MIN", "STRATEGY_LOOP_SLEEP_MAX",
# 루프 숙제별 ms 계측 (기본 OFF · 병목 확정용 · 신호식 불변)
"LOOP_PROFILE_ENABLED",
"LOOP_PROFILE_EVERY_N",
"LOOP_PROFILE_LOG_PATH",
"CHECK_BUY_PROF_MIN_MS",
# 보유 RAM 하이브리드 — 이벤트 갱신 + N초마다만 get_active_trades (0=매루프 레거시)
"HOLDINGS_DB_SYNC_INTERVAL_SEC",
# 당일 trade_history / 유니버스슬롯 history — RAM TTL(초). 0=캐시 끔(매번 DB)
"TODAY_TRADES_CACHE_TTL_SEC",
"UNIVERSE_SLOT_CACHE_TTL_SEC",
"UNIVERSE_SLOT_FLIGHT_WAIT_SEC",
# 전략 config_* RAM TTL — reload 매루프 SELECT 완화 (유니버스 분슬롯 금지와 별개 · 기본 1초)
"STRATEGY_ENV_CACHE_TTL_SEC",
# 틱매도 — WS 현재가 갱신 시 매도 검사 (기본 false · 루프 매도 유지 + 틱 보강)
"TICK_SELL_ENABLED", "TICK_SELL_MIN_INTERVAL_MS", "TICK_SELL_ASYNC_PLACE",
"SHORT_TICK_SELL_ENABLED", "SCALP_TICK_SELL_ENABLED",
"BREAKOUT_TICK_SELL_ENABLED", "MOMENTUM_TICK_SELL_ENABLED",
"RANGE_BREAK_TICK_SELL_ENABLED", "UPDOW_TICK_SELL_ENABLED",
"DBBAND_TICK_SELL_ENABLED",
# ── 시장 급락 서킷브레이커 (KOSPI/KOSDAQ 지수 폭락 시 신규 매수 차단) ──
# 거래소 공식 서킷브레이커는 -8% 이지만, 봇 보호용으로 더 빨리(-2%) 반응.
# PANIC 모드 시 매수만 차단, 매도는 평소처럼 동작 (포지션 정리 가능).
"MARKET_GUARD_ENABLED", # 활성화 토글 (true/false). 백테스트 못하므로 운영 후 켜기
"MARKET_GUARD_5MIN_DROP_PCT", # 5분 내 -N% 하락 시 PANIC 진입 (기본 2.0)
"MARKET_GUARD_DAILY_DROP_PCT", # 일중 누적 -N% 하락 시 PANIC 진입 (기본 3.0)
"MARKET_GUARD_RECOVERY_PCT", # PANIC 중 5분 +N% 반등 시 자동 해제 (기본 1.0)
"MARKET_GUARD_INDEX_CODE", # 감시 지수 ("0001"=KOSPI, "1001"=KOSDAQ, "both"=둘 다)
"MARKET_GUARD_POLL_SEC", # 지수 폴링 주기 초 (기본 30)
"MARKET_GUARD_INDEX_CANDLE_KEEP_MIN", # 재시작 시 ws_candles MG* 1분봉 복원 개수 (기본 20)
"MARKET_GUARD_PERSIST_STATE", # kv_store PANIC 영속 (기본 true)
# ── 전략별 후보 하드캡 (키움 WS 100 + 매수 체크 회전율) ─
# cond/ranking 이 폭주해도 상위 N개만. 0=무제한.
# 키움 100 ≈ 후보합 + 보유 + grace. 영구 KR은 LS. 한투 41은 영구 전용이 아님(기본=후보도 구독).
"SCALP_CAND_LIMIT", "SHORT_CAND_LIMIT", "BREAKOUT_CAND_LIMIT",
"RANGE_BREAK_CAND_LIMIT",
# ── 전략별 매매 시간대 (HHMM 정수, 930=09:30, 1500=15:00) ──────────
# 꼬리(SHORT): config_short 의 TAIL_TIME_START/TAIL_TIME_END (SHORT_TIME_* 레거시 제거)
# 그 외 미설정 시 글로벌 TIME_START/TIME_END → 기본 900~1530.
"SCALP_TIME_START", "SCALP_TIME_END",
"BREAKOUT_TIME_START", "BREAKOUT_TIME_END",
# ── BREAKOUT(돌파) — HTS SCAN(널넬) + 봇 TRIGGER(엄격) ─────────────────
"BREAKOUT_LOOKBACK_MIN", "BREAKOUT_VOL_WIN", "BREAKOUT_VOL_MULT",
"BREAKOUT_PREV_CHG_MIN", "BREAKOUT_PREV_CHG_MAX",
"BREAKOUT_STOP_LOSS_PCT", "BREAKOUT_TAKE_PROFIT_PCT", "BREAKOUT_TRAIL_PCT",
"BREAKOUT_TRAIL_ARM_PCT",
"BREAKOUT_SHOULDER_MIN_HIGH_PCT", "BREAKOUT_SHOULDER_CUT_PCT",
"BREAKOUT_RATCHET_TIERS", "BREAKOUT_MAX_HOLD_BARS",
"BREAKOUT_EOD_ENABLED", "BREAKOUT_EOD_HM",
"MOMENTUM_EOD_ENABLED", "MOMENTUM_EOD_HM",
"TAIL_EOD_ENABLED", "TAIL_EOD_HM",
# 스캘핑 EOD — 실매 기존 15:25 장마감청산과 동일 (strategy_eod SCALP)
"SCALP_EOD_ENABLED", "SCALP_EOD_HM",
# ATR 동적 손절 (sl_mode='atr' 일 때만 활성, 기본 fixed=기존 고정%)
"BREAKOUT_SL_MODE", "BREAKOUT_ATR_PERIOD", "BREAKOUT_ATR_SL_MULT",
"BREAKOUT_ATR_SL_MIN_PCT", "BREAKOUT_ATR_SL_MAX_PCT",
"BREAKOUT_GOLDEN_END_HM",
"BREAKOUT_GRID_COARSE_SHOULDER_SMIN", "BREAKOUT_GRID_COARSE_SHOULDER_CUT",
"BREAKOUT_GRID_FINE_SHOULDER_SMIN", "BREAKOUT_GRID_FINE_SHOULDER_CUT",
"BREAKOUT_GRID_FULL_SHOULDER_SMIN", "BREAKOUT_GRID_FULL_SHOULDER_CUT",
"BREAKOUT_MAX_DAILY_CHG", "BREAKOUT_MIN_PRICE",
"BREAKOUT_MIN_BAR_TRADE_VALUE_KRW",
"BREAKOUT_MIN_TURNOVER_1M_PCT", "BREAKOUT_SHARE_DENOM", "STOCK_SHARE_DENOM",
"BREAKOUT_SKIP_HTS_SCAN_DUPES",
# 가짜돌파(휩쏘) 필터 — 0=OFF. Optuna/웹 apply 시 config_breakout 컬럼 필요
"BREAKOUT_CONFIRM_MARGIN_PCT", "BREAKOUT_BODY_MIN_PCT",
"BREAKOUT_GRID_FAST_SKIP_HTS_SCAN_DUPES",
"BREAKOUT_USE_EMA_FILTER", "BREAKOUT_EMA_FAST_PERIOD", "BREAKOUT_EMA_SLOW_PERIOD",
"BREAKOUT_MAX_LOSS_PER_TRADE_KRW", "BREAKOUT_SLOT_MONEY",
"BREAKOUT_MAX_DAILY", "BREAKOUT_COOLDOWN_SEC",
"BREAKOUT_ENTRY_MODE", "BREAKOUT_INTRABAR_SLIPPAGE_PCT",
"BREAKOUT_TOTAL_BUDGET_KRW", "BREAKOUT_MAX_STOCKS",
"BREAKOUT_MAX_BUY_AMOUNT", "BREAKOUT_MIN_INVEST_RATIO_OF_SLOT",
# ── RANGE_BREAK(박스권 돌파) — HTS momentum SCAN + 봇 TRIGGER ─────────────
"RANGE_BREAK_BOX_LOOKBACK_MIN", "RANGE_BREAK_BOX_MAX_WIDTH_PCT",
"RANGE_BREAK_BOX_MIN_WIDTH_PCT", "RANGE_BREAK_SETUP_VOL_MAX_MULT",
"RANGE_BREAK_SETUP_BEAR_BARS_MIN", "RANGE_BREAK_VOL_MULT", "RANGE_BREAK_VOL_WIN",
"RANGE_BREAK_VOL_BASELINE_WIN", "RANGE_BREAK_BREAK_MARGIN_PCT", "RANGE_BREAK_BODY_MIN_PCT",
"RANGE_BREAK_TIME_START", "RANGE_BREAK_TIME_END_HM",
"RANGE_BREAK_STOP_LOSS_PCT", "RANGE_BREAK_TAKE_PROFIT_PCT", "RANGE_BREAK_TRAIL_PCT",
"RANGE_BREAK_TRAIL_ARM_PCT", "RANGE_BREAK_SHOULDER_MIN_HIGH_PCT", "RANGE_BREAK_SHOULDER_CUT_PCT",
"RANGE_BREAK_MAX_HOLD_BARS", "RANGE_BREAK_MAX_DAILY", "RANGE_BREAK_COOLDOWN_SEC",
"RANGE_BREAK_MAX_DAILY_CHG", "RANGE_BREAK_MIN_PRICE", "RANGE_BREAK_HIGH_CHASE_THR",
"RANGE_BREAK_USE_HIGH_CHASE_FILTER", "RANGE_BREAK_MAX_LOSS_PER_TRADE_KRW",
"RANGE_BREAK_SLOT_MONEY",
"RANGE_BREAK_MAX_BUY_AMOUNT", "RANGE_BREAK_MIN_INVEST_RATIO_OF_SLOT",
"RANGE_BREAK_FORCE_EOD_EXIT",
# ETN/ETF/스팩/우선주 자동 제외 토글 (기본 true) — 후보 매니저 결과를 한 번 더 거른다.
"EXCLUDE_NON_STOCK",
# 우선주 코드 끝자리(숫자) — name==code 일 때도 차단. KRX: 5·7·9 + 알파벳(K/L/M…)
"NON_STOCK_PREF_CODE_SUFFIXES",
"NON_STOCK_EXCLUDE_ALPHA_SUFFIX",
# ETF 등 이름 키워드(콤마). 비우면 코드 기본목록. '파워' 단독은 넣지 말 것(오탐).
"NON_STOCK_NAME_KEYWORDS",
# 봉부족 시 force 갭/로그 스로틀 (특수코드 REST·로그 폭주 방지)
"BREAKOUT_CANDLE_GAP_FORCE_SEC",
"BREAKOUT_CANDLE_SHORT_LOG_SEC",
"SCALP_CANDLE_GAP_FORCE_SEC",
"SCALP_CANDLE_SHORT_LOG_SEC",
# SCALP 일일 진입 횟수
"SCALP_MAX_DAILY",
# ── MOMENTUM 전략 (SCALP reversal 과 분리된 독립 전략) ─────────────
# 키움 'momentum' 조건검색(E∧F∧H∧I). env 키는 MOMENTUM_* 전용.
"STRATEGY_MOMENTUM_ENABLED",
# 해외 모멘텀 (영구구독 US · dry-run 기본)
"STRATEGY_US_MOMENTUM_ENABLED",
"US_MOMENTUM_TIME_START",
"US_MOMENTUM_TIME_END",
"US_MOMENTUM_SESSION_START",
"US_MOMENTUM_BUY_WAIT_LOG_SEC",
"US_MOMENTUM_SELL_TIME_END",
"US_MOMENTUM_DRY_RUN",
"US_MOMENTUM_MAX_STOCKS",
"US_MOMENTUM_SLOT_MONEY",
"US_MOMENTUM_FEE_RATE",
"US_MOMENTUM_SELL_TAX",
"US_MOMENTUM_FX_FEE_RATE",
"US_MOMENTUM_MIN_PRICE",
"US_MOMENTUM_LIVE_MIN_CANDLES",
"US_MOMENTUM_EOD_ENABLED",
"US_MOMENTUM_EOD_HM",
"US_MOMENTUM_MAX_BUY_AMOUNT",
"US_MOMENTUM_TOTAL_BUDGET",
"US_MOMENTUM_MAX_ORDER_QTY",
"US_MOMENTUM_FORCE_BUY_QTY",
"US_MOMENTUM_RELOAD_INTERVAL_SEC",
"US_MOMENTUM_PSAMOUNT_CACHE_SEC",
"US_MOMENTUM_TRADES_CACHE_SEC",
"US_MOMENTUM_LOOP_SLEEP_MIN",
"US_MOMENTUM_LOOP_SLEEP_MAX",
"US_MOMENTUM_SCAN_REJECT_SLEEP_MIN",
"US_MOMENTUM_SCAN_REJECT_SLEEP_MAX",
# 해외 모멘텀 TRIGGER/청산 (국내 MOMENTUM_* 와 분리 저장)
"US_MOMENTUM_RSI_MIN",
"US_MOMENTUM_RSI_MAX",
"US_MOMENTUM_VOL_MULT",
"US_MOMENTUM_VOL_WIN",
"US_MOMENTUM_MAX_FROM_OPEN_PCT",
"US_MOMENTUM_MIN_FROM_OPEN_PCT",
"US_MOMENTUM_STOP_LOSS_PCT",
"US_MOMENTUM_TAKE_PROFIT_PCT",
"US_MOMENTUM_TP_MAX_PCT",
"US_MOMENTUM_SHOULDER_MIN_HIGH_PCT",
"US_MOMENTUM_SHOULDER_CUT_PCT",
"US_MOMENTUM_RATCHET_TIERS",
"US_MOMENTUM_TRAIL_PCT",
"US_MOMENTUM_TRAIL_ARM_PCT",
"US_MOMENTUM_MAX_HOLD_BARS",
"US_MOMENTUM_MAX_DAILY",
"US_MOMENTUM_MAX_DAILY_CHG",
"US_MOMENTUM_HIGH_CHASE_THR",
"US_MOMENTUM_USE_DEFENSE_FILTERS",
"US_MOMENTUM_USE_HIGH_CHASE_FILTER",
"US_MOMENTUM_USE_DAILY_RANGE_FILTER",
"US_MOMENTUM_USE_EMA_FILTER",
"US_MOMENTUM_EMA_FAST_PERIOD",
"US_MOMENTUM_EMA_SLOW_PERIOD",
"US_MOMENTUM_USE_RSI_MAX_FILTER",
"US_MOMENTUM_PATTERN_BREAKOUT",
"US_MOMENTUM_PATTERN_PULLBACK",
"US_MOMENTUM_CHASE_LOOKBACK_MIN",
"US_MOMENTUM_PULLBACK_LOOKBACK_MIN",
"US_MOMENTUM_PULLBACK_MIN_PCT",
"US_MOMENTUM_PULLBACK_MAX_PCT",
"US_MOMENTUM_SETUP_VOL_MAX_MULT",
"US_MOMENTUM_SETUP_BEAR_BARS_MIN",
"US_MOMENTUM_MAX_LOSS_PER_TRADE",
"US_MOMENTUM_MIN_PROFIT_PCT",
"US_MOMENTUM_COOLDOWN_SEC",
"US_MOMENTUM_DAILY_PROFIT_TARGET_ENABLED",
"US_MOMENTUM_DAILY_PROFIT_TRAIL_TIERS",
"US_MOMENTUM_DAILY_PROFIT_MODE",
"KIS_US_MOMENTUM_MM_CHANNEL",
"US_MOMENTUM_DAILY_PROFIT_RISK_BUDGET_ENABLED",
"US_MOMENTUM_DAILY_PROFIT_RISK_BUDGET_COOLDOWN_SEC",
# MM 체결 알림 alias (mm_config.json channels 키) → config_momentum / config_breakout / config_updow
"KIS_MOMENTUM_MM_CHANNEL",
"KIS_BREAKOUT_MM_CHANNEL",
"KIS_RANGE_BREAK_MM_CHANNEL",
"KIS_UPDOW_MM_CHANNEL",
"MOMENTUM_UNIVERSE_SOURCE",
"RANK_MOMENTUM_SORT", "RANK_MOMENTUM_LIMIT",
"CONDITION_MOMENTUM_NAME", "CONDITION_MOMENTUM_SEQ",
"MOMENTUM_CAND_LIMIT",
"MOMENTUM_TIME_START", "MOMENTUM_TIME_END",
"MOMENTUM_MAX_STOCKS",
"MOMENTUM_MAX_BUY_AMOUNT",
"MOMENTUM_MAX_DAILY",
"MOMENTUM_MAX_DAILY_CHG",
"MOMENTUM_RSI_MIN", "MOMENTUM_RSI_MAX",
"MOMENTUM_VOL_MULT", "MOMENTUM_VOL_WIN",
"MOMENTUM_TIME_END_HM",
# 시가 대비 등락(%) 컷 — 파라서치 coarse 그리드 mom_max_from_open_pct 와 동일 의미
"MOMENTUM_MAX_FROM_OPEN_PCT", "MOMENTUM_MIN_FROM_OPEN_PCT",
"MOMENTUM_SKIP_HTS_SCAN_DUPES",
"MOMENTUM_GRID_FAST_SKIP_HTS_SCAN_DUPES",
"MOMENTUM_STOP_LOSS_PCT", "MOMENTUM_TAKE_PROFIT_PCT",
"MOMENTUM_TP_MAX_PCT",
"MOMENTUM_SHOULDER_MIN_HIGH_PCT", "MOMENTUM_SHOULDER_CUT_PCT",
"MOMENTUM_MIN_DROP_RATE",
"MOMENTUM_SLOT_MONEY",
"MOMENTUM_MIN_PRICE",
"MOMENTUM_MAX_LOSS_PER_TRADE_KRW",
"MOMENTUM_MIN_DROP_PCT_FOR_LOSS_CUT",
"MOMENTUM_MIN_PROFIT_PCT",
"MOMENTUM_RATCHET_TIERS",
"MOMENTUM_TRAIL_PCT",
"MOMENTUM_TRAIL_ARM_PCT",
"MOMENTUM_MAX_HOLD_BARS",
"MOMENTUM_USE_DEFENSE_FILTERS",
"MOMENTUM_USE_HIGH_CHASE_FILTER",
"MOMENTUM_USE_DAILY_RANGE_FILTER",
"MOMENTUM_USE_EMA_FILTER",
"MOMENTUM_EMA_FAST_PERIOD",
"MOMENTUM_EMA_SLOW_PERIOD",
"MOMENTUM_USE_RSI_MAX_FILTER",
"MOMENTUM_PATTERN_BREAKOUT",
"MOMENTUM_PATTERN_PULLBACK",
"MOMENTUM_TRIGGER_E_CONFIRM",
"MOMENTUM_TRIGGER_REQUIRE_BULL_BAR",
"MOMENTUM_USE_VOL_TRIGGER",
"MOMENTUM_USE_RSI_FILTER",
"MOMENTUM_E_MIN_CHG_PCT",
"MOMENTUM_CHASE_LOOKBACK_MIN",
"MOMENTUM_PULLBACK_LOOKBACK_MIN",
"MOMENTUM_PULLBACK_MIN_PCT",
"MOMENTUM_PULLBACK_MAX_PCT",
"MOMENTUM_SETUP_VOL_MAX_MULT",
"MOMENTUM_SETUP_BEAR_BARS_MIN",
"MOMENTUM_HIGH_CHASE_THR",
"MOMENTUM_RSI_PERIOD",
"MOMENTUM_COOLDOWN_SEC",
"MOMENTUM_MIN_HOLD_SEC",
"MOMENTUM_FORCE_EOD_EXIT",
# ── UPDOW 전략 (직전봉 몸통 하락 → 다음봉 시가, kis_trader 독립 쓰레드) ─
"UPDOW_UNIVERSE_SOURCE",
"RANK_UPDOW_SORT", "RANK_UPDOW_LIMIT",
"CONDITION_UPDOW_NAME", "CONDITION_UPDOW_SEQ",
"UPDOW_CAND_LIMIT",
"UPDOW_TIME_START", "UPDOW_TIME_END",
"UPDOW_REENTRY_COOLDOWN_SEC",
"UPDOW_MIN_CANDLE_LEN", "MIN_CANDLE_LEN_UPDOW",
"UPDOW_CANDLE_FETCH_N",
"UPDOW_LIVE_MIN_CANDLES",
"UPDOW_MAX_LOSS_PER_TRADE_KRW",
"UPDOW_MAX_BUY_AMOUNT",
# ── 전략별 동시 보유 종목 한도 (기존 글로벌 MAX_STOCKS 분리) ─────────
# 미설정(0/빈값) 시 글로벌 MAX_STOCKS 로 폴백 → 구버전 호환.
# 권장: 합계 ≤ MAX_STOCKS (계좌 슬롯 분산), 예: 3+2+2=7.
"SCALP_MAX_STOCKS", "SHORT_MAX_STOCKS", "BREAKOUT_MAX_STOCKS",
"RANGE_BREAK_MAX_STOCKS", "UPDOW_MAX_STOCKS",
# 전략별 시각순 포트폴리오 총 운용한도 (동시 보유 매입금 합 상한)
"SHORT_TOTAL_BUDGET_KRW",
"SCALP_TOTAL_BUDGET_KRW",
"MOMENTUM_TOTAL_BUDGET_KRW",
"BREAKOUT_TOTAL_BUDGET_KRW",
"RANGE_BREAK_TOTAL_BUDGET_KRW",
"UPDOW_TOTAL_BUDGET_KRW",
# 일일 익절 목표 (서브·전략별) — split_env_keys → config_{strategy}
"SCALP_DAILY_PROFIT_TARGET_ENABLED",
"SCALP_DAILY_PROFIT_TARGET_KRW",
"SCALP_DAILY_PROFIT_TARGET_PCT",
"SCALP_DAILY_PROFIT_HALT_NEW_BUYS",
"SCALP_DAILY_PROFIT_MODE", "SCALP_DAILY_PROFIT_TRAIL_TIERS",
"SCALP_DAILY_PROFIT_TRAIL_DROP_PCT", "SCALP_DAILY_PROFIT_TRAIL_ARM_KRW", "SCALP_DAILY_PROFIT_TRAIL_ARM_PCT",
"SCALP_DAILY_PROFIT_RISK_BUDGET_ENABLED", "SCALP_DAILY_PROFIT_RISK_BUDGET_COOLDOWN_SEC",
"SHORT_DAILY_PROFIT_TARGET_ENABLED",
"SHORT_DAILY_PROFIT_TARGET_KRW",
"SHORT_DAILY_PROFIT_TARGET_PCT",
"SHORT_DAILY_PROFIT_HALT_NEW_BUYS",
"SHORT_DAILY_PROFIT_MODE", "SHORT_DAILY_PROFIT_TRAIL_TIERS",
"SHORT_DAILY_PROFIT_TRAIL_DROP_PCT", "SHORT_DAILY_PROFIT_TRAIL_ARM_KRW", "SHORT_DAILY_PROFIT_TRAIL_ARM_PCT",
"SHORT_DAILY_PROFIT_RISK_BUDGET_ENABLED", "SHORT_DAILY_PROFIT_RISK_BUDGET_COOLDOWN_SEC",
"MOMENTUM_DAILY_PROFIT_TARGET_ENABLED",
"MOMENTUM_DAILY_PROFIT_TARGET_KRW",
"MOMENTUM_DAILY_PROFIT_TARGET_PCT",
"MOMENTUM_DAILY_PROFIT_HALT_NEW_BUYS",
"MOMENTUM_DAILY_PROFIT_MODE", "MOMENTUM_DAILY_PROFIT_TRAIL_TIERS",
"MOMENTUM_DAILY_PROFIT_TRAIL_DROP_PCT", "MOMENTUM_DAILY_PROFIT_TRAIL_ARM_KRW", "MOMENTUM_DAILY_PROFIT_TRAIL_ARM_PCT",
"MOMENTUM_DAILY_PROFIT_RISK_BUDGET_ENABLED", "MOMENTUM_DAILY_PROFIT_RISK_BUDGET_COOLDOWN_SEC",
"BREAKOUT_DAILY_PROFIT_TARGET_ENABLED",
"BREAKOUT_DAILY_PROFIT_TARGET_KRW",
"BREAKOUT_DAILY_PROFIT_TARGET_PCT",
"BREAKOUT_DAILY_PROFIT_HALT_NEW_BUYS",
"BREAKOUT_DAILY_PROFIT_MODE", "BREAKOUT_DAILY_PROFIT_TRAIL_TIERS",
"BREAKOUT_DAILY_PROFIT_TRAIL_DROP_PCT", "BREAKOUT_DAILY_PROFIT_TRAIL_ARM_KRW", "BREAKOUT_DAILY_PROFIT_TRAIL_ARM_PCT",
"BREAKOUT_DAILY_PROFIT_RISK_BUDGET_ENABLED", "BREAKOUT_DAILY_PROFIT_RISK_BUDGET_COOLDOWN_SEC",
"RANGE_BREAK_DAILY_PROFIT_TARGET_ENABLED",
"RANGE_BREAK_DAILY_PROFIT_TARGET_KRW",
"RANGE_BREAK_DAILY_PROFIT_TARGET_PCT",
"RANGE_BREAK_DAILY_PROFIT_HALT_NEW_BUYS",
"RANGE_BREAK_DAILY_PROFIT_MODE", "RANGE_BREAK_DAILY_PROFIT_TRAIL_TIERS",
"RANGE_BREAK_DAILY_PROFIT_TRAIL_DROP_PCT", "RANGE_BREAK_DAILY_PROFIT_TRAIL_ARM_KRW", "RANGE_BREAK_DAILY_PROFIT_TRAIL_ARM_PCT",
"RANGE_BREAK_DAILY_PROFIT_RISK_BUDGET_ENABLED", "RANGE_BREAK_DAILY_PROFIT_RISK_BUDGET_COOLDOWN_SEC",
"UPDOW_DAILY_PROFIT_TARGET_ENABLED",
"UPDOW_DAILY_PROFIT_TARGET_KRW",
"UPDOW_DAILY_PROFIT_TARGET_PCT",
"UPDOW_DAILY_PROFIT_HALT_NEW_BUYS",
"UPDOW_DAILY_PROFIT_MODE", "UPDOW_DAILY_PROFIT_TRAIL_TIERS",
"UPDOW_DAILY_PROFIT_TRAIL_DROP_PCT", "UPDOW_DAILY_PROFIT_TRAIL_ARM_KRW", "UPDOW_DAILY_PROFIT_TRAIL_ARM_PCT",
"UPDOW_DAILY_PROFIT_RISK_BUDGET_ENABLED", "UPDOW_DAILY_PROFIT_RISK_BUDGET_COOLDOWN_SEC",
"DBBAND_DAILY_PROFIT_TARGET_ENABLED",
"DBBAND_DAILY_PROFIT_TARGET_KRW",
"DBBAND_DAILY_PROFIT_TARGET_PCT",
"DBBAND_DAILY_PROFIT_HALT_NEW_BUYS",
"DBBAND_DAILY_PROFIT_MODE", "DBBAND_DAILY_PROFIT_TRAIL_TIERS",
"DBBAND_DAILY_PROFIT_TRAIL_DROP_PCT", "DBBAND_DAILY_PROFIT_TRAIL_ARM_KRW", "DBBAND_DAILY_PROFIT_TRAIL_ARM_PCT",
"DBBAND_DAILY_PROFIT_RISK_BUDGET_ENABLED", "DBBAND_DAILY_PROFIT_RISK_BUDGET_COOLDOWN_SEC",
# ── 시세 WS 스위치 (역할이 다름 — 섞지 말 것) ───────────────────
# WS_PROVIDER: 구식 마이그레이션 토글(kis_only / kis_with_validation / kiwoom_*).
# 지금은 키움 WS 기동이 이 값에 묶이지 않음(_start_ws_validator: 키 있으면 기동).
# 교차검증 테이블은 LIVE_VALIDATOR_ENABLED.
"WS_PROVIDER",
# MINIMAL = 한투 소켓에 *누구를 넣을지* (LIVE_TICK_PROVIDER 와 무관).
# true: 한투 WS=보유만, 후보=키움. false(기본): 한투도 후보∪보유(세션 41).
# 영구구독 KR은 LS. 메인을 한투로 바꾼다고 MINIMAL 명단이 바뀌지 않음.
"WS_SUBSCRIBE_KIS_MINIMAL",
# 구독 한도초과·실패·WS 미연결 시 즉시 벤더 spill (kis↔kiwoom→ls). 기본 true. 기존 LIVE_* 값 변경 없음.
"WS_SUBSCRIBE_SPILL",
# csv 예: kis,kiwoom,ls — 비우면 LIVE_TICK_PROVIDER 1차 + 나머지 + ls(3차)
"WS_TICK_SUBSCRIBE_CHAIN",
"WS_OB_SUBSCRIBE_CHAIN",
# LS WS 마스터 기동 (spill 3차·영구 RAM). false면 VALIDATION/HISTORY/ls_condition 만으로 기동.
"LS_WS_ENABLED",
# LS 로컬 구독 상한. 0=로컬한도 없음. spill 3차용 (기본 100).
"LS_WS_MAX_SUBSCRIPTIONS",
# 검증 비교 주기(초) — 너무 짧으면 부하, 너무 길면 표본 부족. 기본 5.
"WS_VALIDATION_INTERVAL_SEC",
# 차이 경고 임계(%). |diff| 가 이 값 이상이면 WARN 로그. 기본 0.10.
"WS_VALIDATION_DIFF_WARN_PCT",
# 키움 WS 시세는 KIS_MOCK 와 무관하게 항상 실키/실전 도메인 사용.
# 모의 도메인은 장외 데이터 제공 X / 실시간 표본도 빈약 → 시세 비교/마이그레이션 가치 없음.
# true (기본) = 실키 강제, false = KIS_MOCK 따라 자동 (디버그용).
"KIWOOM_WS_FORCE_REAL",
# ── LS 시세 그림자 검증 (실매 경로 미사용) ─────────────────────────
# true 시 LS 실키 WS 기동 → ls_ws_ticks / ls_ws_candles 적재 + ws_price_validation_ls
"LS_WS_VALIDATION_ENABLED",
# LS 조건 이력만 수집 (실매 UNIVERSE_SOURCE 와 분리). true → ls_candidates_history
"LS_CONDITION_HISTORY_ENABLED",
# true → LS 조건 유니버스 diff 를 LS WS 구독(owner=condition)에 반영
"LS_WS_FOLLOW_CONDITION_UNIVERSE",
# true → 조건 구독 시 6자리 숫자 종목만 (우선주/특이코드 스킵)
"LS_WS_CONDITION_PLAIN6_ONLY",
"LS_WS_FORCE_REAL",
# 후보보유영구grace 구독 중이면 ls_ws_ticks INSERT (호가와 동일 게이트).
# INSERT 만 게이트. 호가 틱동기(LS_WS_ORDERBOOK_SAVE) 콜백을 같이 끄지 말 것.
"LS_WS_TICK_SAVE",
# 백테/옵투나: ws_ticks 후 빈 초에 ls_ws_ticks 3차 (나이=LIVE_FEED_FALLBACK)
"BT_TICK_LS_THIRD_FALLBACK",
"LS_WS_CANDLE_SAVE",
# true=LS 틱을 ws_ticks(source=ls) 에도 이중 저장 (용량↑). 기본 false — ls_ws_ticks 만
"LS_WS_TICK_MIRROR_WS_TICKS",
"LS_WS_CANDLE_TF_MIN",
# ls_condition 전략 갭보정 (t8412) — 키움 ka10080 과 분리
"LS_GAP_FILL_ENABLED",
"LS_GAP_FILL_LIMIT",
"LS_GAP_FILL_INTERVAL_SEC",
"LS_GAP_FILL_MAX_RETRIES",
"LS_GAP_FILL_TIMEOUT_SEC",
"LS_GAP_FILL_FAIL_MAX",
"LS_GAP_FILL_WORKERS",
"LS_WS_CONFIRMED_MAX",
"LS_WS_VALIDATION_INTERVAL_SEC",
"LS_WS_VALIDATION_DIFF_WARN_PCT",
"LS_WS_ALSO_HOGA", # UH1 구독. OFF면 호가 RAM·적재 없음. 코드기본 true.
# LS 호가(UH1) → ls_ws_orderbook 적재 (프로그램 UPH 는 미사용·미구독)
"LS_WS_ORDERBOOK_SAVE",
# 종목당 DB INSERT 최소 간격(ms) — 호가 폭주 시 DB 부하 완화 (RAM 캐시는 매 틱)
"LS_WS_ORDERBOOK_SAVE_MS",
# ls_ws_ticks / ls_ws_orderbook 보존 일수 (기본 7)
"LS_WS_TICK_KEEP_DAYS",
"LS_WS_ORDERBOOK_KEEP_DAYS",
# LS VI(UVI) 구독·적재 — 워치독 오탐 방지·백테 구간 마킹 (프로그램·상하한가 미포함)
"LS_WS_UVI_ENABLED",
"LS_WS_VI_SAVE",
# VI 해제 누락 시 RAM 만료(초). 통상 VI ~2분 → 기본 180
"LS_WS_VI_STALE_SEC",
# true=구독 KR이 전부 VI 중이면 틱 silence 강제재연결 스킵
"LS_WS_WATCHDOG_SKIP_WHEN_VI",
"LS_WS_PING_INTERVAL_SEC",
"LS_WS_PING_TIMEOUT_SEC",
"LS_WS_OPEN_REG_DELAY_MS",
"LS_WS_REG_GAP_MS",
# Bye/조기 CLOSE: 안정 OPEN(초) 미만이면 early. streak 도달 시 토큰 갱신+회로 대기
"LS_WS_STABLE_OPEN_SEC",
"LS_WS_EARLY_BYE_STREAK",
"LS_WS_BYE_CIRCUIT_SLEEP_SEC",
# 봇 종료 시 UNREG 동기 전송 (재시작 세션 꼬임 완화). 타임아웃 후 TCP close
"LS_WS_STOP_UNREG_TIMEOUT_SEC",
"LS_WS_STOP_UNREG_GAP_MS",
# s3k3=콘솔 S3_/K3_+6자리(실험), us3(기본)=통합 US3 — 실측 장중·장후 틱은 us3
"LS_WS_TR_MODE",
"LS_WS_WATCHDOG_ENABLED",
"LS_WS_WATCHDOG_SILENCE_SEC",
"LS_WS_WATCHDOG_OPEN_GRACE_SEC",
"LS_WS_WATCHDOG_POLL_SEC",
# 틱 없이 연속 강제재연결 N회 후 silence 배수 백오프 (장외·TR불일치 폭주 방지)
"LS_WS_WATCHDOG_EMPTY_BACKOFF_AFTER",
"LS_WS_WATCHDOG_EMPTY_BACKOFF_MULT",
# true=정규장(JIF/벽시계)에서만 틱 silence 강제재연결 (VI·점심·장외 오탐 방지)
"LS_WS_WATCHDOG_SESSION_GATE",
"LS_WS_WATCHDOG_SESSION_START_HM",
"LS_WS_WATCHDOG_SESSION_END_HM",
# JIF 장운영정보 구독 (스마트 워치독 세션 판정) — 앱 하트비트 TR 아님
"LS_WS_JIF_ENABLED",
"LS_WS_JIF_TR_KEY",
# send 락 대기 상한(초) — half-open 블로킹으로 REG 워커 정지 방지
"LS_WS_SEND_LOCK_TIMEOUT_SEC",
# true 시 LS 복구/비연결 중 신규매수 차단 (그림자 기본 false)
"LS_WS_BLOCK_BUY_WHILE_RECOVERING",
# 웹 보유·실거래「보유중」라이브 시세 (ws_ticks SELECT만 — 실매 WS/주문 경로 비접촉)
"PORTFOLIO_LIVE_QUOTE_ENABLED",
"PORTFOLIO_LIVE_QUOTE_MS",
"PORTFOLIO_LIVE_QUOTE_MAX_AGE_SEC",
# 키움 WS: 그룹당 최대 구독 수·REG 전송 레이트(초기 기동 시 TRNM 건수 초과 완화)
"KIWOOM_WS_MAX_SUBSCRIPTIONS",
"KIWOOM_WS_REG_CHUNK_SIZE",
"KIWOOM_WS_REG_GAP_SEC",
"KIWOOM_WS_REG_DEBOUNCE_SEC",
# REG refresh: 1=기존 구독 유지(기본). 0=LOGIN 일괄 첫 청크만 그룹 해지 후 재등록(테스트)
"KIWOOM_WS_REG_REFRESH",
# 키움 0B FID20(체결시간)이 recv 대비 이 값(초) 초과 stale 이면 wall-clock 으로 tick_time 보정
"KIWOOM_TICK_TIME_MAX_LAG_SEC",
# KIWOOM_TICK_LIVE_MAX_LAG_SEC 삭제 (2026-09-06): WS_TICK_DB_SAVE_LAG_CUT_ENABLED 로 통일
"KIWOOM_TICK_STALE_LOG_GAP_SEC",
# 키움 WS 재연결: 토큰거부(805004) 연타 후 영구 OFF 방지 (기본=쿨다운 리셋)
"KIWOOM_WS_HARD_DISABLE_ON_MAX_RECONNECT",
"KIWOOM_WS_MAX_RECONNECT_COOLDOWN_SEC",
# 키움 WS/CNSR return_code·스레드 진단 (기본 ON — SSL·매수체크 추적)
"KIWOOM_WS_DIAG_ENABLED",
# 전략 매도락 무한 대기 방지(초). 0=무제한(레거시). 틱매도·매도구간 기아 시 매수체크 복구
"STRATEGY_SELL_LOCK_TIMEOUT_SEC",
# 매도 check_sell_signals hold 타임아웃(초). 0=무제한. sem/REST 대기 기아 방지
"STRATEGY_SELL_CHECK_TIMEOUT_SEC",
# 루프 1바퀴가 이 시간(초) 넘으면 phase 로그 (기본 30)
"STRATEGY_LOOP_STALL_SEC",
# WS reconcile debounce(초) — 4전략 owner 갱신 합침
"WS_RECONCILE_DEBOUNCE_SEC",
# 매도 scan 루프에서 ka10007 REST 허용(기본 OFF — sell_lock·sem 블록 방지)
"SELL_SCAN_ALLOW_KIWOOM_REST",
# 키움 ka10080/ka10007 공유 sem 대기(초). 0=무제한
"KIWOOM_KA10080_SEM_TIMEOUT_SEC",
"KIWOOM_KA10007_SEM_TIMEOUT_SEC",
# 계좌 단일 OrderWorker 큐 (B-full — 4전략 BUY/SELL 직렬 place)
"ORDER_WORKER_MAX_QUEUE",
# 전략 루프 매수구간 스킵 사유(후보0·매수허용OFF) rate-limit 로그
"STRATEGY_LOOP_DIAG_ENABLED",
# ── Updow(직전 분봉 음봉·몸통 하락 → 다음 봉 시가 매수) — 백테·웹·CLI 공통 ──
"UPDOW_BODY_DROP_MIN_PCT",
"UPDOW_TP_PCT",
"UPDOW_STOP_LOSS_PCT",
"UPDOW_ATR_USE_DYNAMIC",
"UPDOW_ATR_PERIOD",
"UPDOW_ATR_SL_MULT",
"UPDOW_ATR_TP_MULT",
"UPDOW_ATR_SL_MIN_PCT",
"UPDOW_ATR_SL_MAX_PCT",
"UPDOW_ATR_TP_MIN_PCT",
"UPDOW_ATR_TP_MAX_PCT",
"UPDOW_ATR_MULT_FLOOR",
"UPDOW_ATR_PCT_FLOOR",
"UPDOW_EXIT_PCT_FLOOR",
"UPDOW_SHOULDER_MIN_HIGH_PCT",
"UPDOW_SHOULDER_CUT_PCT",
"UPDOW_MAX_HOLD_BARS",
"UPDOW_EXIT_ON_GREEN",
"UPDOW_SLOT_MONEY",
"UPDOW_TF_MIN",
"UPDOW_GRID_BODY0",
"UPDOW_GRID_BODY1",
"UPDOW_GRID_BODY2",
"UPDOW_GRID_BODY3",
"UPDOW_GRID_TP0",
"UPDOW_GRID_TP1",
"UPDOW_GRID_TP2",
"UPDOW_GRID_TP3",
"UPDOW_GRID_TP4",
"UPDOW_GRID_SL0",
"UPDOW_GRID_SL1",
"UPDOW_GRID_SL2",
"UPDOW_GRID_SL3",
"UPDOW_GRID_HOLD0",
"UPDOW_GRID_HOLD1",
"UPDOW_GRID_HOLD2",
"UPDOW_GRID_HOLD3",
"UPDOW_GRID_HOLD4",
"UPDOW_GRID_SHOULDER_SMIN0",
"UPDOW_GRID_SHOULDER_SMIN1",
"UPDOW_GRID_SHOULDER_SMIN2",
"UPDOW_GRID_SHOULDER_CUT0",
"UPDOW_GRID_SHOULDER_CUT1",
"UPDOW_GRID_SHOULDER_CUT2",
"UPDOW_GRID_REGIME0",
"UPDOW_GRID_REGIME1",
"UPDOW_REGIME_MA_BARS",
"UPDOW_REGIME_MA_EASE_PCT",
"UPDOW_REGIME_MA_EASE_CAP",
"UPDOW_GRID_REGIME_EASE0",
"UPDOW_GRID_REGIME_EASE1",
"UPDOW_KOSPI_1MIN_PROXY_CODE",
"UPDOW_REGIME_PREFER_WS_CANDLES",
"UPDOW_REGIME_WS_CANDLE_MIN",
# ── 익절 호가 (OrderManager) — config_short 에 저장 ──
"SELL_USE_ORDERBOOK_ON_PROFIT",
"SELL_ORDERBOOK_BID_LEVELS",
"SELL_ORDERBOOK_DEPTH_MULT",
# ── 키움 WS 호가(0D) + TRIGGER 호가 필터 (전략별만 · 글로벌 ORDERBOOK_* 수치 폐기) ──
"KIWOOM_WS_ORDERBOOK_ENABLED",
"KIWOOM_WS_PROGRAM_ENABLED",
# 레거시 글로벌(읽기 이관용으로 키만 유지 · 엔진은 전략키만 사용)
"ORDERBOOK_FILTER_ENABLED",
"ORDERBOOK_MAX_SPREAD_PCT",
"ORDERBOOK_ENTRY_BID_LEVELS",
"ORDERBOOK_ENTRY_BID_DEPTH_MULT",
"ORDERBOOK_ENTRY_ASK_MAX_MULT",
"ORDERBOOK_MIN_BID_ASK_RATIO",
"ORDERBOOK_BREAKOUT_ASK_WALL_MAX_QTY",
"SCALP_ORDERBOOK_FILTER_ENABLED",
"SCALP_ORDERBOOK_MAX_SPREAD_PCT",
"SCALP_ORDERBOOK_ENTRY_BID_LEVELS",
"SCALP_ORDERBOOK_ENTRY_BID_DEPTH_MULT",
"SCALP_ORDERBOOK_ENTRY_ASK_MAX_MULT",
"SCALP_ORDERBOOK_MIN_BID_ASK_RATIO",
"MOMENTUM_ORDERBOOK_FILTER_ENABLED",
"MOMENTUM_ORDERBOOK_MAX_SPREAD_PCT",
"MOMENTUM_ORDERBOOK_ENTRY_BID_LEVELS",
"MOMENTUM_ORDERBOOK_ENTRY_BID_DEPTH_MULT",
"MOMENTUM_ORDERBOOK_ENTRY_ASK_MAX_MULT",
"MOMENTUM_ORDERBOOK_MIN_BID_ASK_RATIO",
# 수익구간 호가매도 (exit_ob) + 손절호가 (stop_ob) — MOMENTUM
"MOMENTUM_EXIT_OB_ENABLED",
"MOMENTUM_EXIT_OB_RATIO_MIN",
"MOMENTUM_EXIT_OB_MA_WINDOW",
"MOMENTUM_EXIT_OB_MIN_PROFIT_PCT",
"MOMENTUM_EXIT_OB_MIN_HOLD_BARS",
"MOMENTUM_STOP_OB_ENABLED",
"MOMENTUM_STOP_OB_RATIO_MIN",
"MOMENTUM_STOP_OB_MA_WINDOW",
"MOMENTUM_STOP_OB_MIN_LOSS_PCT",
"MOMENTUM_STOP_OB_MIN_HOLD_BARS",
"BREAKOUT_ORDERBOOK_FILTER_ENABLED",
"BREAKOUT_ORDERBOOK_MAX_SPREAD_PCT",
"BREAKOUT_ORDERBOOK_ENTRY_BID_LEVELS",
"BREAKOUT_ORDERBOOK_ENTRY_BID_DEPTH_MULT",
"BREAKOUT_ORDERBOOK_ENTRY_ASK_MAX_MULT",
"BREAKOUT_ORDERBOOK_MIN_BID_ASK_RATIO",
"BREAKOUT_ORDERBOOK_ASK_WALL_MAX_QTY",
# 수익구간 호가매도 (exit_ob) + 손절호가 (stop_ob) — BREAKOUT
"BREAKOUT_EXIT_OB_ENABLED",
"BREAKOUT_EXIT_OB_RATIO_MIN",
"BREAKOUT_EXIT_OB_MA_WINDOW",
"BREAKOUT_EXIT_OB_MIN_PROFIT_PCT",
"BREAKOUT_EXIT_OB_MIN_HOLD_BARS",
"BREAKOUT_STOP_OB_ENABLED",
"BREAKOUT_STOP_OB_RATIO_MIN",
"BREAKOUT_STOP_OB_MA_WINDOW",
"BREAKOUT_STOP_OB_MIN_LOSS_PCT",
"BREAKOUT_STOP_OB_MIN_HOLD_BARS",
"TAIL_ORDERBOOK_FILTER_ENABLED",
"TAIL_ORDERBOOK_MAX_SPREAD_PCT",
"TAIL_ORDERBOOK_ENTRY_BID_LEVELS",
"TAIL_ORDERBOOK_ENTRY_BID_DEPTH_MULT",
"TAIL_ORDERBOOK_ENTRY_ASK_MAX_MULT",
"TAIL_ORDERBOOK_MIN_BID_ASK_RATIO",
# ── 키움 WS 프로그램매매(0w) + TRIGGER 프로그램 필터 ──
"PROGRAM_FILTER_ENABLED",
"PROGRAM_MIN_NET_BUY_QTY",
"PROGRAM_MIN_NET_BUY_AMT",
"PROGRAM_MAX_SELL_BUY_QTY_RATIO",
"PROGRAM_MIN_NET_DELTA_QTY",
"PROGRAM_SNAPSHOT_MAX_AGE_SEC",
"MOMENTUM_PROGRAM_FILTER_ENABLED",
"MOMENTUM_PROGRAM_MIN_NET_BUY_QTY",
"MOMENTUM_PROGRAM_MIN_NET_BUY_AMT",
"MOMENTUM_PROGRAM_MAX_SELL_BUY_QTY_RATIO",
"MOMENTUM_PROGRAM_MIN_NET_DELTA_QTY",
"MOMENTUM_PROGRAM_SNAPSHOT_MAX_AGE_SEC",
"BREAKOUT_PROGRAM_FILTER_ENABLED",
"BREAKOUT_PROGRAM_MIN_NET_BUY_QTY",
"BREAKOUT_PROGRAM_MIN_NET_BUY_AMT",
"BREAKOUT_PROGRAM_MAX_SELL_BUY_QTY_RATIO",
"BREAKOUT_PROGRAM_MIN_NET_DELTA_QTY",
"BREAKOUT_PROGRAM_SNAPSHOT_MAX_AGE_SEC",
"TAIL_PROGRAM_FILTER_ENABLED",
"TAIL_PROGRAM_MIN_NET_BUY_QTY",
"TAIL_PROGRAM_MIN_NET_BUY_AMT",
"TAIL_PROGRAM_MAX_SELL_BUY_QTY_RATIO",
"TAIL_PROGRAM_MIN_NET_DELTA_QTY",
"TAIL_PROGRAM_SNAPSHOT_MAX_AGE_SEC",
)
# ── 전략별 config_* 테이블 분류 (config_schema.py) ─────────────────────
from config_schema import ( # noqa: E402
CONFIG_TABLE_NAMES,
STRATEGY_ID_TO_TABLE,
_EXPLICIT_KEY_TABLE,
classify_config_key,
split_env_keys,
)
_CONFIG_SPLIT = split_env_keys(ENV_CONFIG_KEYS)
ENV_GLOBAL_KEYS = _CONFIG_SPLIT["env_config"]
CONFIG_SCALP_KEYS = _CONFIG_SPLIT["config_scalp"]
CONFIG_SHORT_KEYS = _CONFIG_SPLIT["config_short"]
CONFIG_MOMENTUM_KEYS = _CONFIG_SPLIT["config_momentum"]
CONFIG_US_MOMENTUM_KEYS = _CONFIG_SPLIT["config_us_momentum"]
CONFIG_BREAKOUT_KEYS = _CONFIG_SPLIT["config_breakout"]
CONFIG_RANGE_BREAK_KEYS = _CONFIG_SPLIT["config_range_break"]
CONFIG_UPDOW_KEYS = _CONFIG_SPLIT["config_updow"]
CONFIG_DBBAND_KEYS = _CONFIG_SPLIT["config_dbband"]
CONFIG_DART_KEYS = _CONFIG_SPLIT["config_dart"]
CONFIG_TABLE_KEYS: Dict[str, Tuple[str, ...]] = {
"env_config": ENV_GLOBAL_KEYS,
"config_scalp": CONFIG_SCALP_KEYS,
"config_short": CONFIG_SHORT_KEYS,
"config_momentum": CONFIG_MOMENTUM_KEYS,
"config_us_momentum": CONFIG_US_MOMENTUM_KEYS,
"config_breakout": CONFIG_BREAKOUT_KEYS,
"config_range_break": CONFIG_RANGE_BREAK_KEYS,
"config_updow": CONFIG_UPDOW_KEYS,
"config_dbband": CONFIG_DBBAND_KEYS,
"config_dart": CONFIG_DART_KEYS,
}
class TradeDB:
"""
트레이딩 봇용 MariaDB 데이터베이스 관리 클래스.
기존 SQLite 인터페이스와 100% 호환 (db_path 인수는 무시됨).
※ 호출자가 매번 ``TradeDB()`` → ``db.close()`` 패턴으로 쓰므로,
DDL(`CREATE TABLE IF NOT EXISTS`) 과 INFO 로그는 **프로세스당 1회만** 실행한다.
(내부 클래스 변수 ``_tables_created`` 가드)
"""
_tables_created = False # 프로세스 내 테이블 생성 1회만
_tables_lock = threading.RLock() # 마이그레이션 중 get_env→TradeDB 재진입 허용
def __init__(self, db_path="quant_bot.db"):
"""
Args:
db_path: 하위 호환용 (무시됨). MariaDB 접속 정보는 환경변수/모듈 상수 사용.
"""
self.db_path = db_path # 호환용 보존
self.conn = _MariaDBConn()
# 마이그레이션(ws_candles RSI/EMA 컬럼 등)이 get_env_from_db → TradeDB() 재호출 시
# 동일 인스턴스를 재사용 — 미등록이면 _tables_lock 데드락으로 기동 무음 hang.
try:
from kis_trader.utils.env import set_db
set_db(self)
except Exception:
pass
# 테이블 생성은 첫 인스턴스에서만 — 매 호출마다 21개 DDL 폭탄 방지
with TradeDB._tables_lock:
if not TradeDB._tables_created:
self._create_tables()
logger.info(
"✅ TradeDB 초기화 완료: MariaDB %s:%s/%s",
_DB_HOST, _DB_PORT, _DB_NAME,
)
TradeDB._tables_created = True
else:
# 이후 인스턴스는 조용히 — 같은 정보 매초 찍어 로그 오염 방지
logger.debug(
"TradeDB 인스턴스 재생성 (테이블 스킵): %s:%s/%s",
_DB_HOST, _DB_PORT, _DB_NAME,
)
def _ensure_safe_index(self, table: str, idx_name: str, columns_sql: str) -> None:
"""인덱스가 없을 때만 추가. 기존 인덱스는 삭제·변경하지 않음.
MariaDB INPLACE + LOCK=NONE 우선 (읽기/쓰기 블로킹 없음).
미지원이면 CREATE INDEX 폴백. Duplicate(1061) 는 무시.
"""
allowed = {
"target_candidates_history": "target_candidates_history",
"ws_ticks": "ws_ticks",
"ws_orderbook": "ws_orderbook",
"ws_ticks_us": "ws_ticks_us",
"ls_ws_orderbook": "ls_ws_orderbook",
}
tbl = allowed.get(str(table or "").strip())
name = str(idx_name or "").strip()
cols = str(columns_sql or "").strip()
if not tbl or not name or not cols.startswith("("):
return
try:
rows = self.conn.execute(f"SHOW INDEX FROM {tbl}").fetchall() or []
existing = set()
for r in rows:
kn = r["Key_name"] if isinstance(r, dict) else r[2]
if kn:
existing.add(str(kn))
if name in existing:
return
except Exception as e:
logger.debug("SHOW INDEX %s 실패: %s", tbl, e)
return
try:
self.conn.execute(
f"ALTER TABLE {tbl} ADD INDEX {name} {cols}, "
"ALGORITHM=INPLACE, LOCK=NONE"
)
logger.info("📌 %s.%s 인덱스 추가 (INPLACE LOCK=NONE)", tbl, name)
return
except Exception as e1:
try:
self.conn.execute(f"CREATE INDEX {name} ON {tbl} {cols}")
logger.info("📌 %s.%s 인덱스 추가", tbl, name)
except Exception as e2:
err = str(e2)
if "1061" not in err and "Duplicate" not in err:
logger.warning(
"idx %s 생성 실패: inplace=%s / create=%s", name, e1, e2
)
def _create_tables(self):
"""DB 테이블 생성 (없을 경우)"""
with self.conn:
# 1. 활성 트레이딩 테이블 (현재 보유 중이거나 매수 중인 종목)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS active_trades (
code VARCHAR(20) NOT NULL, -- 종목코드
name VARCHAR(100) NOT NULL, -- 종목명
strategy VARCHAR(50) NOT NULL DEFAULT 'MANUAL', -- 매매 전략 (SHORT_ANT_SHAKING / SCALP_RSI_REVERSAL 등)
PRIMARY KEY (code, strategy), -- 복합 PK: 같은 종목을 서로 다른 봇이 독립 보유 가능
-- [가격 정보]
avg_buy_price DOUBLE NOT NULL, -- 평단가
current_price DOUBLE, -- 현재가 (업데이트용)
stop_price DOUBLE, -- 손절가
target_price REAL, -- 목표가
max_price REAL, -- 최고가 (트레일링 스탑용)
atr_entry REAL, -- 진입 시 ATR 변동성
-- [수량 및 진행 상태 (분할매수용)]
target_qty INTEGER NOT NULL, -- 목표 매수 수량
current_qty INTEGER NOT NULL,-- 현재 체결 수량
total_invested REAL, -- 총 투입 금액 (수수료 제외)
-- [상태 관리]
status TEXT NOT NULL, -- BUYING(매수중), HOLDING(보유중), SELLING(매도중)
buy_date TEXT NOT NULL, -- 첫 매수 시작 시간
updated_at TEXT NOT NULL, -- 마지막 업데이트 시간
size_class TEXT, -- 대/중/소형 (매수 시점)
is_mock TINYINT(1) NULL -- 1=모의(KIS_MOCK) · 0=실전 · NULL=이관 전 미상
)
""")
# 2. 매매 기록 테이블 (손익 분석 & 켈리 공식용)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS trade_history (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
code VARCHAR(20) NOT NULL,
name VARCHAR(100) NOT NULL,
strategy VARCHAR(50),
buy_price DOUBLE NOT NULL,
sell_price DOUBLE NOT NULL,
qty INT NOT NULL,
profit_rate DOUBLE NOT NULL,
realized_pnl DOUBLE NOT NULL,
hold_minutes INT,
buy_date VARCHAR(30),
sell_date VARCHAR(30) NOT NULL,
sell_reason VARCHAR(200),
env_snapshot TEXT,
size_class VARCHAR(20),
is_mock TINYINT(1) NULL
) CHARACTER SET utf8mb4
""")
# 3. 일일 손익 요약 테이블 (대시보드용)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS daily_summary (
date VARCHAR(10) NOT NULL PRIMARY KEY,
start_asset DOUBLE,
end_asset DOUBLE,
total_trades INT,
win_trades INT,
total_pnl DOUBLE,
win_rate DOUBLE
) CHARACTER SET utf8mb4
""")
# 4. 주문·체결 보강 테이블 (kt00007 / ka10076)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS order_execution_history (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
source VARCHAR(20) NOT NULL,
ord_no VARCHAR(30),
stk_cd VARCHAR(20),
stk_nm VARCHAR(100),
trde_tp VARCHAR(20),
ord_qty VARCHAR(20),
ord_uv VARCHAR(20),
cntr_qty VARCHAR(20),
cntr_uv VARCHAR(20),
ord_tm VARCHAR(20),
cnfm_tm VARCHAR(20),
sell_tp VARCHAR(20),
ord_dt VARCHAR(20),
raw_json TEXT,
fetched_at VARCHAR(30) NOT NULL,
is_mock TINYINT(1) NULL
) CHARACTER SET utf8mb4
""")
# 5. 매수 체결 이력 (일일 한도용 - '산 시점' 날짜 기준 누적)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS buy_execution_log (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
code VARCHAR(20) NOT NULL,
name VARCHAR(100) NOT NULL,
strategy VARCHAR(50) NOT NULL,
buy_date VARCHAR(10) NOT NULL,
executed_at VARCHAR(30) NOT NULL,
amount DOUBLE NOT NULL,
qty INT NOT NULL,
is_mock TINYINT(1) NULL
) CHARACTER SET utf8mb4
""")
# 6. 매수 후보군 테이블 (target_universe 대체)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS target_candidates (
code VARCHAR(20) NOT NULL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
score DOUBLE NOT NULL,
price DOUBLE NOT NULL,
scan_time VARCHAR(30) NOT NULL,
updated_at VARCHAR(30) NOT NULL
) CHARACTER SET utf8mb4
""")
# 6-2. 매수 후보군 이력 (5분마다 쌓아서 백테스트 시 '실제 그 시각 유니버스' 사용 가능)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS target_candidates_history (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
slot_key VARCHAR(12) NOT NULL,
scan_time VARCHAR(30) NOT NULL,
code VARCHAR(20) NOT NULL,
name VARCHAR(100) NOT NULL DEFAULT '',
score DOUBLE NOT NULL DEFAULT 0,
price DOUBLE NOT NULL DEFAULT 0,
market CHAR(1) DEFAULT 'Q',
sector VARCHAR(100),
theme VARCHAR(100),
INDEX idx_slot (slot_key),
INDEX idx_scan (scan_time)
) CHARACTER SET utf8mb4
""")
# 7. 종목 메타데이터 (테마·섹터·시장구분) — 스캐너가 채움, 조인 분석 용
# market: 'K'=KOSPI, 'Q'=KOSDAQ, 'E'=ETF/기타
# sector: 업종명 (KIS bstp_kor_isnm, 예: '반도체')
# theme : 주요 테마 (예: 'AI반도체', '2차전지', '원자력')
# - 직접 UPDATE 또는 별도 스크립트로 채움
# theme_rank: 테마 내 대장주/추종주 순위 (1=핵심, 2=연관, 3=주변)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS stock_meta (
code VARCHAR(20) NOT NULL PRIMARY KEY,
name VARCHAR(100) NOT NULL DEFAULT '',
market CHAR(1) NOT NULL DEFAULT 'Q',
sector_code VARCHAR(20),
sector VARCHAR(100),
theme VARCHAR(100),
theme_rank TINYINT DEFAULT 3,
updated_at VARCHAR(30) NOT NULL
) CHARACTER SET utf8mb4
""")
# 8. env 공통 설정 (API·MM·인프라·전략 스위치) — 컬럼 수 축소
gcols = ", ".join([f"`{k}` TEXT" for k in ENV_GLOBAL_KEYS])
self.conn.execute(f"""
CREATE TABLE IF NOT EXISTS env_config (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
created_at VARCHAR(30) NOT NULL,
{gcols}
) CHARACTER SET utf8mb4
""")
# 8a. 전략별 설정 테이블 (config_scalp / config_short / …)
for tbl, keys in CONFIG_TABLE_KEYS.items():
if tbl == "env_config":
continue
scols = ", ".join([f"`{k}` TEXT" for k in keys])
self.conn.execute(f"""
CREATE TABLE IF NOT EXISTS {tbl} (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
created_at VARCHAR(30) NOT NULL,
{scols}
) CHARACTER SET utf8mb4
""")
# 8. 키-값 저장소 (매터모스트 원격 조종: 마지막 AI 추천문, last_seen 등)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS kv_store (
k VARCHAR(100) NOT NULL PRIMARY KEY,
v MEDIUMTEXT
) CHARACTER SET utf8mb4
""")
# 8b. env_config 컬럼 한도(Row size) 초과 키 — KIWOOM_WS_* · WHIPSAW_* 등
self.conn.execute("""
CREATE TABLE IF NOT EXISTS env_config_ext (
env_key VARCHAR(128) NOT NULL PRIMARY KEY,
env_value TEXT,
updated_at VARCHAR(30) NOT NULL
) CHARACTER SET utf8mb4
""")
# 8b-2. Optuna 스터디별 웹 결과·후처리 (kis_optuna 공식 테이블과 분리)
self.conn.execute_long("""
CREATE TABLE IF NOT EXISTS optuna_study_result (
study_name VARCHAR(255) NOT NULL COMMENT 'Optuna study 이름 (kis_optuna 과 동일)',
strategy VARCHAR(32) NOT NULL DEFAULT '' COMMENT 'tail|momentum|us_momentum|breakout|scalp',
mode VARCHAR(32) NOT NULL DEFAULT '' COMMENT 'tpe 등',
start_date VARCHAR(10) NULL COMMENT '백테 시작 YYYY-MM-DD',
end_date VARCHAR(10) NULL COMMENT '백테 종료 YYYY-MM-DD',
job_id VARCHAR(64) NULL COMMENT '웹 잡 id (참고)',
symbol VARCHAR(16) NULL COMMENT '해외 종목핀일 때만',
study_trials INT NOT NULL DEFAULT 0 COMMENT '스터디 총 완료 목표. 0이면 미설정',
n_complete INT NOT NULL DEFAULT 0 COMMENT 'kis_optuna 완료 trial 수 캐시',
n_running INT NOT NULL DEFAULT 0 COMMENT '진행 중 trial 수 캐시 (표시용)',
payload_json LONGTEXT NULL COMMENT 'results_all/gated/stable, mode_combo, postprocess_topn, overfit (웹 요약과 동일 키)',
briefing_md MEDIUMTEXT NULL COMMENT '브리핑 마크다운',
pp_needed TINYINT NOT NULL DEFAULT 0 COMMENT '1=목표 도달, 후처리 대기',
pp_status VARCHAR(16) NOT NULL DEFAULT 'idle' COMMENT 'idle|running|done|error',
pp_for_n INT NOT NULL DEFAULT 0 COMMENT '이 후처리가 반영한 n_complete',
pp_lock_until DATETIME NULL COMMENT '락 만료 (죽은 워커 복구)',
pp_lock_owner VARCHAR(64) NULL COMMENT '잠근 호스트/pid',
pp_error VARCHAR(500) NULL COMMENT '후처리 실패 메시지',
leftover_note VARCHAR(200) NULL COMMENT '웹 남은횟수 안내 캐시',
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '행 갱신',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '최초 생성',
PRIMARY KEY (study_name),
KEY idx_pp_needed (pp_needed, pp_status),
KEY idx_job (job_id),
KEY idx_updated (updated_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
COMMENT='Optuna 스터디별 웹 결과·후처리 산출 (파일 JSON 대체)'
""", timeout_sec=120)
# 8c. 인증/계좌/ID 전용 컬럼 저장소 (앱키/시크릿/ID 분리)
acols = ", ".join([f"`{k}` TEXT" for k in ENV_AUTH_KEYS])
self.conn.execute("""
CREATE TABLE IF NOT EXISTS env_auth_config (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
created_at VARCHAR(30) NOT NULL,
{acols}
) CHARACTER SET utf8mb4
""".format(acols=acols))
# 9. AI 분석 기록 (Butler !클로드분석/!애미분석 시 프롬프트 요약·응답 저장 → 나중에 꺼내보기)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ai_analysis_log (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
created_at VARCHAR(30) NOT NULL,
model VARCHAR(50) NOT NULL,
context_summary TEXT,
response MEDIUMTEXT
) CHARACTER SET utf8mb4
""")
# 10. WebSocket 실시간 봉 집계 (백테스트용 — CandleAggregator 배치 INSERT)
# - is_confirmed=1 인 확정 봉만 저장 (진행 중 봉은 RAM에만 존재)
# - source: 증권사 (kis/kiwoom). channel: ws|rest|rollup
# - UNIQUE(code, timeframe, candle_time, source, channel)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ws_candles (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
code VARCHAR(20) NOT NULL,
market VARCHAR(8) NOT NULL DEFAULT 'KR',
timeframe TINYINT NOT NULL,
candle_time VARCHAR(12) NOT NULL,
`open` DOUBLE NOT NULL,
high DOUBLE NOT NULL,
low DOUBLE NOT NULL,
close DOUBLE NOT NULL,
volume BIGINT NOT NULL DEFAULT 0,
rsi_2 DOUBLE,
rsi_3 DOUBLE,
rsi_5 DOUBLE,
is_confirmed TINYINT NOT NULL DEFAULT 1,
source VARCHAR(10) NOT NULL DEFAULT 'kis',
channel VARCHAR(10) NOT NULL DEFAULT 'ws',
updated_at VARCHAR(30) NOT NULL,
UNIQUE KEY uq_candle (code, timeframe, candle_time, source, channel),
KEY idx_ws_candles_market (market, code, timeframe, candle_time)
) CHARACTER SET utf8mb4
""")
# 11. WebSocket 실시간 체결 틱 (후보 종목, 배치 INSERT — TickRecorder)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ws_ticks (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
market VARCHAR(8) NOT NULL DEFAULT 'KR',
exchange VARCHAR(16) DEFAULT NULL,
code VARCHAR(32) NOT NULL,
tick_time VARCHAR(14) NOT NULL,
price DOUBLE NOT NULL,
volume BIGINT NOT NULL DEFAULT 0,
tick_seq BIGINT DEFAULT NULL,
session VARCHAR(8) DEFAULT NULL,
currency VARCHAR(8) NOT NULL DEFAULT 'KRW',
source VARCHAR(16) NOT NULL DEFAULT 'kis',
channel VARCHAR(10) NOT NULL DEFAULT 'ws',
recv_ts VARCHAR(30) NOT NULL,
KEY idx_ws_ticks_lookup (market, code, tick_time),
KEY idx_ws_ticks_recv (recv_ts)
) ENGINE=InnoDB CHARACTER SET utf8mb4
""")
# 11a. 해외(US) 체결 틱 — 국내 ws_ticks 와 물리 분리 (인덱스·스캔·정리 격리)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ws_ticks_us (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
market VARCHAR(8) NOT NULL DEFAULT 'US',
exchange VARCHAR(16) DEFAULT NULL,
code VARCHAR(32) NOT NULL,
tick_time VARCHAR(14) NOT NULL,
price DOUBLE NOT NULL,
volume BIGINT NOT NULL DEFAULT 0,
tick_seq BIGINT DEFAULT NULL,
session VARCHAR(8) DEFAULT NULL,
currency VARCHAR(8) NOT NULL DEFAULT 'USD',
source VARCHAR(16) NOT NULL DEFAULT 'kis_us',
channel VARCHAR(10) NOT NULL DEFAULT 'ws',
recv_ts VARCHAR(30) NOT NULL,
KEY idx_ws_ticks_us_lookup (market, code, tick_time),
KEY idx_ws_ticks_us_recv (recv_ts)
) ENGINE=InnoDB CHARACTER SET utf8mb4
""")
self._migrate_ws_ticks_extra_columns("ws_ticks")
self._migrate_ws_ticks_extra_columns("ws_ticks_us")
# 11b. TRIGGER 호가 스냅샷 (키움 0D — 백테·파람서치 재현)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ws_orderbook (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
market VARCHAR(8) NOT NULL DEFAULT 'KR',
code VARCHAR(32) NOT NULL,
snap_time VARCHAR(14) NOT NULL,
best_bid BIGINT NOT NULL DEFAULT 0,
best_ask BIGINT NOT NULL DEFAULT 0,
total_bid_qty BIGINT NOT NULL DEFAULT 0,
total_ask_qty BIGINT NOT NULL DEFAULT 0,
bid_qty_l3 BIGINT NOT NULL DEFAULT 0,
ask_qty_l3 BIGINT NOT NULL DEFAULT 0,
levels_json MEDIUMTEXT,
source VARCHAR(16) NOT NULL DEFAULT 'kiwoom_0d',
recv_ts VARCHAR(30) NOT NULL,
KEY idx_ws_orderbook_lookup (market, code, snap_time),
KEY idx_ws_orderbook_recv (recv_ts)
) CHARACTER SET utf8mb4
""")
# 11c. TRIGGER 프로그램매매 스냅샷 (키움 0w)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ws_program (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
market VARCHAR(8) NOT NULL DEFAULT 'KR',
code VARCHAR(32) NOT NULL,
snap_time VARCHAR(14) NOT NULL,
buy_qty BIGINT NOT NULL DEFAULT 0,
sell_qty BIGINT NOT NULL DEFAULT 0,
net_qty BIGINT NOT NULL DEFAULT 0,
buy_amt BIGINT NOT NULL DEFAULT 0,
sell_amt BIGINT NOT NULL DEFAULT 0,
net_amt BIGINT NOT NULL DEFAULT 0,
source VARCHAR(16) NOT NULL DEFAULT 'kiwoom_0w',
recv_ts VARCHAR(30) NOT NULL,
KEY idx_ws_program_lookup (market, code, snap_time),
KEY idx_ws_program_recv (recv_ts)
) CHARACTER SET utf8mb4
""")
# 12. 종목 유통/상장주식수 (키움 ka10001 — 백테 회전율·전략 공통)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS stock_share_meta (
code VARCHAR(20) NOT NULL PRIMARY KEY,
flo_stk BIGINT NOT NULL DEFAULT 0,
dstr_stk BIGINT NOT NULL DEFAULT 0,
dstr_rt DOUBLE NULL,
source VARCHAR(16) NOT NULL DEFAULT 'ka10001',
updated_at VARCHAR(30) NOT NULL,
KEY idx_share_updated (updated_at)
) CHARACTER SET utf8mb4
""")
# 13. DART 수주/단일판매·공급계약 공시 중복방지 + 이력
self.conn.execute("""
CREATE TABLE IF NOT EXISTS dart_disclosures (
rcept_no VARCHAR(32) NOT NULL PRIMARY KEY,
corp_code VARCHAR(16) NOT NULL DEFAULT '',
corp_name VARCHAR(100) NOT NULL DEFAULT '',
stock_code VARCHAR(20) NOT NULL DEFAULT '',
report_nm VARCHAR(255) NOT NULL DEFAULT '',
rcept_dt VARCHAR(8) NOT NULL DEFAULT '',
url VARCHAR(255) NOT NULL DEFAULT '',
first_seen_at VARCHAR(30) NOT NULL,
mm_sent TINYINT NOT NULL DEFAULT 0,
KEY idx_dart_seen (first_seen_at),
KEY idx_dart_stock (stock_code)
) CHARACTER SET utf8mb4
""")
# 14. DART 임시 워치리스트 (영구구독 아님 — TTL)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS dart_watchlist (
stock_code VARCHAR(20) NOT NULL PRIMARY KEY,
corp_name VARCHAR(100) NOT NULL DEFAULT '',
rcept_no VARCHAR(32) NOT NULL DEFAULT '',
report_nm VARCHAR(255) NOT NULL DEFAULT '',
added_at VARCHAR(30) NOT NULL,
expires_at VARCHAR(30) NOT NULL,
enabled TINYINT NOT NULL DEFAULT 1,
KEY idx_dart_watch_exp (expires_at)
) CHARACTER SET utf8mb4
""")
try:
from kis_trader.database.paper_store import ensure_paper_tables
ensure_paper_tables(self)
except Exception as e:
logger.debug("paper 테이블 ensure 스킵: %s", e)
self._migrate_add_columns()
self._migrate_env_config_to_columns()
logger.info("📊 DB 테이블 생성/확인 완료")
def _migrate_add_columns(self):
"""기존 DB에 누락된 컬럼 추가 (한 번만) — PRAGMA → information_schema 대체"""
try:
cols = self.conn.get_columns("trade_history")
if "env_snapshot" not in cols:
self.conn.execute("ALTER TABLE trade_history ADD COLUMN env_snapshot TEXT")
logger.info("📌 trade_history.env_snapshot 컬럼 추가")
if "size_class" not in cols:
self.conn.execute("ALTER TABLE trade_history ADD COLUMN size_class VARCHAR(20)")
logger.info("📌 trade_history.size_class 컬럼 추가")
if "is_mock" not in cols:
self.conn.execute(
"ALTER TABLE trade_history ADD COLUMN is_mock TINYINT(1) NULL"
)
logger.info("📌 trade_history.is_mock 컬럼 추가 (모의=1/실전=0)")
self._backfill_is_mock_from_env_snapshot("trade_history")
# snapshot에 KIS_MOCK 없으면 잔여 NULL → 현재 KIS_MOCK (이관 시점 운영값)
try:
cur_mock = self.resolve_kis_is_mock(None)
n_left = self.conn.execute(
"UPDATE trade_history SET is_mock=%s WHERE is_mock IS NULL",
(cur_mock,),
).rowcount
if n_left:
logger.info(
"📌 trade_history.is_mock 잔여 NULL→현재KIS_MOCK(%s) %d",
cur_mock, n_left,
)
except Exception as e2:
logger.debug("trade_history is_mock 잔여 백필 스킵: %s", e2)
except Exception as e:
logger.debug(f"migrate trade_history: {e}")
# ── active_trades PK 복합키 마이그레이션 (code → code+strategy) ──────────
# 두 봇(SHORT/SCALP)이 같은 종목을 독립 보유 가능하도록 PK 확장.
# 신규 설치는 DDL에서 처리됨. 기존 테이블은 여기서 한 번만 ALTER.
try:
cursor = self.conn.execute("""
SELECT COLUMN_NAME FROM information_schema.KEY_COLUMN_USAGE
WHERE TABLE_SCHEMA = DATABASE()
AND TABLE_NAME = 'active_trades'
AND CONSTRAINT_NAME = 'PRIMARY'
ORDER BY ORDINAL_POSITION
""")
pk_cols = [row[0] if isinstance(row, (list, tuple)) else row['COLUMN_NAME']
for row in cursor.fetchall()]
if 'strategy' not in pk_cols:
logger.info("⚙️ active_trades PK 복합키 마이그레이션 시작 (code → code+strategy)")
# NULL strategy → 'MANUAL' 로 채움 (NOT NULL 변경 전 필수)
self.conn.execute("UPDATE active_trades SET strategy = 'MANUAL' WHERE strategy IS NULL OR strategy = ''")
# strategy 컬럼 NOT NULL DEFAULT 'MANUAL' 로 변경 후 PK 재구성
self.conn.execute("ALTER TABLE active_trades MODIFY COLUMN strategy VARCHAR(50) NOT NULL DEFAULT 'MANUAL'")
self.conn.execute("ALTER TABLE active_trades DROP PRIMARY KEY, ADD PRIMARY KEY (code, strategy)")
logger.info("✅ active_trades PK 복합키(code, strategy) 변환 완료")
except Exception as e:
logger.debug("active_trades PK 마이그레이션 스킵(이미 완료 또는 신규): %s", e)
try:
cols = self.conn.get_columns("active_trades")
if "size_class" not in cols:
self.conn.execute("ALTER TABLE active_trades ADD COLUMN size_class VARCHAR(20)")
logger.info("📌 active_trades.size_class 컬럼 추가")
if "is_mock" not in cols:
self.conn.execute(
"ALTER TABLE active_trades ADD COLUMN is_mock TINYINT(1) NULL"
)
logger.info("📌 active_trades.is_mock 컬럼 추가 (모의=1/실전=0)")
# 보유 중 행은 현재 KIS_MOCK 로 채움 (탭 필터용 · 과거 청산은 snapshot 백필)
try:
from kis_trader.utils.env import get_env_bool
cur = 1 if get_env_bool("KIS_MOCK", True) else 0
self.conn.execute(
"UPDATE active_trades SET is_mock=%s WHERE is_mock IS NULL",
(cur,),
)
except Exception as e2:
logger.debug("active_trades is_mock 현재값 백필 스킵: %s", e2)
for c in ML_ENTRY_FEATURE_COLUMNS:
if c not in cols:
self.conn.execute(f"ALTER TABLE active_trades ADD COLUMN `{c}` DOUBLE")
logger.info(f"📌 active_trades.{c} 컬럼 추가 (ML 진입 피처)")
except Exception as e:
logger.debug(f"migrate active_trades: {e}")
# buy_execution_log / order_execution_history — 모의·실전 분리
for _tbl in ("buy_execution_log", "order_execution_history"):
try:
cols = self.conn.get_columns(_tbl)
if "is_mock" not in cols:
self.conn.execute(
f"ALTER TABLE {_tbl} ADD COLUMN is_mock TINYINT(1) NULL"
)
logger.info("📌 %s.is_mock 컬럼 추가 (모의=1/실전=0)", _tbl)
try:
cur_mock = self.resolve_kis_is_mock(None)
n_left = self.conn.execute(
f"UPDATE {_tbl} SET is_mock=%s WHERE is_mock IS NULL",
(cur_mock,),
).rowcount
if n_left:
logger.info(
"📌 %s.is_mock 잔여 NULL→현재KIS_MOCK(%s) %d",
_tbl, cur_mock, n_left,
)
except Exception as e2:
logger.debug("%s is_mock 백필 스킵: %s", _tbl, e2)
except Exception as e:
logger.debug("migrate %s is_mock: %s", _tbl, e)
try:
self.migrate_trigger_eval_columns()
except Exception as e:
logger.debug("migrate_trigger_eval_columns: %s", e)
try:
cols = self.conn.get_columns("trade_history")
for c in ML_ENTRY_FEATURE_COLUMNS:
if c not in cols:
self.conn.execute(f"ALTER TABLE trade_history ADD COLUMN `{c}` DOUBLE")
logger.info(f"📌 trade_history.{c} 컬럼 추가 (ML 진입 피처)")
except Exception as e:
logger.debug(f"migrate trade_history ML columns: {e}")
try:
self._migrate_config_table_columns()
self._migrate_env_auth_from_legacy()
self._migrate_strategy_keys_from_env_config()
self._migrate_us_momentum_from_ext()
self._migrate_short_time_to_tail_time()
except Exception as e:
logger.debug(f"migrate config tables: {e}")
try:
cols = self.conn.get_columns("ws_candles")
if "market" not in cols:
self.conn.execute(
"ALTER TABLE ws_candles ADD COLUMN market VARCHAR(8) NOT NULL DEFAULT 'KR' "
"COMMENT 'KR|US' AFTER code"
)
logger.info("📌 ws_candles.market 컬럼 추가")
try:
self.conn.execute(
"CREATE INDEX idx_ws_candles_market ON ws_candles "
"(market, code, timeframe, candle_time)"
)
except Exception:
pass
if "holding_peak" not in cols:
self.conn.execute(
"ALTER TABLE ws_candles ADD COLUMN holding_peak DOUBLE NULL "
"COMMENT '보유 중 트레일 고점(WS 틱 max_price 스냅샷)'"
)
logger.info("📌 ws_candles.holding_peak 컬럼 추가")
# RSI(7,14,21)·EMA(5,9,12,15,21,34) materialized — indicator_cache 정의 기준
try:
from kis_trader.engine.indicator_cache import ensure_ws_candles_indicator_columns
n_added = ensure_ws_candles_indicator_columns(self)
if n_added:
logger.info("📌 ws_candles materialized 지표 컬럼 %d개 추가", n_added)
except Exception as mig_ex:
logger.debug("ws_candles indicator columns migrate: %s", mig_ex)
self._migrate_ws_candles_source_channel()
except Exception as e:
logger.debug(f"migrate ws_candles holding_peak: {e}")
# ── target_candidates 테마/섹터/시장구분 컬럼 추가 ──────────────────
try:
cols = self.conn.get_columns("target_candidates")
for col, ddl in [
("market", "CHAR(1) DEFAULT 'Q'"),
("sector", "VARCHAR(100)"),
("theme", "VARCHAR(100)"),
]:
if col not in cols:
self.conn.execute(
f"ALTER TABLE target_candidates ADD COLUMN `{col}` {ddl}"
)
logger.info(f"📌 target_candidates.{col} 컬럼 추가")
except Exception as e:
logger.debug(f"migrate target_candidates theme cols: {e}")
# ── target_candidates_history (후보 이력, 백테스트용) ──────────────────
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS target_candidates_history (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
slot_key VARCHAR(12) NOT NULL,
scan_time VARCHAR(30) NOT NULL,
code VARCHAR(20) NOT NULL,
name VARCHAR(100) NOT NULL DEFAULT '',
score DOUBLE NOT NULL DEFAULT 0,
price DOUBLE NOT NULL DEFAULT 0,
market CHAR(1) DEFAULT 'Q',
sector VARCHAR(100),
theme VARCHAR(100),
INDEX idx_slot (slot_key),
INDEX idx_scan (scan_time)
) CHARACTER SET utf8mb4
""")
logger.info("📌 target_candidates_history 테이블 확인/생성")
except Exception as e:
logger.warning(f"migrate target_candidates_history 실패(이력 미적재 가능): {e}")
# ── ws_price_validation (KIS↔키움 시세 검증, 마이그레이션 단계용) ────
# 5초마다 같은 종목의 KIS WS 가격과 키움 WS 가격을 비교해 한 행 INSERT.
# diff_pct = (kiwoom - kis) / kis × 100.
# 운영에는 영향 없음 (검증 모드 ON 일 때만 채워짐). 1~2주 누적 후
# 통계 분석 → 본격 마이그레이션 결정 근거.
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ws_price_validation (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
ts DATETIME(3) NOT NULL,
code VARCHAR(20) NOT NULL,
kis_price DOUBLE,
kiwoom_price DOUBLE,
diff_pct DOUBLE,
kis_age_ms INT,
kiwoom_age_ms INT,
INDEX idx_ts (ts),
INDEX idx_code (code),
INDEX idx_diff (diff_pct)
) CHARACTER SET utf8mb4
""")
logger.info("📌 ws_price_validation 테이블 확인/생성")
except Exception as e:
logger.warning(f"migrate ws_price_validation 실패: {e}")
# ── LS 그림자 시세 (틱/1분봉/갭비교) — 실매 ws_candles 와 분리 ─────
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ls_ws_ticks (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
ts DATETIME(3) NOT NULL,
code VARCHAR(20) NOT NULL,
price DOUBLE NOT NULL,
volume DOUBLE NULL,
tot_volume DOUBLE NULL,
chetime VARCHAR(16) NULL,
tr_cd VARCHAR(8) NULL,
tk CHAR(14) NULL,
lag_sec DOUBLE NULL,
INDEX idx_ls_tick_ts (ts),
INDEX idx_ls_tick_code_ts (code, ts),
INDEX idx_ls_tick_code_tk (code, tk)
) CHARACTER SET utf8mb4
""")
# Migration: add tk, lag_sec if not exist
cols = [dict(r)["Field"] for r in self.conn.execute("SHOW COLUMNS FROM ls_ws_ticks").fetchall()]
if "tk" not in cols:
self.conn.execute("ALTER TABLE ls_ws_ticks ADD COLUMN tk CHAR(14) NULL, ADD COLUMN lag_sec DOUBLE NULL, ADD INDEX idx_ls_tick_code_tk (code, tk)")
logger.info("📌 ls_ws_ticks 테이블 확인/생성 및 마이그레이션 완료")
except Exception as e:
logger.warning(f"migrate ls_ws_ticks 실패: {e}")
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ls_ws_candles (
code VARCHAR(20) NOT NULL,
datetime VARCHAR(20) NOT NULL,
tf_min INT NOT NULL DEFAULT 1,
open DOUBLE NOT NULL,
high DOUBLE NOT NULL,
low DOUBLE NOT NULL,
close DOUBLE NOT NULL,
volume DOUBLE NOT NULL DEFAULT 0,
tick_count INT NOT NULL DEFAULT 0,
updated_at DATETIME(3) NOT NULL,
PRIMARY KEY (code, datetime, tf_min),
INDEX idx_ls_cndl_dt (datetime)
) CHARACTER SET utf8mb4
""")
logger.info("📌 ls_ws_candles 테이블 확인/생성")
except Exception as e:
logger.warning(f"migrate ls_ws_candles 실패: {e}")
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ls_ws_orderbook (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
market VARCHAR(8) NOT NULL DEFAULT 'KR',
code VARCHAR(32) NOT NULL,
snap_time VARCHAR(14) NOT NULL,
best_bid BIGINT NOT NULL DEFAULT 0,
best_ask BIGINT NOT NULL DEFAULT 0,
total_bid_qty BIGINT NOT NULL DEFAULT 0,
total_ask_qty BIGINT NOT NULL DEFAULT 0,
bid_qty_l3 BIGINT NOT NULL DEFAULT 0,
ask_qty_l3 BIGINT NOT NULL DEFAULT 0,
levels_json MEDIUMTEXT,
source VARCHAR(16) NOT NULL DEFAULT 'ls_uh1',
recv_ts VARCHAR(30) NOT NULL,
tk CHAR(14) NULL,
lag_sec DOUBLE NULL,
KEY idx_ls_ob_lookup (market, code, snap_time),
KEY idx_ls_ob_recv (recv_ts),
KEY idx_ls_ob_code (code),
KEY idx_ls_ob_code_snap (code, snap_time)
) CHARACTER SET utf8mb4
""")
# Migration: add tk, lag_sec if not exist
cols = [dict(r)["Field"] for r in self.conn.execute("SHOW COLUMNS FROM ls_ws_orderbook").fetchall()]
if "tk" not in cols:
self.conn.execute("ALTER TABLE ls_ws_orderbook ADD COLUMN tk CHAR(14) NULL, ADD COLUMN lag_sec DOUBLE NULL, ADD INDEX idx_ls_ob_code_snap (code, snap_time)")
logger.info("📌 ls_ws_orderbook 테이블 확인/생성 및 마이그레이션 완료")
except Exception as e:
logger.warning(f"migrate ls_ws_orderbook 실패: {e}")
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ls_ws_vi (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
ts DATETIME(3) NOT NULL,
code VARCHAR(20) NOT NULL,
event_time VARCHAR(16) NULL,
vi_gubun VARCHAR(4) NOT NULL DEFAULT '0',
krx_vi_gubun VARCHAR(4) NULL,
nxt_vi_gubun VARCHAR(4) NULL,
svi_recprice DOUBLE NULL,
dvi_recprice DOUBLE NULL,
vi_trgprice DOUBLE NULL,
tr_cd VARCHAR(8) NULL,
exchname VARCHAR(8) NULL,
INDEX idx_ls_vi_ts (ts),
INDEX idx_ls_vi_code_ts (code, ts),
INDEX idx_ls_vi_gubun (vi_gubun)
) CHARACTER SET utf8mb4
""")
logger.info("📌 ls_ws_vi 테이블 확인/생성")
except Exception as e:
logger.warning(f"migrate ls_ws_vi 실패: {e}")
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ws_price_validation_ls (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
ts DATETIME(3) NOT NULL,
code VARCHAR(20) NOT NULL,
kis_price DOUBLE NULL,
kiwoom_price DOUBLE NULL,
ls_price DOUBLE NULL,
diff_kis_ls_pct DOUBLE NULL,
diff_kw_ls_pct DOUBLE NULL,
kis_age_ms INT NULL,
kiwoom_age_ms INT NULL,
ls_age_ms INT NULL,
INDEX idx_lsval_ts (ts),
INDEX idx_lsval_code (code),
INDEX idx_lsval_diff (diff_kis_ls_pct)
) CHARACTER SET utf8mb4
""")
logger.info("📌 ws_price_validation_ls 테이블 확인/생성")
except Exception as e:
logger.warning(f"migrate ws_price_validation_ls 실패: {e}")
# ── LS 유니버스 히스토리 (슬롯별 수집 종목·퀄리티 스냅샷) ───────────
# 실매 target_candidates_history 와 분리. LS 섀도 피드 커버리지/구멍 점검용.
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ls_universe_history (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
slot_key VARCHAR(12) NOT NULL,
scan_time VARCHAR(30) NOT NULL,
code VARCHAR(20) NOT NULL,
name VARCHAR(100) NOT NULL DEFAULT '',
price DOUBLE NOT NULL DEFAULT 0,
tick_count INT NOT NULL DEFAULT 0,
candle_1m_count INT NOT NULL DEFAULT 0,
last_tick_ts DATETIME(3) NULL,
last_candle_dt VARCHAR(20) NULL,
gap_1m_missing INT NOT NULL DEFAULT 0,
source VARCHAR(16) NOT NULL DEFAULT 'ls_ws',
note VARCHAR(200) NULL,
UNIQUE KEY uq_ls_univ_slot_code (slot_key, code),
INDEX idx_ls_univ_slot (slot_key),
INDEX idx_ls_univ_code (code),
INDEX idx_ls_univ_scan (scan_time)
) CHARACTER SET utf8mb4
""")
logger.info("📌 ls_universe_history 테이블 확인/생성")
except Exception as e:
logger.warning(f"migrate ls_universe_history 실패: {e}")
# ── stock_share_meta (유통주식수, 백테·라이브 공통) ─────────────────
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS stock_share_meta (
code VARCHAR(20) NOT NULL PRIMARY KEY,
flo_stk BIGINT NOT NULL DEFAULT 0,
dstr_stk BIGINT NOT NULL DEFAULT 0,
dstr_rt DOUBLE NULL,
source VARCHAR(16) NOT NULL DEFAULT 'ka10001',
updated_at VARCHAR(30) NOT NULL,
KEY idx_share_updated (updated_at)
) CHARACTER SET utf8mb4
""")
logger.info("📌 stock_share_meta 테이블 확인/생성")
except Exception as e:
logger.warning(f"migrate stock_share_meta 실패: {e}")
# 오늘 운영 탭 당일 이력 집계용 — 기존 idx_slot 유지, 복합만 추가
self._ensure_safe_index(
"target_candidates_history",
"idx_tch_slot_sid",
"(slot_key, strategy_id)",
)
self._migrate_feed_collect_stats_indexes()
def _migrate_feed_collect_stats_indexes(self) -> None:
"""수집통계·일자 범위 조회 — tick_time/snap_time 단독 인덱스 (타입 변경 없음)."""
specs = (
("ws_ticks", "idx_ws_ticks_tick_time", "(tick_time)"),
("ws_ticks", "idx_ws_ticks_day_src", "(tick_time, source)"),
("ws_ticks", "idx_ws_ticks_src_code_recv", "(source, code, recv_ts)"),
("ws_orderbook", "idx_ws_ob_snap", "(snap_time)"),
("ws_orderbook", "idx_ws_ob_snap_src", "(snap_time, source)"),
("ws_ticks_us", "idx_ws_ticks_us_tick_time", "(tick_time)"),
("ls_ws_orderbook", "idx_ls_ob_snap", "(snap_time)"),
)
for tbl, name, cols in specs:
self._ensure_safe_index(tbl, name, cols)
def _migrate_ws_candles_source_channel(self) -> None:
"""ws_candles.channel 추가 + UNIQUE(code,tf,time,source,channel) + rest 라벨 정규화."""
cols = self.conn.get_columns("ws_candles")
if "channel" not in cols:
self.conn.execute(
"ALTER TABLE ws_candles ADD COLUMN channel VARCHAR(10) NOT NULL DEFAULT 'ws' "
"COMMENT 'ws|rest|rollup' AFTER source"
)
logger.info("📌 ws_candles.channel 컬럼 추가")
# UNIQUE 를 잠시 내려야 remap 충돌이 안 난다
try:
idx_rows = self.conn.execute(
"SHOW INDEX FROM ws_candles WHERE Key_name='uq_candle'"
).fetchall()
idx_cols = [r["Column_name"] for r in (idx_rows or [])]
except Exception:
idx_cols = []
want = ["code", "timeframe", "candle_time", "source", "channel"]
if idx_cols == want:
return
try:
self.conn.execute("ALTER TABLE ws_candles DROP INDEX uq_candle")
except Exception:
pass
try:
leftover = self.conn.execute(
"SELECT 1 AS x FROM ws_candles "
"WHERE source IN ('rest','kw_rest','rollup_1m','rollup','ws','') LIMIT 1"
).fetchone()
except Exception:
leftover = True
if leftover:
try:
self.conn.execute_long(
"UPDATE ws_candles SET source='kiwoom', channel='rest' "
"WHERE source IN ('rest','kw_rest')"
)
self.conn.execute_long(
"UPDATE ws_candles SET source='kiwoom', channel='rollup' "
"WHERE source IN ('rollup_1m','rollup')"
)
self.conn.execute_long(
"UPDATE ws_candles SET source='kis', channel='ws' "
"WHERE source IN ('ws','')"
)
logger.info(
"📌 ws_candles source/channel 정규화 rest/kw_rest→kiwoom+rest, "
"rollup→kiwoom+rollup, ws→kis+ws"
)
except Exception as e:
logger.debug("ws_candles source remap: %s", e)
try:
self.conn.execute_long(
"ALTER TABLE ws_candles ADD UNIQUE KEY uq_candle "
"(code, timeframe, candle_time, source, channel)"
)
logger.info("📌 ws_candles UNIQUE uq_candle=(code,tf,time,source,channel)")
except Exception as e:
logger.warning("ws_candles UNIQUE 마이그레이션 실패: %s", e)
def _migrate_env_config_to_columns(self):
"""env_config가 예전 JSON 컬럼(snapshot_json)이면 컬럼 스키마로 이전"""
try:
cols = self.conn.get_columns("env_config")
if "snapshot_json" not in cols:
return
rows = self.conn.execute(
"SELECT id, created_at, snapshot_json FROM env_config ORDER BY id"
).fetchall()
col_defs = ", ".join([f"`{k}` TEXT" for k in ENV_CONFIG_KEYS])
self.conn.execute(f"""
CREATE TABLE IF NOT EXISTS env_config_new (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
created_at VARCHAR(30) NOT NULL,
{col_defs}
) CHARACTER SET utf8mb4
""")
key_list = ", ".join(f"`{k}`" for k in ENV_CONFIG_KEYS)
placeholders = ", ".join(["%s"] * (1 + len(ENV_CONFIG_KEYS)))
for row in rows:
snap = json.loads(row["snapshot_json"]) if row["snapshot_json"] else {}
vals = [row["created_at"]] + [snap.get(k) for k in ENV_CONFIG_KEYS]
self.conn.execute(
f"INSERT INTO env_config_new (created_at, {key_list}) VALUES ({placeholders})",
vals,
)
self.conn.execute("DROP TABLE env_config")
self.conn.execute("ALTER TABLE env_config_new RENAME TO env_config")
logger.info("📌 env_config: snapshot_json -> 컬럼 스키마 마이그레이션 완료")
except Exception as e:
logger.debug(f"migrate env_config: {e}")
# ============================================================
# [CRUD] Active Trades (활성 트레이딩 관리)
# ============================================================
@staticmethod
def resolve_kis_is_mock(explicit: Any = None) -> int:
"""모의=1 · 실전=0. explicit 없으면 현재 KIS_MOCK."""
if explicit is not None:
if isinstance(explicit, str):
return 1 if explicit.strip().lower() in ("1", "true", "yes", "y", "on", "mock") else 0
return 1 if bool(explicit) else 0
try:
from kis_trader.utils.env import get_env_bool
return 1 if get_env_bool("KIS_MOCK", True) else 0
except Exception:
return 1
def _backfill_is_mock_from_env_snapshot(self, table: str) -> None:
"""env_snapshot JSON 의 KIS_MOCK 으로 is_mock NULL 행 보정 (한 번)."""
try:
rows = self.conn.execute(
f"SELECT id, env_snapshot FROM {table} "
"WHERE is_mock IS NULL AND env_snapshot IS NOT NULL AND env_snapshot<>''"
).fetchall() or []
except Exception:
# active_trades 는 PK 가 id 아님
try:
rows = self.conn.execute(
f"SELECT code, strategy, env_snapshot FROM {table} "
"WHERE is_mock IS NULL AND env_snapshot IS NOT NULL AND env_snapshot<>''"
).fetchall() or []
except Exception as e:
logger.debug("%s is_mock snapshot 백필 조회 실패: %s", table, e)
return
n_ok = 0
for r in rows:
d = dict(r) if not isinstance(r, dict) else r
raw = d.get("env_snapshot") or ""
try:
snap = json.loads(raw) if isinstance(raw, str) else (raw or {})
except Exception:
continue
if not isinstance(snap, dict) or "KIS_MOCK" not in snap:
continue
flag = self.resolve_kis_is_mock(snap.get("KIS_MOCK"))
try:
if "id" in d and d.get("id") is not None and table == "trade_history":
self.conn.execute(
f"UPDATE {table} SET is_mock=%s WHERE id=%s AND is_mock IS NULL",
(flag, d["id"]),
)
else:
self.conn.execute(
f"UPDATE {table} SET is_mock=%s "
"WHERE code=%s AND strategy=%s AND is_mock IS NULL",
(flag, d.get("code"), d.get("strategy")),
)
n_ok += 1
except Exception:
continue
if n_ok:
logger.info("📌 %s.is_mock env_snapshot 백필 %d", table, n_ok)
def upsert_trade(self, trade_data: Dict):
"""
신규 매수하거나 정보 업데이트 (평단가, 수량 등)
Args:
trade_data: 트레이드 정보 딕셔너리
필수: code, name, avg_buy_price, target_qty, current_qty, status
선택: strategy, stop_price, target_price, max_price, atr_entry, total_invested
ML 학습용: entry_features (dict) 또는 rsi, volume_ratio 등 개별 키
"""
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 기본값 설정
code = trade_data.get('code')
if not code:
logger.error("종목코드 누락: upsert 실패")
return False
size_class = trade_data.get('size_class')
is_mock = self.resolve_kis_is_mock(
trade_data["is_mock"] if "is_mock" in trade_data else None
)
feats = trade_data.get('entry_features') or {}
feat_vals = []
for k in ML_ENTRY_FEATURE_COLUMNS:
v = feats.get(k) if k in feats else trade_data.get(k)
feat_vals.append(v if isinstance(v, (int, float)) else None)
cols = ", ".join(["code", "name", "strategy", "avg_buy_price", "current_price", "stop_price", "target_price",
"max_price", "atr_entry", "target_qty", "current_qty", "total_invested",
"status", "buy_date", "updated_at", "size_class", "env_snapshot", "is_mock"]
+ list(ML_ENTRY_FEATURE_COLUMNS))
placeholders = ", ".join(["%s"] * (18 + len(ML_ENTRY_FEATURE_COLUMNS)))
# MySQL: ON DUPLICATE KEY UPDATE (excluded. → VALUES())
updates = (
"avg_buy_price = VALUES(avg_buy_price), current_price = VALUES(current_price), "
"stop_price = COALESCE(VALUES(stop_price), stop_price), "
"target_price = COALESCE(VALUES(target_price), target_price), "
"atr_entry = COALESCE(VALUES(atr_entry), atr_entry), "
"current_qty = VALUES(current_qty), total_invested = VALUES(total_invested), "
"max_price = GREATEST(max_price, VALUES(max_price)), "
"status = VALUES(status), updated_at = VALUES(updated_at), "
"size_class = COALESCE(VALUES(size_class), size_class), "
"env_snapshot = COALESCE(VALUES(env_snapshot), env_snapshot), "
"is_mock = COALESCE(is_mock, VALUES(is_mock))"
)
for c in ML_ENTRY_FEATURE_COLUMNS:
updates += f", `{c}` = COALESCE(VALUES(`{c}`), `{c}`)"
sql = f"""
INSERT INTO active_trades (
{cols}
) VALUES ({placeholders})
ON DUPLICATE KEY UPDATE
{updates}
"""
try:
from kis_trader.utils.strategy_ids import canonical_strategy_id
_stored_strategy = canonical_strategy_id(trade_data.get("strategy", "MANUAL"))
except Exception:
_stored_strategy = trade_data.get("strategy", "MANUAL") or "MANUAL"
params = (
code,
trade_data.get('name', 'Unknown'),
_stored_strategy,
trade_data.get('avg_buy_price') or trade_data.get('buy_price', 0),
trade_data.get('current_price', 0),
trade_data.get('stop_price', 0),
trade_data.get('target_price', 0),
trade_data.get('max_price', trade_data.get('buy_price', 0)),
trade_data.get('atr_at_entry') or trade_data.get('atr_entry', 0),
trade_data.get('target_qty', trade_data.get('qty', 0)),
trade_data.get('current_qty') or trade_data.get('qty', 0),
trade_data.get('total_invested', 0),
trade_data.get('status', 'HOLDING'),
trade_data.get('buy_date', now),
now,
size_class,
trade_data.get('env_snapshot'),
is_mock,
) + tuple(feat_vals)
try:
with self.conn:
self.conn.execute(sql, params)
return True
except Exception as e:
logger.error(f"❌ upsert_trade 실패 ({code}): {e}")
return False
def get_active_trades(self, strategy_prefix: Optional[str] = None):
"""
활성 트레이딩 목록 조회 (봇 재시작 시 사용)
Args:
strategy_prefix: None이면 전부, 'LONG'이면 strategy LIKE 'LONG%'만, 'SHORT''SHORT%'
(늘림목/단타 섞임 방지)
Returns:
{종목코드: {trade_info}} 형태의 딕셔너리
현재 KIS_MOCK 과 같은 is_mock 행만 반환 — 모의 보유가 실전 봇에 안 섞이게.
"""
try:
mock_flag = self.resolve_kis_is_mock(None)
if strategy_prefix:
cursor = self.conn.execute(
"SELECT * FROM active_trades WHERE strategy LIKE %s AND is_mock=%s",
(strategy_prefix.strip().upper() + "%", mock_flag),
)
else:
cursor = self.conn.execute(
"SELECT * FROM active_trades WHERE is_mock=%s",
(mock_flag,),
)
rows = cursor.fetchall()
# 기존 JSON 포맷과 호환되도록 딕셔너리 변환
result = {}
for row in rows:
code = row['code']
result[code] = {
'code': code,
'name': row['name'],
'strategy': row['strategy'],
'buy_price': row['avg_buy_price'], # JSON 호환
'avg_buy_price': row['avg_buy_price'],
'current_price': row['current_price'],
'stop_price': row['stop_price'],
'target_price': row['target_price'],
'max_price': row['max_price'],
'atr_at_entry': row['atr_entry'],
'qty': row['current_qty'], # JSON 호환
'target_qty': row['target_qty'],
'current_qty': row['current_qty'],
'total_invested': row['total_invested'],
'status': row['status'],
'buy_date': row['buy_date'],
'updated_at': row['updated_at'],
'size_class': row['size_class'] if 'size_class' in row.keys() else None,
'is_mock': (
int(row['is_mock'])
if 'is_mock' in row.keys() and row['is_mock'] is not None
else None
),
}
logger.debug(f"📂 활성 트레이드 로드: {len(result)}개 (is_mock={mock_flag})")
return result
except Exception as e:
logger.error(f"❌ get_active_trades 실패: {e}")
return {}
def get_active_trade(self, code: str) -> Optional[Dict]:
"""
활성 트레이딩 단일 종목 조회.
잔고 동기화 시 DB에 저장된 평단가를 폴백용으로 사용할 때 쓴다.
"""
try:
cursor = self.conn.execute(
"SELECT * FROM active_trades WHERE code = ?",
(code,),
)
row = cursor.fetchone()
if not row:
return None
return {
"code": row["code"],
"name": row["name"],
"strategy": row["strategy"],
"avg_buy_price": row["avg_buy_price"],
"current_price": row["current_price"],
"stop_price": row["stop_price"],
"target_price": row["target_price"],
"max_price": row["max_price"],
"atr_entry": row["atr_entry"],
"target_qty": row["target_qty"],
"current_qty": row["current_qty"],
"total_invested": row["total_invested"],
"status": row["status"],
"buy_date": row["buy_date"],
"updated_at": row["updated_at"],
}
except Exception as e:
logger.error(f"❌ get_active_trade 실패 ({code}): {e}")
return None
def update_current_price(self, code: str, current_price: float):
"""현재가 업데이트 (매도 판단용)"""
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
with self.conn:
self.conn.execute(
"UPDATE active_trades SET current_price=?, updated_at=? WHERE code=?",
(current_price, now, code)
)
except Exception as e:
logger.error(f"❌ 현재가 업데이트 실패 ({code}): {e}")
def update_max_price(self, code: str, new_max_price: float):
"""최고가 갱신 (트레일링 스탑용)"""
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
with self.conn:
# 기존 max_price보다 클 때만 업데이트
self.conn.execute(
"""UPDATE active_trades
SET max_price = MAX(max_price, ?), updated_at = ?
WHERE code = ?""",
(new_max_price, now, code)
)
except Exception as e:
logger.error(f"❌ 최고가 갱신 실패 ({code}): {e}")
def close_trade(
self,
code: str,
sell_price: float,
sell_reason: str = "",
env_snapshot: str = None,
size_class: str = None,
strategy: str = None,
realized_pnl_override: float = None,
sell_qty: int = None,
):
"""
매도 완료 처리: active_trades → trade_history (INSERT만, env 스냅샷 포함)
Args:
code: 종목코드
sell_price: 매도가
sell_reason: 매도 사유
env_snapshot: 매도 시점 env JSON (백테스트/대시보드용)
size_class: 대/중/소형 (매수 시점 저장값)
strategy: 봇 전략 ID (SHORT_ANT_SHAKING / SCALP_RSI_REVERSAL 등)
지정 시 해당 전략 row만 삭제 (다른 봇의 동일 종목 보호).
None이면 code 단독 조회 (단일 봇 운영 환경 호환).
realized_pnl_override: 수수료·세금 반영 순손익 (외부 주입)
sell_qty: 부분매도 수량. None/전량 이상이면 기존처럼 전량 청산.
0 < sell_qty < current_qty 이면 history 기록 후 잔량 유지
(시장가 IOC 부분체결 대비).
"""
try:
# 1. 활성 트레이드 정보 조회 (strategy 지정 시 정확히 해당 row만 조회)
if strategy:
cursor = self.conn.execute(
"SELECT * FROM active_trades WHERE code=%s AND strategy=%s",
(code, strategy),
)
else:
cursor = self.conn.execute("SELECT * FROM active_trades WHERE code=%s", (code,))
trade = cursor.fetchone()
if not trade:
logger.warning(f"⚠️ close_trade: {code} 종목이 active_trades에 없음")
return False
# 2. 손익 계산 (부분매도면 체결분만)
buy_price = trade['avg_buy_price']
pos_qty = int(trade['current_qty'] or 0)
if sell_qty is None:
qty = pos_qty
else:
qty = min(max(0, int(sell_qty)), pos_qty)
if qty <= 0:
logger.warning("⚠️ close_trade: %s sell_qty<=0 (pos=%d)", code, pos_qty)
return False
# realized_pnl_override 가 있으면 수수료·세금 반영 순손익을 외부에서 주입
# 없으면 내부 계산 (수수료 미포함 gross)
if realized_pnl_override is not None:
realized_pnl = realized_pnl_override
else:
realized_pnl = (sell_price - buy_price) * qty
profit_rate = (realized_pnl / (buy_price * qty) * 100) if buy_price * qty > 0 else 0
# 3. 보유 시간 계산
buy_time = datetime.datetime.strptime(trade['buy_date'], '%Y-%m-%d %H:%M:%S')
sell_time = datetime.datetime.now()
hold_minutes = int((sell_time - buy_time).total_seconds() / 60)
# size_class는 active_trades에 있으면 그대로 사용
if size_class is None and 'size_class' in trade.keys() and trade['size_class']:
size_class = trade['size_class']
if 'is_mock' in trade.keys() and trade['is_mock'] is not None:
hist_is_mock = self.resolve_kis_is_mock(trade['is_mock'])
else:
hist_is_mock = self.resolve_kis_is_mock(None)
# 4. trade_history에 저장 — strategy 는 kis_trader canonical ID (실거래 탭·봇 ID 일치)
try:
from kis_trader.utils.strategy_ids import canonical_strategy_id
hist_strategy = canonical_strategy_id(trade.get("strategy"))
except Exception:
hist_strategy = trade.get("strategy")
feat_vals = [trade[c] if c in trade.keys() else None for c in ML_ENTRY_FEATURE_COLUMNS]
cols_th = (
"code, name, strategy, buy_price, sell_price, qty, profit_rate, realized_pnl, "
"hold_minutes, buy_date, sell_date, sell_reason, env_snapshot, size_class, is_mock"
)
if ML_ENTRY_FEATURE_COLUMNS:
cols_th += ", " + ", ".join(ML_ENTRY_FEATURE_COLUMNS)
placeholders = ", ".join(["?"] * (15 + len(ML_ENTRY_FEATURE_COLUMNS)))
remain_qty = pos_qty - qty
inv = float(trade["total_invested"] or 0) if "total_invested" in trade.keys() else 0.0
new_inv = (inv * remain_qty / pos_qty) if pos_qty > 0 and remain_qty > 0 else 0.0
with self.conn:
self.conn.execute(f"""
INSERT INTO trade_history (
{cols_th}
) VALUES ({placeholders})
""", (
trade['code'],
trade['name'],
hist_strategy,
buy_price,
sell_price,
qty,
profit_rate,
realized_pnl,
hold_minutes,
trade['buy_date'],
sell_time.strftime('%Y-%m-%d %H:%M:%S'),
sell_reason,
env_snapshot,
size_class,
hist_is_mock,
) + tuple(feat_vals))
# 5. 전량이면 삭제, 부분이면 잔량·투입금 축소
if remain_qty <= 0:
if strategy:
self.conn.execute(
"DELETE FROM active_trades WHERE code=%s AND strategy=%s",
(code, strategy),
)
else:
self.conn.execute("DELETE FROM active_trades WHERE code=%s", (code,))
else:
if strategy:
self.conn.execute(
"UPDATE active_trades SET current_qty=%s, target_qty=%s, "
"total_invested=%s, updated_at=%s "
"WHERE code=%s AND strategy=%s",
(
remain_qty,
remain_qty,
new_inv,
sell_time.strftime("%Y-%m-%d %H:%M:%S"),
code,
strategy,
),
)
else:
self.conn.execute(
"UPDATE active_trades SET current_qty=%s, target_qty=%s, "
"total_invested=%s, updated_at=%s WHERE code=%s",
(
remain_qty,
remain_qty,
new_inv,
sell_time.strftime("%Y-%m-%d %H:%M:%S"),
code,
),
)
if remain_qty <= 0:
logger.info(f"✅ [{trade['name']}] 매매 종료: 수익률 {profit_rate:.2f}% ({realized_pnl:+,.0f}원)")
else:
logger.info(
"✅ [%s] 부분매도 %d/%d주: 수익률 %.2f%% (%+.0f원) · 잔량 %d",
trade["name"], qty, pos_qty, profit_rate, realized_pnl, remain_qty,
)
return True
except Exception as e:
logger.error(f"❌ close_trade 실패 ({code}): {e}")
return False
def delete_active_trade(self, code: str, strategy: str = None):
"""
활성 트레이드 삭제 (긴급 정리용).
strategy 지정 시 해당 봇 row만 삭제, None이면 해당 종목 전체 삭제.
"""
try:
with self.conn:
if strategy:
self.conn.execute(
"DELETE FROM active_trades WHERE code=%s AND strategy=%s",
(code, strategy),
)
else:
self.conn.execute("DELETE FROM active_trades WHERE code=%s", (code,))
logger.info(f"🗑️ active_trade 삭제: {code}" + (f" [{strategy}]" if strategy else ""))
return True
except Exception as e:
logger.error(f"❌ 삭제 실패 ({code}): {e}")
return False
def insert_buy_execution(
self,
code: str,
name: str,
strategy: str,
amount: float,
qty: int,
):
"""
매수 체결 이력 저장 (일일 한도용). '하루' = 산 날짜(buy_date) 기준.
"""
now = datetime.datetime.now()
buy_date = now.strftime("%Y-%m-%d")
executed_at = now.strftime("%Y-%m-%d %H:%M:%S")
try:
with self.conn:
self.conn.execute("""
INSERT INTO buy_execution_log
(code, name, strategy, buy_date, executed_at, amount, qty, is_mock)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
""", (
code, name, strategy, buy_date, executed_at, amount, qty,
self.resolve_kis_is_mock(None),
))
return True
except Exception as e:
logger.error(f"❌ insert_buy_execution 실패 ({code}): {e}")
return False
def get_daily_buy_amount(self, date_str: str, strategy_prefix: str = "LONG") -> Tuple[float, int]:
"""
해당 날짜(산 시점 기준)에 strategy_prefix에 해당하는 매수 누적 금액·건수.
date_str: YYYY-MM-DD · 현재 KIS_MOCK 계좌만 합산.
Returns:
(누적 금액, 건수)
"""
try:
mock_flag = self.resolve_kis_is_mock(None)
cursor = self.conn.execute("""
SELECT COALESCE(SUM(amount), 0) AS total_amount, COUNT(*) AS cnt
FROM buy_execution_log
WHERE buy_date = %s AND strategy LIKE %s AND is_mock = %s
""", (date_str, strategy_prefix.strip().upper() + "%", mock_flag))
row = cursor.fetchone()
return (float(row["total_amount"]), int(row["cnt"]))
except Exception as e:
logger.error(f"❌ get_daily_buy_amount 실패: {e}")
return (0.0, 0)
# ============================================================
# [보강] 주문·체결 이력 (kt00007 / ka10076)
# ============================================================
def insert_order_execution(
self, source: str, row: dict, ord_dt: str = None, sell_tp: str = None, raw_json: str = None
):
"""주문·체결 1건 INSERT (보강용, 이력만 쌓음)"""
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
self.conn.execute("""
INSERT INTO order_execution_history (
source, ord_no, stk_cd, stk_nm, trde_tp, ord_qty, ord_uv,
cntr_qty, cntr_uv, ord_tm, cnfm_tm, sell_tp, ord_dt, raw_json, fetched_at, is_mock
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""", (
source,
row.get('ord_no') or row.get('orig_ord_no'),
row.get('stk_cd', ''),
row.get('stk_nm', ''),
row.get('trde_tp', ''),
str(row.get('ord_qty', '') or row.get('cntr_qty', '')),
str(row.get('ord_uv', '') or row.get('ord_pric', '') or row.get('cntr_uv', '')),
str(row.get('cntr_qty', '') or row.get('cnfm_qty', '')),
str(row.get('cntr_uv', '') or row.get('cntr_pric', '')),
row.get('ord_tm', ''),
row.get('cnfm_tm', ''),
sell_tp or '',
ord_dt or '',
raw_json,
now,
self.resolve_kis_is_mock(None),
))
self.conn.commit()
return True
except Exception as e:
logger.debug(f"insert_order_execution: {e}")
return False
# ============================================================
# [분석] 켈리 공식 및 통계 계산
# ============================================================
def calculate_half_kelly(self, recent_days: int = 30) -> float:
"""
하프 켈리 공식 계산 (과거 매매 기록 기반)
Args:
recent_days: 최근 N일 데이터만 사용
Returns:
하프 켈리 비율 (0.0 ~ 1.0)
예: 0.15 리턴 -> "예수금의 15%씩 배팅하는 게 최적"
"""
try:
# 최근 N일 데이터 조회
cutoff_date = (datetime.datetime.now() - datetime.timedelta(days=recent_days)).strftime('%Y-%m-%d')
cursor = self.conn.execute(
"SELECT profit_rate FROM trade_history WHERE sell_date >= ? ORDER BY sell_date DESC",
(cutoff_date,)
)
rows = cursor.fetchall()
if len(rows) < 20: # 최소 20건 이상 필요
logger.warning(f"⚠️ 켈리 공식: 데이터 부족 ({len(rows)}건) -> 기본값 10% 리턴")
return 0.10
# 승률 계산
wins = [r['profit_rate'] for r in rows if r['profit_rate'] > 0]
losses = [r['profit_rate'] for r in rows if r['profit_rate'] <= 0]
total_count = len(rows)
win_count = len(wins)
win_rate = win_count / total_count
loss_rate = 1.0 - win_rate
# 손익비 계산 (평균 수익 / 평균 손실)
if not wins or not losses:
logger.warning("⚠️ 켈리 공식: 승 또는 패만 있음 -> 기본값 10%")
return 0.10
avg_win = sum(wins) / len(wins)
avg_loss = abs(sum(losses) / len(losses))
if avg_loss == 0:
return 0.50 # 손실이 0이면 최대치
odds = avg_win / avg_loss
# 켈리 공식: f = (p * b - q) / b
# p=승률, b=손익비, q=패율
kelly_fraction = ((win_rate * odds) - loss_rate) / odds
# 하프 켈리 (안전성 확보)
half_kelly = kelly_fraction * 0.5
# 음수면 0 리턴 (통계적으로 지는 구조)
final_kelly = max(0.0, min(half_kelly, 0.5)) # 최대 50%로 제한
logger.info(
f"📊 [켈리 분석] 승률:{win_rate*100:.1f}% | 손익비:{odds:.2f} | "
f"켈리:{kelly_fraction*100:.1f}% | 하프켈리:{final_kelly*100:.1f}%"
)
return final_kelly
except Exception as e:
logger.error(f"❌ 켈리 계산 실패: {e}")
return 0.10
def get_recent_performance(self, days: int = 7) -> Tuple[float, int, int]:
"""
최근 N일 성과 조회
Returns:
(총손익, 익절횟수, 손절횟수)
"""
try:
cutoff = (datetime.datetime.now() - datetime.timedelta(days=days)).strftime('%Y-%m-%d')
cursor = self.conn.execute(
"SELECT realized_pnl FROM trade_history WHERE sell_date >= ?",
(cutoff,)
)
rows = cursor.fetchall()
total_pnl = sum([r['realized_pnl'] for r in rows])
wins = len([r for r in rows if r['realized_pnl'] > 0])
losses = len([r for r in rows if r['realized_pnl'] <= 0])
return total_pnl, wins, losses
except Exception as e:
logger.error(f"❌ 성과 조회 실패: {e}")
return 0.0, 0, 0
def get_trade_stats(self) -> Dict:
"""전체 매매 통계"""
try:
cursor = self.conn.execute("""
SELECT
COUNT(*) as total,
SUM(CASE WHEN profit_rate > 0 THEN 1 ELSE 0 END) as wins,
AVG(profit_rate) as avg_profit_rate,
SUM(realized_pnl) as total_pnl
FROM trade_history
""")
row = cursor.fetchone()
return {
'total_trades': row['total'] or 0,
'win_trades': row['wins'] or 0,
'win_rate': (row['wins'] / row['total'] * 100) if row['total'] > 0 else 0,
'avg_profit_rate': row['avg_profit_rate'] or 0,
'total_pnl': row['total_pnl'] or 0
}
except Exception as e:
logger.error(f"❌ 통계 조회 실패: {e}")
return {}
# ============================================================
# [유틸] JSON 마이그레이션
# ============================================================
def migrate_from_json(self, json_data: Dict):
"""
기존 JSON 포트폴리오를 DB로 마이그레이션
Args:
json_data: portfolio.json 내용 (딕셔너리)
"""
count = 0
for code, info in json_data.items():
trade_data = info.copy()
trade_data['code'] = code
# 필드 매핑 (JSON -> DB)
if 'target_qty' not in trade_data:
trade_data['target_qty'] = info.get('qty', 0)
if 'current_qty' not in trade_data:
trade_data['current_qty'] = info.get('qty', 0)
if 'total_invested' not in trade_data:
trade_data['total_invested'] = info.get('buy_price', 0) * info.get('qty', 0)
if 'status' not in trade_data:
trade_data['status'] = 'HOLDING'
if self.upsert_trade(trade_data):
count += 1
logger.info(f"✅ JSON -> DB 마이그레이션 완료: {count}개 종목")
return count
# ============================================================
# [CRUD] Target Candidates (매수 후보군 관리)
# ============================================================
def update_target_candidates(self, candidates: List[Dict]):
"""
매수 후보군 업데이트 (5분마다 호출)
Args:
candidates: [{
'code': '005930', 'name': '삼성전자',
'score': 5.2, 'price': 75000,
'market': 'K', # 선택 (없으면 stock_meta에서 조회)
'sector': '반도체', # 선택
'theme': 'AI반도체' # 선택
}, ...]
"""
try:
scan_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 5분 슬롯 키 (백테스트 유니버스 이력 조회용): 2026-03-18 15:55:21 → 202603181555
_now = datetime.datetime.now()
slot_min = (_now.hour * 60 + _now.minute) // 5 * 5
slot_key = _now.strftime("%Y%m%d") + "%02d%02d" % (slot_min // 60, slot_min % 60)
# 이력 테이블 없으면 생성 시도 (마이그레이션 누락/실패 시 자동 보정)
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS target_candidates_history (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
slot_key VARCHAR(12) NOT NULL,
scan_time VARCHAR(30) NOT NULL,
code VARCHAR(20) NOT NULL,
name VARCHAR(100) NOT NULL DEFAULT '',
score DOUBLE NOT NULL DEFAULT 0,
price DOUBLE NOT NULL DEFAULT 0,
market CHAR(1) DEFAULT 'Q',
sector VARCHAR(100),
theme VARCHAR(100),
INDEX idx_slot (slot_key),
INDEX idx_scan (scan_time)
) CHARACTER SET utf8mb4
""")
except Exception as e:
logger.debug("target_candidates_history CREATE 확인: %s", e)
# stock_meta 캐시: 테마/섹터가 없는 항목을 DB에서 보완
meta_cache: Dict[str, Dict] = {}
# 기존 데이터 전체 삭제 (5분마다 새로 갱신)
with self.conn:
self.conn.execute("DELETE FROM target_candidates")
for item in candidates:
code = item['code']
# market/sector/theme 없으면 stock_meta에서 조회
market = item.get('market') or ""
sector = item.get('sector') or ""
theme = item.get('theme') or ""
if not (market and sector and theme):
if code not in meta_cache:
try:
r = self.conn.execute(
"SELECT market, sector, theme FROM stock_meta WHERE code=%s",
(code,)
).fetchone()
meta_cache[code] = dict(r) if r else {}
except Exception:
meta_cache[code] = {}
m = meta_cache[code]
market = market or m.get("market") or "Q"
sector = sector or m.get("sector") or ""
theme = theme or m.get("theme") or ""
self.conn.execute("""
INSERT INTO target_candidates
(code, name, score, price, scan_time, updated_at, market, sector, theme)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
""", (
code,
item.get('name', ''),
item.get('score', 0),
item.get('price', 0),
scan_time,
scan_time,
market,
sector,
theme,
))
# [DEPRECATED] 구봇(kiwoom_universe_scanner 등) 의 5분 슬롯 이력 적재는 제거됨.
# 신봇(kis_trader/**) 은 RankingManager/ConditionSearchManager 가
# ``insert_condition_universe_snapshot()`` 으로 초단위 event_time 스냅샷을
# 기록한다. 구봇이 우발적으로 실행되더라도 이력 혼선을 일으키지 않도록
# 이 경로에서의 history 적재는 의도적으로 하지 않는다.
# (과거 코드: INSERT INTO target_candidates_history ...)
logger.info(f"✅ 매수 후보군 DB 저장: {len(candidates)}개 (slot_key={slot_key}, 이력 적재 안함)")
return True
except Exception as e:
logger.error(f"❌ 후보군 저장 실패: {e}")
return False
def add_target_candidate(self, candidate: Dict):
"""
매수 후보군 개별 추가 (통과 즉시 저장용, UPSERT 방식)
- 500개 스캔 시 시간이 오래 걸려서 통과하는 즉시 DB에 저장
Args:
candidate: {
'code': '005930', 'name': '삼성전자',
'score': 5.2, 'price': 75000,
'market': 'K', 'sector': '반도체', 'theme': 'AI반도체' # 선택
}
"""
try:
scan_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
code = candidate['code']
market = candidate.get('market') or ""
sector = candidate.get('sector') or ""
theme = candidate.get('theme') or ""
# 없으면 stock_meta 에서 보완
if not (market and sector and theme):
try:
r = self.conn.execute(
"SELECT market, sector, theme FROM stock_meta WHERE code=%s",
(code,)
).fetchone()
if r:
m = dict(r)
market = market or m.get("market") or "Q"
sector = sector or m.get("sector") or ""
theme = theme or m.get("theme") or ""
except Exception:
pass
with self.conn:
self.conn.execute("""
INSERT INTO target_candidates
(code, name, score, price, scan_time, updated_at, market, sector, theme)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
name = VALUES(name),
score = VALUES(score),
price = VALUES(price),
scan_time = VALUES(scan_time),
updated_at = VALUES(updated_at),
market = VALUES(market),
sector = VALUES(sector),
theme = VALUES(theme)
""", (
code,
candidate.get('name', ''),
candidate.get('score', 0),
candidate.get('price', 0),
scan_time, scan_time,
market, sector, theme,
))
return True
except Exception as e:
logger.debug(f"후보 개별 저장 실패({candidate.get('code', '')}): {e}")
return False
def get_target_candidates(self) -> List[Dict]:
"""
매수 후보군 조회 (점수 순).
stock_meta JOIN으로 테마/섹터 보완 반환.
Returns:
[{'code': '005930', 'name': '삼성전자', 'score': 5.2, 'price': 75000,
'market': 'K', 'sector': '반도체', 'theme': 'AI반도체'}, ...]
"""
try:
cursor = self.conn.execute("""
SELECT
t.code, t.name, t.score, t.price, t.scan_time,
COALESCE(t.market, m.market, 'Q') AS market,
COALESCE(t.sector, m.sector, '') AS sector,
COALESCE(t.theme, m.theme, '') AS theme
FROM target_candidates t
LEFT JOIN stock_meta m ON m.code = t.code
ORDER BY t.score DESC, t.price ASC
""")
rows = cursor.fetchall()
result = []
for row in rows:
result.append({
'code': row['code'],
'name': row['name'],
'score': row['score'],
'price': row['price'],
'scan_time': row['scan_time'],
'market': row.get('market', 'Q'),
'sector': row.get('sector', ''),
'theme': row.get('theme', ''),
})
return result
except Exception as e:
logger.error(f"❌ 후보군 조회 실패: {e}")
return []
def get_universe_history_for_backtest(
self, start_ymd: str, end_ymd: str
) -> Dict[str, List[str]]:
"""
target_candidates_history에서 기간 내 5분별 유니버스 조회.
백테스트 시 '실제 그 시각에 봇이 보던 후보'로 검사할 때 사용.
Args:
start_ymd: 시작일 YYYYMMDD
end_ymd: 종료일 YYYYMMDD
Returns:
slot_key(YYYYMMDDHHMM) → 해당 슬롯 후보 코드 리스트 (점수 순 유지)
"""
try:
start_key = start_ymd + "0000"
end_key = end_ymd + "2359"
rows = self.conn.execute("""
SELECT slot_key, code
FROM target_candidates_history
WHERE slot_key >= %s AND slot_key <= %s
ORDER BY slot_key, score DESC
""", (start_key, end_key)).fetchall()
out: Dict[str, List[str]] = {}
for r in rows:
# 백테스트 엔진 _slot_key()는 항상 str → 조회 키도 str 통일 (DB 드라이버에 따라 int 올 수 있음)
sk = str(r["slot_key"])
if sk not in out:
out[sk] = []
out[sk].append(r["code"])
return out
except Exception as e:
logger.debug(f"후보 이력 조회 실패: {e}")
return {}
# ------------------------------------------------------------------
# stock_meta 테마/섹터 메타데이터 헬퍼
# ------------------------------------------------------------------
def upsert_stock_meta(
self,
code: str,
name: str = "",
market: str = "Q",
sector_code: str = "",
sector: str = "",
theme: str = "",
theme_rank: int = 3,
) -> bool:
"""
종목 메타데이터 저장/갱신 (스캐너·수동 입력 모두 사용).
Args:
code : 종목코드 (6자리)
name : 종목명
market : 'K'=KOSPI, 'Q'=KOSDAQ, 'E'=ETF
sector_code: 업종코드 (KIS bstp_cls_code 등)
sector : 업종명 (예: '반도체')
theme : 테마명 (예: 'AI반도체', '2차전지')
theme_rank : 테마 내 순위 (1=핵심주, 2=연관, 3=주변)
"""
try:
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
with self.conn:
self.conn.execute("""
INSERT INTO stock_meta
(code, name, market, sector_code, sector, theme, theme_rank, updated_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
name = IF(VALUES(name) != '', VALUES(name), name),
market = IF(VALUES(market) != '', VALUES(market), market),
sector_code = IF(VALUES(sector_code) != '', VALUES(sector_code), sector_code),
sector = IF(VALUES(sector) != '', VALUES(sector), sector),
theme = IF(VALUES(theme) != '', VALUES(theme), theme),
theme_rank = IF(VALUES(theme) != '', VALUES(theme_rank), theme_rank),
updated_at = VALUES(updated_at)
""", (code, name, market, sector_code, sector, theme, theme_rank, now))
return True
except Exception as e:
logger.debug("upsert_stock_meta 실패(%s): %s", code, e)
return False
def get_stock_meta(self, code: str) -> Optional[Dict]:
"""종목 메타데이터 단건 조회. 없으면 None."""
try:
row = self.conn.execute(
"SELECT * FROM stock_meta WHERE code = %s", (code,)
).fetchone()
return dict(row) if row else None
except Exception as e:
logger.debug("get_stock_meta 실패(%s): %s", code, e)
return None
def get_theme_momentum(self, theme: str, tf: int = 60) -> Dict:
"""
테마 종목들의 최신 확정 봉(timeframe=tf) RSI 통계.
"이 테마가 지금 뜨겁냐 식었냐"를 숫자로 파악.
Args:
theme: 테마명 (stock_meta.theme)
tf : 봉 단위 분 (기본 60분봉)
Returns:
{
'theme': 'AI반도체',
'tf': 60,
'count': 5, # 데이터 있는 종목 수
'avg_rsi3': 62.1, # 평균 RSI(3)
'max_rsi3': 78.4, # 최고 RSI(3)
'min_rsi3': 44.2, # 최저 RSI(3)
'hot_count': 2, # RSI>70 (과열) 종목 수
'cold_count': 1, # RSI<30 (과매도) 종목 수
}
"""
try:
row = self.conn.execute("""
SELECT
COUNT(*) AS cnt,
AVG(w.rsi_3) AS avg_rsi,
MAX(w.rsi_3) AS max_rsi,
MIN(w.rsi_3) AS min_rsi,
SUM(w.rsi_3 > 70) AS hot_cnt,
SUM(w.rsi_3 < 30) AS cold_cnt
FROM ws_candles w
INNER JOIN stock_meta m ON m.code = w.code
WHERE m.theme = %s
AND w.timeframe = %s
AND w.source = 'kis'
AND w.is_confirmed = 1
AND w.candle_time = (
SELECT MAX(candle_time)
FROM ws_candles
WHERE code = w.code AND timeframe = w.timeframe AND source = 'kis'
)
""", (theme, tf)).fetchone()
if not row or not row['cnt']:
return {'theme': theme, 'tf': tf, 'count': 0}
return {
'theme': theme,
'tf': tf,
'count': int(row['cnt'] or 0),
'avg_rsi3': round(float(row['avg_rsi'] or 0), 1),
'max_rsi3': round(float(row['max_rsi'] or 0), 1),
'min_rsi3': round(float(row['min_rsi'] or 0), 1),
'hot_count': int(row['hot_cnt'] or 0),
'cold_count':int(row['cold_cnt'] or 0),
}
except Exception as e:
logger.debug("get_theme_momentum 실패(%s, %dM): %s", theme, tf, e)
return {'theme': theme, 'tf': tf, 'count': 0}
def get_all_theme_momentum(self, tf: int = 60) -> List[Dict]:
"""
모든 테마의 최신 모멘텀 요약 (테마 히트맵).
가장 뜨거운 테마부터 정렬 반환.
"""
try:
rows = self.conn.execute("""
SELECT
m.theme,
COUNT(DISTINCT w.code) AS cnt,
AVG(w.rsi_3) AS avg_rsi,
SUM(w.rsi_3 > 70) AS hot_cnt,
SUM(w.rsi_3 < 30) AS cold_cnt
FROM ws_candles w
INNER JOIN stock_meta m ON m.code = w.code
WHERE m.theme IS NOT NULL AND m.theme != ''
AND w.timeframe = %s
AND w.source = 'kis'
AND w.is_confirmed = 1
AND w.candle_time = (
SELECT MAX(candle_time)
FROM ws_candles
WHERE code = w.code AND timeframe = w.timeframe AND source = 'kis'
)
GROUP BY m.theme
ORDER BY avg_rsi DESC
""", (tf,)).fetchall()
return [{
'theme': r['theme'],
'count': int(r['cnt'] or 0),
'avg_rsi3': round(float(r['avg_rsi'] or 0), 1),
'hot_count': int(r['hot_cnt'] or 0),
'cold_count':int(r['cold_cnt'] or 0),
} for r in rows]
except Exception as e:
logger.debug("get_all_theme_momentum 실패: %s", e)
return []
def get_market_regime(self, codes: list = None, tf: int = 60) -> Dict:
"""
영구 구독 ETF (KODEX200 등)의 최신 60분봉 RSI로 시장 방향 판단.
스캘핑/꼬리잡기 진입 전 "지금 상승장인가?" 필터로 활용.
Args:
codes: ETF 코드 리스트 (None=env PERMANENT_WS_CODES)
tf : 봉 단위 분 (기본 60)
Returns:
{
'is_bull': True, # avg_rsi > 50 → 상승 국면
'avg_rsi': 58.4,
'details': [{'code': '069500', 'rsi': 61.2}, ...]
}
"""
try:
if not codes:
raw = self.conn.execute(
"SELECT PERMANENT_WS_CODES FROM env_config ORDER BY id DESC LIMIT 1"
).fetchone()
perm = str(raw['PERMANENT_WS_CODES'] if raw else "") if raw else ""
codes = [c.strip() for c in perm.split(",") if c.strip()] or ["069500", "229200"]
rows = self.conn.execute("""
SELECT code, rsi_3
FROM ws_candles
WHERE code IN ({})
AND timeframe = %s
AND source = 'kis'
AND is_confirmed = 1
AND candle_time = (
SELECT MAX(candle_time) FROM ws_candles
WHERE code = ws_candles.code AND timeframe = ws_candles.timeframe AND source = 'kis'
)
""".format(",".join(["%s"] * len(codes))), (*codes, tf)).fetchall()
details = [{'code': r['code'], 'rsi': round(float(r['rsi_3'] or 50), 1)} for r in rows]
avg_rsi = sum(d['rsi'] for d in details) / len(details) if details else 50.0
return {
'is_bull': avg_rsi > 50,
'avg_rsi': round(avg_rsi, 1),
'details': details,
}
except Exception as e:
logger.debug("get_market_regime 실패: %s", e)
return {'is_bull': True, 'avg_rsi': 50.0, 'details': []}
def get_trades_by_date(self, date_str: str) -> List[Dict]:
"""
특정 날짜의 매매 기록 조회
Args:
date_str: 날짜 (YYYYMMDD 또는 YYYY-MM-DD 모두 허용)
Returns:
매매 기록 리스트
"""
try:
# YYYYMMDD(8자) → YYYY-MM-DD 변환, 이미 YYYY-MM-DD(10자)면 그대로 사용
if len(date_str) == 8 and "-" not in date_str:
date_formatted = f"{date_str[:4]}-{date_str[4:6]}-{date_str[6:]}"
else:
date_formatted = date_str[:10] # 'YYYY-MM-DD'만 사용 (시간 부분 제거)
cursor = self.conn.execute("""
SELECT * FROM trade_history
WHERE DATE(sell_date) = %s
ORDER BY sell_date DESC
""", (date_formatted,))
rows = cursor.fetchall()
result = []
for row in rows:
result.append({
'id': row['id'],
'code': row['code'],
'name': row['name'],
'strategy': row['strategy'],
'buy_price': row['buy_price'],
'sell_price': row['sell_price'],
'qty': row['qty'],
'profit_rate': row['profit_rate'],
'realized_pnl': row['realized_pnl'],
'hold_minutes': row['hold_minutes'],
'buy_date': row['buy_date'],
'sell_date': row['sell_date'],
'sell_reason': row['sell_reason']
})
return result
except Exception as e:
logger.error(f"❌ 날짜별 조회 실패: {e}")
return []
# ============================================================
# [ws_price_validation] KIS↔키움 시세 비교 검증
# ============================================================
def insert_ws_price_validation(
self,
*,
code: str,
kis_price: Optional[float],
kiwoom_price: Optional[float],
kis_age_ms: Optional[int] = None,
kiwoom_age_ms: Optional[int] = None,
) -> bool:
"""단일 비교 결과 1행 INSERT.
둘 다 None 이면 저장 안 함. 한쪽만 있어도 저장(소스별 가용성 분석용).
diff_pct 는 둘 다 있을 때만 계산.
"""
if kis_price is None and kiwoom_price is None:
return False
diff_pct: Optional[float] = None
if kis_price not in (None, 0) and kiwoom_price is not None:
try:
diff_pct = (float(kiwoom_price) - float(kis_price)) / float(kis_price) * 100.0
except (ValueError, ZeroDivisionError):
diff_pct = None
try:
now = datetime.datetime.now()
self.conn.execute(
"INSERT INTO ws_price_validation "
"(ts, code, kis_price, kiwoom_price, diff_pct, kis_age_ms, kiwoom_age_ms) "
"VALUES (%s, %s, %s, %s, %s, %s, %s)",
(now, code, kis_price, kiwoom_price, diff_pct, kis_age_ms, kiwoom_age_ms),
)
return True
except Exception as e:
logger.debug("ws_price_validation INSERT 실패: %s", e)
return False
def insert_ws_price_validation_ls(
self,
*,
code: str,
kis_price: Optional[float],
kiwoom_price: Optional[float],
ls_price: Optional[float],
kis_age_ms: Optional[int] = None,
kiwoom_age_ms: Optional[int] = None,
ls_age_ms: Optional[int] = None,
) -> bool:
"""KIS/키움/LS 3자 비교 1행 (그림자 검증 전용)."""
if kis_price is None and kiwoom_price is None and ls_price is None:
return False
diff_kis_ls = None
diff_kw_ls = None
try:
if kis_price not in (None, 0) and ls_price is not None:
diff_kis_ls = (float(ls_price) - float(kis_price)) / float(kis_price) * 100.0
if kiwoom_price not in (None, 0) and ls_price is not None:
diff_kw_ls = (float(ls_price) - float(kiwoom_price)) / float(kiwoom_price) * 100.0
except (ValueError, ZeroDivisionError):
pass
try:
now = datetime.datetime.now()
self.conn.execute(
"INSERT INTO ws_price_validation_ls "
"(ts, code, kis_price, kiwoom_price, ls_price, "
"diff_kis_ls_pct, diff_kw_ls_pct, kis_age_ms, kiwoom_age_ms, ls_age_ms) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
(
now, code, kis_price, kiwoom_price, ls_price,
diff_kis_ls, diff_kw_ls, kis_age_ms, kiwoom_age_ms, ls_age_ms,
),
)
return True
except Exception as e:
logger.debug("ws_price_validation_ls INSERT 실패: %s", e)
return False
def insert_ls_ws_tick(
self,
*,
code: str,
ts,
price: float,
volume: Optional[float] = None,
tot_volume: Optional[float] = None,
chetime: str = "",
tr_cd: str = "",
tk: Optional[str] = None,
lag_sec: Optional[float] = None,
) -> bool:
try:
self.conn.execute(
"INSERT INTO ls_ws_ticks "
"(ts, code, price, volume, tot_volume, chetime, tr_cd, tk, lag_sec) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)",
(ts, code, price, volume, tot_volume, chetime or None, tr_cd or None, tk, lag_sec),
)
return True
except Exception as e:
logger.debug("ls_ws_ticks INSERT 실패: %s", e)
return False
def upsert_ls_ws_candle(
self,
*,
code: str,
candle: Dict[str, Any],
) -> bool:
"""ls_ws_candles upsert (확정된 1분 버킷)."""
try:
now = datetime.datetime.now()
self.conn.execute(
"INSERT INTO ls_ws_candles "
"(code, datetime, tf_min, open, high, low, close, volume, tick_count, updated_at) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
"ON DUPLICATE KEY UPDATE "
"open=VALUES(open), high=VALUES(high), low=VALUES(low), close=VALUES(close), "
"volume=VALUES(volume), tick_count=VALUES(tick_count), updated_at=VALUES(updated_at)",
(
code,
candle.get("datetime"),
int(candle.get("tf_min") or 1),
float(candle["open"]),
float(candle["high"]),
float(candle["low"]),
float(candle["close"]),
float(candle.get("volume") or 0),
int(candle.get("tick_count") or 0),
now,
),
)
return True
except Exception as e:
logger.debug("ls_ws_candles upsert 실패: %s", e)
return False
def insert_ls_ws_candle_if_absent(
self,
*,
code: str,
candle: Dict[str, Any],
) -> bool:
"""ls_ws_candles 구멍만 INSERT. 이미 있으면 덮지 않음 (영구구독 확정봉 버튼)."""
try:
now = datetime.datetime.now()
cur = self.conn.execute(
"INSERT IGNORE INTO ls_ws_candles "
"(code, datetime, tf_min, open, high, low, close, volume, tick_count, updated_at) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
(
code,
candle.get("datetime"),
int(candle.get("tf_min") or 1),
float(candle["open"]),
float(candle["high"]),
float(candle["low"]),
float(candle["close"]),
float(candle.get("volume") or 0),
int(candle.get("tick_count") or 0),
now,
),
)
try:
return bool(cur.rowcount)
except Exception:
return True
except Exception as e:
logger.debug("ls_ws_candles INSERT IGNORE 실패: %s", e)
return False
@staticmethod
def _ls_datetime_to_candle_time(dt_s: str) -> str:
"""``YYYY-MM-DD HH:MM:00`` → ``YYYYMMDDHHMM`` (전략/백테 호환)."""
s = (dt_s or "").strip()
if len(s) >= 16 and s[4] == "-" and s[10] == " ":
return s[0:4] + s[5:7] + s[8:10] + s[11:13] + s[14:16]
digits = "".join(ch for ch in s if ch.isdigit())
return digits[:12]
@staticmethod
def _candle_time_to_ls_datetime(ct: str) -> str:
"""``YYYYMMDDHHMM`` → ``YYYY-MM-DD HH:MM:00``."""
d = "".join(ch for ch in (ct or "") if ch.isdigit())[:12]
if len(d) < 12:
return ""
return f"{d[0:4]}-{d[4:6]}-{d[6:8]} {d[8:10]}:{d[10:12]}:00"
def get_ls_ws_candles(
self,
code: str,
tf_min: int = 1,
limit: int = 100,
*,
start_dt: Optional[str] = None,
end_dt: Optional[str] = None,
as_strategy_schema: bool = True,
) -> list:
"""
ls_ws_candles 최근/기간 조회.
as_strategy_schema=True → candle_time(YYYYMMDDHHMM), is_confirmed=1 등
CandleAggregator/get_ws_candles 호환 dict.
"""
try:
tf = max(1, int(tf_min or 1))
lim = max(1, min(int(limit or 100), 2000))
params: list = [code, tf]
cond = "code=%s AND tf_min=%s"
if start_dt:
cond += " AND datetime >= %s"
params.append(start_dt)
if end_dt:
cond += " AND datetime <= %s"
params.append(end_dt)
params.append(lim)
rows = self.conn.execute(
f"SELECT datetime, open, high, low, close, volume, tick_count, tf_min "
f"FROM ls_ws_candles WHERE {cond} "
f"ORDER BY datetime DESC LIMIT %s",
tuple(params),
).fetchall()
result = []
for r in reversed(list(rows or [])):
d = dict(r)
if as_strategy_schema:
ct = self._ls_datetime_to_candle_time(str(d.get("datetime") or ""))
result.append({
"candle_time": ct,
"open": float(d.get("open") or 0),
"high": float(d.get("high") or 0),
"low": float(d.get("low") or 0),
"close": float(d.get("close") or 0),
"volume": float(d.get("volume") or 0),
"tick_count": int(d.get("tick_count") or 0),
"is_confirmed": 1,
"source": "ls",
"rsi_2": None,
"rsi_3": None,
"rsi_5": None,
})
else:
result.append(d)
return result
except Exception as e:
logger.error("get_ls_ws_candles 실패(%s): %s", code, e)
return []
def get_ls_ws_candles_range(
self,
code: str,
start_key: str,
end_key: str,
tf_min: int = 1,
) -> list:
"""백테용: candle_time 12자리 범위 → ls_ws_candles 조회 (스키마 변환)."""
s_dt = self._candle_time_to_ls_datetime(start_key)
e_dt = self._candle_time_to_ls_datetime(end_key)
if not s_dt or not e_dt:
return []
return self.get_ls_ws_candles(
code,
tf_min=tf_min,
limit=5000,
start_dt=s_dt,
end_dt=e_dt,
as_strategy_schema=True,
)
def insert_ls_ws_orderbook(
self,
*,
code: str,
snap: Dict[str, Any],
market: str = "KR",
tk: Optional[str] = None,
lag_sec: Optional[float] = None,
) -> bool:
"""LS UH1 호가 스냅샷 → ls_ws_orderbook (키움 ws_orderbook 스키마 대칭)."""
try:
recv_ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
snap_time = str(snap.get("snap_time") or "").strip()
if len(snap_time) < 14:
snap_time = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
self.conn.execute(
"INSERT INTO ls_ws_orderbook "
"(market, code, snap_time, best_bid, best_ask, "
"total_bid_qty, total_ask_qty, bid_qty_l3, ask_qty_l3, "
"levels_json, source, recv_ts, tk, lag_sec) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
(
(market or "KR")[:8],
code,
snap_time[:14],
int(snap.get("best_bid") or 0),
int(snap.get("best_ask") or 0),
int(snap.get("total_bid_qty") or 0),
int(snap.get("total_ask_qty") or 0),
int(snap.get("bid_qty_l3") or 0),
int(snap.get("ask_qty_l3") or 0),
snap.get("levels_json"),
str(snap.get("source") or "ls_uh1")[:16],
recv_ts,
tk,
lag_sec,
),
)
return True
except Exception as e:
logger.debug("ls_ws_orderbook INSERT 실패: %s", e)
return False
def insert_ls_ws_vi(
self,
*,
code: str,
ts=None,
event_time: str = "",
vi_gubun: str = "0",
krx_vi_gubun: Optional[str] = None,
nxt_vi_gubun: Optional[str] = None,
svi_recprice: Optional[float] = None,
dvi_recprice: Optional[float] = None,
vi_trgprice: Optional[float] = None,
tr_cd: str = "",
exchname: str = "",
) -> bool:
"""LS UVI/VI_ 발동·해제 이벤트 → ls_ws_vi (백테 '이때 VI' 구간용)."""
try:
if ts is None:
ts = datetime.datetime.now()
self.conn.execute(
"INSERT INTO ls_ws_vi "
"(ts, code, event_time, vi_gubun, krx_vi_gubun, nxt_vi_gubun, "
"svi_recprice, dvi_recprice, vi_trgprice, tr_cd, exchname) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
(
ts,
code,
(event_time or None),
str(vi_gubun or "0")[:4],
(str(krx_vi_gubun)[:4] if krx_vi_gubun not in (None, "") else None),
(str(nxt_vi_gubun)[:4] if nxt_vi_gubun not in (None, "") else None),
svi_recprice,
dvi_recprice,
vi_trgprice,
(tr_cd or None),
(exchname or None),
),
)
return True
except Exception as e:
logger.debug("ls_ws_vi INSERT 실패: %s", e)
return False
def get_ws_validation_stats(
self, *, hours: int = 24, code: Optional[str] = None,
) -> Dict[str, Any]:
"""최근 N시간 검증 통계 (운영자용 분석).
Returns:
{
"samples": 1234,
"both_present": 1100, # KIS·키움 둘 다 가격 있던 비율
"avg_diff_pct": 0.012,
"max_abs_diff_pct": 0.45,
"stddev_diff_pct": 0.08,
"kis_only": 80, # KIS 만 가격 있던 횟수 (키움 미수신)
"kiwoom_only": 30, # 키움 만 가격 있던 횟수
}
"""
try:
where = ["ts >= NOW() - INTERVAL %s HOUR"]
args: List[Any] = [hours]
if code:
where.append("code = %s")
args.append(code)
wsql = " AND ".join(where)
row = self.conn.execute(f"""
SELECT
COUNT(*) AS samples,
SUM(kis_price IS NOT NULL AND kiwoom_price IS NOT NULL) AS both_present,
AVG(diff_pct) AS avg_diff_pct,
MAX(ABS(diff_pct)) AS max_abs_diff_pct,
STDDEV(diff_pct) AS stddev_diff_pct,
SUM(kis_price IS NOT NULL AND kiwoom_price IS NULL) AS kis_only,
SUM(kis_price IS NULL AND kiwoom_price IS NOT NULL) AS kiwoom_only
FROM ws_price_validation
WHERE {wsql}
""", args).fetchone()
return dict(row) if row else {}
except Exception as e:
logger.debug("ws_validation_stats 조회 실패: %s", e)
return {}
# ============================================================
# [env_config + config_*] 관리자용 env (INSERT만 / 최신 1건 = 현재 설정)
# ============================================================
def _migrate_config_table_columns(self) -> None:
"""config_* · env_config(글로벌) 테이블에 누락 컬럼 추가."""
for tbl, keys in CONFIG_TABLE_KEYS.items():
try:
cols = set(self.conn.get_columns(tbl))
except Exception:
continue
for key in keys:
if key not in cols:
try:
self.conn.execute(
f"ALTER TABLE {tbl} ADD COLUMN `{key}` TEXT"
)
logger.info("📌 %s.%s 컬럼 추가", tbl, key)
except Exception as e:
logger.debug("%s.%s 컬럼 추가 스킵: %s", tbl, key, e)
def _migrate_strategy_keys_from_env_config(self) -> None:
"""
env_config 레거시 컬럼에만 있는 전략 ON/OFF·MM alias → config_* 최신 행으로 이전.
get_merged_env_snapshot() 폴백과 별도로, 전략 테이블에 명시 저장.
"""
by_table: Dict[str, Tuple[str, ...]] = {}
for k, tbl in _EXPLICIT_KEY_TABLE.items():
if tbl not in by_table:
by_table[tbl] = ()
by_table[tbl] = by_table[tbl] + (k,)
try:
env_row = self.conn.execute(
"SELECT * FROM env_config ORDER BY id DESC LIMIT 1"
).fetchone()
if not env_row:
return
rk = env_row.keys() if hasattr(env_row, "keys") else []
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
for tbl, keys in by_table.items():
patch: Dict[str, str] = {}
all_keys = CONFIG_TABLE_KEYS.get(tbl, ())
latest = self._read_latest_config_table(tbl, all_keys)
for k in keys:
if latest.get(k):
continue
if k not in rk:
continue
v = env_row[k]
if v is None or str(v).strip() == "":
continue
patch[k] = str(v)
if not patch:
continue
merged = dict(latest)
merged.update(patch)
self._insert_config_table_row(tbl, merged, all_keys, now)
logger.info("📌 %s ← env_config 값 이전: %s", tbl, list(patch.keys()))
except Exception as e:
logger.debug("strategy keys env→config migrate: %s", e)
def _migrate_us_momentum_from_ext(self) -> None:
"""env_config_ext 의 US_MOMENTUM_* → config_us_momentum 이전 후 ext 삭제."""
try:
tbl = "config_us_momentum"
keys = CONFIG_TABLE_KEYS.get(tbl, ())
if not keys:
return
rows = self.conn.execute(
"SELECT env_key, env_value FROM env_config_ext WHERE env_key LIKE %s",
("US_MOMENTUM%",),
).fetchall()
# STRATEGY_US_MOMENTUM_ENABLED 도
rows2 = self.conn.execute(
"SELECT env_key, env_value FROM env_config_ext WHERE env_key=%s",
("STRATEGY_US_MOMENTUM_ENABLED",),
).fetchall()
all_rows = list(rows or []) + list(rows2 or [])
if not all_rows:
return
patch: Dict[str, str] = {}
for er in all_rows:
ek = er["env_key"] if isinstance(er, dict) else er[0]
ev = er["env_value"] if isinstance(er, dict) else er[1]
if not ek:
continue
if classify_config_key(str(ek)) != tbl:
continue
if ev is None or str(ev).strip() == "":
continue
patch[str(ek)] = str(ev)
if not patch:
return
latest = self._read_latest_config_table(tbl, keys)
merged = dict(latest)
# 최신 테이블에 이미 값이 있으면 유지, 비어 있을 때만 ext 로 채움
for k, v in patch.items():
if not str(merged.get(k) or "").strip():
merged[k] = v
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
rid = self._insert_config_table_row(tbl, merged, keys, now)
if rid:
for k in patch:
try:
self.conn.execute(
"DELETE FROM env_config_ext WHERE env_key=%s", (k,)
)
except Exception:
pass
logger.info(
"📌 config_us_momentum ← env_config_ext 이전 %d키 (row id=%s)",
len(patch), rid,
)
try:
from kis_trader.utils.env import invalidate_merged_env_cache
invalidate_merged_env_cache()
except Exception:
pass
except Exception as e:
logger.debug("us_momentum ext→config migrate: %s", e)
def _migrate_short_time_to_tail_time(self) -> None:
"""
레거시 ``SHORT_TIME_START`` / ``SHORT_TIME_END`` → ``TAIL_TIME_*`` 이전 후 단일화.
config_short 최신 행에 SHORT_TIME 값만 있고 TAIL_TIME 이 비어 있으면 복사 INSERT.
"""
try:
keys = CONFIG_TABLE_KEYS.get("config_short", ())
if not keys:
return
latest = self._read_latest_config_table("config_short", keys)
st_short = str(latest.get("SHORT_TIME_START") or "").strip()
en_short = str(latest.get("SHORT_TIME_END") or "").strip()
st_tail = str(latest.get("TAIL_TIME_START") or "").strip()
en_tail = str(latest.get("TAIL_TIME_END") or "").strip()
patch: Dict[str, str] = {}
if st_short and not st_tail:
patch["TAIL_TIME_START"] = st_short
if en_short and not en_tail:
patch["TAIL_TIME_END"] = en_short
if not patch:
return
merged = dict(latest)
merged.update(patch)
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
rid = self._insert_config_table_row("config_short", merged, keys, now)
if rid:
logger.info(
"📌 config_short SHORT_TIME_* → TAIL_TIME_* 이전: %s (row id=%s)",
list(patch.keys()), rid,
)
try:
from kis_trader.utils.env import invalidate_merged_env_cache
invalidate_merged_env_cache()
except Exception:
pass
except Exception as e:
logger.debug("SHORT_TIME→TAIL_TIME migrate: %s", e)
def _insert_config_table_row(
self,
table: str,
snapshot: Dict[str, Any],
keys: Tuple[str, ...],
created_at: str,
) -> Optional[int]:
"""단일 config 테이블에 스냅샷 INSERT."""
if not keys:
return None
try:
table_cols = set(self.conn.get_columns(table))
except Exception:
return None
prev = self._read_latest_config_table(table, keys)
row_data: Dict[str, Any] = {}
for k in keys:
if k not in table_cols:
continue
v = snapshot.get(k) if k in snapshot else None
if v not in (None, ""):
row_data[k] = v
elif prev.get(k):
row_data[k] = prev[k]
if not row_data:
return None
valid = list(dict.fromkeys(row_data.keys()))
key_list = ", ".join(f"`{k}`" for k in valid)
placeholders = ", ".join(["%s"] * (1 + len(valid)))
vals = [created_at] + [row_data[k] for k in valid]
cur = self.conn.execute(
f"INSERT INTO {table} (created_at, {key_list}) VALUES ({placeholders})",
vals,
)
return cur.lastrowid
def _read_latest_config_table(self, table: str, keys: Tuple[str, ...]) -> Dict[str, str]:
"""테이블 최신 행들에서 키별 마지막 비어 있지 않은 값 (부분 INSERT NULL 덮어쓰기 방지)."""
out: Dict[str, str] = {}
if not keys:
return out
try:
rows = self.conn.execute(
f"SELECT * FROM {table} ORDER BY id DESC LIMIT 30"
).fetchall()
except Exception:
return out
if not rows:
return out
rk = rows[0].keys() if hasattr(rows[0], "keys") else []
for k in keys:
if k not in rk:
continue
for row in rows:
v = row[k]
if v is not None and str(v).strip() != "":
out[k] = str(v)
break
if k not in out:
out[k] = ""
return out
def _env_config_column_set(self) -> set:
"""env_config 실제 DB 컬럼 (MariaDB row size 한도로 ENV_GLOBAL_KEYS 일부 미생성)."""
cached = getattr(self, "_env_config_cols_cache", None)
if cached is not None:
return cached
try:
cached = set(self.conn.get_columns("env_config"))
except Exception:
cached = set()
self._env_config_cols_cache = cached
return cached
def _persist_env_config_overflow(
self, snapshot: Dict[str, Any], created_at: str,
) -> int:
"""
env_config 테이블에 컬럼이 없는 글로벌 키 → env_config_ext UPSERT.
(Row size too large 로 ALTER 불가한 KIWOOM_WS_* · WHIPSAW_* · PROGRAM_* 등)
"""
cols = self._env_config_column_set()
saved = 0
for k, v in (snapshot or {}).items():
key = str(k).strip()
if not key or classify_config_key(key) != "env_config":
continue
# 앱키/시크릿/ID 는 env_auth_config 컬럼 전용 (ext 행으로 중복 저장하지 않음)
if key in ENV_AUTH_KEYS:
continue
if key in cols:
continue
if v is None:
continue
sv = str(v).strip()
if sv == "":
# 빈 문자열 = 명시적 클리어/OFF.
# ext 에 이전 값이 남아 있으면 머지 폴백에서 되살아나므로(예:
# UPDOWN_BOX_RATCHET_TIERS='' → 래칫 OFF 인데 직전 '2:1.5,5:1.0' 이 부활)
# 반드시 ext row 를 삭제해 "꺼짐"이 실제 반영되도록 한다.
try:
self.conn.execute(
"DELETE FROM env_config_ext WHERE env_key=%s", (key,)
)
except Exception as ex:
logger.warning("env_config_ext 클리어 실패 %s: %s", key, ex)
continue
try:
self.conn.execute(
"INSERT INTO env_config_ext (env_key, env_value, updated_at) "
"VALUES (%s, %s, %s) "
"ON DUPLICATE KEY UPDATE env_value=VALUES(env_value), "
"updated_at=VALUES(updated_at)",
(key, sv, created_at),
)
saved += 1
except Exception as ex:
logger.warning("env_config_ext 저장 실패 %s: %s", key, ex)
if saved:
logger.info("📌 env_config_ext overflow 저장 %d", saved)
return saved
def _insert_env_auth_row(self, snapshot: Dict[str, Any], created_at: str) -> Optional[int]:
"""env_auth_config 1행 INSERT (앱키/시크릿/ID/계좌 전용).
snapshot 에 실제 값이 있는 인증키가 하나도 없으면 빈 행 방지를 위해 INSERT 생략.
"""
try:
# 빈 행 방지 가드: 실제 값을 가진 키가 하나도 없으면 저장 생략
has_any = any(
snapshot.get(k) is not None and str(snapshot.get(k)).strip() != ""
for k in ENV_AUTH_KEYS
)
if not has_any:
logger.debug("env_auth_config INSERT 생략 — snapshot에 인증키 값 없음 (빈 행 방지)")
return None
key_list = ", ".join(f"`{k}`" for k in ENV_AUTH_KEYS)
placeholders = ", ".join(["%s"] * (1 + len(ENV_AUTH_KEYS)))
vals = [created_at] + [snapshot.get(k, None) for k in ENV_AUTH_KEYS]
cur = self.conn.execute(
f"INSERT INTO env_auth_config (created_at, {key_list}) VALUES ({placeholders})",
vals,
)
return int(cur.lastrowid) if getattr(cur, "lastrowid", None) else None
except Exception as ex:
logger.debug("env_auth_config INSERT 실패: %s", ex)
return None
def _read_latest_env_auth(self) -> Dict[str, str]:
"""env_auth_config 최신 행의 비어있지 않은 값만."""
out: Dict[str, str] = {}
try:
row = self.conn.execute(
"SELECT * FROM env_auth_config ORDER BY id DESC LIMIT 1",
).fetchone()
if not row:
return out
keys = row.keys() if hasattr(row, "keys") else []
for k in ENV_AUTH_KEYS:
if k in keys:
v = row[k]
if v is not None and str(v).strip() != "":
out[k] = str(v)
except Exception:
return out
return out
def get_auth_env(self) -> Dict[str, str]:
"""앱키/시크릿/ID/계좌 + KIS_MOCK.
인증키는 env_auth_config 최신 행 우선(get_merged_env_snapshot).
env_config 컬럼을 직접 SELECT 하지 말 것.
"""
merged = self.get_merged_env_snapshot() or {}
out: Dict[str, str] = {}
for k in ENV_AUTH_KEYS:
v = merged.get(k)
if v is not None and str(v).strip() != "":
out[k] = str(v)
mock = merged.get("KIS_MOCK")
if mock is not None and str(mock).strip() != "":
out["KIS_MOCK"] = str(mock)
return out
def _read_env_auth_as_of(self, as_of: str) -> Dict[str, str]:
"""env_auth_config as_of 이하 최신 행."""
out: Dict[str, str] = {}
try:
row = self.conn.execute(
"SELECT * FROM env_auth_config WHERE created_at <= %s ORDER BY id DESC LIMIT 1",
(as_of,),
).fetchone()
if not row:
return out
keys = row.keys() if hasattr(row, "keys") else []
for k in ENV_AUTH_KEYS:
if k in keys:
v = row[k]
if v is not None and str(v).strip() != "":
out[k] = str(v)
except Exception:
return out
return out
def _migrate_env_auth_from_legacy(self) -> None:
"""
env_auth_config 이 비어 있을 때, env_config 최신 행 + env_config_ext 키로 백필.
"""
try:
c = self.conn.execute("SELECT COUNT(*) c FROM env_auth_config").fetchone()
if c and int(c["c"]) > 0:
return
latest = self.conn.execute(
"SELECT * FROM env_config ORDER BY id DESC LIMIT 1",
).fetchone()
snap: Dict[str, Any] = {}
if latest:
keys = latest.keys() if hasattr(latest, "keys") else []
for k in ENV_AUTH_KEYS:
if k in keys:
v = latest[k]
if v is not None and str(v).strip() != "":
snap[k] = str(v)
ext_rows = self.conn.execute(
"SELECT env_key, env_value FROM env_config_ext",
).fetchall()
for er in ext_rows or []:
ek = er["env_key"] if isinstance(er, dict) else er[0]
ev = er["env_value"] if isinstance(er, dict) else er[1]
if not ek:
continue
sk = str(ek).strip()
if sk in ENV_AUTH_KEYS and ev is not None and str(ev).strip() != "":
snap[sk] = str(ev)
if not snap:
return
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
rid = self._insert_env_auth_row(snap, now)
if rid:
logger.info("📌 env_auth_config 백필 완료 (id=%s, keys=%d)", rid, len(snap))
except Exception as ex:
logger.debug("migrate env_auth_config: %s", ex)
def get_merged_env_snapshot(self) -> Dict[str, str]:
"""env_config + config_* 최신 행 병합 flat dict (get_env_from_db 호환)."""
merged: Dict[str, str] = {}
for tbl, keys in CONFIG_TABLE_KEYS.items():
merged.update(self._read_latest_config_table(tbl, keys))
# 인증/계좌/ID 전용 컬럼 테이블 우선
merged.update(self._read_latest_env_auth())
# 레거시 env_config(381컬럼) 에만 있는 값 — 신규 테이블에 없을 때 폴백
try:
row = self.conn.execute(
"SELECT * FROM env_config ORDER BY id DESC LIMIT 1"
).fetchone()
if row:
rk = row.keys() if hasattr(row, "keys") else []
for k in ENV_CONFIG_KEYS:
if k not in merged and k in rk:
v = row[k]
if v is not None and str(v).strip() != "":
merged[k] = str(v)
except Exception:
pass
# env_config_ext — 컬럼 한도 초과 키 + 레거시 overflow
try:
ext_rows = self.conn.execute(
"SELECT env_key, env_value FROM env_config_ext"
).fetchall()
env_cols = self._env_config_column_set()
for er in ext_rows or []:
ek = er["env_key"] if isinstance(er, dict) else er[0]
ev = er["env_value"] if isinstance(er, dict) else er[1]
if not ek:
continue
sk = str(ek)
# 전략 config_* 테이블 키는 ext 가 덮어쓰지 않는다 (전략 테이블이 우선).
# 레거시로 ext 에 남은 TAIL_*/MOMENTUM_* 등이 config_short/_momentum 값을
# 가리던 버그 방지 — ext 는 순수 글로벌(env_config) 키 폴백 전용.
if classify_config_key(sk) != "env_config":
continue
# ext 우선: env_config 컬럼 없음 · 또는 병합값 비어 있음
if sk not in env_cols or not str(merged.get(sk) or "").strip():
merged[sk] = "" if ev is None else str(ev)
except Exception:
pass
return merged
def _read_config_table_as_of(
self, table: str, keys: Tuple[str, ...], as_of: str,
) -> Dict[str, str]:
"""테이블에서 created_at <= as_of 인 행만 보고 키별 마지막 비어 있지 않은 값."""
out: Dict[str, str] = {}
if not keys or not as_of:
return out
try:
rows = self.conn.execute(
f"SELECT * FROM {table} WHERE created_at <= %s ORDER BY id DESC LIMIT 30",
(as_of,),
).fetchall()
except Exception:
return out
if not rows:
return out
rk = rows[0].keys() if hasattr(rows[0], "keys") else []
for k in keys:
if k not in rk:
continue
for row in rows:
v = row[k]
if v is not None and str(v).strip() != "":
out[k] = str(v)
break
if k not in out:
out[k] = ""
return out
def get_merged_env_snapshot_as_of(self, as_of: str) -> Dict[str, str]:
"""env_config + config_* — as_of 시각(YYYY-MM-DD HH:MM:SS) 이하 최신 행 병합."""
merged: Dict[str, str] = {}
if not as_of:
return self.get_merged_env_snapshot()
for tbl, keys in CONFIG_TABLE_KEYS.items():
merged.update(self._read_config_table_as_of(tbl, keys, as_of))
merged.update(self._read_env_auth_as_of(as_of))
try:
row = self.conn.execute(
"SELECT * FROM env_config WHERE created_at <= %s ORDER BY id DESC LIMIT 1",
(as_of,),
).fetchone()
if row:
rk = row.keys() if hasattr(row, "keys") else []
for k in ENV_CONFIG_KEYS:
if k not in merged and k in rk:
v = row[k]
if v is not None and str(v).strip() != "":
merged[k] = str(v)
except Exception:
pass
try:
ext_rows = self.conn.execute(
"SELECT env_key, env_value FROM env_config_ext WHERE updated_at <= %s",
(as_of,),
).fetchall()
env_cols = self._env_config_column_set()
for er in ext_rows or []:
ek = er["env_key"] if isinstance(er, dict) else er[0]
ev = er["env_value"] if isinstance(er, dict) else er[1]
if not ek:
continue
sk = str(ek)
if classify_config_key(sk) != "env_config":
continue
if sk not in env_cols or not str(merged.get(sk) or "").strip():
merged[sk] = "" if ev is None else str(ev)
except Exception:
pass
return merged
def list_env_change_times_in_range(
self, start_ymd: str, end_ymd: str,
) -> List[str]:
"""구간 내 config_* / env_config INSERT 시각(created_at) 목록 — 오름차순."""
sy = str(start_ymd or "")[:8]
ey = str(end_ymd or "")[:8]
if len(sy) < 8 or len(ey) < 8:
return []
start_bound = f"{sy[:4]}-{sy[4:6]}-{sy[6:8]} 00:00:00"
end_bound = f"{ey[:4]}-{ey[4:6]}-{ey[6:8]} 23:59:59"
found: set = set()
found.add(start_bound)
for tbl in CONFIG_TABLE_KEYS:
try:
rows = self.conn.execute(
f"SELECT DISTINCT created_at FROM {tbl} "
"WHERE created_at >= %s AND created_at <= %s ORDER BY created_at ASC",
(start_bound, end_bound),
).fetchall()
except Exception:
continue
for row in rows or []:
ca = row["created_at"] if isinstance(row, dict) else row[0]
if ca:
found.add(str(ca))
return sorted(found)
def insert_env_snapshot(self, snapshot) -> Optional[int]:
"""
env 설정 INSERT — 테이블별 분리 저장.
env_config(공통) + config_scalp/short/momentum/breakout/updow.
Returns: env_config 행 id (없으면 마지막 성공 config_* id)
"""
try:
if isinstance(snapshot, str):
snapshot = json.loads(snapshot) if snapshot else {}
if not isinstance(snapshot, dict):
return None
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
primary_id: Optional[int] = None
for tbl, keys in CONFIG_TABLE_KEYS.items():
rid = self._insert_config_table_row(tbl, snapshot, keys, now)
if rid and tbl == "env_config":
primary_id = rid
elif rid and primary_id is None:
primary_id = rid
if primary_id is None:
logger.error("config INSERT: 모든 테이블 저장 실패")
return None
ext_n = self._persist_env_config_overflow(snapshot, now)
auth_id = self._insert_env_auth_row(snapshot, now)
logger.info(
"📌 env 스냅샷 저장 (env_config id=%s, auth id=%s, 테이블 %d개, ext+%d)",
primary_id,
auth_id,
len(CONFIG_TABLE_KEYS),
ext_n,
)
try:
from kis_trader.utils.env import invalidate_merged_env_cache
invalidate_merged_env_cache()
except Exception:
pass
return primary_id
except Exception as e:
logger.error(f"❌ env/config INSERT 실패: {e}")
return None
def get_latest_env(self) -> Optional[Dict]:
"""
최신 env — env_config + config_* 병합 snapshot.
Returns:
{"id": int, "created_at": str, "snapshot": dict} 또는 None
"""
try:
row = self.conn.execute(
"SELECT id, created_at FROM env_config ORDER BY id DESC LIMIT 1"
).fetchone()
merged = self.get_merged_env_snapshot()
if not merged and not row:
return None
return {
"id": int(row["id"]) if row else 0,
"created_at": row["created_at"] if row else "",
"snapshot": merged,
}
except Exception as e:
logger.error(f"❌ env 최신 조회 실패: {e}")
return None
def get_strategy_config_snapshot(self, strategy_id: str) -> Dict[str, str]:
"""전략 ID(SCALP/SHORT/…) 해당 config_* 테이블만 조회."""
tbl = STRATEGY_ID_TO_TABLE.get((strategy_id or "").upper())
if not tbl:
return {}
keys = CONFIG_TABLE_KEYS.get(tbl, ())
snap = self._read_latest_config_table(tbl, keys)
if snap:
return snap
return {
k: v
for k, v in self.get_merged_env_snapshot().items()
if classify_config_key(k) == tbl
}
def get_strategy_env_change_history(self, limit: int = 3) -> List[Dict[str, Any]]:
"""
전략별 config_* 최근 N개 **서로 다른** 스냅샷 (기본 3) + 변경 키.
연속 동일 행(전역만 저장해도 config_* 가 복제 INSERT 되는 경우)은 건너뜀.
"""
try:
lim = int(limit)
except (TypeError, ValueError):
lim = 3
if lim < 1:
lim = 1
if lim > 10:
lim = 10
labels = {
"SCALP": "스캘핑",
"SHORT": "꼬리잡기",
"MOMENTUM": "모멘텀",
"US_MOMENTUM": "해외모멘텀",
"BREAKOUT": "돌파",
"RANGE_BREAK": "박스권돌파",
"UPDOW": "업다운",
"DBBAND": "DBBAND",
"DART": "DART",
}
secret_bits = ("SECRET", "APP_KEY", "TOKEN", "PASSWORD", "PASSWD", "API_KEY")
scan_n = max(40, lim * 15)
out: List[Dict[str, Any]] = []
for sid, tbl in STRATEGY_ID_TO_TABLE.items():
try:
rows_raw = self.conn.execute(
f"SELECT * FROM `{tbl}` ORDER BY id DESC LIMIT %s",
(scan_n,),
).fetchall() or []
except Exception as e:
logger.debug("env history %s: %s", tbl, e)
rows_raw = []
distinct: List[Dict[str, Any]] = []
prev_vals: Optional[Dict[str, str]] = None
for r0 in rows_raw:
r = dict(r0)
snap: Dict[str, str] = {}
for k, v in r.items():
if k in ("id", "created_at"):
continue
ku = str(k).upper()
if any(b in ku for b in secret_bits):
continue
if v is None:
continue
sv = str(v).strip()
if sv == "":
continue
snap[str(k)] = sv
if prev_vals is not None and snap == prev_vals:
continue
distinct.append({
"id": int(r.get("id") or 0),
"created_at": str(r.get("created_at") or ""),
"values": snap,
})
prev_vals = snap
if len(distinct) >= lim:
break
# 표시: 오래된 → 최신
snaps = list(reversed(distinct))
changed_keys: List[str] = []
all_keys: set = set()
for s in snaps:
all_keys.update(s["values"].keys())
for i in range(1, len(snaps)):
a = snaps[i - 1]["values"]
b = snaps[i]["values"]
for k in sorted(set(a) | set(b)):
if a.get(k, "") != b.get(k, ""):
if k not in changed_keys:
changed_keys.append(k)
out.append({
"strategy_id": sid,
"label": labels.get(sid, sid),
"table": tbl,
"limit": lim,
"snapshots": snaps,
"changed_keys": changed_keys,
"key_count": len(all_keys),
})
return out
# ============================================================
# [kv_store] 매터모스트 원격 조종용 키-값 (마지막 AI 추천, last_seen 등)
# ============================================================
def get_kv(self, key: str) -> Optional[str]:
"""kv_store에서 값 조회. 없으면 None."""
try:
row = self.conn.execute("SELECT v FROM kv_store WHERE k = ?", (key,)).fetchone()
return row["v"] if row else None
except Exception as e:
logger.debug(f"get_kv 실패: {e}")
return None
def set_kv(self, key: str, value: str) -> bool:
"""kv_store에 값 저장 (REPLACE INTO = INSERT OR REPLACE MySQL 버전)."""
try:
self.conn.execute("REPLACE INTO kv_store (k, v) VALUES (%s, %s)", (key, value))
return True
except Exception as e:
logger.error(f"❌ set_kv 실패: {e}")
return False
def set_last_ai_recommendations(self, text: str) -> bool:
"""마지막 AI 수치 추천문 저장 (!적용 시 사용)."""
return self.set_kv("last_ai_recommendations", text)
def get_last_ai_recommendations(self) -> Optional[str]:
"""마지막 AI 수치 추천문 조회. 없으면 None."""
return self.get_kv("last_ai_recommendations")
# ------------------------------------------------------------------
# [ai_analysis_log] Butler 분석 기록 (프롬프트 요약·응답 저장/조회)
# ------------------------------------------------------------------
def insert_ai_analysis_log(self, model: str, context_summary: str, response: str) -> Optional[int]:
"""AI 분석 한 건 저장. model=claude|gemini, context_summary=현재상태 요약, response=AI 전체 응답."""
try:
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
cur = self.conn.execute(
"INSERT INTO ai_analysis_log (created_at, model, context_summary, response) VALUES (%s, %s, %s, %s)",
(now, model, (context_summary or "")[:2000], response or ""),
)
return cur.lastrowid
except Exception as e:
logger.error("insert_ai_analysis_log 실패: %s", e)
return None
def get_ai_analysis_log_list(self, limit: int = 10) -> list:
"""최근 N건 목록. 각 항목: id, created_at, model, context_summary, response(앞 400자)."""
try:
rows = self.conn.execute(
"""
SELECT id, created_at, model, context_summary, response
FROM ai_analysis_log
ORDER BY id DESC
LIMIT ?
""",
(limit,),
).fetchall()
out = []
for r in rows:
resp = (r["response"] or "")[:400]
if len(r["response"] or "") > 400:
resp += "..."
out.append({"id": r["id"], "created_at": r["created_at"], "model": r["model"],
"context_summary": r["context_summary"], "response_preview": resp})
return out
except Exception as e:
logger.error("get_ai_analysis_log_list 실패: %s", e)
return []
def get_ai_analysis_log_by_id(self, log_id: int) -> Optional[dict]:
"""id로 전체 한 건 조회. 없으면 None."""
try:
row = self.conn.execute(
"SELECT id, created_at, model, context_summary, response FROM ai_analysis_log WHERE id = ?",
(log_id,),
).fetchone()
if not row:
return None
return {"id": row["id"], "created_at": row["created_at"], "model": row["model"],
"context_summary": row["context_summary"], "response": row["response"]}
except Exception as e:
logger.error("get_ai_analysis_log_by_id 실패: %s", e)
return None
# ==================================================================
# ws_candles 헬퍼 (kis_scalping_ver1 전용)
# ==================================================================
def upsert_ws_candle(self, code: str, timeframe: int, candle_time: str,
open_: float, high: float, low: float, close: float,
volume: int, is_confirmed: int = 0, source: str = "kis",
rsi_2: float = None, rsi_3: float = None, rsi_5: float = None,
holding_peak: float = None, market: str = "KR",
channel: str = ""):
"""
봉 1개를 INSERT OR REPLACE 로 저장.
source=증권사(kis/kiwoom), channel=ws|rest|rollup.
"""
now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
mk = (market or "KR").strip().upper()[:8] or "KR"
try:
from kis_trader.ws.candle_series import normalize_source_channel
src, ch = normalize_source_channel(source, channel)
except Exception:
src, ch = (source or "kis")[:16], (channel or "ws")[:10] or "ws"
try:
with self.conn:
self.conn.execute("""
INSERT INTO ws_candles
(code, market, timeframe, candle_time, open, high, low, close, volume,
rsi_2, rsi_3, rsi_5, is_confirmed, source, channel, holding_peak, updated_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
market = IF(VALUES(market) IS NULL OR VALUES(market)='', market, VALUES(market)),
high = GREATEST(high, VALUES(high)),
low = LEAST(low, VALUES(low)),
close = VALUES(close),
volume = VALUES(volume),
rsi_2 = COALESCE(VALUES(rsi_2), rsi_2),
rsi_3 = COALESCE(VALUES(rsi_3), rsi_3),
rsi_5 = COALESCE(VALUES(rsi_5), rsi_5),
is_confirmed = VALUES(is_confirmed),
source = VALUES(source),
channel = VALUES(channel),
holding_peak = GREATEST(COALESCE(holding_peak, 0), COALESCE(VALUES(holding_peak), 0)),
updated_at = VALUES(updated_at)
""", (code, mk, timeframe, candle_time, open_, high, low, close, volume,
rsi_2, rsi_3, rsi_5, is_confirmed, src, ch, holding_peak, now_str))
except Exception as e:
logger.error("upsert_ws_candle 실패(%s): %s", code, e)
def get_ws_candles(self, code: str, timeframe: int, limit: int = 100,
confirmed_only: bool = False, source: str = "kis") -> list:
"""
최근 N개 봉 리스트 반환 (오래된 순 → 최신 순).
confirmed_only=True 면 확정 봉만 반환 (RSI 계산용).
"""
try:
cond = "AND is_confirmed = 1" if confirmed_only else ""
rows = self.conn.execute(f"""
SELECT candle_time, open, high, low, close, volume,
rsi_2, rsi_3, rsi_5, is_confirmed, source
FROM ws_candles
WHERE code = %s AND timeframe = %s AND source = %s {cond}
ORDER BY candle_time DESC
LIMIT %s
""", (code, timeframe, source, limit)).fetchall()
# 최신→오래된 순으로 왔으므로 역정렬해서 반환 (오래된→최신)
result = [dict(r) for r in rows]
result.reverse()
return result
except Exception as e:
logger.error("get_ws_candles 실패(%s): %s", code, e)
return []
def get_latest_ws_candle(self, code: str, timeframe: int, source: str = "kis") -> Optional[dict]:
"""가장 최신 봉 1개 반환 (진행 중 봉 포함)."""
try:
row = self.conn.execute("""
SELECT candle_time, open, high, low, close, volume,
rsi_2, rsi_3, rsi_5, is_confirmed
FROM ws_candles
WHERE code = %s AND timeframe = %s AND source = %s
ORDER BY candle_time DESC LIMIT 1
""", (code, timeframe, source)).fetchone()
return dict(row) if row else None
except Exception as e:
logger.error("get_latest_ws_candle 실패(%s): %s", code, e)
return None
def get_latest_confirmed_ws_candle(self, code: str, timeframe: int) -> Optional[dict]:
"""가장 최신 확정 봉 1개 반환 (RSI 포함)."""
try:
row = self.conn.execute("""
SELECT candle_time, open, high, low, close, volume,
rsi_2, rsi_3, rsi_5
FROM ws_candles
WHERE code = ? AND timeframe = ? AND is_confirmed = 1
ORDER BY candle_time DESC LIMIT 1
""", (code, timeframe)).fetchone()
return dict(row) if row else None
except Exception as e:
logger.error("get_latest_confirmed_ws_candle 실패(%s): %s", code, e)
return None
def cleanup_old_ws_candles(self, keep_days: int = 7) -> None:
"""오래된 ws_candles 정리 (기본 7일 이상 지난 봉 삭제, 청크 방식)."""
if keep_days <= 0:
return
cutoff = (datetime.datetime.now() - datetime.timedelta(days=keep_days)).strftime("%Y%m%d%H%M")
chunk = max(1000, int(os.environ.get("WS_CANDLE_CLEANUP_CHUNK", "5000") or 5000))
max_loops = max(1, int(os.environ.get("WS_CANDLE_CLEANUP_MAX_LOOPS", "200") or 200))
total = 0
try:
import time as _time
for _ in range(max_loops):
cur = self.conn.execute(
"DELETE FROM ws_candles WHERE candle_time < %s LIMIT %s",
(cutoff, chunk),
)
n = int(getattr(cur, "rowcount", 0) or 0)
total += n
if n < chunk:
break
_time.sleep(0.05)
if total:
logger.info(
"🧹 ws_candles 정리 %d행 (candle_time < %s, chunk=%d)",
total, cutoff, chunk,
)
except Exception as e:
logger.error("cleanup_old_ws_candles 실패: %s", e)
def cleanup_old_ls_ws_ticks(self, keep_days: int = 7) -> None:
"""ls_ws_ticks 오래된 틱 정리 (ts 컬럼 기준, 청크 방식)."""
if keep_days <= 0:
return
cutoff = (
datetime.datetime.now() - datetime.timedelta(days=keep_days)
).strftime("%Y-%m-%d %H:%M:%S")
chunk = max(1000, int(os.environ.get("LS_TICK_CLEANUP_CHUNK", "5000") or 5000))
max_loops = max(1, int(os.environ.get("LS_TICK_CLEANUP_MAX_LOOPS", "200") or 200))
total = 0
try:
import time as _time
for _ in range(max_loops):
cur = self.conn.execute(
"DELETE FROM ls_ws_ticks WHERE ts < %s LIMIT %s",
(cutoff, chunk),
)
n = int(getattr(cur, "rowcount", 0) or 0)
total += n
if n < chunk:
break
_time.sleep(0.05)
if total:
logger.info(
"🧹 ls_ws_ticks 정리 %d행 (ts < %s, chunk=%d)",
total, cutoff, chunk,
)
except Exception as e:
logger.error("cleanup_old_ls_ws_ticks 실패: %s", e)
def cleanup_old_ls_ws_orderbook(self, keep_days: int = 7) -> None:
"""ls_ws_orderbook 오래된 스냅샷 정리 (recv_ts 기준, 청크 방식)."""
if keep_days <= 0:
return
cutoff = (
datetime.datetime.now() - datetime.timedelta(days=keep_days)
).strftime("%Y-%m-%d %H:%M:%S")
chunk = max(1000, int(os.environ.get("LS_OB_CLEANUP_CHUNK", "5000") or 5000))
max_loops = max(1, int(os.environ.get("LS_OB_CLEANUP_MAX_LOOPS", "200") or 200))
total = 0
try:
import time as _time
for _ in range(max_loops):
cur = self.conn.execute(
"DELETE FROM ls_ws_orderbook WHERE recv_ts < %s LIMIT %s",
(cutoff, chunk),
)
n = int(getattr(cur, "rowcount", 0) or 0)
total += n
if n < chunk:
break
_time.sleep(0.05)
if total:
logger.info(
"🧹 ls_ws_orderbook 정리 %d행 (recv_ts < %s, chunk=%d)",
total, cutoff, chunk,
)
except Exception as e:
logger.error("cleanup_old_ls_ws_orderbook 실패: %s", e)
# ==================================================================
# ws_ticks — 실시간 체결 틱 (TickRecorder 배치 INSERT)
# ==================================================================
@staticmethod
def new_raw_connection():
"""틱 writer 등 전용 커넥션 (앱 TradeDB._lock 과 분리)."""
return _MariaDBConn()
@staticmethod
def ws_ticks_table_for_market(market: str = "KR") -> str:
"""국내=ws_ticks / 해외=ws_ticks_us."""
return "ws_ticks_us" if str(market or "KR").strip().upper() == "US" else "ws_ticks"
def ensure_ws_ticks_table(self) -> None:
"""런타임 테이블 보장 (구 DB 마이그레이션)."""
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ws_ticks (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
market VARCHAR(8) NOT NULL DEFAULT 'KR',
exchange VARCHAR(16) DEFAULT NULL,
code VARCHAR(32) NOT NULL,
tick_time VARCHAR(14) NOT NULL,
price DOUBLE NOT NULL,
volume BIGINT NOT NULL DEFAULT 0,
tick_seq BIGINT DEFAULT NULL,
session VARCHAR(8) DEFAULT NULL,
currency VARCHAR(8) NOT NULL DEFAULT 'KRW',
source VARCHAR(16) NOT NULL DEFAULT 'kis',
channel VARCHAR(10) NOT NULL DEFAULT 'ws',
recv_ts VARCHAR(30) NOT NULL,
cntr_str DOUBLE DEFAULT NULL,
upper_limit_time VARCHAR(8) DEFAULT NULL,
tick_time_raw VARCHAR(64) DEFAULT NULL,
KEY idx_ws_ticks_lookup (market, code, tick_time),
KEY idx_ws_ticks_recv (recv_ts)
) ENGINE=InnoDB CHARACTER SET utf8mb4
""")
self._migrate_ws_ticks_extra_columns("ws_ticks")
except Exception as e:
logger.debug("ensure_ws_ticks_table: %s", e)
def _migrate_ws_ticks_extra_columns(self, table: str) -> None:
"""키움 0B 부가 FID 컬럼 (구 DB ALTER)."""
table = (table or "ws_ticks").strip()
if table not in ("ws_ticks", "ws_ticks_us"):
return
try:
existing = {
r["Field"]
for r in self.conn.execute(f"SHOW COLUMNS FROM {table}").fetchall()
}
except Exception:
return
for col, ddl in (
("cntr_str", "DOUBLE DEFAULT NULL"),
("upper_limit_time", "VARCHAR(8) DEFAULT NULL"),
("tick_time_raw", "VARCHAR(64) DEFAULT NULL"),
# 봉 ws_candles 와 동일: source=증권사, channel=경로(ws|rest). 틱은 WS만 적재 → 기본 ws.
("channel", "VARCHAR(10) NOT NULL DEFAULT 'ws' COMMENT 'ws|rest' AFTER `source`"),
):
if col not in existing:
try:
self.conn.execute(f"ALTER TABLE {table} ADD COLUMN `{col}` {ddl}")
logger.info("%s.%s 컬럼 추가", table, col)
except Exception as ex:
logger.warning("%s.%s ALTER 실패: %s", table, col, ex)
# tick_time_raw 20→64 확장 (KIS BSOP_DATE+체결시각·키움 FID20 원본)
try:
col_info = {
r["Field"]: r
for r in self.conn.execute(f"SHOW COLUMNS FROM {table}").fetchall()
}
ttr = col_info.get("tick_time_raw")
if ttr and "varchar(20)" in str(ttr.get("Type", "")).lower():
self.conn.execute(
f"ALTER TABLE {table} MODIFY COLUMN tick_time_raw VARCHAR(64) DEFAULT NULL"
)
logger.info("%s.tick_time_raw VARCHAR(64) 확장", table)
except Exception as ex:
logger.debug("%s.tick_time_raw widen skip: %s", table, ex)
def ensure_ws_ticks_us_table(self) -> None:
"""해외 체결 틱 테이블 보장 (InnoDB, 국내 ws_ticks 와 분리)."""
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ws_ticks_us (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
market VARCHAR(8) NOT NULL DEFAULT 'US',
exchange VARCHAR(16) DEFAULT NULL,
code VARCHAR(32) NOT NULL,
tick_time VARCHAR(14) NOT NULL,
price DOUBLE NOT NULL,
volume BIGINT NOT NULL DEFAULT 0,
tick_seq BIGINT DEFAULT NULL,
session VARCHAR(8) DEFAULT NULL,
currency VARCHAR(8) NOT NULL DEFAULT 'USD',
source VARCHAR(16) NOT NULL DEFAULT 'kis_us',
channel VARCHAR(10) NOT NULL DEFAULT 'ws',
recv_ts VARCHAR(30) NOT NULL,
cntr_str DOUBLE DEFAULT NULL,
upper_limit_time VARCHAR(8) DEFAULT NULL,
tick_time_raw VARCHAR(64) DEFAULT NULL,
KEY idx_ws_ticks_us_lookup (market, code, tick_time),
KEY idx_ws_ticks_us_recv (recv_ts)
) ENGINE=InnoDB CHARACTER SET utf8mb4
""")
self._migrate_ws_ticks_extra_columns("ws_ticks_us")
except Exception as e:
logger.debug("ensure_ws_ticks_us_table: %s", e)
def _insert_ws_ticks_batch_into(
self,
table: str,
batch: List[Dict],
*,
default_market: str = "KR",
default_currency: str = "KRW",
default_source: str = "kis",
conn=None,
) -> int:
"""공통 배치 INSERT (table=ws_ticks|ws_ticks_us)."""
if not batch:
return 0
table = (table or "ws_ticks").strip()
if table not in ("ws_ticks", "ws_ticks_us"):
logger.error("insert_ws_ticks: 허용되지 않은 테이블 %s", table)
return 0
rows = []
for it in batch:
ch = str(it.get("channel") or "ws").strip().lower() or "ws"
if ch not in ("ws", "rest"):
ch = "ws"
rows.append((
(it.get("market") or default_market)[:8],
(it.get("exchange") or None),
(it.get("code") or "")[:32],
(it.get("tick_time") or "")[:14],
float(it.get("price") or 0),
int(it.get("volume") or 0),
it.get("tick_seq"),
(it.get("session") or None),
(it.get("currency") or default_currency)[:8],
(it.get("source") or default_source)[:16],
ch[:10],
it.get("recv_ts") or datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
float(it["cntr_str"]) if it.get("cntr_str") is not None else None,
(it.get("upper_limit_time") or None),
(it.get("tick_time_raw") or None),
))
exec_conn = conn or self.conn
try:
ph = "(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
placeholders = ", ".join([ph] * len(rows))
flat: List[Any] = []
for row in rows:
flat.extend(row)
exec_conn.execute(
f"""
INSERT INTO {table}
(market, exchange, code, tick_time, price, volume,
tick_seq, session, currency, source, channel, recv_ts,
cntr_str, upper_limit_time, tick_time_raw)
VALUES {placeholders}
""",
tuple(flat),
)
return len(rows)
except Exception as e:
logger.error("insert_%s_batch 실패: %s", table, e)
return 0
def insert_ws_ticks_batch(self, batch: List[Dict], *, conn=None) -> int:
"""국내 체결 틱 배치 append INSERT (중복 키 없음 — 이벤트 로그)."""
return self._insert_ws_ticks_batch_into(
"ws_ticks", batch,
default_market="KR", default_currency="KRW", default_source="kis",
conn=conn,
)
def insert_ws_ticks_us_batch(self, batch: List[Dict], *, conn=None) -> int:
"""해외 체결 틱 → ws_ticks_us (전용 conn 권장 — TradeDB 락 분리)."""
return self._insert_ws_ticks_batch_into(
"ws_ticks_us", batch,
default_market="US", default_currency="USD", default_source="kis_us",
conn=conn,
)
def get_latest_ws_tick_quotes(
self,
codes: List[str],
*,
market: str = "KR",
) -> Dict[str, Dict[str, Any]]:
"""
종목별 최신 1틱 (웹 보유 라이브 표시용 SELECT).
실매 메모리 캐시와 무관 — INSERT 경로만 공유, 주문/전략 비접촉.
"""
uniq: List[str] = []
seen = set()
for c in codes or []:
cc = str(c or "").strip()
if not cc or cc in seen:
continue
seen.add(cc)
uniq.append(cc[:32])
if not uniq:
return {}
mkt = (market or "KR").strip().upper()
table = self.ws_ticks_table_for_market(mkt)
if table == "ws_ticks_us":
self.ensure_ws_ticks_us_table()
else:
self.ensure_ws_ticks_table()
ph = ", ".join(["%s"] * len(uniq))
try:
rows = self.conn.execute(
f"""
SELECT t.code, t.price, t.recv_ts, t.tick_time, t.source, t.currency
FROM {table} t
INNER JOIN (
SELECT code, MAX(id) AS mid
FROM {table}
WHERE code IN ({ph})
GROUP BY code
) x ON t.id = x.mid
""",
tuple(uniq),
).fetchall()
except Exception as e:
logger.debug("get_latest_ws_tick_quotes 실패 market=%s: %s", mkt, e)
return {}
out: Dict[str, Dict[str, Any]] = {}
for r in rows or []:
code = str(r.get("code") or "").strip()
if not code:
continue
try:
px = float(r.get("price") or 0)
except (TypeError, ValueError):
px = 0.0
if px <= 0:
continue
out[code] = {
"price": px,
"recv_ts": str(r.get("recv_ts") or ""),
"tick_time": str(r.get("tick_time") or ""),
"source": str(r.get("source") or ""),
"currency": str(r.get("currency") or ("USD" if mkt == "US" else "KRW")),
"market": mkt,
}
return out
def get_prev_close_map(
self,
codes: List[str],
*,
market: str = "KR",
timeframe: int = 1,
) -> Dict[str, float]:
"""당일 등락률용 전일(직전 세션) 종가 — ws_candles 1분봉 마지막 close.
KR: 당일 09:00 미만 최신 봉. US: 당일 00:00 미만 최신 봉.
REST 없이 SELECT 만 (웹 라이브 표시용).
"""
uniq: List[str] = []
seen = set()
for c in codes or []:
cc = str(c or "").strip()
if not cc or cc in seen:
continue
seen.add(cc)
uniq.append(cc[:32])
if not uniq:
return {}
mkt = (market or "KR").strip().upper()
tf = max(1, int(timeframe or 1))
today = datetime.datetime.now().strftime("%Y%m%d")
# candle_time = YYYYMMDDHHMM
cutoff = today + ("0900" if mkt == "KR" else "0000")
ph = ", ".join(["%s"] * len(uniq))
try:
rows = self.conn.execute(
f"""
SELECT c.code, c.close
FROM ws_candles c
INNER JOIN (
SELECT code, MAX(candle_time) AS mt
FROM ws_candles
WHERE market=%s AND timeframe=%s AND source='kis'
AND code IN ({ph})
AND candle_time < %s
GROUP BY code
) x ON c.code = x.code AND c.candle_time = x.mt
WHERE c.market=%s AND c.timeframe=%s AND c.source='kis'
""",
(mkt, tf) + tuple(uniq) + (cutoff, mkt, tf),
).fetchall()
except Exception as e:
logger.debug("get_prev_close_map 실패 market=%s: %s", mkt, e)
return {}
out: Dict[str, float] = {}
for r in rows or []:
code = str(r.get("code") or "").strip()
try:
px = float(r.get("close") or 0)
except (TypeError, ValueError):
px = 0.0
if code and px > 0:
out[code] = px
return out
def get_ws_ticks(
self,
code: str,
*,
market: str = "KR",
start_tick_time: str = "",
end_tick_time: str = "",
limit: int = 5000,
) -> List[Dict]:
"""백테·분석용 틱 조회 (오래된→최신). US → ws_ticks_us."""
code = (code or "").strip()
if not code:
return []
limit = max(1, min(int(limit), 500000))
mkt = (market or "KR")[:8].upper()
table = self.ws_ticks_table_for_market(mkt)
try:
cond = "market=%s AND code=%s"
params: List[Any] = [mkt, code[:32]]
if start_tick_time:
cond += " AND tick_time >= %s"
params.append(start_tick_time[:14])
if end_tick_time:
cond += " AND tick_time <= %s"
params.append(end_tick_time[:14])
rows = self.conn.execute(
f"""
SELECT market, exchange, code, tick_time, price, volume,
tick_seq, session, currency, source, recv_ts
FROM {table}
WHERE {cond}
ORDER BY tick_time ASC
LIMIT %s
""",
(*params, limit),
).fetchall()
return [dict(r) for r in rows]
except Exception as e:
logger.error("get_ws_ticks 실패(%s/%s): %s", mkt, code, e)
return []
def _cleanup_old_ticks_table(self, table: str, keep_days: int) -> None:
if keep_days <= 0:
return
table = (table or "").strip()
if table not in ("ws_ticks", "ws_ticks_us"):
return
cutoff = (
datetime.datetime.now() - datetime.timedelta(days=keep_days)
).strftime("%Y-%m-%d %H:%M:%S")
chunk = max(1000, int(os.environ.get("WS_TICK_CLEANUP_CHUNK", "5000") or 5000))
max_loops = max(1, int(os.environ.get("WS_TICK_CLEANUP_MAX_LOOPS", "200") or 200))
total = 0
try:
import time as _time
for _ in range(max_loops):
cur = self.conn.execute(
f"DELETE FROM {table} WHERE recv_ts < %s LIMIT %s",
(cutoff, chunk),
)
n = int(getattr(cur, "rowcount", 0) or 0)
total += n
if n < chunk:
break
_time.sleep(0.05)
if total:
logger.info(
"🧹 %s 정리 %d행 (recv_ts < %s, chunk=%d)",
table, total, cutoff, chunk,
)
except Exception as e:
logger.error("cleanup_old_%s 실패: %s", table, e)
def cleanup_old_ws_ticks(self, keep_days: int = 7) -> None:
"""recv_ts 기준 오래된 틱 삭제 (청크 — 단일 거대 DELETE 가 TradeDB 락을 장시간 점유하지 않게)."""
self._cleanup_old_ticks_table("ws_ticks", keep_days)
def cleanup_old_ws_ticks_us(self, keep_days: int = 7) -> None:
"""해외 ws_ticks_us 오래된 틱 정리."""
self._cleanup_old_ticks_table("ws_ticks_us", keep_days)
# ==================================================================
# ws_orderbook / ws_program — TRIGGER 스냅샷 (TriggerSnapshotRecorder)
# ==================================================================
def ensure_ws_orderbook_table(self) -> None:
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ws_orderbook (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
market VARCHAR(8) NOT NULL DEFAULT 'KR',
code VARCHAR(32) NOT NULL,
snap_time VARCHAR(14) NOT NULL,
best_bid BIGINT NOT NULL DEFAULT 0,
best_ask BIGINT NOT NULL DEFAULT 0,
total_bid_qty BIGINT NOT NULL DEFAULT 0,
total_ask_qty BIGINT NOT NULL DEFAULT 0,
bid_qty_l3 BIGINT NOT NULL DEFAULT 0,
ask_qty_l3 BIGINT NOT NULL DEFAULT 0,
levels_json MEDIUMTEXT,
source VARCHAR(16) NOT NULL DEFAULT 'kiwoom_0d',
recv_ts VARCHAR(30) NOT NULL,
KEY idx_ws_orderbook_lookup (market, code, snap_time),
KEY idx_ws_orderbook_recv (recv_ts)
) CHARACTER SET utf8mb4
""")
except Exception as e:
logger.debug("ensure_ws_orderbook_table: %s", e)
def ensure_ws_program_table(self) -> None:
try:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ws_program (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
market VARCHAR(8) NOT NULL DEFAULT 'KR',
code VARCHAR(32) NOT NULL,
snap_time VARCHAR(14) NOT NULL,
buy_qty BIGINT NOT NULL DEFAULT 0,
sell_qty BIGINT NOT NULL DEFAULT 0,
net_qty BIGINT NOT NULL DEFAULT 0,
buy_amt BIGINT NOT NULL DEFAULT 0,
sell_amt BIGINT NOT NULL DEFAULT 0,
net_amt BIGINT NOT NULL DEFAULT 0,
source VARCHAR(16) NOT NULL DEFAULT 'kiwoom_0w',
recv_ts VARCHAR(30) NOT NULL,
KEY idx_ws_program_lookup (market, code, snap_time),
KEY idx_ws_program_recv (recv_ts)
) CHARACTER SET utf8mb4
""")
except Exception as e:
logger.debug("ensure_ws_program_table: %s", e)
def migrate_trigger_eval_columns(self) -> None:
"""ws_orderbook / ws_program — 판정 스냅샷 메타 컬럼 (filter_eval)."""
for table in ("ws_orderbook", "ws_program"):
try:
cols = self.conn.get_columns(table)
except Exception:
continue
if "strategy" not in cols:
self.conn.execute(
f"ALTER TABLE {table} ADD COLUMN strategy VARCHAR(16) DEFAULT NULL"
)
logger.info("📌 %s.strategy 컬럼 추가", table)
if "reject_code" not in cols:
self.conn.execute(
f"ALTER TABLE {table} ADD COLUMN reject_code VARCHAR(64) DEFAULT NULL"
)
logger.info("📌 %s.reject_code 컬럼 추가", table)
if "reject_msg" not in cols:
self.conn.execute(
f"ALTER TABLE {table} ADD COLUMN reject_msg VARCHAR(255) DEFAULT NULL"
)
logger.info("📌 %s.reject_msg 컬럼 추가", table)
if "eval_price" not in cols:
self.conn.execute(
f"ALTER TABLE {table} ADD COLUMN eval_price BIGINT NOT NULL DEFAULT 0"
)
logger.info("📌 %s.eval_price 컬럼 추가", table)
def insert_ws_orderbook_eval_batch(self, batch: List[Dict]) -> int:
"""판정 시점 호가 스냅샷 INSERT (filter_eval)."""
if not batch:
return 0
rows = []
for it in batch:
rows.append((
(it.get("market") or "KR")[:8],
(it.get("code") or "")[:32],
(it.get("snap_time") or "")[:14],
int(it.get("best_bid") or 0),
int(it.get("best_ask") or 0),
int(it.get("total_bid_qty") or 0),
int(it.get("total_ask_qty") or 0),
int(it.get("bid_qty_l3") or 0),
int(it.get("ask_qty_l3") or 0),
it.get("levels_json") or "{}",
(it.get("source") or "filter_eval")[:16],
it.get("recv_ts") or datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
(it.get("strategy") or "")[:16] or None,
(it.get("reject_code") or "")[:64] or None,
(it.get("reject_msg") or "")[:255] or None,
int(it.get("eval_price") or 0),
))
try:
ph = "(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
placeholders = ", ".join([ph] * len(rows))
flat: List[Any] = []
for row in rows:
flat.extend(row)
self.conn.execute(
f"""
INSERT INTO ws_orderbook
(market, code, snap_time, best_bid, best_ask,
total_bid_qty, total_ask_qty, bid_qty_l3, ask_qty_l3,
levels_json, source, recv_ts, strategy, reject_code, reject_msg, eval_price)
VALUES {placeholders}
""",
tuple(flat),
)
return len(rows)
except Exception as e:
logger.error("insert_ws_orderbook_eval_batch 실패: %s", e)
return self.insert_ws_orderbook_batch(batch)
def insert_ws_program_eval_batch(self, batch: List[Dict]) -> int:
"""판정 시점 프로그램 스냅샷 INSERT (filter_eval)."""
if not batch:
return 0
rows = []
for it in batch:
rows.append((
(it.get("market") or "KR")[:8],
(it.get("code") or "")[:32],
(it.get("snap_time") or "")[:14],
int(it.get("buy_qty") or 0),
int(it.get("sell_qty") or 0),
int(it.get("net_qty") or 0),
int(it.get("buy_amt") or 0),
int(it.get("sell_amt") or 0),
int(it.get("net_amt") or 0),
(it.get("source") or "filter_eval")[:16],
it.get("recv_ts") or datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
(it.get("strategy") or "")[:16] or None,
(it.get("reject_code") or "")[:64] or None,
(it.get("reject_msg") or "")[:255] or None,
int(it.get("eval_price") or 0),
))
try:
ph = "(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
placeholders = ", ".join([ph] * len(rows))
flat: List[Any] = []
for row in rows:
flat.extend(row)
self.conn.execute(
f"""
INSERT INTO ws_program
(market, code, snap_time, buy_qty, sell_qty, net_qty,
buy_amt, sell_amt, net_amt, source, recv_ts,
strategy, reject_code, reject_msg, eval_price)
VALUES {placeholders}
""",
tuple(flat),
)
return len(rows)
except Exception as e:
logger.error("insert_ws_program_eval_batch 실패: %s", e)
return self.insert_ws_program_batch(batch)
def cleanup_old_ws_orderbook_eval(self, keep_days: int = 7) -> None:
"""filter_eval 호가 스냅만 정리."""
if keep_days <= 0:
return
cutoff = (
datetime.datetime.now() - datetime.timedelta(days=keep_days)
).strftime("%Y-%m-%d %H:%M:%S")
try:
self.conn.execute(
"DELETE FROM ws_orderbook WHERE source = 'filter_eval' AND recv_ts < %s",
(cutoff,),
)
logger.info("🧹 ws_orderbook filter_eval 정리 완료 (recv_ts < %s)", cutoff)
except Exception as e:
logger.error("cleanup_old_ws_orderbook_eval 실패: %s", e)
def cleanup_old_ws_program_eval(self, keep_days: int = 7) -> None:
"""filter_eval 프로그램 스냅만 정리."""
if keep_days <= 0:
return
cutoff = (
datetime.datetime.now() - datetime.timedelta(days=keep_days)
).strftime("%Y-%m-%d %H:%M:%S")
try:
self.conn.execute(
"DELETE FROM ws_program WHERE source = 'filter_eval' AND recv_ts < %s",
(cutoff,),
)
logger.info("🧹 ws_program filter_eval 정리 완료 (recv_ts < %s)", cutoff)
except Exception as e:
logger.error("cleanup_old_ws_program_eval 실패: %s", e)
def insert_ws_orderbook_batch(self, batch: List[Dict]) -> int:
if not batch:
return 0
rows = []
for it in batch:
rows.append((
(it.get("market") or "KR")[:8],
(it.get("code") or "")[:32],
(it.get("snap_time") or "")[:14],
int(it.get("best_bid") or 0),
int(it.get("best_ask") or 0),
int(it.get("total_bid_qty") or 0),
int(it.get("total_ask_qty") or 0),
int(it.get("bid_qty_l3") or 0),
int(it.get("ask_qty_l3") or 0),
it.get("levels_json") or "{}",
(it.get("source") or "kiwoom_0d")[:16],
it.get("recv_ts") or datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
))
try:
ph = "(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
placeholders = ", ".join([ph] * len(rows))
flat: List[Any] = []
for row in rows:
flat.extend(row)
self.conn.execute(
f"""
INSERT INTO ws_orderbook
(market, code, snap_time, best_bid, best_ask,
total_bid_qty, total_ask_qty, bid_qty_l3, ask_qty_l3,
levels_json, source, recv_ts)
VALUES {placeholders}
""",
tuple(flat),
)
return len(rows)
except Exception as e:
logger.error("insert_ws_orderbook_batch 실패: %s", e)
return 0
def insert_ws_program_batch(self, batch: List[Dict]) -> int:
if not batch:
return 0
rows = []
for it in batch:
rows.append((
(it.get("market") or "KR")[:8],
(it.get("code") or "")[:32],
(it.get("snap_time") or "")[:14],
int(it.get("buy_qty") or 0),
int(it.get("sell_qty") or 0),
int(it.get("net_qty") or 0),
int(it.get("buy_amt") or 0),
int(it.get("sell_amt") or 0),
int(it.get("net_amt") or 0),
(it.get("source") or "kiwoom_0w")[:16],
it.get("recv_ts") or datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
))
try:
ph = "(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
placeholders = ", ".join([ph] * len(rows))
flat: List[Any] = []
for row in rows:
flat.extend(row)
self.conn.execute(
f"""
INSERT INTO ws_program
(market, code, snap_time, buy_qty, sell_qty, net_qty,
buy_amt, sell_amt, net_amt, source, recv_ts)
VALUES {placeholders}
""",
tuple(flat),
)
return len(rows)
except Exception as e:
logger.error("insert_ws_program_batch 실패: %s", e)
return 0
def cleanup_old_ws_orderbook(self, keep_days: int = 7) -> None:
if keep_days <= 0:
return
cutoff = (
datetime.datetime.now() - datetime.timedelta(days=keep_days)
).strftime("%Y-%m-%d %H:%M:%S")
try:
self.conn.execute(
"DELETE FROM ws_orderbook WHERE recv_ts < %s", (cutoff,)
)
logger.info("🧹 ws_orderbook 정리 완료 (recv_ts < %s)", cutoff)
except Exception as e:
logger.error("cleanup_old_ws_orderbook 실패: %s", e)
def cleanup_old_ws_program(self, keep_days: int = 7) -> None:
if keep_days <= 0:
return
cutoff = (
datetime.datetime.now() - datetime.timedelta(days=keep_days)
).strftime("%Y-%m-%d %H:%M:%S")
try:
self.conn.execute(
"DELETE FROM ws_program WHERE recv_ts < %s", (cutoff,)
)
logger.info("🧹 ws_program 정리 완료 (recv_ts < %s)", cutoff)
except Exception as e:
logger.error("cleanup_old_ws_program 실패: %s", e)
def close(self):
"""DB 연결 종료 (호출자가 매번 close 하므로 로그는 DEBUG 로 강등)."""
if self.conn:
self.conn.close()
logger.debug("🔒 DB 연결 종료")