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
412 lines
18 KiB
Python
412 lines
18 KiB
Python
"""데이트레이딩 실제 체결가 확인 (백그라운드).
|
||
|
||
[2026-10-01] DT 는 시장가 주문 후 신호 시점 가격(1분봉 종가)을 체결가로 기록했다.
|
||
실제 체결가와의 차이(슬리피지)가 원장·일일 손익·성과 통계에서 사라진다.
|
||
|
||
설계:
|
||
- 주문 응답(KIS order-cash)에는 체결가가 없다 → 당일 주문체결 조회
|
||
(`korea_investment_api.get_todays_orders`, inquire-daily-ccld)에서 주문번호(ODNO)로 찾는다.
|
||
`get_order_execution()` 은 쓰지 않는다 — 국내 경로가 같은 TR(TTTC8001R)을
|
||
`/trading/inquire-ccnl` 로 호출해(같은 TR 의 다른 메서드 `get_todays_orders` 는
|
||
`/trading/inquire-daily-ccld`) 경로가 어긋나 있을 가능성이 높고, 실패를 debug 로 삼켜
|
||
체결 0 과 구분되지 않는다(운영 확인 필요 — 보고 사항).
|
||
- **실시간 루프를 막지 않는다**: `schedule()` 이 asyncio 태스크로 띄우고 즉시 반환.
|
||
전체를 `DAY_TRADE_FILL_CHECK_TIMEOUT_SEC` 로 감싼다 — KIS 토큰 rate-limit 의
|
||
`asyncio.sleep(≈70초)` 에 걸려도 상한에서 끊긴다.
|
||
- DB I/O 없음. 반영은 호출부 콜백이 인메모리 + db_worker 큐로 한다(§3).
|
||
- 확인 실패 시 None → 호출부는 신호가 기록을 유지하고 warning 을 남긴다.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
from typing import Awaitable, Callable, Optional
|
||
|
||
from app.core.config import settings
|
||
from app.services.korea_investment_api import korea_investment_api
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# create_task 결과의 강한 참조 — 참조가 없으면 GC 로 태스크가 중간에 사라질 수 있다.
|
||
_pending_tasks: "set[asyncio.Task]" = set()
|
||
|
||
|
||
def _norm_odno(value) -> str:
|
||
"""KIS 주문번호 정규화 — 응답(ODNO)은 0 패딩, 조회(odno)는 경우에 따라 다르다."""
|
||
s = str(value or "").strip()
|
||
return s.lstrip("0") or ("0" if s else "")
|
||
|
||
|
||
def parse_fill_row(rows: list, order_id: str) -> Optional[tuple[int, float]]:
|
||
"""inquire-daily-ccld output1 에서 주문번호가 일치하는 행의 (체결수량, 평균체결가).
|
||
|
||
체결수량 0 이거나 평균가를 구할 수 없으면 None.
|
||
"""
|
||
target = _norm_odno(order_id)
|
||
if not target:
|
||
return None
|
||
for row in rows or []:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
if _norm_odno(row.get("odno")) != target:
|
||
continue
|
||
try:
|
||
qty = int(float(row.get("tot_ccld_qty", 0) or 0))
|
||
avg = float(row.get("avg_prvs", 0) or 0)
|
||
if avg <= 0 and qty > 0:
|
||
amt = float(row.get("tot_ccld_amt", 0) or 0)
|
||
avg = amt / qty if amt > 0 else 0.0
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if qty > 0 and avg > 0:
|
||
return qty, avg
|
||
return None
|
||
return None
|
||
|
||
|
||
async def _poll_fill(stock_code: str, order_id: str) -> Optional[tuple[int, float]]:
|
||
"""체결 조회 폴링 (타임아웃은 호출부 wait_for 가 건다)."""
|
||
attempts = max(1, int(settings.DAY_TRADE_FILL_CHECK_ATTEMPTS))
|
||
interval = max(0.0, float(settings.DAY_TRADE_FILL_CHECK_INTERVAL_SEC))
|
||
await asyncio.sleep(max(0.0, float(settings.DAY_TRADE_FILL_CHECK_DELAY_SEC)))
|
||
for i in range(attempts):
|
||
rows = await korea_investment_api.get_todays_orders(stock_code)
|
||
fill = parse_fill_row(rows, order_id)
|
||
if fill is not None:
|
||
return fill
|
||
if i < attempts - 1:
|
||
await asyncio.sleep(interval)
|
||
return None
|
||
|
||
|
||
async def fetch_fill(stock_code: str, order_id: str, side: str) -> Optional[tuple[int, float]]:
|
||
"""주문번호의 실제 (체결수량, 평균체결가). 실패·시간초과·미체결이면 None + warning."""
|
||
if not order_id:
|
||
logger.warning(f"[DTFill] {stock_code} {side}: 주문번호 없음 — 신호가 기록 유지")
|
||
return None
|
||
timeout = float(settings.DAY_TRADE_FILL_CHECK_TIMEOUT_SEC)
|
||
try:
|
||
fill = await asyncio.wait_for(_poll_fill(stock_code, order_id), timeout=timeout)
|
||
except asyncio.TimeoutError:
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 체결 조회 시간초과({timeout:.0f}초) "
|
||
f"order_id={order_id} — 신호가 기록 유지"
|
||
)
|
||
return None
|
||
except Exception as e:
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 체결 조회 실패 order_id={order_id}: "
|
||
f"{type(e).__name__}: {e} — 신호가 기록 유지"
|
||
)
|
||
return None
|
||
if fill is None:
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 체결 미확인 "
|
||
f"({int(settings.DAY_TRADE_FILL_CHECK_ATTEMPTS)}회 조회) order_id={order_id} — 신호가 기록 유지"
|
||
)
|
||
return fill
|
||
|
||
|
||
def schedule(
|
||
stock_code: str,
|
||
order_id: str,
|
||
side: str,
|
||
on_fill: Callable[[int, float], Awaitable[None]],
|
||
) -> Optional[asyncio.Task]:
|
||
"""체결 확인을 백그라운드로 띄우고 즉시 반환. 확인되면 `on_fill(qty, avg_price)` 호출.
|
||
|
||
`DAY_TRADE_FILL_CHECK_ENABLED=False` 면 아무것도 하지 않는다(None).
|
||
"""
|
||
if not settings.DAY_TRADE_FILL_CHECK_ENABLED:
|
||
return None
|
||
|
||
async def _run() -> None:
|
||
fill = await fetch_fill(stock_code, order_id, side)
|
||
if fill is None:
|
||
return
|
||
try:
|
||
await on_fill(*fill)
|
||
except Exception as e:
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 체결가 반영 실패 order_id={order_id}: "
|
||
f"{type(e).__name__}: {e}",
|
||
exc_info=True,
|
||
)
|
||
|
||
try:
|
||
task = asyncio.get_running_loop().create_task(_run())
|
||
except RuntimeError as e:
|
||
logger.warning(f"[DTFill] {stock_code} {side} 체결 확인 예약 실패: {e}")
|
||
return None
|
||
_pending_tasks.add(task)
|
||
task.add_done_callback(_pending_tasks.discard)
|
||
return task
|
||
|
||
|
||
# ── [M7 2026-10-02] 부분체결·미체결 판정용 v2 ─────────────────────────────────
|
||
# 기존 parse_fill_row/fetch_fill/schedule 은 **체결수량>0 인 첫 조회 결과**만 돌려줘
|
||
# (a) 체결 0 과 조회 실패를 구분하지 못하고 (b) 아직 체결 중인 부분체결도 확정으로 본다.
|
||
# 매수 쪽 수량 보정은 "주문이 종결됐다"는 근거가 있을 때만 해야 한다 — 미종결 부분체결을
|
||
# 수량으로 확정하면 나머지가 체결된 뒤 실보유 > 인메모리가 되어 EOD 청산 뒤 잔량이 남는다.
|
||
# 종결 판정: 체결수량 ≥ 주문수량, 또는 응답의 잔여수량(rmn_qty)=0(취소·거부 포함).
|
||
# rmn_qty 가 없는 응답에서 체결<주문이면 '미확정' 으로 남긴다(보정하지 않음, warning).
|
||
|
||
def parse_fill_row_v2(rows: list, order_id: str) -> Optional[dict]:
|
||
"""주문번호가 일치하는 행 → {filled_qty, avg_price, ord_qty, rmn_qty}. 없거나 파싱 불가면 None.
|
||
|
||
체결수량 0 행도 돌려준다(미체결·취소 판정용). ord_qty/rmn_qty 는 응답에 없으면 None.
|
||
"""
|
||
target = _norm_odno(order_id)
|
||
if not target:
|
||
return None
|
||
for row in rows or []:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
if _norm_odno(row.get("odno")) != target:
|
||
continue
|
||
try:
|
||
qty = int(float(row.get("tot_ccld_qty", 0) or 0))
|
||
avg = float(row.get("avg_prvs", 0) or 0)
|
||
if avg <= 0 and qty > 0:
|
||
amt = float(row.get("tot_ccld_amt", 0) or 0)
|
||
avg = amt / qty if amt > 0 else 0.0
|
||
_oq = row.get("ord_qty")
|
||
ord_qty = int(float(_oq)) if _oq not in (None, "") else None
|
||
_rq = row.get("rmn_qty")
|
||
rmn_qty = int(float(_rq)) if _rq not in (None, "") else None
|
||
_jq = row.get("rjct_qty")
|
||
rjct_qty = int(float(_jq)) if _jq not in (None, "") else None
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if qty > 0 and avg <= 0:
|
||
return None
|
||
cncl_yn = str(row.get("cncl_yn", "") or "").upper() or None
|
||
return {"filled_qty": max(0, qty), "avg_price": avg, "ord_qty": ord_qty,
|
||
"rmn_qty": rmn_qty, "cncl_yn": cncl_yn, "rjct_qty": rjct_qty}
|
||
return None
|
||
|
||
|
||
def is_terminal_fill(info: dict, ordered_qty: int) -> bool:
|
||
"""주문 종결 여부 — 전량 체결 / 잔여수량 0 / 취소(cncl_yn=Y) / 거부수량>0.
|
||
|
||
[Low 2026-10-02] 거부·취소 행은 응답에 따라 rmn_qty 가 비거나 0 이 아닐 수 있어
|
||
cncl_yn·rjct_qty 도 본다(체결 0주 종결 판정 누락 방지).
|
||
"""
|
||
filled = int(info.get("filled_qty", 0) or 0)
|
||
if ordered_qty > 0 and filled >= ordered_qty:
|
||
return True
|
||
if str(info.get("cncl_yn") or "").upper() == "Y":
|
||
return True
|
||
rj = info.get("rjct_qty")
|
||
rmn = info.get("rmn_qty")
|
||
if rj is not None and int(rj) > 0 and (rmn is None or int(rmn) <= 0):
|
||
return True
|
||
return rmn is not None and int(rmn) <= 0
|
||
|
||
|
||
async def _poll_fill_v2(stock_code: str, order_id: str, ordered_qty: int) -> Optional[dict]:
|
||
"""종결될 때까지(또는 시도 소진까지) 폴링. 마지막으로 본 행 정보를 돌려준다."""
|
||
return await _poll_fill_v2_impl(stock_code, order_id, ordered_qty, None)
|
||
|
||
|
||
async def _poll_fill_v2_impl(
|
||
stock_code: str, order_id: str, ordered_qty: int, seen: Optional[dict],
|
||
) -> Optional[dict]:
|
||
"""`_poll_fill_v2` 본체. `seen` 이 주어지면 마지막으로 본 행을 `seen["last"]` 에도 남긴다.
|
||
|
||
[R1 2026-10-09] fetch_fill_v2 의 wait_for 가 시간초과로 이 코루틴을 취소하면 반환값이
|
||
사라져, 이미 본 부분체결 행까지 '행 미확인'과 같은 None 이 됐다. seen 으로 그 행을 살린다.
|
||
"""
|
||
attempts = max(1, int(settings.DAY_TRADE_FILL_CHECK_ATTEMPTS))
|
||
interval = max(0.0, float(settings.DAY_TRADE_FILL_CHECK_INTERVAL_SEC))
|
||
await asyncio.sleep(max(0.0, float(settings.DAY_TRADE_FILL_CHECK_DELAY_SEC)))
|
||
last: Optional[dict] = None
|
||
for i in range(attempts):
|
||
rows = await korea_investment_api.get_todays_orders(stock_code)
|
||
info = parse_fill_row_v2(rows, order_id)
|
||
if info is not None:
|
||
last = info
|
||
if seen is not None:
|
||
seen["last"] = info
|
||
if is_terminal_fill(info, ordered_qty):
|
||
return info
|
||
if i < attempts - 1:
|
||
await asyncio.sleep(interval)
|
||
return last
|
||
|
||
|
||
async def fetch_fill_v2(
|
||
stock_code: str, order_id: str, side: str, ordered_qty: int,
|
||
) -> Optional[dict]:
|
||
"""[M7] 주문 체결 상태. 조회 실패·시간초과·행 없음이면 None + warning.
|
||
|
||
반환 dict 에 `terminal`(bool) 이 추가된다.
|
||
"""
|
||
if not order_id:
|
||
logger.warning(f"[DTFill] {stock_code} {side}: 주문번호 없음 — 신호가 기록 유지")
|
||
return None
|
||
timeout = float(settings.DAY_TRADE_FILL_CHECK_TIMEOUT_SEC)
|
||
_seen: dict = {}
|
||
try:
|
||
info = await asyncio.wait_for(
|
||
_poll_fill_v2_impl(stock_code, order_id, ordered_qty, _seen), timeout=timeout,
|
||
)
|
||
except asyncio.TimeoutError:
|
||
_last = _seen.get("last")
|
||
if _last is None:
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 체결 조회 시간초과({timeout:.0f}초) "
|
||
f"order_id={order_id} — 신호가 기록 유지"
|
||
)
|
||
return None
|
||
# [R1 2026-10-09] 시간초과 전에 본 행이 있으면 버리지 않는다(시도 소진 시와 같은 의미:
|
||
# 미종결이면 terminal=False 로 넘어가 호출부가 수량 보정을 보류한다).
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 체결 조회 시간초과({timeout:.0f}초) "
|
||
f"order_id={order_id} — 직전 확인 행(체결 {int(_last.get('filled_qty', 0) or 0)}주) 사용"
|
||
)
|
||
info = _last
|
||
except Exception as e:
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 체결 조회 실패 order_id={order_id}: "
|
||
f"{type(e).__name__}: {e} — 신호가 기록 유지"
|
||
)
|
||
return None
|
||
if info is None:
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 주문 행 미확인 "
|
||
f"({int(settings.DAY_TRADE_FILL_CHECK_ATTEMPTS)}회 조회) order_id={order_id} — 신호가 기록 유지"
|
||
)
|
||
return None
|
||
info = dict(info)
|
||
info["terminal"] = is_terminal_fill(info, ordered_qty)
|
||
return info
|
||
|
||
|
||
def schedule_v2(
|
||
stock_code: str,
|
||
order_id: str,
|
||
side: str,
|
||
ordered_qty: int,
|
||
on_result: Callable[[dict], Awaitable[None]],
|
||
) -> Optional[asyncio.Task]:
|
||
"""[M7] fetch_fill_v2 결과(체결 0·부분체결 포함)를 `on_result(info)` 로 넘긴다.
|
||
|
||
`DAY_TRADE_FILL_CHECK_ENABLED=False` 면 아무것도 하지 않는다(None).
|
||
"""
|
||
if not settings.DAY_TRADE_FILL_CHECK_ENABLED:
|
||
return None
|
||
|
||
async def _run() -> None:
|
||
# [R1 2026-10-09] 즉시 확인(≈6초)에서 종결을 못 보면 지연 재확인까지 한 뒤 **1회만** 넘긴다
|
||
# (on_result 는 기록 손익 대비 차액을 더하므로 두 번 부르면 이중 보정된다).
|
||
info = await fetch_fill_with_late_recheck_v2(stock_code, order_id, side, ordered_qty)
|
||
if info is None:
|
||
return
|
||
try:
|
||
await on_result(info)
|
||
except Exception as e:
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 체결 결과 반영 실패 order_id={order_id}: "
|
||
f"{type(e).__name__}: {e}",
|
||
exc_info=True,
|
||
)
|
||
|
||
try:
|
||
task = asyncio.get_running_loop().create_task(_run())
|
||
except RuntimeError as e:
|
||
logger.warning(f"[DTFill] {stock_code} {side} 체결 확인 예약 실패: {e}")
|
||
return None
|
||
_pending_tasks.add(task)
|
||
task.add_done_callback(_pending_tasks.discard)
|
||
return task
|
||
|
||
|
||
# ── [R1 2026-10-09] 지연 재확인 ───────────────────────────────────────────────
|
||
# 운영 실측(trade_history daytrading, 2026-10-06~08): SELL 8/18·BUY 3/19 의 slippage 가 NULL
|
||
# = 체결 미확인(신호가 기록 유지). 즉시 확인 창은 DELAY 2 + INTERVAL 2×2 ≈ 6초(+조회 왕복)로,
|
||
# 모의 서버 체결 지연·조회 반영 지연·rate-limit 으로 행이 늦게 보이면 그대로 놓친다(원인 추정 —
|
||
# 운영 로그 미확인). 놓치면 daily_pnl(-3% 한도)·연패 판정·원장이 신호가 기준으로 남는다.
|
||
# 즉시 확인이 종결을 못 보면 간격을 넓혀 몇 번 더 본다. 전략 파라미터가 아니라 조회 정책이다.
|
||
# 기본값: 시도 4회, 간격 = 즉시 확인 간격 × 5(운영 2초 → 10초), 호출당 상한 = 즉시 확인 전체 상한.
|
||
# config 키(DAY_TRADE_FILL_LATE_RECHECK_ATTEMPTS / _INTERVAL_SEC)가 있으면 그 값을 쓴다.
|
||
_LATE_RECHECK_DEFAULT_ATTEMPTS = 4
|
||
_LATE_RECHECK_INTERVAL_FACTOR = 5.0
|
||
|
||
|
||
def _late_recheck_params_impl() -> tuple[int, float, float]:
|
||
"""(시도 횟수, 간격 초, 호출당 상한 초). 음수·잘못된 값은 0 으로 막는다."""
|
||
try:
|
||
attempts = int(getattr(settings, "DAY_TRADE_FILL_LATE_RECHECK_ATTEMPTS",
|
||
_LATE_RECHECK_DEFAULT_ATTEMPTS))
|
||
except (TypeError, ValueError):
|
||
attempts = _LATE_RECHECK_DEFAULT_ATTEMPTS
|
||
_iv = getattr(settings, "DAY_TRADE_FILL_LATE_RECHECK_INTERVAL_SEC", None)
|
||
try:
|
||
interval = (float(_iv) if _iv is not None
|
||
else float(settings.DAY_TRADE_FILL_CHECK_INTERVAL_SEC) * _LATE_RECHECK_INTERVAL_FACTOR)
|
||
except (TypeError, ValueError):
|
||
interval = float(settings.DAY_TRADE_FILL_CHECK_INTERVAL_SEC) * _LATE_RECHECK_INTERVAL_FACTOR
|
||
per_call = max(0.0, float(settings.DAY_TRADE_FILL_CHECK_TIMEOUT_SEC))
|
||
return max(0, attempts), max(0.0, interval), per_call
|
||
|
||
|
||
async def fetch_fill_with_late_recheck_v2(
|
||
stock_code: str, order_id: str, side: str, ordered_qty: int,
|
||
) -> Optional[dict]:
|
||
"""fetch_fill_v2 + (종결 미확인이면) 지연 재확인. 반환 형식은 fetch_fill_v2 와 같다.
|
||
|
||
- 즉시 확인이 종결이면 그대로 반환(추가 조회 없음).
|
||
- 아니면 `_late_recheck_params_impl()` 만큼 더 조회한다. 조회 1회의 예외·시간초과는 그 회차만
|
||
건너뛴다(warning). 종결 행을 보면 즉시 반환, 끝까지 못 보면 가장 최근에 본 행
|
||
(terminal=False) — 없으면 None.
|
||
- 로그: 지연 재확인으로 확정되면 info, 최종 미확인이면 warning 1건
|
||
('최종 체결 미확인' — 운영 집계용. 즉시 확인 단계의 warning 은 그대로 남는다).
|
||
종가 단일가(15:20~15:30) 매도는 exit_service 가 장 마감 후 재확인을 따로 하므로 이 함수를 쓰지 않는다.
|
||
"""
|
||
info = await fetch_fill_v2(stock_code, order_id, side, ordered_qty)
|
||
if not order_id:
|
||
return info
|
||
if info is not None and info.get("terminal"):
|
||
return info
|
||
attempts, interval, per_call = _late_recheck_params_impl()
|
||
if attempts <= 0:
|
||
return info
|
||
last = info
|
||
for i in range(attempts):
|
||
await asyncio.sleep(interval)
|
||
try:
|
||
rows = await asyncio.wait_for(
|
||
korea_investment_api.get_todays_orders(stock_code), timeout=per_call,
|
||
)
|
||
except asyncio.TimeoutError:
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 지연 재확인 {i + 1}/{attempts} 시간초과({per_call:.0f}초) "
|
||
f"order_id={order_id}"
|
||
)
|
||
continue
|
||
except Exception as e:
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 지연 재확인 {i + 1}/{attempts} 실패 order_id={order_id}: "
|
||
f"{type(e).__name__}: {e}"
|
||
)
|
||
continue
|
||
row = parse_fill_row_v2(rows, order_id)
|
||
if row is None:
|
||
continue
|
||
last = dict(row)
|
||
last["terminal"] = is_terminal_fill(row, ordered_qty)
|
||
if last["terminal"]:
|
||
logger.info(
|
||
f"[DTFill] {stock_code} {side} 지연 재확인 {i + 1}/{attempts}회차에 종결 확인 "
|
||
f"(체결 {int(last.get('filled_qty', 0) or 0)}/{ordered_qty}주) order_id={order_id}"
|
||
)
|
||
return last
|
||
logger.warning(
|
||
f"[DTFill] {stock_code} {side} 최종 체결 미확인(즉시 확인 + 지연 재확인 {attempts}회, "
|
||
f"간격 {interval:.0f}초) order_id={order_id} — "
|
||
+ ("주문 행 없음, 신호가 기록 유지" if last is None
|
||
else f"미종결 행 사용(체결 {int(last.get('filled_qty', 0) or 0)}/{ordered_qty}주)")
|
||
)
|
||
return last
|