diff --git a/src/quant_engine/execution_slippage_store_v1.py b/src/quant_engine/execution_slippage_store_v1.py new file mode 100644 index 0000000..07fe088 --- /dev/null +++ b/src/quant_engine/execution_slippage_store_v1.py @@ -0,0 +1,144 @@ +"""WBS-7.6(2026-06-21) — 실거래 슬리피지 실측 캡처 스캐폴딩. + +spec/55_execution_simulator_contract.yaml의 slippage_model(bps=5)은 이론치이며 +"추후 실측 데이터로 보정 예정"이라는 메모만 있고 실제 캡처 경로가 없었다. 이 모듈은 +주문은 사람이 HTS에서 직접 실행한다는 governance/rules/06 원칙을 그대로 유지한 채 +(API로 체결을 가져오지 않는다), 실행 후 사람이 수동으로 기록한 실제 체결가를 +누적해 가정치(5bps)와 비교할 수 있게 한다. 5건 미만이면 항상 DATA_GATED로 보고한다 +— 추정 금지 원칙(spec/00_execution_contract.yaml)을 따른다. 표준 라이브러리 +sqlite3만 사용한다. +""" +from __future__ import annotations + +import sqlite3 +from pathlib import Path +from typing import Any + +from src.quant_engine.storage_backend_v1 import StoreSpec, default_sqlite_store_path, normalize_store_spec + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS realized_slippage_samples ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ticker TEXT NOT NULL, + side TEXT NOT NULL CHECK (side IN ('BUY', 'SELL')), + intended_price REAL NOT NULL, + actual_fill_price REAL NOT NULL, + slippage_bps_actual REAL NOT NULL, + recorded_at TEXT NOT NULL, + note TEXT, + inserted_at TEXT DEFAULT (datetime('now')) +); +""" + +ASSUMED_SLIPPAGE_BPS = 5.0 +MIN_SAMPLE_FOR_COMPARISON = 5 + + +def default_execution_slippage_store_path(root: Path) -> Path: + return default_sqlite_store_path(root, "execution_slippage/execution_slippage.db") + + +def resolve_store_path(spec: StoreSpec, root: Path) -> Path: + backend, location = normalize_store_spec( + spec, root, default_sqlite_name="execution_slippage/execution_slippage.db" + ) + if backend != "sqlite": + raise ValueError("execution_slippage_store_v1 currently executes on sqlite only.") + return Path(location) + + +def init_db(db_path: Path) -> None: + db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(db_path) + try: + conn.executescript(SCHEMA) + conn.commit() + finally: + conn.close() + + +def _compute_slippage_bps(intended_price: float, actual_fill_price: float, side: str) -> float: + """체결가가 의도가(지정가)보다 불리한 방향으로 움직인 만큼을 양수 bps로 환산한다. + + BUY: 실제 체결가가 의도가보다 높으면(더 비싸게 샀으면) 양수 슬리피지. + SELL: 실제 체결가가 의도가보다 낮으면(더 싸게 팔았으면) 양수 슬리피지. + """ + if intended_price <= 0: + raise ValueError("intended_price must be > 0") + direction = 1 if side.upper() == "BUY" else -1 + return direction * (actual_fill_price - intended_price) / intended_price * 10_000.0 + + +def insert_realized_slippage_sample( + db_path: Path, + *, + ticker: str, + side: str, + intended_price: float, + actual_fill_price: float, + recorded_at: str, + note: str | None = None, +) -> dict[str, Any]: + init_db(db_path) + slippage_bps = _compute_slippage_bps(intended_price, actual_fill_price, side) + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO realized_slippage_samples " + "(ticker, side, intended_price, actual_fill_price, slippage_bps_actual, recorded_at, note) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (ticker, side.upper(), intended_price, actual_fill_price, slippage_bps, recorded_at, note), + ) + conn.commit() + finally: + conn.close() + return { + "ticker": ticker, + "side": side.upper(), + "intended_price": intended_price, + "actual_fill_price": actual_fill_price, + "slippage_bps_actual": round(slippage_bps, 4), + "recorded_at": recorded_at, + } + + +def fetch_all_samples(db_path: Path) -> list[dict[str, Any]]: + if not db_path.exists(): + return [] + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + try: + rows = conn.execute( + "SELECT ticker, side, intended_price, actual_fill_price, slippage_bps_actual, recorded_at, note " + "FROM realized_slippage_samples ORDER BY recorded_at ASC" + ).fetchall() + return [dict(row) for row in rows] + finally: + conn.close() + + +def build_slippage_comparison_report(db_path: Path) -> dict[str, Any]: + """WBS-7.6 성공 하네스 — 5건 미만이면 DATA_GATED를 정직하게 반환한다(추정 금지).""" + samples = fetch_all_samples(db_path) + sample_n = len(samples) + if sample_n < MIN_SAMPLE_FOR_COMPARISON: + return { + "status": "DATA_GATED", + "sample_n": sample_n, + "min_required": MIN_SAMPLE_FOR_COMPARISON, + "assumed_slippage_bps": ASSUMED_SLIPPAGE_BPS, + "actual_mean_slippage_bps": None, + "note": f"실측 표본 {sample_n}/{MIN_SAMPLE_FOR_COMPARISON}건 — 비교 불가, 가정치(5bps) 유지", + } + actual_mean = sum(s["slippage_bps_actual"] for s in samples) / sample_n + gap = abs(actual_mean - ASSUMED_SLIPPAGE_BPS) + return { + "status": "OK", + "sample_n": sample_n, + "assumed_slippage_bps": ASSUMED_SLIPPAGE_BPS, + "actual_mean_slippage_bps": round(actual_mean, 4), + "gap_bps": round(gap, 4), + "recommendation": ( + "가정치(5bps) 유지" if gap <= 3.0 else "spec/55_execution_simulator_contract.yaml의 bps 값을 실측 평균으로 갱신 검토" + ), + } diff --git a/tests/unit/test_execution_slippage_store_v1.py b/tests/unit/test_execution_slippage_store_v1.py new file mode 100644 index 0000000..26b08f5 --- /dev/null +++ b/tests/unit/test_execution_slippage_store_v1.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from src.quant_engine.execution_slippage_store_v1 import ( + ASSUMED_SLIPPAGE_BPS, + MIN_SAMPLE_FOR_COMPARISON, + build_slippage_comparison_report, + fetch_all_samples, + insert_realized_slippage_sample, +) + + +def test_report_is_data_gated_below_minimum_sample(tmp_path): + db_path = tmp_path / "execution_slippage.db" + report = build_slippage_comparison_report(db_path) + assert report["status"] == "DATA_GATED" + assert report["sample_n"] == 0 + assert report["actual_mean_slippage_bps"] is None + + +def test_buy_slippage_sign_is_positive_when_filled_worse(tmp_path): + db_path = tmp_path / "execution_slippage.db" + result = insert_realized_slippage_sample( + db_path, + ticker="005930", + side="buy", + intended_price=70000, + actual_fill_price=70070, + recorded_at="2026-06-21", + ) + # BUY 체결가가 의도가보다 비싸게 체결됐으면 양수 슬리피지(불리) + assert result["slippage_bps_actual"] > 0 + assert abs(result["slippage_bps_actual"] - 10.0) < 1e-6 # 70/70000 = 10bps + + +def test_sell_slippage_sign_is_positive_when_filled_worse(tmp_path): + db_path = tmp_path / "execution_slippage.db" + result = insert_realized_slippage_sample( + db_path, + ticker="000660", + side="SELL", + intended_price=200000, + actual_fill_price=199900, + recorded_at="2026-06-21", + ) + # SELL 체결가가 의도가보다 싸게 체결됐으면 양수 슬리피지(불리) + assert result["slippage_bps_actual"] > 0 + + +def test_report_compares_against_assumed_bps_once_min_sample_reached(tmp_path): + db_path = tmp_path / "execution_slippage.db" + for i in range(MIN_SAMPLE_FOR_COMPARISON): + insert_realized_slippage_sample( + db_path, + ticker="005930", + side="BUY", + intended_price=70000, + actual_fill_price=70070, # 항상 10bps 불리하게 체결 + recorded_at=f"2026-06-{21 + i}", + ) + + samples = fetch_all_samples(db_path) + assert len(samples) == MIN_SAMPLE_FOR_COMPARISON + + report = build_slippage_comparison_report(db_path) + assert report["status"] == "OK" + assert abs(report["actual_mean_slippage_bps"] - 10.0) < 1e-6 + assert abs(report["gap_bps"] - abs(10.0 - ASSUMED_SLIPPAGE_BPS)) < 1e-6 + assert report["recommendation"] + + +def test_intended_price_must_be_positive(tmp_path): + db_path = tmp_path / "execution_slippage.db" + import pytest + + with pytest.raises(ValueError): + insert_realized_slippage_sample( + db_path, + ticker="005930", + side="BUY", + intended_price=0, + actual_fill_price=100, + recorded_at="2026-06-21", + ) diff --git a/tools/evaluate_execution_slippage_v1.py b/tools/evaluate_execution_slippage_v1.py new file mode 100644 index 0000000..4890755 --- /dev/null +++ b/tools/evaluate_execution_slippage_v1.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +"""WBS-7.6(2026-06-21) — 실거래 슬리피지 실측 캡처/비교 CLI. + +사용법: + 실측 1건 기록(주문 실행은 여전히 사람이 HTS에서 수동 실행 — 이 도구는 API로 + 체결을 가져오지 않는다. governance/rules/06_no_direct_api_trading.yaml 준수): + python tools/evaluate_execution_slippage_v1.py record --ticker 005930 --side BUY \ + --intended-price 71000 --actual-price 71050 --recorded-at 2026-06-21 + + 누적 표본과 가정치(5bps) 비교 리포트: + python tools/evaluate_execution_slippage_v1.py report +""" +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"): + sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1) + +from src.quant_engine.execution_slippage_store_v1 import ( + build_slippage_comparison_report, + default_execution_slippage_store_path, + insert_realized_slippage_sample, +) + +OUTPUT = ROOT / "Temp" / "execution_slippage_report_v1.json" + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--db", type=Path, default=None) + sub = parser.add_subparsers(dest="command", required=True) + + record = sub.add_parser("record") + record.add_argument("--ticker", required=True) + record.add_argument("--side", required=True, choices=["BUY", "SELL", "buy", "sell"]) + record.add_argument("--intended-price", type=float, required=True) + record.add_argument("--actual-price", type=float, required=True) + record.add_argument("--recorded-at", required=True) + record.add_argument("--note", default=None) + + sub.add_parser("report") + + args = parser.parse_args() + db_path = args.db or default_execution_slippage_store_path(ROOT) + + if args.command == "record": + result = insert_realized_slippage_sample( + db_path, + ticker=args.ticker, + side=args.side, + intended_price=args.intended_price, + actual_fill_price=args.actual_price, + recorded_at=args.recorded_at, + note=args.note, + ) + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 + + report = build_slippage_comparison_report(db_path) + OUTPUT.parent.mkdir(parents=True, exist_ok=True) + OUTPUT.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + print(json.dumps(report, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())