feat(safety-sell): dcy-007 core 자율 안전 매도 — BAT-010 기반 + BAT-011 트리거

dcy-007(core 자율 안전 매도) 전체 delivery. BAT-010 기반 인프라가 미커밋
상태였고 BAT-011 트리거 레이어와 공유 파일에서 섞여 있어 한 커밋으로 묶음.

BAT-010 (REQ-034/035/036/041, release-candidate):
- WatchdogStore(application/watchdog.py) + snapshot RPC 워치독 갱신
- Slot stop_loss_price/take_profit_price 필드 + 004_add_price_thresholds.sql
- SlotService.sell() 자율 매도 내부 경로 + idempotency
- SafetySellConfig env 5종 파싱

BAT-011 (REQ-037/038/039/040/042, release-candidate):
- SafetySellService(application/safety_sell.py): forced_close 만기 / stop-loss·
  take-profit KIS polling / 장 마감 강제 청산 + trex-safety-sell 7번째 스레드
- BrokerPort.get_current_price + KIS inquire-price(TTTC8397R/VTTC8397R, stck_prpr)
  + paper stub
- SlotService.update_thresholds() + intent RPC update_thresholds 액션
  (부분 갱신/null 해제, 워치독 last_seen_at 리셋)
- baseline 3문서 정합(interface-contract/runtime-flows/project-structure)

거버넌스: REQ-034~042 Implemented 동기화, BAT-010/011 release-candidate,
dcy-007 discovery 포함.

검증: 신규 38 테스트 green, 단위 236 passed/23 skipped,
인프라 게이트(PG+RabbitMQ) 259 passed(마이그레이션 001~004 round-trip 포함),
ruff clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

trex_core/adapters/broker/kis.py

@@ -11,6 +11,7 @@ Endpoints: /oauth2/tokenP (token), /uapi/domestic-stock/v1/trading/order-cash
 from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
+from decimal import Decimal, InvalidOperation
 from typing import Optional
 
 import httpx
@@ -119,6 +120,10 @@ class KISBroker:
             return "VTTC0012U" if side == "buy" else "VTTC0011U"
         return "TTTC0012U" if side == "buy" else "TTTC0011U"
 
+    def _tr_price(self) -> str:
+        """tr_id for current-price inquiry (REQ-038 AC-7)."""
+        return "VTTC8397R" if self._mock else "TTTC8397R"
+
     # --- BrokerPort ---
     def submit_order(self, req: OrderRequest) -> OrderAck:
         cano, product = self._account()
@@ -234,6 +239,33 @@ class KISBroker:
             out.append(BalanceItem(symbol=symbol, hldg_qty=qty))
         return out
 
+    def get_current_price(self, symbol: str) -> Optional[Decimal]:
+        """Current traded price `stck_prpr` via KIS inquire-price (REQ-038).
+
+        tr_id: `TTTC8397R` (live) / `VTTC8397R` (mock) per REQ-038 AC-7.
+        Returns None on network/parse error or missing price so the
+        safety-sell worker skips the slot this tick (no false trigger).
+        """
+        params = {"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": symbol}
+        try:
+            resp = httpx.get(
+                f"{self._base_url}/uapi/domestic-stock/v1/quotations/inquire-price",
+                headers=self._headers(self._tr_price()), params=params, timeout=self._timeout,
+            )
+            payload = resp.json()
+        except (httpx.HTTPError, ValueError):
+            return None
+        if str(payload.get("rt_cd")) != "0":
+            return None
+        out = payload.get("output") or {}
+        raw = out.get("stck_prpr")
+        if raw is None or str(raw).strip() == "":
+            return None
+        try:
+            return Decimal(str(raw).strip())
+        except (InvalidOperation, ValueError):
+            return None
+
     def cancel_order(self, broker_order_id: str, *, side: str, symbol: str, qty: int) -> bool:
         cano, product = self._account()
         parts = broker_order_id.split(":")

trex_core/adapters/broker/paper.py

@@ -8,6 +8,7 @@ execution code path that production (KIS) takes.
 from __future__ import annotations
 
 import threading
