fix(P2): 주문 멱등성·취소 보호·다중행·v2/수동매도 정합
- korea_investment_api: 주문 계열 엔드포인트는 내부 재시도 금지(_NON_IDEMPOTENT_ENDPOINTS). 상위 _place_order_with_retry 와 중첩돼 타임아웃 시 동일 주문 최대 12회 전송 가능했음. get_todays_orders() 추가. - trade_executor: 예외 후 재시도 전에 당일 주문내역으로 접수 여부 확인(_find_recent_order) → 접수됐으면 재주문하지 않고 기존 주문 반환. - daytrading/exit_service: 주문 체결 이후 상태 반영을 _finalize_exit 로 분리하고 asyncio.shield 로 보호 — 스케줄러 단계 타임아웃이 이 구간을 취소하면 브로커는 팔았는데 인메모리는 그대로라 다음 사이클이 재매도. - swing/position_manager: active 행이 2개면 MultipleResultsFound 로 해당 종목 청산이 통째로 스킵됐음(손절 미작동) → 최신 행 사용. - swing/exit_service: 큐 지연·DLQ 시 같은 분할매도가 반복되지 않도록 180초 인메모리 쿨다운. - portfolio_v2: 부분매도 ActiveTrade 잔량 갱신 + 전략명 매핑(daytrading→day_trade, us_swing→us_mean_reversion) — 매칭 실패로 active 행이 영구 잔존했음. - portfolio.sell_stock: safe_place_order 경유(세마포어·분산락) + DT 인메모리 잔량 반영. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -244,6 +244,21 @@ async def _execute_exit(
|
||||
logger.error(f"[DayExit] 매도 주문 실패 ({stock_code}): {err}")
|
||||
return None
|
||||
|
||||
# [2026-09-23] 여기서부터는 이미 체결된 주문의 상태 반영이다. 스케줄러의 단계
|
||||
# 타임아웃(_guard)이 이 지점을 취소하면 "브로커는 팔았는데 인메모리는 그대로" 상태가
|
||||
# 되어 다음 사이클이 같은 물량을 다시 판다 → shield 로 취소로부터 보호한다.
|
||||
return await asyncio.shield(_finalize_exit(
|
||||
stock_code, pos, price, reason, sell_pct, order_resp, quantity, total_qty, is_full_exit
|
||||
))
|
||||
|
||||
|
||||
async def _finalize_exit(
|
||||
stock_code: str, pos: dict, price: float, reason: str, sell_pct: float,
|
||||
order_resp: dict, quantity: int, total_qty: int, is_full_exit: bool,
|
||||
) -> dict | None:
|
||||
"""체결 이후 상태 반영 (인메모리·DB·캐시). 취소되면 안 되는 구간."""
|
||||
entry_price = pos.get("entry_price", 0)
|
||||
stock_name = pos.get("stock_name", stock_code)
|
||||
order_id = order_resp.get("output", {}).get("ODNO", "")
|
||||
profit_loss = (price - entry_price) * quantity
|
||||
profit_loss_rate = ((price - entry_price) / entry_price * 100) if entry_price > 0 else 0.0
|
||||
|
||||
@@ -500,13 +500,24 @@ class KoreaInvestmentAPI:
|
||||
async with self.semaphore:
|
||||
return await self._make_request_with_retry(method, endpoint, tr_id, extra_headers=extra_headers, **kwargs)
|
||||
|
||||
# 주문 계열 엔드포인트 — 재전송 시 중복 체결 위험이 있어 내부 재시도를 하지 않는다.
|
||||
# (상위 trade_executor._place_order_with_retry 가 체결 확인 후 재시도를 판단한다)
|
||||
_NON_IDEMPOTENT_ENDPOINTS = (
|
||||
"/uapi/domestic-stock/v1/trading/order-cash",
|
||||
"/uapi/domestic-stock/v1/trading/order-rvsecncl",
|
||||
"/uapi/overseas-stock/v1/trading/order",
|
||||
)
|
||||
|
||||
async def _make_request_with_retry(self, method: str, endpoint: str, tr_id: str, extra_headers: Optional[Dict[str, str]] = None, **kwargs) -> (Dict[str, Any], Any):
|
||||
"""재시도 로직을 포함한 실제 요청 실행"""
|
||||
"""재시도 로직을 포함한 실제 요청 실행 (주문 엔드포인트는 재시도 없음)"""
|
||||
start_time = time.time()
|
||||
retry_count = 0
|
||||
last_exception = None
|
||||
|
||||
for attempt in range(self.max_retries + 1):
|
||||
|
||||
# 주문은 1회만 전송 — 타임아웃/5xx 에도 재전송하지 않는다(중복 주문 방지)
|
||||
_max_retries = 0 if endpoint in self._NON_IDEMPOTENT_ENDPOINTS else self.max_retries
|
||||
|
||||
for attempt in range(_max_retries + 1):
|
||||
try:
|
||||
# 쓰로틀링: 최소 요청 간격 보장
|
||||
# _throttle_lock으로 "간격 계산 + sleep + last_request_time 갱신"을 직렬화하여
|
||||
@@ -556,9 +567,9 @@ class KoreaInvestmentAPI:
|
||||
# 서버 오류 (5xx, 429) — 재시도
|
||||
elif response.status_code in [429, 500, 502, 503, 504]:
|
||||
retry_count += 1
|
||||
if attempt < self.max_retries:
|
||||
if attempt < _max_retries:
|
||||
delay = self.retry_delay * (self.retry_backoff ** attempt) + random.uniform(0, 0.3)
|
||||
logger.warning(f"⏳ API 재시도 {attempt + 1}/{self.max_retries}: {response.status_code}, {delay:.1f}초 후")
|
||||
logger.warning(f"⏳ API 재시도 {attempt + 1}/{_max_retries}: {response.status_code}, {delay:.1f}초 후")
|
||||
await asyncio.sleep(delay)
|
||||
continue
|
||||
else:
|
||||
@@ -577,9 +588,9 @@ class KoreaInvestmentAPI:
|
||||
retry_count += 1
|
||||
last_exception = e
|
||||
|
||||
if attempt < self.max_retries:
|
||||
if attempt < _max_retries:
|
||||
delay = self.retry_delay * (self.retry_backoff ** attempt) + random.uniform(0, 0.3)
|
||||
logger.warning(f"⏳ API 타임아웃, {delay:.1f}초 후 재시도 ({attempt + 1}/{self.max_retries})")
|
||||
logger.warning(f"⏳ API 타임아웃, {delay:.1f}초 후 재시도 ({attempt + 1}/{_max_retries})")
|
||||
await asyncio.sleep(delay)
|
||||
continue
|
||||
else:
|
||||
@@ -1092,6 +1103,46 @@ class KoreaInvestmentAPI:
|
||||
logger.error(f"Error fetching balance: {e}")
|
||||
raise
|
||||
|
||||
async def get_todays_orders(self, stock_code: str = "") -> list[dict]:
|
||||
"""당일 주문체결 내역 조회 (inquire-daily-ccld).
|
||||
|
||||
주문 전송 후 응답을 못 받았을 때 '실제로 접수됐는지' 확인해 중복 주문을 막는 용도.
|
||||
실전: TTTC8001R / 모의: VTTC8001R. 실패 시 빈 리스트.
|
||||
"""
|
||||
try:
|
||||
account_number_prefix = self.account_number[:8]
|
||||
account_number_suffix = self.account_number[8:]
|
||||
is_virtual = "vts" in self.base_url.lower()
|
||||
today = datetime.now().strftime("%Y%m%d")
|
||||
result, _ = await self._make_request(
|
||||
"GET",
|
||||
"/uapi/domestic-stock/v1/trading/inquire-daily-ccld",
|
||||
"VTTC8001R" if is_virtual else "TTTC8001R",
|
||||
params={
|
||||
"CANO": account_number_prefix,
|
||||
"ACNT_PRDT_CD": account_number_suffix,
|
||||
"INQR_STRT_DT": today,
|
||||
"INQR_END_DT": today,
|
||||
"SLL_BUY_DVSN_CD": "00",
|
||||
"INQR_DVSN": "00",
|
||||
"PDNO": stock_code or "",
|
||||
"CCLD_DVSN": "00",
|
||||
"ORD_GNO_BRNO": "",
|
||||
"ODNO": "",
|
||||
"INQR_DVSN_3": "00",
|
||||
"INQR_DVSN_1": "",
|
||||
"CTX_AREA_FK100": "",
|
||||
"CTX_AREA_NK100": "",
|
||||
},
|
||||
)
|
||||
if result.get("rt_cd") != "0":
|
||||
return []
|
||||
out = result.get("output1") or []
|
||||
return out if isinstance(out, list) else [out]
|
||||
except Exception as e:
|
||||
logger.debug(f"[get_todays_orders] 조회 실패: {e}")
|
||||
return []
|
||||
|
||||
async def get_buyable_cash(self, stock_code: str = "005930") -> float:
|
||||
"""주문가능현금(ord_psbl_cash) 조회 — 매수가능조회 API.
|
||||
|
||||
|
||||
@@ -20,6 +20,7 @@ strategy_type='swing' 포지션만 처리하도록 분리한 서비스.
|
||||
- 매도 주문: safe_place_order() 사용 (세마포어+분산락 보호)
|
||||
"""
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import select
|
||||
@@ -41,6 +42,10 @@ from app.core.json_helpers import _dumps
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 종목별 마지막 청산 시각(epoch) — 큐 지연으로 같은 분할매도가 반복되는 것을 막는 인메모리 가드
|
||||
_recent_exits: dict[str, float] = {}
|
||||
EXIT_COOLDOWN_SEC = 180
|
||||
|
||||
|
||||
def _check_momentum_exit(
|
||||
current_price: float,
|
||||
@@ -160,6 +165,7 @@ async def check_and_execute_exit_v2(db: AsyncSession) -> list[dict]:
|
||||
|
||||
try:
|
||||
# strategy_type='swing' 포지션만 조회
|
||||
_now_ts = time.time()
|
||||
holdings = await get_swing_holdings(db)
|
||||
if not holdings:
|
||||
logger.debug("[SwingExit] 스윙 보유 종목 없음")
|
||||
@@ -285,6 +291,17 @@ async def check_and_execute_exit_v2(db: AsyncSession) -> list[dict]:
|
||||
if not exit_signal:
|
||||
continue
|
||||
|
||||
# [2026-09-23] 상태 저장(remaining_pct·Portfolio)이 db_worker 큐 경유라
|
||||
# 큐 지연·DLQ 시 다음 사이클이 옛 값을 읽어 같은 분할매도를 반복한다.
|
||||
# 인메모리 쿨다운으로 직전 청산 직후 재매도를 차단한다.
|
||||
_last_exit = _recent_exits.get(stock_code)
|
||||
if _last_exit is not None and (_now_ts - _last_exit) < EXIT_COOLDOWN_SEC:
|
||||
logger.info(
|
||||
f"[SwingExit] {stock_code}: 직전 청산 {int(_now_ts - _last_exit)}초 전 "
|
||||
f"— 쿨다운({EXIT_COOLDOWN_SEC}초) 중 스킵"
|
||||
)
|
||||
continue
|
||||
|
||||
# ── 매도 수량 결정 (소량 분할 방지) ─────────────────────
|
||||
sell_pct = getattr(exit_signal, "sell_pct", 1.0)
|
||||
if sell_pct < 1.0 and quantity <= 3:
|
||||
@@ -339,6 +356,8 @@ async def check_and_execute_exit_v2(db: AsyncSession) -> list[dict]:
|
||||
"strategy_type": "swing",
|
||||
})
|
||||
|
||||
_recent_exits[stock_code] = time.time()
|
||||
|
||||
# 포트폴리오 화면 즉시 반영 (Portfolio 수량 차감 + 캐시)
|
||||
from app.services.portfolio_cache_bridge import reflect_sell
|
||||
await reflect_sell(stock_code, sell_qty, latest_price, remaining_qty=quantity - sell_qty)
|
||||
|
||||
@@ -78,14 +78,20 @@ async def get_swing_active_trade(db: AsyncSession, stock_code: str) -> ActiveTra
|
||||
Returns:
|
||||
ActiveTrade | None: 활성 매매 ORM 객체, 없으면 None
|
||||
"""
|
||||
# [2026-09-23] scalar_one_or_none() 은 행이 2개 이상이면 MultipleResultsFound 를 던지고,
|
||||
# 호출측 except 가 이를 삼켜 해당 종목의 청산(손절 포함)이 통째로 스킵됐다.
|
||||
# db_worker legacy 경로와 동일하게 가장 최근 행을 사용한다.
|
||||
result = await db.execute(
|
||||
select(ActiveTrade).where(
|
||||
select(ActiveTrade)
|
||||
.where(
|
||||
ActiveTrade.stock_code == stock_code,
|
||||
ActiveTrade.status == "active",
|
||||
ActiveTrade.strategy_name != "day_trade",
|
||||
)
|
||||
.order_by(ActiveTrade.id.desc())
|
||||
.limit(1)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
return result.scalars().first()
|
||||
|
||||
|
||||
async def count_swing_positions(db: AsyncSession) -> int:
|
||||
|
||||
@@ -11,7 +11,7 @@ import asyncio
|
||||
import logging
|
||||
import time
|
||||
from decimal import Decimal
|
||||
from datetime import timedelta
|
||||
from datetime import datetime, timedelta
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import select, desc
|
||||
|
||||
@@ -155,7 +155,17 @@ async def _place_order_with_retry(
|
||||
await asyncio.sleep(wait)
|
||||
except Exception as e:
|
||||
# 네트워크 타임아웃 등 예외 → 재시도 대상
|
||||
# [2026-09-23] 단, KIS가 주문을 접수한 뒤 응답만 못 받은 경우가 있으므로
|
||||
# 재시도 전에 당일 체결 내역으로 중복 여부를 확인한다(멱등성 보강).
|
||||
last_result = {"rt_cd": "-1", "msg1": str(e), "odno": None}
|
||||
if is_domestic:
|
||||
_dup = await _find_recent_order(stock_code, order_type, quantity)
|
||||
if _dup:
|
||||
logger.warning(
|
||||
f"[ORDER_DUP] {stock_code} {order_type} 예외 발생했으나 주문 접수 확인됨 "
|
||||
f"(odno={_dup.get('odno')}) → 재시도 중단"
|
||||
)
|
||||
return _dup
|
||||
if attempt < max_retries - 1:
|
||||
wait = 1 * (2 ** attempt)
|
||||
logger.warning(f"[ORDER_RETRY] {stock_code} {order_type} 예외 ({attempt + 1}/{max_retries}), {wait}초 후 재시도: {type(e).__name__}: {e}")
|
||||
@@ -165,6 +175,49 @@ async def _place_order_with_retry(
|
||||
return last_result or {"rt_cd": "-1", "msg1": "max_retries_exceeded", "odno": None}
|
||||
|
||||
|
||||
async def _find_recent_order(stock_code: str, order_type: str, quantity: int) -> dict | None:
|
||||
"""직전 주문이 실제로 KIS에 접수됐는지 당일 주문체결 내역으로 확인.
|
||||
|
||||
타임아웃/연결 끊김으로 응답을 못 받았지만 주문은 접수된 경우, 그대로 재시도하면
|
||||
중복 체결이 된다. 최근 2분 내 같은 종목·매매구분·수량 주문이 있으면 그 주문을 반환한다.
|
||||
조회 실패 시 None (판단 불가 → 호출측의 기존 재시도 로직을 따른다).
|
||||
"""
|
||||
try:
|
||||
rows = await korea_investment_api.get_todays_orders(stock_code)
|
||||
except Exception as e:
|
||||
logger.debug(f"[ORDER_DUP] 주문내역 조회 실패 ({stock_code}): {e}")
|
||||
return None
|
||||
if not rows:
|
||||
return None
|
||||
|
||||
_now = datetime.now()
|
||||
_side = "02" if order_type.upper() == "BUY" else "01" # KIS: 01=매도, 02=매수
|
||||
for row in rows:
|
||||
try:
|
||||
if row.get("pdno") != stock_code:
|
||||
continue
|
||||
if row.get("sll_buy_dvsn_cd") not in (_side, None, ""):
|
||||
continue
|
||||
if int(row.get("ord_qty", 0) or 0) != int(quantity):
|
||||
continue
|
||||
_t = str(row.get("ord_tmd") or "") # HHMMSS
|
||||
if len(_t) == 6:
|
||||
_ordered = _now.replace(
|
||||
hour=int(_t[:2]), minute=int(_t[2:4]), second=int(_t[4:]), microsecond=0
|
||||
)
|
||||
if (_now - _ordered).total_seconds() > 120:
|
||||
continue
|
||||
return {
|
||||
"rt_cd": "0",
|
||||
"msg1": "기존 주문 확인(중복 방지)",
|
||||
"odno": row.get("odno"),
|
||||
"output": {"ODNO": row.get("odno"), "ord_prc": row.get("ord_unpr")},
|
||||
}
|
||||
except Exception:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
# ── [C1] 공통 주문 래퍼 — 분산락+세마포어 보호 (daytrading/swing에서 import) ──
|
||||
async def safe_place_order(
|
||||
stock_code: str,
|
||||
|
||||
Reference in New Issue
Block a user