kjh2064
27730704ae
Snapshot Admin Web Validation / validate-snapshot-admin-smoke (push) Has been cancelled
Snapshot Admin Web Validation / validate-snapshot-admin-full (push) Has been cancelled
Quant Engine CI/CD Pipeline / validate-core (pull_request) Has been cancelled
Quant Engine CI/CD Pipeline / validate-ui-and-storage (pull_request) Has been cancelled
WBS-9.3 - NULL Policy CI Gate / NULL Policy Validation (pull_request) Has been cancelled
118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import subprocess
|
|
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",
|
|
)
|
|
|
|
|
|
def test_cli_report_is_data_gated_below_minimum_sample(tmp_path):
|
|
db_path = tmp_path / "execution_slippage.db"
|
|
report_path = ROOT / "Temp" / "execution_slippage_report_v1.json"
|
|
if report_path.exists():
|
|
report_path.unlink()
|
|
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"tools/evaluate_execution_slippage_v1.py",
|
|
"--db",
|
|
str(db_path),
|
|
"report",
|
|
],
|
|
cwd=ROOT,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assert "DATA_GATED" in result.stdout
|
|
payload = report_path.read_text(encoding="utf-8")
|
|
assert '"status": "DATA_GATED"' in payload
|