+from decimal import Decimal
 from typing import Optional
 
 from trex_core.application.broker_port import (
@@ -52,6 +53,8 @@ class PaperBroker:
         self._orders: dict[str, _Order] = {}
         self._seq = 0
         self._lock = threading.Lock()
+        # REQ-038: programmable current price for safety-sell scenario tests.
+        self._prices: dict[str, Decimal] = {}
 
     def name(self) -> str:
         return "paper"
@@ -111,3 +114,14 @@ class PaperBroker:
                 balances[o.symbol] = balances.get(o.symbol, 0) + sign * int(o.filled)
         return [BalanceItem(symbol=sym, hldg_qty=qty)
                 for sym, qty in balances.items() if qty != 0]
+
+    # --- REQ-038: programmable current price (safety-sell scenario double) ---
+    def set_current_price(self, symbol: str, price: Decimal) -> None:
+        """Program the current price returned by get_current_price(symbol)."""
+        with self._lock:
+            self._prices[symbol] = Decimal(str(price))
+
+    def get_current_price(self, symbol: str) -> Optional[Decimal]:
+        """Return the programmed current price for symbol, or None if unset."""
+        with self._lock:
+            return self._prices.get(symbol)

trex_core/adapters/persistence/postgresql.py

@@ -29,7 +29,8 @@ from trex_core.domain.slot import Slot, SlotNotFoundError, SlotState, SlotTransi
 _SLOT_COLUMNS = (
     "slot_id, trading_date, status, symbol, qty, avg_price, entry_price, "
     "allocated_capital, order_id, strategy_id, broker_order_id, portfolio_id, "
-    "sell_strategy_id, forced_close_at, created_at, updated_at"
+    "sell_strategy_id, forced_close_at, stop_loss_price, take_profit_price, "
+    "created_at, updated_at"
 )
 
 
@@ -122,12 +123,14 @@ class PostgreSQLSlotRepository:
                     "UPDATE daily_slot SET status=%s, symbol=%s, qty=%s, avg_price=%s, "
                     "entry_price=%s, allocated_capital=%s, order_id=%s, strategy_id=%s, "
                     "broker_order_id=%s, portfolio_id=%s, sell_strategy_id=%s, "
-                    "forced_close_at=%s, updated_at=%s WHERE slot_id=%s",
+                    "forced_close_at=%s, stop_loss_price=%s, take_profit_price=%s, "
+                    "updated_at=%s WHERE slot_id=%s",
                     (
                         slot.status.value, slot.symbol, slot.qty, slot.avg_price,
                         slot.entry_price, slot.allocated_capital, slot.order_id,
                         slot.strategy_id, slot.broker_order_id, slot.portfolio_id,
                         slot.sell_strategy_id, slot.forced_close_at,
+                        slot.stop_loss_price, slot.take_profit_price,
                         slot.updated_at or datetime.now(tz=timezone.utc), slot.slot_id,
                     ),
                 )
@@ -229,7 +232,14 @@ class PostgreSQLSlotRepository:
 
     @staticmethod
     def _row_to_slot(row: tuple) -> Slot:
+        from decimal import Decimal
         f = lambda v: float(v) if v is not None else None  # noqa: E731
+        d = lambda v: Decimal(str(v)) if v is not None else None  # noqa: E731
+        # Column order: slot_id(0) trading_date(1) status(2) symbol(3) qty(4)
+        # avg_price(5) entry_price(6) allocated_capital(7) order_id(8)
+        # strategy_id(9) broker_order_id(10) portfolio_id(11) sell_strategy_id(12)
+        # forced_close_at(13) stop_loss_price(14) take_profit_price(15)
+        # created_at(16) updated_at(17)
         return Slot(
             slot_id=row[0] if isinstance(row[0], UUID) else UUID(str(row[0])),
             trading_date=row[1],
@@ -245,8 +255,10 @@ class PostgreSQLSlotRepository:
             portfolio_id=row[11],
             sell_strategy_id=row[12],
             forced_close_at=row[13],
-            created_at=row[14],
-            updated_at=row[15],
+            stop_loss_price=d(row[14]),
+            take_profit_price=d(row[15]),
+            created_at=row[16],
+            updated_at=row[17],
         )
 
 

trex_core/adapters/persistence/sql/004_add_price_thresholds.sql

@@ -0,0 +1,4 @@
+-- BAT-010 / REQ-035: additive price threshold columns for stop-loss / take-profit.
+-- Existing rows get NULL, which is the correct default (no threshold set).
+ALTER TABLE daily_slot ADD COLUMN IF NOT EXISTS stop_loss_price  NUMERIC;
+ALTER TABLE daily_slot ADD COLUMN IF NOT EXISTS take_profit_price NUMERIC;

