kjh2064/QuantEngineByItz
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WBS-7.6: 슬리피지 실측 캡처 스캐폴딩

Browse Source 
spec/55_execution_simulator_contract.yaml의 5bps 슬리피지 가정치를
검증할 실측 캡처 경로가 없었다. 주문 실행은 여전히 사람이 HTS에서
직접 한다(governance/rules/06 준수, API로 체결을 가져오지 않음) —
실행 후 사람이 의도가/실제체결가를 수동 기록하면 SQLite에 누적되고,
5건 미만이면 항상 DATA_GATED를 정직하게 반환한다(추정 금지).
This commit is contained in:
김재현
2026-06-21 20:09:16 +09:00
parent 5166750b53
commit 449721433b
3 changed files with 309 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/quant_engine/execution_slippage_store_v1.py
+144
View File
@@ -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 값을 실측 평균으로 갱신 검토"
),
}
tests/unit/test_execution_slippage_store_v1.py
+90
View File
@@ -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",
)
tools/evaluate_execution_slippage_v1.py
+75
View File
@@ -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())