Files
QuantEngineByItz/tools/evaluate_execution_slippage_v1.py
T
kjh2064 449721433b WBS-7.6: 슬리피지 실측 캡처 스캐폴딩
spec/55_execution_simulator_contract.yaml의 5bps 슬리피지 가정치를
검증할 실측 캡처 경로가 없었다. 주문 실행은 여전히 사람이 HTS에서
직접 한다(governance/rules/06 준수, API로 체결을 가져오지 않음) —
실행 후 사람이 의도가/실제체결가를 수동 기록하면 SQLite에 누적되고,
5건 미만이면 항상 DATA_GATED를 정직하게 반환한다(추정 금지).
2026-06-21 20:09:16 +09:00

76 lines
2.6 KiB
Python

#!/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())