trex_core/application/broker_port.py

@@ -8,6 +8,7 @@ trex_core.adapters.broker. Fills are observed asynchronously via poll_fill
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from decimal import Decimal
 from typing import Any, Optional, Protocol
 
 
@@ -90,6 +91,15 @@ class BrokerPort(Protocol):
         """
         ...
 
+    def get_current_price(self, symbol: str) -> Optional[Decimal]:
+        """Return the current traded price (`stck_prpr`) for a symbol (REQ-038).
+
+        Used by the safety-sell worker to evaluate stop-loss/take-profit
+        thresholds. Returns None when the price is unavailable (network error,
+        unknown symbol, or — for the paper double — no programmed price).
+        """
+        ...
+
 
 # --- shared fee model (REQ-018, orbit calculate_fee pattern) -----------------
 # Conservative KR equities approximation (bps). Brokers can override, but we

trex_core/application/intents.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 from datetime import date, datetime
+from decimal import Decimal
 from typing import Any, Optional
 from uuid import UUID
 
@@ -54,6 +55,8 @@ class BuyIntent:
     sell_strategy_id: Optional[str] = None
     portfolio_id: Optional[str] = None
     forced_close_at: Optional[datetime] = None
+    stop_loss_price: Optional[Decimal] = None   # REQ-035
+    take_profit_price: Optional[Decimal] = None  # REQ-035
     extra: dict[str, Any] = field(default_factory=dict)
 
 
@@ -157,6 +160,8 @@ class IntentService:
                     sell_strategy_id=intent.sell_strategy_id,
                     portfolio_id=intent.portfolio_id,
                     forced_close_at=intent.forced_close_at,
+                    stop_loss_price=intent.stop_loss_price,
+                    take_profit_price=intent.take_profit_price,
                     extra={"intent_id": intent.intent_id, "symbol": intent.symbol},
                 )
             except SlotStateMismatchError:

trex_core/application/messaging_handlers.py

@@ -20,14 +20,25 @@ from trex_core.application.portfolio import (
     View,
     build_snapshot_envelope,
 )
-from trex_core.domain.slot import SlotRepository, SlotService
+from trex_core.application.watchdog import WatchdogStore
+from trex_core.domain.slot import (
+    SlotNotFoundError,
+    SlotRepository,
+    SlotService,
+    SlotState,
+)
 
 _log = logging.getLogger(__name__)
 
 
 # ---- intent RPC handler -----------------------------------------------------
 
-def make_intent_rpc_handler(intent_service: IntentService):
+def make_intent_rpc_handler(
+    intent_service: IntentService,
+    *,
+    slot_service: Optional[SlotService] = None,
+    watchdog_store: Optional[WatchdogStore] = None,
+):
     """Build the RPC handler for `trex.intent.req`.
 
     Payload shape:
@@ -36,9 +47,14 @@ def make_intent_rpc_handler(intent_service: IntentService):
           "intent_id":   "<idempotency key, = correlation_id>",
           ... fields per BuyIntent / SellIntent ...
         }
+      or (REQ-040):
+        {"action": "update_thresholds", "slot_id": ...,
+         "stop_loss_price"?, "take_profit_price"?, "forced_close_at"?}
 
     Response shape:
         {"intent_id": str, "status": "admitted"|"rejected", "slot_id": str|None, "reason": str|None}
+      or for update_thresholds:
+        {"status": "updated"|"ignored"|"error", "slot_id": str, "reason"?: str}
 
     Dedup uses `intent_log.try_insert(intent_id, ...)` (REQ-007 gate). The
     request's `correlation_id` is reused as `intent_id` so the messaging
@@ -48,6 +64,9 @@ def make_intent_rpc_handler(intent_service: IntentService):
 
     def handler(req: RpcRequest) -> RpcResponse:
         body = req.payload or {}
+        # REQ-040: update_thresholds action (separate from buy/sell intents).
+        if str(body.get("action") or "").lower() == "update_thresholds":
+            return _handle_update_thresholds(body, slot_service, watchdog_store)
         intent_id = str(body.get("intent_id") or req.correlation_id)
         intent_type = str(body.get("intent_type") or "").lower()
         try:
