e1ac8d119b
- 프론트엔드 UI 업데이트 (backtest.html, backtest.js) 엔진 라디오 버튼 통합 관련 반영 - Rust 플러그인(kis_rust_core) 및 컴파일 소스코드 추가 - CLI 백테스트 스크립트 수정 및 최신화 - 기타 스크래치 테스트 스크립트, 로그 요약 마크다운(.md) 등 누락 파일 일괄 반영 - 추가적으로 아직 발견되지 않은 엣지 케이스나 렌더링 오류가 포함되어 있을 가능성이 있음
1107 lines
47 KiB
Python
1107 lines
47 KiB
Python
"""
|
|
kis_trader/database/db_manager.py — 통합 DB 관리자
|
|
==================================================
|
|
기존 ``database.TradeDB``(MariaDB 2,000줄)를 **그대로 재사용**하고,
|
|
본 모듈에서는 주문(order_id=ODNO) 단위 추적을 위한 ``orders`` 테이블을
|
|
추가·관리한다.
|
|
|
|
핵심 설계:
|
|
1. ODNO(한투 주문번호)를 PK 로 사용 → DB 레벨에서 중복 주문 차단.
|
|
2. UNIQUE(strategy_id, code, order_date, side) 인덱스 → 같은 전략이
|
|
같은 날 같은 종목·같은 방향으로 중복 주문 시도 시 서버단 차단.
|
|
3. 전략별 filter → 꼬리잡기봇과 스캘핑봇의 "남의 집 물건 팔기" 방지.
|
|
|
|
본 모듈은 TradeDB 를 수정하지 않고 래핑한다 (원본 코드 존중 규칙).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import logging
|
|
from typing import Dict, List, Optional
|
|
|
|
from database import TradeDB # 프로젝트 루트 모듈
|
|
|
|
logger = logging.getLogger("kis_trader.db")
|
|
|
|
# 싱글톤: 프로세스 내 TradeDBExt 재사용 (연결 낭비 방지)
|
|
_SINGLETON: Optional["TradeDBExt"] = None
|
|
|
|
|
|
def get_db() -> "TradeDBExt":
|
|
global _SINGLETON
|
|
if _SINGLETON is None:
|
|
_SINGLETON = TradeDBExt()
|
|
return _SINGLETON
|
|
|
|
|
|
class TradeDBExt:
|
|
"""
|
|
기존 TradeDB 를 합성(Composition)한 래퍼. ``self.raw`` 로 원본 객체 접근.
|
|
``orders`` 테이블 생성/마이그레이션을 책임지고, 주문 이력을 기록/조회하는
|
|
메서드를 추가로 제공한다.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.raw = TradeDB()
|
|
self.conn = self.raw.conn
|
|
self._ensure_orders_table()
|
|
|
|
# ------------------------------------------------------------------
|
|
# 테이블 마이그레이션: orders
|
|
# ------------------------------------------------------------------
|
|
def _ensure_orders_table(self) -> None:
|
|
"""
|
|
orders 테이블:
|
|
ord_no : 한투 주문번호(ODNO). PK — 서버단 중복 차단.
|
|
strategy_id : SCALP_RSI_REVERSAL, SHORT_ANT_SHAKING 등
|
|
code/name : 종목코드/명
|
|
side : BUY / SELL
|
|
qty : 주문 수량
|
|
price : 주문 시 참고가 (시장가=0 일 수 있음)
|
|
filled_qty : 실제 체결 수량
|
|
filled_avg_price : 평균 체결가
|
|
status : SUBMITTED / FILLED / PARTIAL / REJECTED / CANCELLED
|
|
msg_cd/msg1 : 한투 응답 메시지(실패 원인 추적)
|
|
submitted_at : 주문 접수 시각
|
|
filled_at : 체결 확인 시각
|
|
|
|
PK: 서로게이트 id(auto_increment). 중복 차단은
|
|
(ord_no, strategy_id, code, side, ord_date) 복합 UNIQUE 로 한정한다.
|
|
(구) UNIQUE(strategy,code,side,ord_date) 는 당일 재매수·부분체결 재주문 시
|
|
HTS 에는 체결됐는데 orders/active_trades 미기록 버그 유발 → 제거.
|
|
|
|
※ ord_no 단독 PK 였던 과거엔 "다른 영업일·다른 전략" 주문이 같은 ODNO 를
|
|
받아도(모의투자 서버가 과거 이미 쓴 ODNO 를 재발급하는 사례 확인됨) DB가
|
|
전역 중복으로 오판 → insert 실패 → 그 매수 시도를 통째로 포기해
|
|
active_trades 미기록(장마감 고아복구 전까지 손절 무방비) 사고가 발생했다.
|
|
복합 UNIQUE 로 좁혀 "진짜 같은 날·같은 전략·같은 종목·같은 방향의 재전송"만
|
|
차단하고, 그 외 ODNO 재사용은 정상 신규 주문으로 기록되게 한다.
|
|
"""
|
|
try:
|
|
self.conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS orders (
|
|
id BIGINT NOT NULL AUTO_INCREMENT,
|
|
ord_no VARCHAR(30) NOT NULL,
|
|
strategy_id VARCHAR(40) NOT NULL,
|
|
code VARCHAR(20) NOT NULL,
|
|
name VARCHAR(100) NOT NULL DEFAULT '',
|
|
side VARCHAR(8) NOT NULL,
|
|
qty INT NOT NULL,
|
|
price DOUBLE DEFAULT 0,
|
|
filled_qty INT NOT NULL DEFAULT 0,
|
|
filled_avg_price DOUBLE DEFAULT 0,
|
|
status VARCHAR(16) NOT NULL DEFAULT 'SUBMITTED',
|
|
msg_cd VARCHAR(20) DEFAULT NULL,
|
|
msg1 VARCHAR(300) DEFAULT NULL,
|
|
ord_date VARCHAR(10) NOT NULL,
|
|
submitted_at VARCHAR(30) NOT NULL,
|
|
filled_at VARCHAR(30) DEFAULT NULL,
|
|
raw_json MEDIUMTEXT DEFAULT NULL,
|
|
is_mock TINYINT(1) NULL,
|
|
target_price DOUBLE DEFAULT NULL,
|
|
slippage_pct DOUBLE DEFAULT NULL,
|
|
PRIMARY KEY (id),
|
|
UNIQUE KEY uq_ord_no_ctx (ord_no, strategy_id, code, side, ord_date, is_mock),
|
|
INDEX idx_strategy_date (strategy_id, ord_date),
|
|
INDEX idx_code_date (code, ord_date),
|
|
INDEX idx_orders_mock_date (is_mock, ord_date)
|
|
) CHARACTER SET utf8mb4
|
|
""")
|
|
self._migrate_orders_drop_daily_side_unique()
|
|
self._migrate_orders_pk_scope()
|
|
self._migrate_orders_is_mock()
|
|
self._migrate_orders_broker_debug_columns()
|
|
self._migrate_orders_slippage_columns()
|
|
logger.info("📊 orders 테이블 확인/생성 완료")
|
|
except Exception as e:
|
|
logger.warning("orders 테이블 생성 실패(무시·폴백): %s", e)
|
|
|
|
def _resolve_is_mock(self, explicit=None) -> int:
|
|
"""모의=1 · 실전=0 — TradeDB.resolve_kis_is_mock 과 동일."""
|
|
return int(TradeDB.resolve_kis_is_mock(explicit))
|
|
|
|
def _migrate_orders_drop_daily_side_unique(self) -> None:
|
|
"""당일 1회 매수 UNIQUE 제거 — 재진입·삼성전자 누적매수 DB 미기록 방지."""
|
|
try:
|
|
rows = self.conn.execute(
|
|
"SHOW INDEX FROM orders WHERE Key_name = %s",
|
|
("uq_strategy_code_side_date",),
|
|
).fetchall()
|
|
if rows:
|
|
self.conn.execute(
|
|
"ALTER TABLE orders DROP INDEX uq_strategy_code_side_date"
|
|
)
|
|
logger.info(
|
|
"✅ orders.uq_strategy_code_side_date 제거 "
|
|
"(당일 재매수 시 DB 동기화 가능)"
|
|
)
|
|
except Exception as e:
|
|
logger.debug("orders UNIQUE 마이그레이션 스킵: %s", e)
|
|
|
|
def _migrate_orders_pk_scope(self) -> None:
|
|
"""기존(구버전) 테이블: ord_no 단독 PK → surrogate id PK + 복합 UNIQUE 로 전환.
|
|
|
|
모의투자 서버의 ODNO 재사용 버그로 다른 날짜/전략 주문이 같은 ODNO 를
|
|
받으면 종전 PK(ord_no 단독)에서는 무조건 "중복"으로 차단됐다. 이미
|
|
운영 중인 DB(구 스키마)를 새 스키마로 안전 전환한다.
|
|
"""
|
|
try:
|
|
cols = self.conn.get_columns("orders")
|
|
if "id" not in cols:
|
|
self.conn.execute(
|
|
"ALTER TABLE orders "
|
|
"ADD COLUMN id BIGINT NOT NULL AUTO_INCREMENT FIRST, "
|
|
"DROP PRIMARY KEY, "
|
|
"ADD PRIMARY KEY (id)"
|
|
)
|
|
logger.info(
|
|
"✅ orders.id(surrogate PK) 추가 — ord_no 단독 PK 제거"
|
|
)
|
|
rows = self.conn.execute(
|
|
"SHOW INDEX FROM orders WHERE Key_name = %s",
|
|
("uq_ord_no_ctx",),
|
|
).fetchall()
|
|
if not rows:
|
|
self.conn.execute(
|
|
"ALTER TABLE orders ADD UNIQUE INDEX uq_ord_no_ctx "
|
|
"(ord_no, strategy_id, code, side, ord_date)"
|
|
)
|
|
logger.info(
|
|
"✅ orders.uq_ord_no_ctx 추가 — 중복 판정을 "
|
|
"(ord_no, strategy_id, code, side, ord_date) 로 한정"
|
|
)
|
|
except Exception as e:
|
|
logger.warning("orders PK 범위 마이그레이션 실패(무시·폴백): %s", e)
|
|
|
|
def _migrate_orders_is_mock(self) -> None:
|
|
"""orders.is_mock — 모의/실전 분리. UNIQUE 에 is_mock 포함(모의·실전 동시 허용)."""
|
|
try:
|
|
cols = self.conn.get_columns("orders")
|
|
if "is_mock" not in cols:
|
|
self.conn.execute(
|
|
"ALTER TABLE orders ADD COLUMN is_mock TINYINT(1) NULL"
|
|
)
|
|
logger.info("📌 orders.is_mock 컬럼 추가 (모의=1/실전=0)")
|
|
# 과거 이력은 모의 구간으로 1회 백필 (스냅샷 없음)
|
|
n_null = self.conn.execute(
|
|
"SELECT COUNT(*) AS c FROM orders WHERE is_mock IS NULL"
|
|
).fetchone()
|
|
n_left = int((n_null or {}).get("c") or 0)
|
|
if n_left > 0:
|
|
cur_mock = self._resolve_is_mock(None)
|
|
self.conn.execute(
|
|
"UPDATE orders SET is_mock=%s WHERE is_mock IS NULL",
|
|
(cur_mock,),
|
|
)
|
|
logger.info(
|
|
"📌 orders.is_mock 잔여 NULL→현재KIS_MOCK(%s) %d건",
|
|
cur_mock, n_left,
|
|
)
|
|
# UNIQUE: (…, is_mock) — 모의 ODNO 와 실전 ODNO 가 같아도 공존
|
|
rows = self.conn.execute(
|
|
"SHOW INDEX FROM orders WHERE Key_name = %s",
|
|
("uq_ord_no_ctx",),
|
|
).fetchall()
|
|
idx_cols = []
|
|
for r in (rows or []):
|
|
col = r["Column_name"] if isinstance(r, dict) else r[4]
|
|
seq = int(r["Seq_in_index"] if isinstance(r, dict) else r[3])
|
|
idx_cols.append((seq, str(col)))
|
|
idx_cols = [c for _, c in sorted(idx_cols)]
|
|
want = ["ord_no", "strategy_id", "code", "side", "ord_date", "is_mock"]
|
|
if idx_cols != want:
|
|
if rows:
|
|
self.conn.execute("ALTER TABLE orders DROP INDEX uq_ord_no_ctx")
|
|
self.conn.execute(
|
|
"ALTER TABLE orders ADD UNIQUE INDEX uq_ord_no_ctx "
|
|
"(ord_no, strategy_id, code, side, ord_date, is_mock)"
|
|
)
|
|
logger.info(
|
|
"✅ orders.uq_ord_no_ctx → (ord_no,strategy_id,code,side,ord_date,is_mock)"
|
|
)
|
|
# 조회용 인덱스
|
|
mock_idx = self.conn.execute(
|
|
"SHOW INDEX FROM orders WHERE Key_name = %s",
|
|
("idx_orders_mock_date",),
|
|
).fetchall()
|
|
if not mock_idx:
|
|
self.conn.execute(
|
|
"ALTER TABLE orders ADD INDEX idx_orders_mock_date (is_mock, ord_date)"
|
|
)
|
|
logger.info("✅ orders.idx_orders_mock_date 추가")
|
|
except Exception as e:
|
|
logger.warning("orders.is_mock 마이그레이션 실패(무시·폴백): %s", e)
|
|
|
|
def _migrate_orders_broker_debug_columns(self) -> None:
|
|
"""orders — cancelable reconcile 디버그 스냅샷 컬럼."""
|
|
specs = {
|
|
"broker_open_qty": "INT NULL",
|
|
"broker_open_odno": "VARCHAR(30) NULL",
|
|
"broker_reconcile_at": "VARCHAR(30) NULL",
|
|
"broker_reconcile_note": "VARCHAR(200) NULL",
|
|
}
|
|
try:
|
|
cols = self.conn.get_columns("orders")
|
|
for col, ddl in specs.items():
|
|
if col not in cols:
|
|
self.conn.execute(
|
|
f"ALTER TABLE orders ADD COLUMN `{col}` {ddl}"
|
|
)
|
|
logger.info("📌 orders.%s 컬럼 추가 (reconcile 디버그)", col)
|
|
except Exception as e:
|
|
logger.warning("orders broker debug 마이그레이션 실패(무시·폴백): %s", e)
|
|
|
|
def _migrate_orders_slippage_columns(self) -> None:
|
|
"""orders — 실매매 vs 백테스트 슬리피지 갭 모니터링용 컬럼."""
|
|
specs = {
|
|
"target_price": "DOUBLE NULL",
|
|
"slippage_pct": "DOUBLE NULL",
|
|
}
|
|
try:
|
|
cols = self.conn.get_columns("orders")
|
|
for col, ddl in specs.items():
|
|
if col not in cols:
|
|
self.conn.execute(
|
|
f"ALTER TABLE orders ADD COLUMN `{col}` {ddl}"
|
|
)
|
|
logger.info("📌 orders.%s 컬럼 추가 (슬리피지 모니터링)", col)
|
|
except Exception as e:
|
|
logger.warning("orders slippage 마이그레이션 실패(무시·폴백): %s", e)
|
|
|
|
# ------------------------------------------------------------------
|
|
# orders CRUD
|
|
# ------------------------------------------------------------------
|
|
def insert_order(
|
|
self,
|
|
*,
|
|
ord_no: str,
|
|
strategy_id: str,
|
|
code: str,
|
|
name: str,
|
|
side: str,
|
|
qty: int,
|
|
price: float = 0.0,
|
|
status: str = "SUBMITTED",
|
|
msg_cd: Optional[str] = None,
|
|
msg1: Optional[str] = None,
|
|
raw_json: Optional[str] = None,
|
|
is_mock=None,
|
|
target_price: Optional[float] = None,
|
|
) -> bool:
|
|
"""
|
|
주문 기록 INSERT.
|
|
중복 판정은 (ord_no, strategy_id, code, side, ord_date, is_mock) 복합 UNIQUE.
|
|
다른 날짜·전략·종목·모의/실전이 같은 ODNO 를 받아도 오탐 차단하지 않는다.
|
|
side: 'BUY' | 'SELL'
|
|
"""
|
|
side = (side or "").upper()
|
|
assert side in ("BUY", "SELL"), f"side must be BUY|SELL, got {side}"
|
|
now = datetime.datetime.now()
|
|
mock_flag = self._resolve_is_mock(is_mock)
|
|
try:
|
|
with self.conn:
|
|
self.conn.execute(
|
|
"""
|
|
INSERT INTO orders (
|
|
ord_no, strategy_id, code, name, side, qty, price,
|
|
status, msg_cd, msg1, ord_date, submitted_at, raw_json, is_mock, target_price
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
""",
|
|
(
|
|
ord_no, strategy_id, code, name, side, qty, price,
|
|
status, msg_cd, msg1,
|
|
now.strftime("%Y-%m-%d"),
|
|
now.strftime("%Y-%m-%d %H:%M:%S"),
|
|
raw_json,
|
|
mock_flag,
|
|
target_price,
|
|
),
|
|
)
|
|
return True
|
|
except Exception as e:
|
|
# 중복 PK(1062) / 유니크 인덱스 위반은 정상적 차단으로 간주
|
|
msg = str(e)
|
|
if "1062" in msg or "Duplicate entry" in msg:
|
|
logger.warning(
|
|
"⚠️ [주문중복차단] strategy=%s code=%s side=%s ord_no=%s is_mock=%s "
|
|
"(같은 날짜·전략·종목·방향·계좌모드로 이미 DB 존재)",
|
|
strategy_id, code, side, ord_no, mock_flag,
|
|
)
|
|
return False
|
|
logger.error("insert_order 실패 (ord_no=%s): %s", ord_no, e)
|
|
return False
|
|
|
|
def update_order_fill(
|
|
self,
|
|
*,
|
|
ord_no: str,
|
|
strategy_id: str,
|
|
code: str,
|
|
filled_qty: int,
|
|
filled_avg_price: float,
|
|
status: str = "FILLED",
|
|
ord_date: Optional[str] = None,
|
|
) -> bool:
|
|
"""주문 체결 확인 후 체결가/체결수량/상태 갱신.
|
|
|
|
strategy_id·code·ord_date 로 정확히 그 행만 갱신 — ODNO 재사용 시
|
|
다른 날짜/전략의 동일 ODNO 행을 잘못 덮어쓰는 사고 방지.
|
|
"""
|
|
now_dt = datetime.datetime.now()
|
|
now = now_dt.strftime("%Y-%m-%d %H:%M:%S")
|
|
od = ord_date or now_dt.strftime("%Y-%m-%d")
|
|
mock_flag = self._resolve_is_mock(None)
|
|
try:
|
|
with self.conn:
|
|
self.conn.execute(
|
|
"""
|
|
UPDATE orders
|
|
SET filled_qty=%s,
|
|
filled_avg_price=%s,
|
|
slippage_pct=IF(target_price IS NOT NULL AND target_price > 0, ((%s - target_price) / target_price) * 100, slippage_pct),
|
|
status=%s,
|
|
filled_at=%s
|
|
WHERE ord_no=%s AND strategy_id=%s AND code=%s AND ord_date=%s
|
|
AND is_mock=%s
|
|
""",
|
|
(filled_qty, filled_avg_price, filled_avg_price, status, now,
|
|
ord_no, strategy_id, code, od, mock_flag),
|
|
)
|
|
return True
|
|
except Exception as e:
|
|
logger.error("update_order_fill 실패 (%s): %s", ord_no, e)
|
|
return False
|
|
|
|
def mark_order_rejected(
|
|
self,
|
|
*,
|
|
ord_no: str,
|
|
strategy_id: str,
|
|
code: str,
|
|
msg_cd: str = "",
|
|
msg1: str = "",
|
|
ord_date: Optional[str] = None,
|
|
) -> None:
|
|
"""주문 실패/거부 시 상태 REJECTED 처리 (strategy_id·code·ord_date·is_mock 로 행 한정)."""
|
|
od = ord_date or datetime.datetime.now().strftime("%Y-%m-%d")
|
|
mock_flag = self._resolve_is_mock(None)
|
|
try:
|
|
with self.conn:
|
|
self.conn.execute(
|
|
"""
|
|
UPDATE orders SET status='REJECTED', msg_cd=%s, msg1=%s
|
|
WHERE ord_no=%s AND strategy_id=%s AND code=%s AND ord_date=%s
|
|
AND is_mock=%s
|
|
""",
|
|
(msg_cd, msg1, ord_no, strategy_id, code, od, mock_flag),
|
|
)
|
|
except Exception as e:
|
|
logger.debug("mark_order_rejected 실패 (%s): %s", ord_no, e)
|
|
|
|
def update_order_status(
|
|
self,
|
|
*,
|
|
ord_no: str,
|
|
strategy_id: str,
|
|
code: str,
|
|
status: str,
|
|
ord_date: Optional[str] = None,
|
|
broker_reconcile_note: Optional[str] = None,
|
|
) -> bool:
|
|
"""체결 대기 등 — filled_qty 없이 status 만 갱신 (strategy_id·code·ord_date·is_mock 로 행 한정)."""
|
|
od = ord_date or datetime.datetime.now().strftime("%Y-%m-%d")
|
|
mock_flag = self._resolve_is_mock(None)
|
|
note = str(broker_reconcile_note or "").strip() or None
|
|
try:
|
|
with self.conn:
|
|
if note:
|
|
self.conn.execute(
|
|
"UPDATE orders SET status=%s, broker_reconcile_note=%s, "
|
|
"broker_reconcile_at=%s "
|
|
"WHERE ord_no=%s AND strategy_id=%s AND code=%s AND ord_date=%s "
|
|
"AND is_mock=%s",
|
|
(
|
|
status,
|
|
note[:200],
|
|
datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
ord_no,
|
|
strategy_id,
|
|
code,
|
|
od,
|
|
mock_flag,
|
|
),
|
|
)
|
|
else:
|
|
self.conn.execute(
|
|
"UPDATE orders SET status=%s "
|
|
"WHERE ord_no=%s AND strategy_id=%s AND code=%s AND ord_date=%s "
|
|
"AND is_mock=%s",
|
|
(status, ord_no, strategy_id, code, od, mock_flag),
|
|
)
|
|
return True
|
|
except Exception as e:
|
|
logger.error("update_order_status 실패 (%s): %s", ord_no, e)
|
|
return False
|
|
|
|
def update_order_broker_snapshot(
|
|
self,
|
|
*,
|
|
ord_no: str,
|
|
strategy_id: str,
|
|
code: str,
|
|
open_qty: Optional[int] = None,
|
|
open_odno: Optional[str] = None,
|
|
note: Optional[str] = None,
|
|
ord_date: Optional[str] = None,
|
|
) -> bool:
|
|
"""cancelable reconcile — 브로커 미체결 스냅샷 기록."""
|
|
od = ord_date or datetime.datetime.now().strftime("%Y-%m-%d")
|
|
mock_flag = self._resolve_is_mock(None)
|
|
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
try:
|
|
with self.conn:
|
|
self.conn.execute(
|
|
"""
|
|
UPDATE orders
|
|
SET broker_open_qty=%s,
|
|
broker_open_odno=%s,
|
|
broker_reconcile_at=%s,
|
|
broker_reconcile_note=%s
|
|
WHERE ord_no=%s AND strategy_id=%s AND code=%s
|
|
AND ord_date=%s AND is_mock=%s
|
|
""",
|
|
(
|
|
int(open_qty) if open_qty is not None else None,
|
|
str(open_odno or "").strip() or None,
|
|
now,
|
|
str(note or "").strip()[:200] or None,
|
|
ord_no,
|
|
strategy_id,
|
|
code,
|
|
od,
|
|
mock_flag,
|
|
),
|
|
)
|
|
return True
|
|
except Exception as e:
|
|
logger.error("update_order_broker_snapshot 실패 (%s): %s", ord_no, e)
|
|
return False
|
|
|
|
def get_latest_sell_order_for_code(
|
|
self,
|
|
strategy_id: str,
|
|
code: str,
|
|
*,
|
|
statuses: Optional[tuple] = None,
|
|
) -> Optional[Dict]:
|
|
"""당일·전략·종목 최신 매도 1건 (reconcile용)."""
|
|
today = datetime.datetime.now().strftime("%Y-%m-%d")
|
|
mock_flag = self._resolve_is_mock(None)
|
|
st = statuses or ("CANCELLED", "PENDING_FILL", "SUBMITTED", "PARTIAL", "FILLED")
|
|
placeholders = ",".join(["%s"] * len(st))
|
|
try:
|
|
row = self.conn.execute(
|
|
f"""
|
|
SELECT * FROM orders
|
|
WHERE ord_date=%s AND is_mock=%s
|
|
AND strategy_id=%s AND code=%s AND side='SELL'
|
|
AND status IN ({placeholders})
|
|
ORDER BY submitted_at DESC
|
|
LIMIT 1
|
|
""",
|
|
(today, mock_flag, strategy_id, code, *st),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
except Exception as e:
|
|
logger.error(
|
|
"get_latest_sell_order_for_code 실패 (%s/%s): %s",
|
|
strategy_id,
|
|
code,
|
|
e,
|
|
)
|
|
return None
|
|
|
|
def get_pending_fill_orders(self) -> List[Dict]:
|
|
"""
|
|
당일 미확인·부분체결 주문 (체결 qty < 주문 qty).
|
|
heartbeat ``poll_pending_fills`` 재조회용.
|
|
"""
|
|
today = datetime.datetime.now().strftime("%Y-%m-%d")
|
|
mock_flag = self._resolve_is_mock(None)
|
|
try:
|
|
rows = self.conn.execute(
|
|
"""
|
|
SELECT * FROM orders
|
|
WHERE ord_date=%s
|
|
AND is_mock=%s
|
|
AND status IN ('SUBMITTED', 'PENDING_FILL', 'PARTIAL')
|
|
AND COALESCE(filled_qty, 0) < qty
|
|
ORDER BY submitted_at ASC
|
|
""",
|
|
(today, mock_flag),
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
except Exception as e:
|
|
logger.error("get_pending_fill_orders 실패: %s", e)
|
|
return []
|
|
|
|
def get_pending_sell_order(
|
|
self, strategy_id: str, code: str
|
|
) -> Optional[Dict]:
|
|
"""동일 전략·종목 미체결 매도 1건 (중복 주문 방지용)."""
|
|
today = datetime.datetime.now().strftime("%Y-%m-%d")
|
|
mock_flag = self._resolve_is_mock(None)
|
|
try:
|
|
row = self.conn.execute(
|
|
"""
|
|
SELECT * FROM orders
|
|
WHERE ord_date=%s
|
|
AND is_mock=%s
|
|
AND strategy_id=%s
|
|
AND code=%s
|
|
AND side='SELL'
|
|
AND status IN ('SUBMITTED', 'PENDING_FILL', 'PARTIAL')
|
|
AND COALESCE(filled_qty, 0) < qty
|
|
ORDER BY submitted_at DESC
|
|
LIMIT 1
|
|
""",
|
|
(today, mock_flag, strategy_id, code),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
except Exception as e:
|
|
logger.error("get_pending_sell_order 실패 (%s/%s): %s", strategy_id, code, e)
|
|
return None
|
|
|
|
def get_pending_buy_order(
|
|
self, strategy_id: str, code: str
|
|
) -> Optional[Dict]:
|
|
"""동일 전략·종목 미체결 매수 1건 (중복 주문 방지용).
|
|
|
|
체결확인 API(inquire-daily-ccld) 장애로 fill 미확인(PENDING_FILL) 상태일 때,
|
|
전략 루프가 같은 종목을 매 턴 재주문하는 폭주를 막는다.
|
|
만료 시 poll_pending_fills 가 status 를 CANCELLED 로 바꿔 자동 해제된다.
|
|
"""
|
|
today = datetime.datetime.now().strftime("%Y-%m-%d")
|
|
mock_flag = self._resolve_is_mock(None)
|
|
try:
|
|
row = self.conn.execute(
|
|
"""
|
|
SELECT * FROM orders
|
|
WHERE ord_date=%s
|
|
AND is_mock=%s
|
|
AND strategy_id=%s
|
|
AND code=%s
|
|
AND side='BUY'
|
|
AND status IN ('SUBMITTED', 'PENDING_FILL', 'PARTIAL')
|
|
AND COALESCE(filled_qty, 0) < qty
|
|
ORDER BY submitted_at DESC
|
|
LIMIT 1
|
|
""",
|
|
(today, mock_flag, strategy_id, code),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
except Exception as e:
|
|
logger.error("get_pending_buy_order 실패 (%s/%s): %s", strategy_id, code, e)
|
|
return None
|
|
|
|
def get_order_by_odno(
|
|
self,
|
|
ord_no: str,
|
|
*,
|
|
strategy_id: Optional[str] = None,
|
|
code: Optional[str] = None,
|
|
) -> Optional[Dict]:
|
|
"""ODNO 로 주문 조회. ODNO 는 더 이상 전역 유일하지 않으므로
|
|
(모의투자 ODNO 재사용 버그) strategy_id·code 를 함께 주면 정확히
|
|
그 행만, 안 주면 최신 순 1건을 반환한다. 현재 KIS_MOCK 우선."""
|
|
mock_flag = self._resolve_is_mock(None)
|
|
try:
|
|
if strategy_id and code:
|
|
row = self.conn.execute(
|
|
"SELECT * FROM orders WHERE ord_no=%s AND strategy_id=%s AND code=%s "
|
|
"AND is_mock=%s "
|
|
"ORDER BY submitted_at DESC LIMIT 1",
|
|
(ord_no, strategy_id, code, mock_flag),
|
|
).fetchone()
|
|
else:
|
|
row = self.conn.execute(
|
|
"SELECT * FROM orders WHERE ord_no=%s AND is_mock=%s "
|
|
"ORDER BY submitted_at DESC LIMIT 1",
|
|
(ord_no, mock_flag),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
except Exception as e:
|
|
logger.error("get_order_by_odno 실패: %s", e)
|
|
return None
|
|
|
|
def get_orders_today(
|
|
self, strategy_id: Optional[str] = None
|
|
) -> List[Dict]:
|
|
today = datetime.datetime.now().strftime("%Y-%m-%d")
|
|
mock_flag = self._resolve_is_mock(None)
|
|
try:
|
|
if strategy_id:
|
|
rows = self.conn.execute(
|
|
"""
|
|
SELECT * FROM orders
|
|
WHERE ord_date=%s AND strategy_id=%s AND is_mock=%s
|
|
ORDER BY submitted_at DESC
|
|
""",
|
|
(today, strategy_id, mock_flag),
|
|
).fetchall()
|
|
else:
|
|
rows = self.conn.execute(
|
|
"""
|
|
SELECT * FROM orders
|
|
WHERE ord_date=%s AND is_mock=%s
|
|
ORDER BY submitted_at DESC
|
|
""",
|
|
(today, mock_flag),
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
except Exception as e:
|
|
logger.error("get_orders_today 실패: %s", e)
|
|
return []
|
|
|
|
# ------------------------------------------------------------------
|
|
# 조건검색 유니버스 스냅샷 (백테스트용 이력)
|
|
# ------------------------------------------------------------------
|
|
# 설계 원칙:
|
|
# - 5분 슬롯 스냅샷은 정보 손실이 크다(그 사이 들락날락한 종목 유실).
|
|
# - 따라서 **변동(ENTER/EXIT) 발생 tick 마다 풀 스냅샷**을 저장한다.
|
|
# - 변동 없는 tick 은 저장하지 않는다 (공간 절약).
|
|
# - 각 스냅샷은 같은 ``event_time`` 을 공유하는 N 행으로 표현.
|
|
# - 백테스트 시 특정 시점 유니버스 = 그 시점 이전 최근 event_time.
|
|
_HISTORY_MIGRATED = False
|
|
|
|
def _ensure_history_columns(self) -> None:
|
|
"""
|
|
target_candidates_history 에 필요한 확장 컬럼을 자동 추가:
|
|
* strategy_id — 전략 구분 (기존 키움 스캐너 행은 NULL)
|
|
* event_time — 스냅샷 시각 ('YYYY-MM-DD HH:MM:SS.ffffff', varchar(26))
|
|
"""
|
|
if TradeDBExt._HISTORY_MIGRATED:
|
|
return
|
|
try:
|
|
cols = self.conn.get_columns("target_candidates_history")
|
|
if "strategy_id" not in cols:
|
|
self.conn.execute(
|
|
"ALTER TABLE target_candidates_history "
|
|
"ADD COLUMN strategy_id VARCHAR(30) DEFAULT NULL"
|
|
)
|
|
logger.info("📌 target_candidates_history.strategy_id 컬럼 추가")
|
|
if "event_time" not in cols:
|
|
self.conn.execute(
|
|
"ALTER TABLE target_candidates_history "
|
|
"ADD COLUMN event_time VARCHAR(26) DEFAULT NULL"
|
|
)
|
|
logger.info("📌 target_candidates_history.event_time 컬럼 추가")
|
|
else:
|
|
try:
|
|
meta = self.conn.execute(
|
|
"SHOW COLUMNS FROM target_candidates_history LIKE %s",
|
|
("event_time",),
|
|
).fetchone()
|
|
typ = str((meta or {}).get("Type") or "").lower()
|
|
n = 0
|
|
if "varchar" in typ:
|
|
import re
|
|
m = re.search(r"varchar\((\d+)\)", typ)
|
|
n = int(m.group(1)) if m else 0
|
|
if n and n < 26:
|
|
self.conn.execute(
|
|
"ALTER TABLE target_candidates_history "
|
|
"MODIFY COLUMN event_time VARCHAR(26) DEFAULT NULL"
|
|
)
|
|
logger.info(
|
|
"📌 target_candidates_history.event_time 길이 %s→26",
|
|
n,
|
|
)
|
|
except Exception as e:
|
|
logger.debug("event_time 길이 ALTER: %s", e)
|
|
# 인덱스 — 기존 유지 + 당일 slot_key 집계용 복합 (DROP 없음)
|
|
for idx_name, idx_def in (
|
|
("idx_strategy_event",
|
|
"(strategy_id, event_time)"),
|
|
("idx_tch_slot_sid",
|
|
"(slot_key, strategy_id)"),
|
|
# resolve_stock_display_name · 조건 REAL 편입 시 code→name (206만+행 풀스캔 방지)
|
|
("idx_tch_code_id",
|
|
"(code, id)"),
|
|
):
|
|
try:
|
|
self.conn.execute(
|
|
f"CREATE INDEX {idx_name} "
|
|
f"ON target_candidates_history {idx_def}"
|
|
)
|
|
except Exception as e:
|
|
# 이미 존재 / 버전 미지원 → 무시
|
|
if "1061" not in str(e) and "Duplicate" not in str(e):
|
|
logger.debug("idx %s 생성 실패: %s", idx_name, e)
|
|
except Exception as e:
|
|
logger.debug("history 컬럼 마이그레이션: %s", e)
|
|
TradeDBExt._HISTORY_MIGRATED = True
|
|
|
|
def insert_condition_universe_snapshot(
|
|
self,
|
|
*,
|
|
strategy_id: str,
|
|
event_time: str,
|
|
items: List[Dict],
|
|
slot_key: Optional[str] = None,
|
|
) -> int:
|
|
"""
|
|
조건검색 변동(ENTER/EXIT) 발생 tick 마다 호출되는 **풀 스냅샷** 저장.
|
|
같은 ``event_time`` 으로 들어온 N 행 = 그 시점의 유니버스 전체.
|
|
INSERT만 (같은 초 DELETE 바꿔치기 없음). 한 스냅샷 N행은 한 트랜잭션.
|
|
|
|
Args:
|
|
strategy_id : 'SCALP' | 'SHORT' | 'BREAKOUT' ...
|
|
event_time : 'YYYY-MM-DD HH:MM:SS.ffffff'. 백테스트 기준 시각.
|
|
items : [{"code": "...", "name": "..."}, ...] — 현재 유니버스 전체
|
|
slot_key : (선택) 기존 5분 슬롯 키. 과거 대시보드/쿼리 호환용.
|
|
|
|
Returns: 실제 INSERT 된 행수
|
|
"""
|
|
if not items or not event_time:
|
|
return 0
|
|
self._ensure_history_columns()
|
|
|
|
# slot_key 기본값 — 기존 스키마 NOT NULL 이므로 최소한 값 채움
|
|
if not slot_key:
|
|
try:
|
|
dp = event_time.replace("-", "").replace(":", "").replace(" ", "")
|
|
slot_key = dp[:12] # YYYYMMDDHHMM
|
|
except Exception:
|
|
slot_key = event_time[:12]
|
|
|
|
rows = []
|
|
for it in items:
|
|
code = (it.get("code") or "").strip()
|
|
if not code:
|
|
continue
|
|
raw_name = (it.get("name") or code)[:100]
|
|
name = raw_name
|
|
if not name or name == code:
|
|
try:
|
|
from kis_trader.utils.stock_name import resolve_stock_display_name
|
|
name = resolve_stock_display_name(
|
|
self, code, fallback=code, cache_to_meta=True,
|
|
)[:100]
|
|
except Exception:
|
|
name = code
|
|
rows.append(
|
|
(slot_key, event_time, code, name, 0.0, 0.0,
|
|
"Q", "", "", strategy_id, event_time),
|
|
)
|
|
if not rows:
|
|
return 0
|
|
sql = """
|
|
INSERT INTO target_candidates_history
|
|
(slot_key, scan_time, code, name, score, price,
|
|
market, sector, theme, strategy_id, event_time)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
"""
|
|
try:
|
|
n = self.conn.executemany_tx(sql, rows)
|
|
return int(n if n and n > 0 else len(rows))
|
|
except Exception as e:
|
|
logger.warning("history 스냅샷 저장 실패: %s", e)
|
|
return 0
|
|
|
|
# 범용 별칭 — 유니버스 소스가 condition 이든 ranking 이든 같은 테이블을 공유
|
|
# (함수명 혼동 줄이기 위한 alias. 기존 호출부는 그대로 동작)
|
|
def insert_universe_snapshot(
|
|
self,
|
|
*,
|
|
strategy_id: str,
|
|
event_time: str,
|
|
items: List[Dict],
|
|
slot_key: Optional[str] = None,
|
|
) -> int:
|
|
return self.insert_condition_universe_snapshot(
|
|
strategy_id=strategy_id, event_time=event_time,
|
|
items=items, slot_key=slot_key,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 백테스트 헬퍼: 특정 시점의 유니버스 복원
|
|
# ------------------------------------------------------------------
|
|
@staticmethod
|
|
def _universe_history_table(history_source: str = "kiwoom") -> str:
|
|
from kis_trader.backtest.universe_history_source import history_table_for_source
|
|
return history_table_for_source(history_source)
|
|
|
|
def get_universe_at(
|
|
self, *, strategy_id: str, at_time: str, history_source: str = "kiwoom",
|
|
) -> List[Dict]:
|
|
"""
|
|
``at_time`` ('YYYY-MM-DD HH:MM:SS' 또는 마이크로초) 시점에 봇이 보던 유니버스를 복원.
|
|
= 그 시각 이전의 가장 최근 event_time 스냅샷.
|
|
초 단위 조회는 그 초의 **마지막** 마이크로 스냅샷.
|
|
|
|
``history_source``: ``kiwoom``(target_candidates_history) | ``ls``(ls_candidates_history)
|
|
"""
|
|
self._ensure_history_columns()
|
|
table = self._universe_history_table(history_source)
|
|
from kis_trader.backtest.universe_timeline import event_time_query_upper
|
|
at_q = event_time_query_upper(at_time)
|
|
try:
|
|
row = self.conn.execute(
|
|
f"""
|
|
SELECT MAX(event_time) AS et
|
|
FROM {table}
|
|
WHERE strategy_id=%s AND event_time <= %s
|
|
""",
|
|
(strategy_id, at_q),
|
|
).fetchone()
|
|
et = (row or {}).get("et") if row else None
|
|
if not et:
|
|
return []
|
|
rows = self.conn.execute(
|
|
f"""
|
|
SELECT code, name FROM {table}
|
|
WHERE strategy_id=%s AND event_time=%s
|
|
ORDER BY code
|
|
""",
|
|
(strategy_id, et),
|
|
).fetchall()
|
|
return [{"code": r["code"], "name": r["name"]} for r in rows]
|
|
except Exception as e:
|
|
logger.error("get_universe_at 실패(src=%s): %s", history_source, e)
|
|
return []
|
|
|
|
def iter_universe_events(
|
|
self,
|
|
*,
|
|
strategy_id: str,
|
|
start_time: str,
|
|
end_time: str,
|
|
preserve_insert_order: bool = False,
|
|
history_source: str = "kiwoom",
|
|
) -> List[Dict]:
|
|
"""
|
|
``[start_time, end_time]`` 구간의 모든 스냅샷 이벤트를 시간순 반환.
|
|
각 이벤트 = {"event_time": ..., "codes": [...]}.
|
|
백테스트 루프에서 시점별 유니버스를 순회할 때 사용.
|
|
|
|
``preserve_insert_order=True``: 스냅샷 내 종목 순서를 DB insert(id) 순으로
|
|
유지 (백테 매수 우선순위 정합). 기본 False 는 code 가나다순.
|
|
``history_source``: ``kiwoom`` | ``ls``
|
|
"""
|
|
self._ensure_history_columns()
|
|
table = self._universe_history_table(history_source)
|
|
from kis_trader.backtest.universe_timeline import event_time_query_upper
|
|
end_q = event_time_query_upper(end_time)
|
|
order_clause = "event_time, id" if preserve_insert_order else "event_time, code"
|
|
try:
|
|
rows = self.conn.execute(
|
|
f"""
|
|
SELECT event_time, code, name
|
|
FROM {table}
|
|
WHERE strategy_id=%s
|
|
AND event_time BETWEEN %s AND %s
|
|
ORDER BY {order_clause}
|
|
""",
|
|
(strategy_id, start_time, end_q),
|
|
).fetchall()
|
|
grouped: Dict[str, List[Dict]] = {}
|
|
for r in rows:
|
|
et = r["event_time"]
|
|
grouped.setdefault(et, []).append(
|
|
{"code": r["code"], "name": r["name"]}
|
|
)
|
|
return [
|
|
{"event_time": et, "items": items}
|
|
for et, items in sorted(grouped.items())
|
|
]
|
|
except Exception as e:
|
|
logger.error("iter_universe_events 실패(src=%s): %s", history_source, e)
|
|
return []
|
|
|
|
@staticmethod
|
|
def _add_minutes_to_candle_slot(candle_slot: str, minutes: int) -> str:
|
|
"""YYYYMMDDHHMM 슬롯에 분 단위 가산."""
|
|
from datetime import datetime, timedelta
|
|
dt = datetime.strptime(candle_slot, "%Y%m%d%H%M")
|
|
return (dt + timedelta(minutes=int(minutes))).strftime("%Y%m%d%H%M")
|
|
|
|
def get_universe_by_candle_time(
|
|
self,
|
|
*,
|
|
strategy_id: str,
|
|
start_ymd: str,
|
|
end_ymd: str,
|
|
strict: bool = False,
|
|
strict_lag_minutes: int = 1,
|
|
exit_debounce_sec: int = 0,
|
|
history_source: str = "kiwoom",
|
|
) -> Dict[str, List[str]]:
|
|
"""
|
|
백테스트 편의용 — 1분봉 캔들 시각(YYYYMMDDHHMM)을 키로 하는 유니버스 dict.
|
|
|
|
실매매는 조건검색 변동 tick 마다 마이크로초 ``event_time``
|
|
(YYYY-MM-DD HH:MM:SS.ffffff) 으로 ``target_candidates_history`` 에 적재한다.
|
|
반면 백테스트 엔진은 1분봉 단위로 돌아가므로, 각 캔들에 대해
|
|
**그 캔들의 종가 형성 시각 (HH:MM:59 초의 마지막 스냅샷) 이전의 가장 최근 스냅샷** 을
|
|
선택해 "그 시점에 봇이 보던 유니버스" 를 재현한다.
|
|
|
|
``strict=True`` (모멘텀 백테 실매 정합):
|
|
분 전체에 스냅샷을 미리 적용하는 lookahead 를 제거한다.
|
|
이전 스냅샷 대비 **신규 편입** 종목만 ``event_time`` 분 + lag 이후 분봉부터 포함.
|
|
(예: 09:42:25 편입 + lag=1 → 09:43 분봉부터, 09:42 분봉에서는 제외)
|
|
|
|
Args:
|
|
strategy_id: 'SCALP' | 'SHORT' | 'BREAKOUT' | 'MOMENTUM'
|
|
start_ymd: 시작일 YYYYMMDD
|
|
end_ymd: 종료일 YYYYMMDD
|
|
strict: 종목별 첫 event_time 기준 편입 지연 적용
|
|
strict_lag_minutes: 첫 편입 분 이후 추가 대기 분 (기본 1)
|
|
exit_debounce_sec: 짧은 EXIT→재편입 무시(초). 0=OFF.
|
|
history_source: ``kiwoom`` | ``ls``
|
|
|
|
Returns:
|
|
``{candle_time(YYYYMMDDHHMM): [code, ...]}`` —
|
|
해당 1분 캔들 종료 시점에 유효했던 유니버스 코드 리스트.
|
|
스냅샷이 하나도 없으면 빈 dict.
|
|
"""
|
|
self._ensure_history_columns()
|
|
|
|
from kis_trader.backtest.universe_history_source import (
|
|
apply_ls_session_filter_to_start,
|
|
normalize_universe_history_source,
|
|
)
|
|
|
|
history_source = normalize_universe_history_source(history_source)
|
|
|
|
# 조회 범위: 전일 마지막 스냅샷도 포함하기 위해 시작일 00:00:00 이전 1건은
|
|
# 엔진 쪽에서 "이전에 유효했던 유니버스" 로 물려받는 게 자연스럽다.
|
|
# 단순화를 위해 start_ymd 00:00:00 ~ end_ymd 23:59:59 범위로 쿼리.
|
|
# LS + SESSION_ONLY: 장전 sticky 스냅 제외 → 당일 09:00 부터.
|
|
start_time = (
|
|
f"{start_ymd[:4]}-{start_ymd[4:6]}-{start_ymd[6:8]} 00:00:00"
|
|
)
|
|
start_time = apply_ls_session_filter_to_start(
|
|
start_time, source=history_source,
|
|
)
|
|
end_time = (
|
|
f"{end_ymd[:4]}-{end_ymd[4:6]}-{end_ymd[6:8]} 23:59:59.999999"
|
|
)
|
|
|
|
events = self.iter_universe_events(
|
|
strategy_id=strategy_id,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
preserve_insert_order=True,
|
|
history_source=history_source,
|
|
)
|
|
if not events:
|
|
return {}
|
|
|
|
from kis_trader.backtest.universe_timeline import _event_time_to_key
|
|
|
|
# 각 스냅샷의 event_time 을 'YYYYMMDDHHMMSSffffff' 로 정규화.
|
|
# event_time 문자열은 이미 시간순 정렬됨 (iter_universe_events).
|
|
normalized: List[tuple] = []
|
|
for ev in events:
|
|
et_key = _event_time_to_key(str(ev.get("event_time") or ""))
|
|
if not et_key:
|
|
continue
|
|
codes = [it["code"] for it in ev.get("items", []) if it.get("code")]
|
|
normalized.append((et_key, codes))
|
|
|
|
if not normalized:
|
|
return {}
|
|
|
|
if exit_debounce_sec > 0:
|
|
try:
|
|
from kis_trader.backtest.momentum_universe_timeline import (
|
|
debounce_universe_snapshots,
|
|
)
|
|
normalized = debounce_universe_snapshots(
|
|
normalized, int(exit_debounce_sec),
|
|
)
|
|
except Exception as e:
|
|
logger.warning("유니버스 EXIT 디바운스 실패(원본 사용): %s", e)
|
|
|
|
out: Dict[str, List[str]] = {}
|
|
# candle_time 은 분단위 (YYYYMMDDHHMM). 각 캔들의 "종가 형성 시점" 은
|
|
# 그 분의 59초로 본다 → candle_end_key = candle_time + "59".
|
|
# 스냅샷 포인터를 전진하며 O(N+M) 으로 매칭.
|
|
ev_idx = 0
|
|
n_ev = len(normalized)
|
|
active_codes: List[str] = []
|
|
|
|
# 첫 이벤트 시각을 분단위로 내림 → 그 이전 캔들은 유니버스 없음
|
|
first_ev_key = normalized[0][0]
|
|
strict_lag = max(0, int(strict_lag_minutes))
|
|
|
|
prev_codes_set: set = set()
|
|
code_avail: Dict[str, str] = {}
|
|
if strict:
|
|
try:
|
|
prev_items = self.get_universe_at(
|
|
strategy_id=strategy_id, at_time=start_time,
|
|
history_source=history_source,
|
|
)
|
|
prev_codes_set = {
|
|
str(it["code"]) for it in prev_items if it.get("code")
|
|
}
|
|
code_avail = {c: "000000000000" for c in prev_codes_set}
|
|
except Exception as e:
|
|
logger.warning("strict 유니버스 초기 스냅샷 실패: %s", e)
|
|
|
|
# 날짜별 장중 시간대 (09:00~15:30) 1분 단위로 캔들 시각 생성.
|
|
# DB 에 없는 시각도 dict 에 key 가 생기면 메모리 낭비이므로, 엔진이 실제
|
|
# 요청하는 캔들 시각에 대해서만 값을 반환하도록 **lazy** 하게 넘기고 싶지만,
|
|
# 엔진 쪽은 dict lookup 만 하므로 여기서 미리 만들어 주는 편이 단순.
|
|
from datetime import datetime, timedelta
|
|
d0 = datetime.strptime(start_ymd, "%Y%m%d").date()
|
|
d1 = datetime.strptime(end_ymd, "%Y%m%d").date()
|
|
day = d0
|
|
while day <= d1:
|
|
# 장중 대략 범위: 08:00 ~ 16:00 (pre/post 여유)
|
|
t = datetime.combine(day, datetime.min.time()).replace(hour=8)
|
|
t_end = datetime.combine(day, datetime.min.time()).replace(hour=16)
|
|
while t <= t_end:
|
|
candle_time = t.strftime("%Y%m%d%H%M")
|
|
candle_end_key = candle_time + "59999999" # YYYYMMDDHHMMSS + 999999
|
|
# 그 이전 또는 같은 시각의 가장 최근 event_time 까지 포인터 전진
|
|
while ev_idx < n_ev and normalized[ev_idx][0] <= candle_end_key:
|
|
et_key, codes = normalized[ev_idx]
|
|
if strict:
|
|
cur_set = {str(c) for c in codes if c}
|
|
et_slot = et_key[:12]
|
|
avail_slot = (
|
|
self._add_minutes_to_candle_slot(et_slot, strict_lag)
|
|
if strict_lag > 0 else et_slot
|
|
)
|
|
for c in cur_set - prev_codes_set:
|
|
code_avail[c] = avail_slot
|
|
for c in prev_codes_set - cur_set:
|
|
code_avail.pop(c, None)
|
|
prev_codes_set = cur_set
|
|
active_codes = codes
|
|
ev_idx += 1
|
|
if active_codes and candle_end_key >= first_ev_key:
|
|
if strict:
|
|
filtered = [
|
|
c for c in active_codes
|
|
if candle_time >= code_avail.get(str(c), "000000000000")
|
|
]
|
|
if filtered:
|
|
out[candle_time] = filtered
|
|
else:
|
|
out[candle_time] = active_codes
|
|
t += timedelta(minutes=1)
|
|
day += timedelta(days=1)
|
|
|
|
return out
|
|
|
|
# ------------------------------------------------------------------
|
|
# TradeDB 위임 (원본 API 그대로 사용)
|
|
# ------------------------------------------------------------------
|
|
def __getattr__(self, name):
|
|
"""TradeDBExt 에 없는 속성/메서드는 self.raw 에 위임."""
|
|
return getattr(self.raw, name)
|