shuim
a264561576
- 스윙 체결가 확인 전 executed_price NULL + fill_sweep, 진입가 resolve(ActiveTrade↔Portfolio), 0체결/부분체결 매수 장부 원복 - DT 고아 감지(근거 한정)·0체결 매도 복원·stale 점검 확대, 시간손절 -0.3% 원복 - 폴백 손절 ATR 캐시, 거래량부족-차단 신호 매수 차단, 현금 5% 유보 재적용, MDD peak/reset 수정 - KIS 서킷브레이커 청산 우회·경보, db_worker 재큐잉 순서, heartbeat·health-watch 사이드카 - 실시간 WS 재구독·41건 우선순위·다건 프레임 분할, 차트 샘플 제거·당일봉 합성, WS 서브프로토콜 인증 - 프론트: 미관리 그룹·확인 모달·차트 실시간 반영·예수금 표시 정정 - 전문가 에이전트 6종 추가(.claude/agents) - 유닛 스위트 2667 passed (소켓 차단 하), tsc OK Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y83Za98XsNw9zuA4FTUytN
78 lines
3.0 KiB
Python
78 lines
3.0 KiB
Python
"""실행 품질(슬리피지 bp·체결률) 집계 — 순수 계산 모듈 (settings/DB/네트워크 참조 없음).
|
||
|
||
TradeHistory.slippage 는 (체결가 − 주문가)/주문가×100 [%], side 무관 원시값이다
|
||
(trade_executor.slippage_cost_bp_v2 주석). BUY/SELL 을 그대로 평균하면 상쇄되므로
|
||
side 보정(불리 = +) bp 로 바꿔 집계한다. 보정식은 slippage_cost_bp_v2 와 동일하며
|
||
(trade_executor 는 import 비용이 커서 복제) 동일성은 테스트로 고정한다.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from collections import defaultdict
|
||
from typing import Any, Iterable, Mapping
|
||
|
||
LOW_SAMPLE_THRESHOLD = 30 # n < 30 이면 low_sample
|
||
|
||
|
||
def slippage_cost_bp_v3(order_type: str, slippage_pct) -> float | None:
|
||
"""원시 슬리피지(%) → side 보정 비용(bp, 불리 = +). 해석 불가면 None."""
|
||
try:
|
||
val = float(slippage_pct)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if not math.isfinite(val):
|
||
return None
|
||
side = str(order_type or "").upper()
|
||
if side == "BUY":
|
||
return val * 100.0
|
||
if side == "SELL":
|
||
return -val * 100.0
|
||
return None
|
||
|
||
|
||
def _hour_of(row: Mapping[str, Any]):
|
||
ts = row.get("executed_at") or row.get("created_at")
|
||
return getattr(ts, "hour", None)
|
||
|
||
|
||
def _bucket_stats(rows: list, min_n: int) -> dict:
|
||
costs: list[float] = []
|
||
for r in rows:
|
||
c = slippage_cost_bp_v3(r.get("order_type"), r.get("slippage"))
|
||
if c is not None:
|
||
costs.append(c)
|
||
n_all = len(rows)
|
||
n = len(costs)
|
||
verified = [r.get("fill_verified") for r in rows if r.get("fill_verified") is not None]
|
||
filled = sum(1 for r in rows if (r.get("executed_quantity") or 0) > 0)
|
||
return {
|
||
"n_orders": n_all,
|
||
"n_slippage": n,
|
||
"fill_rate": round(filled / n_all, 4) if n_all else None,
|
||
"mean_cost_bp": round(sum(costs) / n, 3) if n else None,
|
||
"mean_abs_bp": round(sum(abs(c) for c in costs) / n, 3) if n else None,
|
||
"fill_verified_ratio": round(sum(1 for v in verified if v) / len(verified), 4) if verified else None,
|
||
"low_sample": n < min_n,
|
||
}
|
||
|
||
|
||
def aggregate_execution_quality_v1(
|
||
rows: Iterable[Mapping[str, Any]], min_n: int = LOW_SAMPLE_THRESHOLD
|
||
) -> dict:
|
||
"""행(dict: order_type, slippage, strategy_type, executed_quantity, executed_at, fill_verified)을
|
||
전체/전략별/시간대별로 집계한다. slippage NULL 행은 n_orders 에만 포함된다."""
|
||
rows = list(rows)
|
||
by_strategy: dict[str, list] = defaultdict(list)
|
||
by_hour: dict[int, list] = defaultdict(list)
|
||
for r in rows:
|
||
by_strategy[str(r.get("strategy_type") or "unknown")].append(r)
|
||
h = _hour_of(r)
|
||
if h is not None:
|
||
by_hour[int(h)].append(r)
|
||
return {
|
||
"overall": _bucket_stats(rows, min_n),
|
||
"by_strategy": {k: _bucket_stats(v, min_n) for k, v in sorted(by_strategy.items())},
|
||
"by_hour": {h: _bucket_stats(v, min_n) for h, v in sorted(by_hour.items())},
|
||
"min_n": min_n,
|
||
}
|