@@ -63,6 +82,8 @@ def make_intent_rpc_handler(intent_service: IntentService):
                     sell_strategy_id=_opt_str(body.get("sell_strategy_id")),
                     portfolio_id=_opt_str(body.get("portfolio_id")),
                     forced_close_at=_opt_dt(body.get("forced_close_at")),
+                    stop_loss_price=_opt_decimal(body.get("stop_loss_price")),
+                    take_profit_price=_opt_decimal(body.get("take_profit_price")),
                 )
                 d = intent_service.submit_buy(buy)
             elif intent_type == "sell":
@@ -93,6 +114,46 @@ def make_intent_rpc_handler(intent_service: IntentService):
     return handler
 
 
+def _handle_update_thresholds(
+    body: dict,
+    slot_service: Optional[SlotService],
+    watchdog_store: Optional[WatchdogStore],
+) -> RpcResponse:
+    """REQ-040: update stop_loss/take_profit/forced_close_at on a holding slot.
+
+    Only the keys actually present in the payload are forwarded, so an absent
+    field keeps its value while an explicit ``null`` clears it (AC-4 vs AC-5).
+    On success the slot's watchdog last_seen_at is reset (AC-3) — receiving an
+    update_thresholds is itself a strategy-module liveness signal.
+    """
+    if slot_service is None:
+        return RpcResponse(payload={"status": "error", "reason": "update_thresholds_unsupported"})
+    try:
+        slot_id = UUID(str(body["slot_id"]))
+    except (KeyError, ValueError) as exc:
+        return RpcResponse(payload={"status": "error", "reason": f"bad_request:{exc}"})
+    kwargs: dict = {}
+    if "stop_loss_price" in body:
+        kwargs["stop_loss_price"] = _opt_decimal(body["stop_loss_price"])
+    if "take_profit_price" in body:
+        kwargs["take_profit_price"] = _opt_decimal(body["take_profit_price"])
+    if "forced_close_at" in body:
+        kwargs["forced_close_at"] = _opt_dt(body["forced_close_at"])
+    actor = str(body.get("actor") or "strategy")
+    try:
+        status = slot_service.update_thresholds(slot_id, actor=actor, **kwargs)
+    except SlotNotFoundError:
+        return RpcResponse(payload={
+            "status": "ignored", "slot_id": str(slot_id), "reason": "slot_not_found",
+        })
+    if status == "updated" and watchdog_store is not None:
+        watchdog_store.update(str(slot_id))  # AC-3: liveness signal
+    payload = {"status": status, "slot_id": str(slot_id)}
+    if status == "ignored":
+        payload["reason"] = "not_holding"  # AC-6
+    return RpcResponse(payload=payload)
+
+
 # ---- snapshot RPC handler ---------------------------------------------------
 
 def make_snapshot_rpc_handler(
@@ -102,6 +163,7 @@ def make_snapshot_rpc_handler(
     snapshot_store: Optional[SnapshotStore] = None,
     pause_store: Optional[PauseStore] = None,
     default_total_slots: int = 10,
+    watchdog_store: Optional[WatchdogStore] = None,
 ):
     """Build the RPC handler for `trex.snapshot.req` (REQ-028).
 
@@ -126,6 +188,12 @@ def make_snapshot_rpc_handler(
         body = req.payload or {}
         day = _parse_date(body.get("trading_date")) if body.get("trading_date") else _today()
 
+        # REQ-034: update watchdog for all holding slots visible in this RPC.
+        if watchdog_store is not None:
+            for s in slot_repo.list_slots(day):
+                if s.status is SlotState.HOLDING:
+                    watchdog_store.update(str(s.slot_id))
+
         # --- BAT-008 envelope path (snapshot store wired) ---
         if snapshot_store is not None and pause_store is not None:
             requested_view = str(body.get("view") or "all").lower()
@@ -244,6 +312,13 @@ def _opt_str(v) -> Optional[str]:
     return str(v)
 
 
+def _opt_decimal(v):
+    if v is None:
+        return None
+    from decimal import Decimal
+    return Decimal(str(v))
+
+
 def _opt_float(v) -> Optional[float]:
     if v is None:
         return None

trex_core/application/safety_sell.py

@@ -0,0 +1,190 @@
+"""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,
+    ) -> None:
+        self._repo = slot_repo
+        self._svc = slot_service
+        self._broker = broker
+        self._watchdog = watchdog
+        self._cfg = config
+        self._last_liquidation_date: date_cls | None = None
+        self._last_forced_close_at: datetime | None = None
+        self._last_price_check_at: datetime | None = None
+
+    @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
+
+    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 and self._safe_sell(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:
+            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._safe_sell(s, "stop_loss"):
+                    fired.append(s.slot_id)
+            elif s.take_profit_price is not None and price >= s.take_profit_price:
+                if self._safe_sell(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"):
+                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)
+        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

trex_core/application/watchdog.py

@@ -0,0 +1,49 @@
+"""Per-slot watchdog store for safety-sell (REQ-034 / BAT-010).
+
+WatchdogStore tracks the last time a snapshot RPC response included each
+holding slot. BAT-011 trigger loops use is_expired() to decide whether a
+slot is unmonitored (strategy module absent) and therefore eligible for
+autonomous safety-sell actions.
+
+Thread-safety: dict reads and writes in CPython are GIL-atomic for simple
+assignments, which is sufficient here — one writer (snapshot-rpc thread)
+and one or more readers (BAT-011 trigger-loop threads). No explicit lock
+is needed.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+
+class WatchdogStore:
+    """In-memory per-slot last_seen_at tracker (REQ-034)."""
+
+    def __init__(self) -> None:
+        self._store: dict[str, datetime] = {}
+        self._started_at: datetime = datetime.now(tz=timezone.utc)
+
+    def update(self, slot_id: str) -> None:
+        """Record that slot_id appeared in a snapshot RPC response."""
+        self._store[slot_id] = datetime.now(tz=timezone.utc)
+
+    def is_expired(self, slot_id: str, timeout_sec: int) -> bool:
+        """Return True if slot_id has not been seen for longer than timeout_sec.
+
+        A slot that has never been seen is treated as expired (returns True)
+        unless in_grace_period() is True — callers must check grace period
+        before calling this method when they want to suppress false positives.
+        """
+        last = self._store.get(slot_id)
+        if last is None:
+            return True
+        elapsed = (datetime.now(tz=timezone.utc) - last).total_seconds()
+        return elapsed > timeout_sec
+
+    def in_grace_period(self, grace_sec: int) -> bool:
+        """Return True if the daemon started less than grace_sec seconds ago.
+
+        During the grace period ALL expiry judgements must be deferred —
+        no slot should be treated as watchdog-expired (REQ-034 AC-4).
+        """
+        elapsed = (datetime.now(tz=timezone.utc) - self._started_at).total_seconds()
+        return elapsed < grace_sec

trex_core/domain/slot/models.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 from datetime import date, datetime
+from decimal import Decimal
 from typing import Optional
 from uuid import UUID
 
@@ -32,6 +33,8 @@ class Slot:
     portfolio_id: Optional[str] = None  # BAT-006 / REQ-025 (trex-orbit lineage)
     sell_strategy_id: Optional[str] = None  # BAT-008 / REQ-047 (trex-orbit lineage)
     forced_close_at: Optional[datetime] = None  # BAT-014 / REQ-066 (trex-orbit lineage)
+    stop_loss_price: Optional[Decimal] = None   # REQ-035 / BAT-010
+    take_profit_price: Optional[Decimal] = None  # REQ-035 / BAT-010
     created_at: Optional[datetime] = None
     updated_at: Optional[datetime] = None
 
@@ -55,6 +58,8 @@ EXPECTED_SLOT_FIELDS: frozenset[str] = frozenset({
     "portfolio_id",
     "sell_strategy_id",
     "forced_close_at",
+    "stop_loss_price",    # REQ-035
+    "take_profit_price",  # REQ-035
     "created_at",
     "updated_at",
 })

trex_core/domain/slot/service.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 
 from dataclasses import replace
 from datetime import date, datetime, timezone
+from decimal import Decimal
 from typing import Any, Optional
 from uuid import UUID
 
@@ -24,6 +25,10 @@ from trex_core.domain.slot.models import Slot, SlotCapacity, SlotTransitionRecor
 from trex_core.domain.slot.repository import SlotRepository
 from trex_core.domain.slot.state import SLOT_ADVISORY_LOCK_ID, SlotState, is_allowed_transition
 
+# Sentinel distinguishing "field omitted (keep current)" from "field set to
+# None (clear threshold)" in update_thresholds (REQ-040 AC-4 vs AC-5).
+_UNSET: Any = object()
+
 
 class SlotService:
     """Slot lifecycle service. Race-free transitions via advisory lock + transaction."""
@@ -69,6 +74,8 @@ class SlotService:
         portfolio_id: Optional[str] = None,
         sell_strategy_id: Optional[str] = None,
         forced_close_at: Optional[datetime] = None,
+        stop_loss_price: Optional[Decimal] = None,
+        take_profit_price: Optional[Decimal] = None,
         extra: Optional[dict[str, Any]] = None,
     ) -> Slot:
         """Transition slot from one state to another atomically.
@@ -115,6 +122,8 @@ class SlotService:
                 portfolio_id=portfolio_id,
                 sell_strategy_id=sell_strategy_id,
                 forced_close_at=forced_close_at,
+                stop_loss_price=stop_loss_price,
+                take_profit_price=take_profit_price,
             )
             self._repo.update_slot(updated)
 
