genichinandClaude Opus 4.8
6d7bca1460
전략 이탈로 슬롯이 내부 청산 루틴(워치독) 소유로 넘어간 상태가 무성이라 운영자가 인지할 수 없던 문제(issue #12)를 가시화한다. 거버넌스: 전부 change-req (신규 REQ 없음, dcy-017 OQ-3; Redis 미채택). - REQ-034/021: WatchdogStore.last_seen + snapshot RPC 응답에 holding 슬롯 워치독 상태(expired/last_seen) read-only 동봉(best-effort) - REQ-032: trex-core slots 가 expired 슬롯을 WATCHDOG-OWNED(전략 이탈) 표시 - REQ-037/038/039: safety-sell 워치독 소유 진입(WARN)/이탈(INFO) + 발동 (forced_close·stop_loss=WARN, take_profit·market_close=INFO) 로그 + in-process 카운터(SafetySellService.counters), 구조화 필드(slot_id/ portfolio_id/reason/phase/instance_key). per-tick 스킵은 카운터만 - REQ-042: runtime-flows/interface-contract 정합 best-effort 격리(NFR-1): 모든 가시화는 예외 격리 — 워치독 판정·청산 불변. dcy-016 억제 읽기로 관측이 워치독을 리셋하지 않음(비침습). 검증: 단위 335 passed/34 skipped, 인프라 게이트(PG+RabbitMQ) 369 passed, ruff clean. 신규 test_safety_sell_visibility(6) + snapshot watchdog(1) + CLI(1). BAT-021 rc. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
292 lines
13 KiB
Python
292 lines
13 KiB
Python
"""Safety-sell trigger service (REQ-037, REQ-038, REQ-039 / BAT-011).
|
|
|
|
A single service that the `trex-safety-sell` daemon worker drives once per
|
|
tick. It evaluates three autonomous sell triggers on `holding` slots:
|
|
|
|
- forced_close_at expiry (REQ-037) — watchdog-gated
|
|
- stop-loss / take-profit price (REQ-038) — watchdog-gated, KIS polling
|
|
- market-close liquidation (REQ-039) — watchdog-INDEPENDENT, once/day
|
|
|
|
Watchdog gating: forced_close and price checks only act on slots whose
|
|
watchdog has expired (strategy module is absent). During the grace period
|
|
no expiry-gated trigger fires (REQ-034). Market-close liquidation ignores
|
|
the watchdog and the grace period entirely (REQ-039 AC-2/AC-6).
|
|
|
|
All sells go through `SlotService.sell()` (REQ-036), which is idempotent on
|
|
`pending_sell` — so concurrent/duplicate triggers never double-submit.
|
|
Each trigger is exception-isolated: a single slot's failure (KIS error,
|
|
transition race) is logged and the loop continues (REQ-024 degrade policy).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import date as date_cls
|
|
from datetime import datetime, time as time_cls, timedelta, timezone
|
|
|
|
from trex_core.application.broker_port import BrokerPort
|
|
from trex_core.application.watchdog import WatchdogStore
|
|
from trex_core.domain.slot import SlotRepository, SlotService, SlotState
|
|
from trex_core.domain.slot.models import Slot
|
|
from trex_core.runtime.config import SafetySellConfig
|
|
|
|
_log = logging.getLogger(__name__)
|
|
_KST = timezone(timedelta(hours=9))
|
|
|
|
# Conservative KIS current-price quota guard (REQ-038 AC-8). When the
|
|
# estimated hourly polling volume crosses this, we log a warning so the
|
|
# operator can widen TREX_STOP_LOSS_POLL_INTERVAL.
|
|
_KIS_PRICE_QUOTA_PER_HOUR = 1000
|
|
|
|
_ACTOR = "trex-safety-sell"
|
|
|
|
|
|
def _parse_kst_time(spec: str) -> time_cls:
|
|
parts = spec.split(":")
|
|
return time_cls(hour=int(parts[0]), minute=int(parts[1]) if len(parts) > 1 else 0)
|
|
|
|
|
|
class SafetySellService:
|
|
"""Evaluates safety-sell triggers for one daemon tick (REQ-037/038/039)."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
slot_repo: SlotRepository,
|
|
slot_service: SlotService,
|
|
broker: BrokerPort,
|
|
watchdog: WatchdogStore,
|
|
config: SafetySellConfig,
|
|
instance_key: str = "default",
|
|
) -> None:
|
|
self._repo = slot_repo
|
|
self._svc = slot_service
|
|
self._broker = broker
|
|
self._watchdog = watchdog
|
|
self._cfg = config
|
|
self._instance_key = instance_key # dcy-017 FR-5: multi-instance label
|
|
self._last_liquidation_date: date_cls | None = None
|
|
self._last_forced_close_at: datetime | None = None
|
|
self._last_price_check_at: datetime | None = None
|
|
# BAT-014 / REQ-052: per-slot retry cooldown for persistent rejection.
|
|
self._last_attempt: dict = {} # slot_id -> last sell-attempt datetime
|
|
self._attempt_count: dict = {} # slot_id -> consecutive attempt count
|
|
# dcy-017 / REQ-037 (issue #12): watchdog-ownership visibility.
|
|
self._owned: set = set() # holding slot_ids currently watchdog-owned
|
|
self.counters: dict[str, int] = {} # phase/event counters (in-process metrics)
|
|
|
|
# ----------------------------------------------- dcy-017 visibility helpers
|
|
def _count(self, key: str) -> None:
|
|
self.counters[key] = self.counters.get(key, 0) + 1
|
|
|
|
# WARN-level fire reasons (attention events); others log at INFO.
|
|
_WARN_REASONS = frozenset({"forced_close_at", "stop_loss"})
|
|
|
|
def _log_fire(self, slot: Slot, reason: str) -> None:
|
|
"""Visibility for an actual safety-sell fire (dcy-017 FR-1, OQ-2)."""
|
|
self._count(f"fire:{reason}")
|
|
level = logging.WARNING if reason in self._WARN_REASONS else logging.INFO
|
|
_log.log(
|
|
level,
|
|
"safety-sell fired: phase=%s slot=%s portfolio_id=%s reason=%s instance_key=%s",
|
|
reason, slot.slot_id, slot.portfolio_id, reason, self._instance_key,
|
|
)
|
|
|
|
def _update_ownership(self, now: datetime) -> None:
|
|
"""Log watchdog-ownership entry (strategy absent) / exit (returned).
|
|
|
|
dcy-017 FR-1: WARN on entry (a holding slot's strategy went silent so the
|
|
safety net now owns it), INFO on exit. Counters track both. Best-effort —
|
|
never raises into the tick (NFR-1)."""
|
|
try:
|
|
if self._in_grace():
|
|
return # no expiry judgement during grace (REQ-034)
|
|
current = {
|
|
s.slot_id for s in self._holding(now) if self._watchdog_expired(s)
|
|
}
|
|
by_id = {s.slot_id: s for s in self._holding(now)}
|
|
for sid in current - self._owned:
|
|
self._count("owned:enter")
|
|
s = by_id.get(sid)
|
|
_log.warning(
|
|
"watchdog ownership entered (strategy absent): slot=%s "
|
|
"portfolio_id=%s phase=watchdog_owned instance_key=%s",
|
|
sid, (s.portfolio_id if s else None), self._instance_key,
|
|
)
|
|
for sid in self._owned - current:
|
|
self._count("owned:exit")
|
|
_log.info(
|
|
"watchdog ownership released (strategy returned): slot=%s "
|
|
"phase=watchdog_released instance_key=%s",
|
|
sid, self._instance_key,
|
|
)
|
|
self._owned = current
|
|
except Exception: # noqa: BLE001 - visibility must not break the tick
|
|
return
|
|
|
|
@property
|
|
def base_tick_sec(self) -> float:
|
|
"""Worker loop sleep — the finest of the configured check cadences."""
|
|
return float(max(
|
|
1,
|
|
min(self._cfg.forced_close_check_interval_sec,
|
|
self._cfg.stop_loss_poll_interval_sec),
|
|
))
|
|
|
|
# ---------------------------------------------------------------- helpers
|
|
@staticmethod
|
|
def _trading_date(now: datetime) -> date_cls:
|
|
return now.astimezone(timezone.utc).date()
|
|
|
|
def _holding(self, now: datetime) -> list[Slot]:
|
|
return [s for s in self._repo.list_slots(self._trading_date(now))
|
|
if s.status is SlotState.HOLDING]
|
|
|
|
def _watchdog_expired(self, slot: Slot) -> bool:
|
|
return self._watchdog.is_expired(str(slot.slot_id), self._cfg.watchdog_timeout_sec)
|
|
|
|
def _in_grace(self) -> bool:
|
|
return self._watchdog.in_grace_period(self._cfg.watchdog_grace_period_sec)
|
|
|
|
def _safe_sell(self, slot: Slot, reason: str) -> bool:
|
|
try:
|
|
self._svc.sell(slot.slot_id, actor=_ACTOR, reason=reason)
|
|
return True
|
|
except Exception: # noqa: BLE001 - one slot's failure must not stop the loop
|
|
_log.warning("safety-sell failed (slot=%s reason=%s)",
|
|
slot.slot_id, reason, exc_info=True)
|
|
return False
|
|
|
|
# ------------------------------------------------ REQ-052 retry cooldown
|
|
def _in_cooldown(self, slot_id, now: datetime) -> bool:
|
|
"""True if the slot was attempted within min_retry_interval_sec (skip retry)."""
|
|
last = self._last_attempt.get(slot_id)
|
|
if last is None:
|
|
return False
|
|
return (now - last).total_seconds() < self._cfg.min_retry_interval_sec
|
|
|
|
def _attempt_sell(self, slot: Slot, reason: str, now: datetime) -> bool:
|
|
"""Record a sell attempt (cooldown already cleared by caller) + warn on repeat.
|
|
|
|
Updates last_attempt/attempt_count regardless of success (dcy-010 FR-2).
|
|
"""
|
|
sid = slot.slot_id
|
|
count = self._attempt_count.get(sid, 0) + 1
|
|
if count >= self._cfg.max_retry_warn_count:
|
|
_log.warning(
|
|
"safety-sell repeated rejection: slot=%s count=%d reason=%s last_attempt_at=%s",
|
|
sid, count, reason, self._last_attempt.get(sid),
|
|
)
|
|
self._last_attempt[sid] = now
|
|
self._attempt_count[sid] = count
|
|
return self._safe_sell(slot, reason)
|
|
|
|
def _prune_cooldown(self, now: datetime) -> None:
|
|
"""Drop cooldown state for slots no longer holding (dcy-010 FR-4)."""
|
|
holding_ids = {s.slot_id for s in self._holding(now)}
|
|
for sid in list(self._last_attempt):
|
|
if sid not in holding_ids:
|
|
self._last_attempt.pop(sid, None)
|
|
self._attempt_count.pop(sid, None)
|
|
|
|
def estimated_hourly_kis_calls(self, n_slots: int) -> int:
|
|
"""REQ-038 AC-8: worst-case current-price polls per hour for n_slots."""
|
|
interval = max(1, self._cfg.stop_loss_poll_interval_sec)
|
|
return n_slots * (3600 // interval)
|
|
|
|
# ---------------------------------------------------- REQ-037 forced_close
|
|
def check_forced_close(self, now: datetime) -> list:
|
|
"""Sell holding+watchdog-expired slots whose forced_close_at has passed."""
|
|
if self._in_grace():
|
|
return []
|
|
fired = []
|
|
for s in self._holding(now):
|
|
if s.forced_close_at is None:
|
|
continue
|
|
if not self._watchdog_expired(s):
|
|
continue # strategy module active — do not interfere (AC-3)
|
|
if s.forced_close_at > now:
|
|
continue
|
|
if self._in_cooldown(s.slot_id, now):
|
|
continue # REQ-052: persistent-rejection backoff
|
|
if self._attempt_sell(s, "forced_close_at", now):
|
|
self._log_fire(s, "forced_close_at")
|
|
fired.append(s.slot_id)
|
|
return fired
|
|
|
|
# ----------------------------------------------- REQ-038 stop-loss/take-profit
|
|
def check_prices(self, now: datetime) -> list:
|
|
"""Poll KIS current price for eligible slots and trigger SL/TP sells."""
|
|
if self._in_grace():
|
|
return []
|
|
candidates = [
|
|
s for s in self._holding(now)
|
|
if (s.stop_loss_price is not None or s.take_profit_price is not None)
|
|
and self._watchdog_expired(s) # no KIS call for watchdog-active (AC-4)
|
|
]
|
|
if self.estimated_hourly_kis_calls(len(candidates)) > _KIS_PRICE_QUOTA_PER_HOUR:
|
|
_log.warning(
|
|
"safety-sell KIS price polling may exceed quota: ~%d calls/hour "
|
|
"(%d slots, interval=%ds). Consider widening TREX_STOP_LOSS_POLL_INTERVAL.",
|
|
self.estimated_hourly_kis_calls(len(candidates)),
|
|
len(candidates), self._cfg.stop_loss_poll_interval_sec,
|
|
)
|
|
fired = []
|
|
for s in candidates:
|
|
if self._in_cooldown(s.slot_id, now):
|
|
continue # REQ-052: skip KIS poll + retry during cooldown
|
|
price = self._broker.get_current_price(s.symbol)
|
|
if price is None:
|
|
continue # unavailable — skip, retry next tick (no false trigger)
|
|
if s.stop_loss_price is not None and price <= s.stop_loss_price:
|
|
if self._attempt_sell(s, "stop_loss", now):
|
|
self._log_fire(s, "stop_loss")
|
|
fired.append(s.slot_id)
|
|
elif s.take_profit_price is not None and price >= s.take_profit_price:
|
|
if self._attempt_sell(s, "take_profit", now):
|
|
self._log_fire(s, "take_profit")
|
|
fired.append(s.slot_id)
|
|
return fired
|
|
|
|
# ------------------------------------------------ REQ-039 market-close liquidation
|
|
def check_market_close(self, now: datetime) -> list:
|
|
"""Liquidate ALL holding slots at the configured KST time, once per day."""
|
|
spec = self._cfg.market_close_liquidation_time
|
|
if not spec:
|
|
return [] # feature disabled (AC-3)
|
|
now_kst = now.astimezone(_KST)
|
|
today = now_kst.date()
|
|
if self._last_liquidation_date == today:
|
|
return [] # already fired this trading day (AC-4)
|
|
if now_kst.time() < _parse_kst_time(spec):
|
|
return [] # not yet
|
|
fired = []
|
|
for s in self._holding(now): # watchdog/grace ignored (AC-2/AC-6)
|
|
if self._safe_sell(s, "market_close_liquidation"):
|
|
self._log_fire(s, "market_close_liquidation")
|
|
fired.append(s.slot_id)
|
|
self._last_liquidation_date = today
|
|
return fired
|
|
|
|
# ----------------------------------------------------------------- run_once
|
|
def run_once(self, now: datetime | None = None) -> None:
|
|
"""One worker tick — market-close, then cadence-gated forced_close + prices."""
|
|
now = now or datetime.now(tz=timezone.utc)
|
|
self._prune_cooldown(now) # REQ-052 FR-4: drop state for non-holding slots
|
|
self._update_ownership(now) # dcy-017 FR-1: watchdog ownership enter/exit visibility
|
|
try:
|
|
self.check_market_close(now)
|
|
except Exception: # noqa: BLE001
|
|
_log.warning("market-close liquidation tick failed", exc_info=True)
|
|
if self._due(self._last_forced_close_at, now, self._cfg.forced_close_check_interval_sec):
|
|
self._last_forced_close_at = now
|
|
self.check_forced_close(now)
|
|
if self._due(self._last_price_check_at, now, self._cfg.stop_loss_poll_interval_sec):
|
|
self._last_price_check_at = now
|
|
self.check_prices(now)
|
|
|
|
@staticmethod
|
|
def _due(last: datetime | None, now: datetime, interval_sec: int) -> bool:
|
|
if last is None:
|
|
return True
|
|
return (now - last).total_seconds() >= interval_sec
|