@@ -155,6 +164,8 @@ class SlotService:
         portfolio_id: Optional[str] = None,
         sell_strategy_id: Optional[str] = None,
         forced_close_at: Optional[datetime] = None,
+        stop_loss_price: Optional[Decimal] = None,
+        take_profit_price: Optional[Decimal] = None,
     ) -> Slot:
         """Compute the post-transition Slot row.
 
@@ -181,6 +192,8 @@ class SlotService:
                 portfolio_id=None,
                 sell_strategy_id=None,
                 forced_close_at=None,
+                stop_loss_price=None,
+                take_profit_price=None,
                 updated_at=datetime.now(tz=timezone.utc),
             )
 
@@ -211,9 +224,86 @@ class SlotService:
             forced_close_at=(
                 forced_close_at if forced_close_at is not None else slot.forced_close_at
             ),
+            stop_loss_price=(
+                stop_loss_price if stop_loss_price is not None else slot.stop_loss_price
+            ),
+            take_profit_price=(
+                take_profit_price if take_profit_price is not None else slot.take_profit_price
+            ),
             updated_at=datetime.now(tz=timezone.utc),
         )
 
+    # ------------------------------------------------------------
+    # sell  (REQ-036 / BAT-010)
+    # ------------------------------------------------------------
+    def sell(self, slot_id: UUID, actor: str, reason: str) -> None:
+        """Trigger an autonomous safety-sell for a holding slot (REQ-036).
+
+        Idempotent: silently returns if the slot is already pending_sell.
+        Raises SlotNotFoundError if slot_id is unknown.
+        Raises SlotTransitionError if the slot is neither holding nor pending_sell.
+        """
+        slot = self._repo.fetch_slot(slot_id)
+        if slot is None:
+            raise SlotNotFoundError(f"slot not found: {slot_id}")
+        if slot.status is SlotState.PENDING_SELL:
+            return  # already in progress — idempotent (REQ-036 AC-2)
+        if slot.status is not SlotState.HOLDING:
+            raise SlotTransitionError(
+                f"sell() requires holding state, got: {slot.status.value}"
+            )
+        self.transit_slot(
+            slot_id,
+            SlotState.HOLDING,
+            SlotState.PENDING_SELL,
+            reason=reason,
+            actor=actor,
+        )
+
+    # ------------------------------------------------------------
+    # update_thresholds  (REQ-040 / BAT-011)
+    # ------------------------------------------------------------
+    def update_thresholds(
+        self,
+        slot_id: UUID,
+        *,
+        stop_loss_price: Any = _UNSET,
+        take_profit_price: Any = _UNSET,
+        forced_close_at: Any = _UNSET,
+        actor: str = "strategy",
+    ) -> str:
+        """Partially update a holding slot's safety-sell thresholds (REQ-040).
+
+        Each of the three fields is independent: pass `_UNSET` (the default —
+        i.e. omit the kwarg) to keep the current value, or pass an explicit
+        value (including ``None``) to set/clear it (AC-4 vs AC-5).
+
+        Returns "updated" on success, or "ignored" when the slot is not in
+        the holding state (AC-6). Raises SlotNotFoundError for unknown slots.
+        The caller (RPC handler) resets the watchdog last_seen_at (AC-3).
+        """
+        with self._repo.transactional_advisory_lock(SLOT_ADVISORY_LOCK_ID):
+            slot = self._repo.fetch_slot(slot_id)
+            if slot is None:
+                raise SlotNotFoundError(f"slot not found: {slot_id}")
+            if slot.status is not SlotState.HOLDING:
+                return "ignored"  # not_holding (AC-6)
+            updated = replace(
+                slot,
+                stop_loss_price=(
+                    slot.stop_loss_price if stop_loss_price is _UNSET else stop_loss_price
+                ),
+                take_profit_price=(
+                    slot.take_profit_price if take_profit_price is _UNSET else take_profit_price
+                ),
+                forced_close_at=(
+                    slot.forced_close_at if forced_close_at is _UNSET else forced_close_at
+                ),
+                updated_at=datetime.now(tz=timezone.utc),
+            )
+            self._repo.update_slot(updated)
+            return "updated"
+
     # ------------------------------------------------------------
     # get_today_entry_count
     # ------------------------------------------------------------

trex_core/runtime/config.py

@@ -69,6 +69,22 @@ class SnapshotScheduleConfig:
     tick_sec: float = 60.0
 
 
+@dataclass(frozen=True)
+class SafetySellConfig:
+    """Safety-sell feature tuning (REQ-041 / BAT-010).
+
+    All five env vars are parsed here even though only WATCHDOG_TIMEOUT and
+    WATCHDOG_GRACE_PERIOD are consumed in BAT-010; the remaining three are
+    used by BAT-011 trigger loops once they land.
+    """
+
+    watchdog_timeout_sec: int = 120
+    watchdog_grace_period_sec: int = 60
+    forced_close_check_interval_sec: int = 60
+    stop_loss_poll_interval_sec: int = 30
+    market_close_liquidation_time: Optional[str] = None  # "HH:MM" KST, optional
+
+
 @dataclass(frozen=True)
 class DaemonConfig:
     dsn: str
@@ -87,6 +103,8 @@ class DaemonConfig:
     rabbitmq: Optional[RabbitMQConfig] = None
     # REQ-030 (BAT-008): daily snapshot refresh.
     snapshot_schedule: SnapshotScheduleConfig = SnapshotScheduleConfig()
+    # REQ-041 (BAT-010): safety-sell feature config.
+    safety_sell: SafetySellConfig = SafetySellConfig()
 
     @staticmethod
     def from_env() -> "DaemonConfig":
@@ -136,6 +154,18 @@ class DaemonConfig:
             kst_time=os.environ.get("TREX_SNAPSHOT_SCHEDULE_KST", "07:30").strip() or "07:30",
             tick_sec=float(os.environ.get("TREX_SNAPSHOT_TICK_SEC", "60")),
         )
+        _mc_time = os.environ.get("TREX_MARKET_CLOSE_LIQUIDATION_TIME", "").strip() or None
+        safety_sell = SafetySellConfig(
+            watchdog_timeout_sec=int(os.environ.get("TREX_WATCHDOG_TIMEOUT", "120")),
+            watchdog_grace_period_sec=int(os.environ.get("TREX_WATCHDOG_GRACE_PERIOD", "60")),
+            forced_close_check_interval_sec=int(
+                os.environ.get("TREX_FORCED_CLOSE_CHECK_INTERVAL", "60")
+            ),
+            stop_loss_poll_interval_sec=int(
+                os.environ.get("TREX_STOP_LOSS_POLL_INTERVAL", "30")
+            ),
+            market_close_liquidation_time=_mc_time,
+        )
         return DaemonConfig(
             dsn=dsn,
             total_slots=int(os.environ.get("TREX_TOTAL_SLOTS", "10")),
@@ -147,4 +177,5 @@ class DaemonConfig:
             ledger_interval_sec=float(os.environ.get("TREX_LEDGER_INTERVAL_SEC", "1.0")),
             rabbitmq=rabbitmq,
             snapshot_schedule=snapshot_schedule,
+            safety_sell=safety_sell,
         )

trex_core/runtime/daemon.py

@@ -5,9 +5,12 @@ strategy-facing communication runs over RabbitMQ (REQ-021/022/023).
 
 Threads (started in `run()`):
   - intent RPC consumer    — `trex.intent.req`,  prefetch=1, FIFO, dedup via intent_log
-  - snapshot RPC consumer  — `trex.snapshot.req`
+  - snapshot RPC consumer  — `trex.snapshot.req` (also refreshes the watchdog)
+  - pause RPC consumer      — `trex.pause.req` (REQ-032)
   - execution loop         — broker dispatch + poll + slot transitions + event emit
   - ledger writer          — ledger outbox drain (REQ-019)
+  - snapshot scheduler      — daily portfolio-hub snapshot refresh (REQ-030)
+  - safety-sell worker      — forced_close / stop-loss·take-profit / market-close (REQ-037/038/039)
 
 Each consumer/publisher uses its OWN RabbitMQTransport instance because
 pika BlockingConnection is not thread-safe across channels.
@@ -56,7 +59,9 @@ from trex_core.application.portfolio import (
     PortfolioHubReadPort,
     SnapshotStore,
 )
+from trex_core.application.safety_sell import SafetySellService
 from trex_core.application.snapshot_scheduler import SnapshotScheduler
+from trex_core.application.watchdog import WatchdogStore
 from trex_core.domain.slot import SlotService
 from trex_core.runtime.config import DaemonConfig
 
@@ -140,6 +145,8 @@ class Daemon:
 
         # BAT-008: portfolio-hub read + snapshot store + durable pause + scheduler.
         self.snapshot_store = SnapshotStore()
+        # BAT-010/011: per-slot watchdog (snapshot RPC writer, safety-sell reader).
+        self.watchdog = WatchdogStore()
         self.pause_store = PostgreSQLPauseStore(config.dsn)
         self.portfolio_read = _build_portfolio_read(config)
         self.snapshot_scheduler = SnapshotScheduler(
@@ -160,6 +167,15 @@ class Daemon:
             event_emitter=self._make_event_emitter(),
         )
 
+        # BAT-011: safety-sell trigger service (forced_close / SL-TP / market-close).
+        self.safety_sell = SafetySellService(
+            slot_repo=self.slot_repo,
+            slot_service=self.slot_service,
+            broker=self.broker,
+            watchdog=self.watchdog,
+            config=config.safety_sell,
+        )
+
     # ------------------------------------------------------------ event hook
     def _make_event_emitter(self):
         def emit(event_type: str, slot) -> None:
@@ -195,7 +211,11 @@ class Daemon:
             transport.declare_topology()
         except Exception:  # noqa: BLE001
             pass
-        handler = make_intent_rpc_handler(self.intent_service)
+        handler = make_intent_rpc_handler(
+            self.intent_service,
+            slot_service=self.slot_service,
+            watchdog_store=self.watchdog,
+        )
         transport.consume_rpc_forever(topology.INTENT_REQ_QUEUE, handler, self.stop)
 
     def _snapshot_rpc_loop(self) -> None:
@@ -210,6 +230,7 @@ class Daemon:
             snapshot_store=self.snapshot_store,
             pause_store=self.pause_store,
             default_total_slots=self.config.total_slots,
+            watchdog_store=self.watchdog,
         )
         transport.consume_rpc_forever(topology.SNAPSHOT_REQ_QUEUE, handler, self.stop)
 
@@ -223,6 +244,14 @@ class Daemon:
         handler = make_pause_rpc_handler(self.pause_store)
         transport.consume_rpc_forever(topology.PAUSE_REQ_QUEUE, handler, self.stop)
 
+    def _safety_sell_loop(self) -> None:
+        # REQ-037/038/039: forced_close / stop-loss·take-profit / market-close.
+        while not self.stop.wait(self.safety_sell.base_tick_sec):
+            try:
+                self.safety_sell.run_once()
+            except Exception:  # noqa: BLE001 - loop must survive transient errors
+                continue
+
     def _snapshot_scheduler_loop(self) -> None:
         try:
             self.snapshot_scheduler.run_forever(
@@ -251,6 +280,7 @@ class Daemon:
             ("trex-snapshot-rpc", self._snapshot_rpc_loop),
             ("trex-pause-rpc", self._pause_rpc_loop),
             ("trex-snapshot-scheduler", self._snapshot_scheduler_loop),
+            ("trex-safety-sell", self._safety_sell_loop),
         ]
         for name, target in workers:
             t = threading.Thread(target=target, name=name, daemon=True)