analysis: apply execution-adjusted cost model in v2 backtest pipeline (#368)

Reviewed-on: #386

docs/ouroboros/01_requirements_registry.md

@@ -1,6 +1,6 @@
 <!--
 Doc-ID: DOC-REQ-001
-Version: 1.0.3
+Version: 1.0.7
 Status: active
 Owner: strategy
 Updated: 2026-03-02
@@ -26,7 +26,7 @@ Updated: 2026-03-02
 - `REQ-V3-001`: 모든 신호/주문/로그는 `session_id`를 포함해야 한다.
 - `REQ-V3-002`: 세션 전환 시 리스크 파라미터 재로딩이 수행되어야 한다.
 - `REQ-V3-003`: 브로커 블랙아웃 시간대에는 신규 주문이 금지되어야 한다.
-- `REQ-V3-004`: 블랙아웃 중 신호는 Queue에 적재되고, 복구 후 유효성 재검증을 거친다.
+- `REQ-V3-004`: 블랙아웃 중 신호는 bounded Queue에 적재되며, 포화 시 oldest-drop 정책으로 최신 intent를 보존하고 복구 후 유효성 재검증을 거친다.
 - `REQ-V3-005`: 저유동 세션(`NXT_AFTER`, `US_PRE`, `US_DAY`, `US_AFTER`)은 시장가 주문 금지다.
 - `REQ-V3-006`: 백테스트 체결가는 불리한 방향 체결 가정을 기본으로 한다.
 - `REQ-V3-007`: US 운용은 환율 손익 분리 추적과 통화 버퍼 정책을 포함해야 한다.

docs/ouroboros/80_implementation_audit.md

@@ -43,12 +43,12 @@ Updated: 2026-03-02
 | REQ-ID | 요구사항 | 상태 | 비고 |
 |--------|----------|------|------|
 | REQ-V3-001 | 모든 신호/주문/로그에 session_id 포함 | ⚠️ 부분 | 큐 intent에 `session_id` 누락 (`#375`) |
-| REQ-V3-002 | 세션 전환 훅 + 리스크 파라미터 재로딩 | ⚠️ 부분 | 구현 존재, 세션 경계 E2E 회귀 보강 필요 (`#376`) |
+| REQ-V3-002 | 세션 전환 훅 + 리스크 파라미터 재로딩 | ✅ 완료 | 세션 경계 E2E 회귀(override 적용/해제 + 재로딩 실패 폴백) 보강 (`#376`) |
 | REQ-V3-003 | 블랙아웃 윈도우 정책 | ✅ 완료 | `src/core/blackout_manager.py` |
-| REQ-V3-004 | 블랙아웃 큐 + 복구 시 재검증 | ⚠️ 부분 | 큐 포화 시 intent 유실 경로 존재 (`#371`), 재검증 강화를 `#328`에서 추적 |
+| REQ-V3-004 | 블랙아웃 큐 + 복구 시 재검증 | ⚠️ 부분 | 큐 포화는 oldest-drop 정책으로 정합화 (`#371`), 재검증 강화는 `#328` 추적 |
 | REQ-V3-005 | 저유동 세션 시장가 금지 | ✅ 완료 | `src/core/order_policy.py` |
 | REQ-V3-006 | 보수적 백테스트 체결 (불리 방향) | ✅ 완료 | `src/analysis/backtest_execution_model.py` |
-| REQ-V3-007 | FX 손익 분리 (전략 PnL vs 환율 PnL) | ⚠️ 부분 | 스키마 존재, 런타임 분리 계산/전달 미적용 (`#370`) |
+| REQ-V3-007 | FX 손익 분리 (전략 PnL vs 환율 PnL) | ⚠️ 부분 | 런타임 분리 계산/전달 적용 (`#370`), buy-side `fx_rate` 미관측 시 `fx_pnl=0` fallback |
 | REQ-V3-008 | 오버나잇 예외 vs Kill Switch 우선순위 | ✅ 완료 | `src/main.py` — `_should_force_exit_for_overnight()`, `_apply_staged_exit_override_for_hold()` |
 
 ### 1.4 운영 거버넌스: 부분 완료 (2026-03-02 재평가)
@@ -80,13 +80,13 @@ Updated: 2026-03-02
 - **해소**: #326 머지 — `log_trade()` 호출 시 런타임 `session_id` 명시적 전달
 - **요구사항**: REQ-V3-001
 
-### GAP-3: 세션 전환 시 리스크 파라미터 재로딩 없음 → ⚠️ 부분 해소 (#327)
+### GAP-3: 세션 전환 시 리스크 파라미터 재로딩 없음 → ✅ 해소 (#327, #376)
 
 - **위치**: `src/main.py`, `src/config.py`
 - **해소 내용**: #327 머지 — `SESSION_RISK_PROFILES_JSON` 기반 세션별 파라미터 재로딩 메커니즘 구현
   - `SESSION_RISK_RELOAD_ENABLED=true` 시 세션 경계에서 파라미터 재로딩
   - 재로딩 실패 시 기존 파라미터 유지 (안전 폴백)
-- **잔여 갭**: 세션 경계 실시간 전환 E2E 통합 테스트 보강 필요 (`test_main.py`에 설정 오버라이드/폴백 단위 테스트는 존재)
+- **해소**: 세션 경계 E2E 회귀 테스트를 추가해 override 적용/해제, 재로딩 실패 시 폴백 유지를 검증함 (`#376`)
 - **요구사항**: REQ-V3-002
 
 ### GAP-4: 블랙아웃 복구 DB 기록 + 재검증 → ⚠️ 부분 해소 (#324, #328, #371)
@@ -95,7 +95,7 @@ Updated: 2026-03-02
 - **현 상태**:
   - #324 추적 범위(DB 기록)는 구현 경로가 존재
   - #328 범위(가격/세션 재검증 강화)는 추적 이슈 오픈 상태
-  - #371: 큐 포화 시 intent 유실 경로가 남아 있어 `REQ-V3-004`를 완료로 보기 어려움
+  - #371: 큐 포화 정책을 oldest-drop으로 명시/구현해 최신 intent 유실 경로를 제거
 - **요구사항**: REQ-V3-004
 
 ### GAP-5: 시간장벽이 봉 개수 고정 → ✅ 해소 (#329)
@@ -107,10 +107,12 @@ Updated: 2026-03-02
   - `max_holding_bars` deprecated 경고 유지 (하위 호환)
 - **요구사항**: REQ-V2-005 / v3 확장
 
-### GAP-6 (신규): FX PnL 분리 미완료 (MEDIUM — 부분 구현)
+### GAP-6 (신규): FX PnL 분리 부분 해소 (MEDIUM)
 
 - **위치**: `src/db.py` (`fx_pnl`, `strategy_pnl` 컬럼 존재)
-- **문제**: 스키마와 함수는 존재하지만 런타임 경로에서 `strategy_pnl`/`fx_pnl` 분리 계산 전달이 누락됨 (`#370`)
+- **현 상태**: 런타임 SELL 경로에서 `strategy_pnl`/`fx_pnl` 분리 계산 및 전달을 적용함 (`#370`).
+- **운영 메모**: `trading_cycle`은 scanner 기반 `selection_context`에 `fx_rate`를 추가하고, `run_daily_session`은 scanner 컨텍스트 없이 `fx_rate` 스냅샷만 기록한다.
+- **잔여**: 과거 BUY 레코드에 `fx_rate`가 없으면 해외 구간도 `fx_pnl=0` fallback으로 기록됨.
 - **영향**: USD 거래에서 환율 손익과 전략 손익이 분리되지 않아 성과 분석 부정확
 - **요구사항**: REQ-V3-007
 
@@ -392,8 +394,7 @@ Phase 3 (중기): v3 세션 최적화
 
 ### 테스트 미존재 (잔여)
 
-- ❌ 세션 전환 훅 콜백 (GAP-3 잔여)
-- ❌ 세션 경계 리스크 파라미터 재로딩 단위 테스트 (GAP-3 잔여)
+- ✅ 세션 전환 훅 콜백/세션 경계 리스크 재로딩 E2E 회귀 (`#376`)
 - ❌ 실거래 경로 ↔ v2 상태기계 통합 테스트 (피처 공급 포함)
 - ❌ FX PnL 운영 활성화 검증 (GAP-6)
 

src/analysis/backtest_cost_guard.py

@@ -32,7 +32,7 @@ def validate_backtest_cost_model(
 
     slippage = model.slippage_bps_by_session or {}
     failure = model.failure_rate_by_session or {}
-    partial = model.partial_fill_rate_by_session or {}
+    partial_fill = model.partial_fill_rate_by_session or {}
 
     missing_slippage = [s for s in required_sessions if s not in slippage]
     if missing_slippage:
@@ -45,11 +45,12 @@ def validate_backtest_cost_model(
         raise ValueError(
             f"missing failure_rate_by_session for sessions: {', '.join(missing_failure)}"
         )
-    missing_partial = [s for s in required_sessions if s not in partial]
-    if missing_partial:
+
+    missing_partial_fill = [s for s in required_sessions if s not in partial_fill]
+    if missing_partial_fill:
         raise ValueError(
             "missing partial_fill_rate_by_session for sessions: "
-            f"{', '.join(missing_partial)}"
+            f"{', '.join(missing_partial_fill)}"
         )
 
     for sess, bps in slippage.items():
@@ -58,6 +59,6 @@ def validate_backtest_cost_model(
     for sess, rate in failure.items():
         if not math.isfinite(rate) or rate < 0 or rate > 1:
             raise ValueError(f"failure rate must be within [0,1] for session={sess}")
-    for sess, rate in partial.items():
+    for sess, rate in partial_fill.items():
         if not math.isfinite(rate) or rate < 0 or rate > 1:
             raise ValueError(f"partial fill rate must be within [0,1] for session={sess}")

src/analysis/backtest_pipeline.py

@@ -45,6 +45,7 @@ class WalkForwardConfig:
 class BaselineScore:
     name: Literal["B0", "B1", "M1"]
     accuracy: float
+    cost_adjusted_accuracy: float
 
 
 @dataclass(frozen=True)
@@ -132,6 +133,8 @@ def run_v2_backtest_pipeline(
         ).label
 
     ordered_labels = [labels_by_bar_index[idx] for idx in normalized_entries]
+    ordered_sessions = [bars[idx].session_id for idx in normalized_entries]
+    ordered_prices = [bars[idx].close for idx in normalized_entries]
     folds = generate_walk_forward_splits(
         n_samples=len(normalized_entries),
         train_size=walk_forward.train_size,
@@ -146,8 +149,17 @@ def run_v2_backtest_pipeline(
     for fold_idx, fold in enumerate(folds):
         train_labels = [ordered_labels[i] for i in fold.train_indices]
         test_labels = [ordered_labels[i] for i in fold.test_indices]
+        test_sessions = [ordered_sessions[i] for i in fold.test_indices]
+        test_prices = [ordered_prices[i] for i in fold.test_indices]
         if not test_labels:
             continue
+        execution_model = _build_execution_model(cost_model=cost_model, fold_seed=fold_idx)
+        execution_return_model = _build_execution_model(
+            cost_model=cost_model,
+            fold_seed=fold_idx,
+        )
+        b0_pred = _baseline_b0_pred(train_labels)
+        m1_pred = _m1_pred(train_labels)
         execution_returns_bps: list[float] = []
         execution_rejected = 0
         execution_partial = 0
@@ -155,7 +167,7 @@ def run_v2_backtest_pipeline(
             entry_bar_index = normalized_entries[rel_idx]
             bar = bars[entry_bar_index]
             trade = _simulate_execution_adjusted_return_bps(
-                execution_model=execution_model,
+                execution_model=execution_return_model,
                 bar=bar,
                 label=ordered_labels[rel_idx],
                 side=side,
@@ -176,11 +188,41 @@ def run_v2_backtest_pipeline(
                 train_label_distribution=_label_dist(train_labels),
                 test_label_distribution=_label_dist(test_labels),
                 baseline_scores=[
-                    BaselineScore(name="B0", accuracy=_baseline_b0(train_labels, test_labels)),
-                    BaselineScore(name="B1", accuracy=_score_constant(1, test_labels)),
+                    BaselineScore(
+                        name="B0",
+                        accuracy=_score_constant(b0_pred, test_labels),
+                        cost_adjusted_accuracy=_score_with_execution(
+                            prediction=b0_pred,
+                            actual=test_labels,
+                            sessions=test_sessions,
+                            reference_prices=test_prices,
+                            execution_model=execution_model,
+                            commission_bps=float(cost_model.commission_bps or 0.0),
+                        ),
+                    ),
+                    BaselineScore(
+                        name="B1",
+                        accuracy=_score_constant(1, test_labels),
+                        cost_adjusted_accuracy=_score_with_execution(
+                            prediction=1,
+                            actual=test_labels,
+                            sessions=test_sessions,
+                            reference_prices=test_prices,
+                            execution_model=execution_model,
+                            commission_bps=float(cost_model.commission_bps or 0.0),
+                        ),
+                    ),
                     BaselineScore(
                         name="M1",
-                        accuracy=_score_constant(_m1_pred(train_labels), test_labels),
+                        accuracy=_score_constant(m1_pred, test_labels),
+                        cost_adjusted_accuracy=_score_with_execution(
+                            prediction=m1_pred,
+                            actual=test_labels,
+                            sessions=test_sessions,
+                            reference_prices=test_prices,
+                            execution_model=execution_model,
+                            commission_bps=float(cost_model.commission_bps or 0.0),
+                        ),
                     ),
                 ],
                 execution_adjusted_avg_return_bps=(
@@ -219,12 +261,15 @@ def _score_constant(pred: int, actual: Sequence[int]) -> float:
 
 
 def _baseline_b0(train_labels: Sequence[int], test_labels: Sequence[int]) -> float:
+    return _score_constant(_baseline_b0_pred(train_labels), test_labels)
+
+
+def _baseline_b0_pred(train_labels: Sequence[int]) -> int:
     if not train_labels:
-        return _score_constant(0, test_labels)
+        return 0
     # Majority-class baseline from training fold.
     choices = (-1, 0, 1)
-    pred = max(choices, key=lambda c: train_labels.count(c))
-    return _score_constant(pred, test_labels)
+    return max(choices, key=lambda c: train_labels.count(c))
 
 
 def _m1_pred(train_labels: Sequence[int]) -> int:
@@ -233,6 +278,56 @@ def _m1_pred(train_labels: Sequence[int]) -> int:
     return train_labels[-1]
 
 
+def _build_execution_model(
+    *,
+    cost_model: BacktestCostModel,
+    fold_seed: int,
+) -> BacktestExecutionModel:
+    return BacktestExecutionModel(
+        ExecutionAssumptions(
+            slippage_bps_by_session=dict(cost_model.slippage_bps_by_session or {}),
+            failure_rate_by_session=dict(cost_model.failure_rate_by_session or {}),
+            partial_fill_rate_by_session=dict(cost_model.partial_fill_rate_by_session or {}),
+            seed=fold_seed,
+        )
+    )
+
+
+def _score_with_execution(
+    *,
+    prediction: int,
+    actual: Sequence[int],
+    sessions: Sequence[str],
+    reference_prices: Sequence[float],
+    execution_model: BacktestExecutionModel,
+    commission_bps: float,
+) -> float:
+    if not actual:
+        return 0.0
+    contributions: list[float] = []
+    for label, session_id, reference_price in zip(actual, sessions, reference_prices, strict=True):
+        if prediction == 0:
+            contributions.append(1.0 if label == 0 else 0.0)
+            continue
+        side = "BUY" if prediction > 0 else "SELL"
+        execution = execution_model.simulate(
+            ExecutionRequest(
+                side=side,
+                session_id=session_id,
+                qty=100,
+                reference_price=reference_price,
+            )
+        )
+        if execution.status == "REJECTED":
+            contributions.append(0.0)
+            continue
+        fill_ratio = execution.filled_qty / 100.0
+        cost_penalty = min(0.99, (commission_bps + execution.slippage_bps) / 10000.0)
+        correctness = 1.0 if prediction == label else 0.0
+        contributions.append(correctness * fill_ratio * (1.0 - cost_penalty))
+    return mean(contributions)
+
+
 def _build_run_id(*, n_entries: int, n_folds: int, sessions: Sequence[str]) -> str:
     sess_key = "_".join(sessions)
     return f"v2p-e{n_entries}-f{n_folds}-s{sess_key}"

src/core/blackout_manager.py

@@ -23,6 +23,7 @@ class BlackoutWindow:
 class QueuedOrderIntent:
     market_code: str
     exchange_code: str
+    session_id: str
     stock_code: str
     order_type: str
     quantity: int
@@ -68,11 +69,16 @@ class BlackoutOrderManager:
         self._queue: deque[QueuedOrderIntent] = deque()
         self._was_blackout = False
         self._max_queue_size = max_queue_size
+        self._overflow_drop_count = 0
 
     @property
     def pending_count(self) -> int:
         return len(self._queue)
 
+    @property
+    def overflow_drop_count(self) -> int:
+        return self._overflow_drop_count
+
     def in_blackout(self, now: datetime | None = None) -> bool:
         if not self.enabled or not self._windows:
             return False
@@ -81,8 +87,11 @@ class BlackoutOrderManager:
         return any(window.contains(kst_now) for window in self._windows)
 
     def enqueue(self, intent: QueuedOrderIntent) -> bool:
-        if len(self._queue) >= self._max_queue_size:
+        if self._max_queue_size <= 0:
             return False
+        if len(self._queue) >= self._max_queue_size:
+            self._queue.popleft()
+            self._overflow_drop_count += 1
         self._queue.append(intent)
         return True
 

src/db.py

@@ -318,7 +318,7 @@ def get_latest_buy_trade(
     if exchange_code:
         cursor = conn.execute(
             """
-            SELECT decision_id, price, quantity
+            SELECT decision_id, price, quantity, selection_context
             FROM trades
             WHERE stock_code = ?
               AND market = ?
@@ -339,7 +339,7 @@ def get_latest_buy_trade(
     else:
         cursor = conn.execute(
             """
-            SELECT decision_id, price, quantity
+            SELECT decision_id, price, quantity, selection_context
             FROM trades
             WHERE stock_code = ?
               AND market = ?

src/main.py

@@ -128,6 +128,84 @@ def _resolve_sell_qty_for_pnl(*, sell_qty: int | None, buy_qty: int | None) -> i
     return max(0, int(buy_qty or 0))
 
 
+def _extract_fx_rate_from_sources(*sources: dict[str, Any] | None) -> float | None:
+    """Best-effort FX rate extraction from broker payloads."""
+    # KIS overseas payloads expose exchange-rate fields with varying key names
+    # across endpoints/responses (price, balance, buying power). Keep this list
+    # centralised so schema drifts can be patched in one place.
+    rate_keys = (
+        "frst_bltn_exrt",
+        "bass_exrt",
+        "ovrs_exrt",
+        "aply_xchg_rt",
+        "xchg_rt",
+        "exchange_rate",
+        "fx_rate",
+    )
+    for source in sources:
+        if not isinstance(source, dict):
+            continue
+        for key in rate_keys:
+            rate = safe_float(source.get(key), 0.0)
+            if rate > 0:
+                return rate
+    return None
+
+
+def _split_trade_pnl_components(
+    *,
+    market: MarketInfo,
+    trade_pnl: float,
+    buy_price: float,
+    sell_price: float,
+    quantity: int,
+    buy_fx_rate: float | None = None,
+    sell_fx_rate: float | None = None,
+) -> tuple[float, float]:
+    """Split total trade pnl into strategy/fx components.
+
+    For overseas symbols, use buy/sell FX rates when both are available.
+    Otherwise preserve backward-compatible behaviour (all strategy pnl).
+    """
+    if trade_pnl == 0.0:
+        return 0.0, 0.0
+    if market.is_domestic:
+        return trade_pnl, 0.0
+
+    if (
+        buy_fx_rate is not None
+        and sell_fx_rate is not None
+        and buy_fx_rate > 0
+        and sell_fx_rate > 0
+        and quantity > 0
+        and buy_price > 0
+        and sell_price > 0
+    ):
+        buy_notional = buy_price * quantity
+        fx_return = (sell_fx_rate - buy_fx_rate) / buy_fx_rate
+        fx_pnl = buy_notional * fx_return
+        strategy_pnl = trade_pnl - fx_pnl
+        return strategy_pnl, fx_pnl
+
+    return trade_pnl, 0.0
+
+
+def _extract_buy_fx_rate(buy_trade: dict[str, Any] | None) -> float | None:
+    if not buy_trade:
+        return None
+    raw_ctx = buy_trade.get("selection_context")
+    if not isinstance(raw_ctx, str) or not raw_ctx.strip():
+        return None
+    try:
+        decoded = json.loads(raw_ctx)
+    except (TypeError, ValueError):
+        return None
+    if not isinstance(decoded, dict):
+        return None
+    rate = safe_float(decoded.get("fx_rate"), 0.0)
+    return rate if rate > 0 else None
+
+
 def _compute_kr_dynamic_stop_loss_pct(
     *,
     market: MarketInfo | None = None,
@@ -926,6 +1004,7 @@ async def build_overseas_symbol_universe(
 def _build_queued_order_intent(
     *,
     market: MarketInfo,
+    session_id: str,
     stock_code: str,
     order_type: str,
     quantity: int,
@@ -935,6 +1014,7 @@ def _build_queued_order_intent(
     return QueuedOrderIntent(
         market_code=market.code,
         exchange_code=market.exchange_code,
+        session_id=session_id,
         stock_code=stock_code,
         order_type=order_type,
         quantity=quantity,
@@ -947,6 +1027,7 @@ def _build_queued_order_intent(
 def _maybe_queue_order_intent(
     *,
     market: MarketInfo,
+    session_id: str,
     stock_code: str,
     order_type: str,
     quantity: int,
@@ -956,9 +1037,11 @@ def _maybe_queue_order_intent(
     if not BLACKOUT_ORDER_MANAGER.in_blackout():
         return False
 
+    before_overflow_drops = BLACKOUT_ORDER_MANAGER.overflow_drop_count
     queued = BLACKOUT_ORDER_MANAGER.enqueue(
         _build_queued_order_intent(
             market=market,
+            session_id=session_id,
             stock_code=stock_code,
             order_type=order_type,
             quantity=quantity,
@@ -967,6 +1050,7 @@ def _maybe_queue_order_intent(
         )
     )
     if queued:
+        after_overflow_drops = BLACKOUT_ORDER_MANAGER.overflow_drop_count
         logger.warning(
             (
                 "Blackout active: queued order intent %s %s (%s) "
@@ -980,9 +1064,22 @@ def _maybe_queue_order_intent(
             source,
             BLACKOUT_ORDER_MANAGER.pending_count,
         )
+        if after_overflow_drops > before_overflow_drops:
+            logger.error(
+                (
+                    "Blackout queue overflow policy applied: evicted oldest intent "
+                    "to keep latest %s %s (%s) source=%s pending=%d total_evicted=%d"
+                ),
+                order_type,
+                stock_code,
+                market.code,
+                source,
+                BLACKOUT_ORDER_MANAGER.pending_count,
+                after_overflow_drops,
+            )
     else:
         logger.error(
-            "Blackout queue full: dropped order intent %s %s (%s) qty=%d source=%s",
+            "Blackout queue unavailable: could not queue order intent %s %s (%s) qty=%d source=%s",
             order_type,
             stock_code,
             market.code,
@@ -1371,6 +1468,7 @@ async def trading_cycle(
     _session_risk_overrides(market=market, settings=settings)
 
     # 1. Fetch market data
+    balance_info: dict[str, Any] = {}
     price_output: dict[str, Any] = {}  # Populated for overseas markets; used for fallback metrics
     if market.is_domestic:
         current_price, price_change_pct, foreigner_net = await broker.get_current_price(stock_code)
@@ -1393,8 +1491,6 @@ async def trading_cycle(
             balance_info = output2[0]
         elif isinstance(output2, dict):
             balance_info = output2
-        else:
-            balance_info = {}
 
         total_eval = safe_float(balance_info.get("frcr_evlu_tota", "0") or "0")
         purchase_total = safe_float(balance_info.get("frcr_buy_amt_smtl", "0") or "0")
@@ -1814,6 +1910,9 @@ async def trading_cycle(
     quantity = 0
     trade_price = current_price
     trade_pnl = 0.0
+    buy_trade: dict[str, Any] | None = None
+    buy_price = 0.0
+    sell_qty = 0
     if decision.action in ("BUY", "SELL"):
         if KILL_SWITCH.new_orders_blocked and decision.action == "BUY":
             logger.critical(
@@ -1961,6 +2060,7 @@ async def trading_cycle(
                 return
             if _maybe_queue_order_intent(
                 market=market,
+                session_id=runtime_session_id,
                 stock_code=stock_code,
                 order_type=decision.action,
                 quantity=quantity,
@@ -2008,6 +2108,7 @@ async def trading_cycle(
                 return
             if _maybe_queue_order_intent(
                 market=market,
+                session_id=runtime_session_id,
                 stock_code=stock_code,
                 order_type=decision.action,
                 quantity=quantity,
@@ -2128,6 +2229,26 @@ async def trading_cycle(
             "signal": candidate.signal,
             "score": candidate.score,
         }
+    sell_fx_rate = _extract_fx_rate_from_sources(price_output, balance_info)
+    if sell_fx_rate is not None and not market.is_domestic:
+        if selection_context is None:
+            selection_context = {"fx_rate": sell_fx_rate}
+        else:
+            selection_context["fx_rate"] = sell_fx_rate
+
+    strategy_pnl: float | None = None
+    fx_pnl: float | None = None
+    if decision.action == "SELL" and order_succeeded:
+        buy_fx_rate = _extract_buy_fx_rate(buy_trade)
+        strategy_pnl, fx_pnl = _split_trade_pnl_components(
+            market=market,
+            trade_pnl=trade_pnl,
+            buy_price=buy_price,
+            sell_price=trade_price,
+            quantity=sell_qty or quantity,
+            buy_fx_rate=buy_fx_rate,
+            sell_fx_rate=sell_fx_rate,
+        )
 
     log_trade(
         conn=db_conn,
@@ -2138,6 +2259,8 @@ async def trading_cycle(
         quantity=quantity,
         price=trade_price,
         pnl=trade_pnl,
+        strategy_pnl=strategy_pnl,
+        fx_pnl=fx_pnl,
         market=market.code,
         exchange_code=market.exchange_code,
         session_id=runtime_session_id,
@@ -2736,6 +2859,7 @@ async def run_daily_session(
             )
             continue
 
+        balance_info: dict[str, Any] = {}
         if market.is_domestic:
             output2 = balance_data.get("output2", [{}])
             total_eval = safe_float(output2[0].get("tot_evlu_amt", "0")) if output2 else 0
@@ -2990,6 +3114,9 @@ async def run_daily_session(
             quantity = 0
             trade_price = stock_data["current_price"]
             trade_pnl = 0.0
+            buy_trade: dict[str, Any] | None = None
+            buy_price = 0.0
+            sell_qty = 0
             order_succeeded = True
             if decision.action in ("BUY", "SELL"):
                 if KILL_SWITCH.new_orders_blocked and decision.action == "BUY":
@@ -3142,6 +3269,7 @@ async def run_daily_session(
                             continue
                         if _maybe_queue_order_intent(
                             market=market,
+                            session_id=runtime_session_id,
                             stock_code=stock_code,
                             order_type=decision.action,
                             quantity=quantity,
@@ -3179,6 +3307,7 @@ async def run_daily_session(
                             continue
                         if _maybe_queue_order_intent(
                             market=market,
+                            session_id=runtime_session_id,
                             stock_code=stock_code,
                             order_type=decision.action,
                             quantity=quantity,
@@ -3272,6 +3401,30 @@ async def run_daily_session(
             # Log trade (skip if order was rejected by API)
             if decision.action in ("BUY", "SELL") and not order_succeeded:
                 continue
+            strategy_pnl: float | None = None
+            fx_pnl: float | None = None
+            selection_context: dict[str, Any] | None = None
+            if decision.action == "SELL" and order_succeeded:
+                buy_fx_rate = _extract_buy_fx_rate(buy_trade)
+                sell_fx_rate = _extract_fx_rate_from_sources(balance_info, stock_data)
+                strategy_pnl, fx_pnl = _split_trade_pnl_components(
+                    market=market,
+                    trade_pnl=trade_pnl,
+                    buy_price=buy_price,
+                    sell_price=trade_price,
+                    quantity=sell_qty or quantity,
+                    buy_fx_rate=buy_fx_rate,
+                    sell_fx_rate=sell_fx_rate,
+                )
+                if sell_fx_rate is not None and not market.is_domestic:
+                    # Daily path does not carry scanner candidate metrics, so this
+                    # context intentionally stores FX snapshot only.
+                    selection_context = {"fx_rate": sell_fx_rate}
+            elif not market.is_domestic:
+                snapshot_fx_rate = _extract_fx_rate_from_sources(balance_info, stock_data)
+                if snapshot_fx_rate is not None:
+                    # BUY/HOLD in daily path: persist FX snapshot for later SELL split.
+                    selection_context = {"fx_rate": snapshot_fx_rate}
             log_trade(
                 conn=db_conn,
                 stock_code=stock_code,
@@ -3281,9 +3434,12 @@ async def run_daily_session(
                 quantity=quantity,
                 price=trade_price,
                 pnl=trade_pnl,
+                strategy_pnl=strategy_pnl,
+                fx_pnl=fx_pnl,
                 market=market.code,
                 exchange_code=market.exchange_code,
                 session_id=runtime_session_id,
+                selection_context=selection_context,
                 decision_id=decision_id,
                 mode=settings.MODE,
             )

tests/test_backtest_cost_guard.py

@@ -45,7 +45,7 @@ def test_invalid_failure_rate_range_raises() -> None:
         commission_bps=5.0,
         slippage_bps_by_session={"KRX_REG": 10.0},
         failure_rate_by_session={"KRX_REG": 1.2},
-        partial_fill_rate_by_session={"KRX_REG": 0.1},
+        partial_fill_rate_by_session={"KRX_REG": 0.2},
         unfavorable_fill_required=True,
     )
     with pytest.raises(ValueError, match="failure rate must be within"):
@@ -57,7 +57,7 @@ def test_unfavorable_fill_requirement_cannot_be_disabled() -> None:
         commission_bps=5.0,
         slippage_bps_by_session={"KRX_REG": 10.0},
         failure_rate_by_session={"KRX_REG": 0.02},
-        partial_fill_rate_by_session={"KRX_REG": 0.1},
+        partial_fill_rate_by_session={"KRX_REG": 0.2},
         unfavorable_fill_required=False,
     )
     with pytest.raises(ValueError, match="unfavorable_fill_required must be True"):
@@ -70,7 +70,7 @@ def test_non_finite_commission_rejected(bad_commission: float) -> None:
         commission_bps=bad_commission,
         slippage_bps_by_session={"KRX_REG": 10.0},
         failure_rate_by_session={"KRX_REG": 0.02},
-        partial_fill_rate_by_session={"KRX_REG": 0.1},
+        partial_fill_rate_by_session={"KRX_REG": 0.2},
         unfavorable_fill_required=True,
     )
     with pytest.raises(ValueError, match="commission_bps"):
@@ -83,7 +83,7 @@ def test_non_finite_slippage_rejected(bad_slippage: float) -> None:
         commission_bps=5.0,
         slippage_bps_by_session={"KRX_REG": bad_slippage},
         failure_rate_by_session={"KRX_REG": 0.02},
-        partial_fill_rate_by_session={"KRX_REG": 0.1},
+        partial_fill_rate_by_session={"KRX_REG": 0.2},
         unfavorable_fill_required=True,
     )
     with pytest.raises(ValueError, match="slippage bps"):
@@ -102,13 +102,13 @@ def test_missing_required_partial_fill_session_raises() -> None:
         validate_backtest_cost_model(model=model, required_sessions=["KRX_REG", "US_PRE"])
 
 
-@pytest.mark.parametrize("bad_rate", [-0.1, 1.1, float("nan")])
-def test_invalid_partial_fill_rate_range_raises(bad_rate: float) -> None:
+@pytest.mark.parametrize("bad_partial_fill", [float("nan"), float("inf"), float("-inf"), -0.1, 1.1])
+def test_invalid_partial_fill_rate_rejected(bad_partial_fill: float) -> None:
     model = BacktestCostModel(
         commission_bps=5.0,
         slippage_bps_by_session={"KRX_REG": 10.0},
         failure_rate_by_session={"KRX_REG": 0.02},
-        partial_fill_rate_by_session={"KRX_REG": bad_rate},
+        partial_fill_rate_by_session={"KRX_REG": bad_partial_fill},
         unfavorable_fill_required=True,
     )
     with pytest.raises(ValueError, match="partial fill rate must be within"):

tests/test_backtest_pipeline_integration.py

@@ -35,7 +35,7 @@ def _cost_model() -> BacktestCostModel:
         commission_bps=3.0,
         slippage_bps_by_session={"KRX_REG": 10.0, "US_PRE": 50.0},
         failure_rate_by_session={"KRX_REG": 0.01, "US_PRE": 0.08},
-        partial_fill_rate_by_session={"KRX_REG": 0.1, "US_PRE": 0.2},
+        partial_fill_rate_by_session={"KRX_REG": 0.05, "US_PRE": 0.2},
         unfavorable_fill_required=True,
     )
 
@@ -72,6 +72,7 @@ def test_pipeline_happy_path_returns_fold_and_artifact_contract() -> None:
         assert names == {"B0", "B1", "M1"}
         for score in fold.baseline_scores:
             assert 0.0 <= score.accuracy <= 1.0
+            assert 0.0 <= score.cost_adjusted_accuracy <= 1.0
         assert fold.execution_adjusted_trade_count >= 0
         assert fold.execution_rejected_count >= 0
         assert fold.execution_partial_count >= 0
@@ -82,7 +83,7 @@ def test_pipeline_cost_guard_fail_fast() -> None:
         commission_bps=3.0,
         slippage_bps_by_session={"KRX_REG": 10.0},
         failure_rate_by_session={"KRX_REG": 0.01},
-        partial_fill_rate_by_session={"KRX_REG": 0.1},
+        partial_fill_rate_by_session={"KRX_REG": 0.05},
         unfavorable_fill_required=True,
     )
     try:
@@ -173,8 +174,8 @@ def test_pipeline_rejects_minutes_spec_when_timestamp_missing() -> None:
         raise AssertionError("expected timestamp validation error")
 
 
-def test_pipeline_execution_adjusted_returns_reflect_cost_and_fill_assumptions() -> None:
-    base_cfg = dict(
+def test_pipeline_fold_scores_reflect_cost_and_execution_effects() -> None:
+    cfg = dict(
         bars=_bars(),
         entry_indices=[0, 1, 2, 3, 4, 5, 6, 7],
         side=1,
@@ -192,7 +193,6 @@ def test_pipeline_execution_adjusted_returns_reflect_cost_and_fill_assumptions()
             min_train_size=3,
         ),
     )
-
     optimistic = BacktestCostModel(
         commission_bps=0.0,
         slippage_bps_by_session={"KRX_REG": 0.0, "US_PRE": 0.0},
@@ -202,25 +202,19 @@ def test_pipeline_execution_adjusted_returns_reflect_cost_and_fill_assumptions()
     )
     conservative = BacktestCostModel(
         commission_bps=10.0,
-        slippage_bps_by_session={"KRX_REG": 20.0, "US_PRE": 60.0},
+        slippage_bps_by_session={"KRX_REG": 30.0, "US_PRE": 80.0},
         failure_rate_by_session={"KRX_REG": 0.2, "US_PRE": 0.4},
         partial_fill_rate_by_session={"KRX_REG": 0.5, "US_PRE": 0.7},
         unfavorable_fill_required=True,
     )
+    optimistic_out = run_v2_backtest_pipeline(cost_model=optimistic, **cfg)
+    conservative_out = run_v2_backtest_pipeline(cost_model=conservative, **cfg)
 
-    opt_out = run_v2_backtest_pipeline(cost_model=optimistic, **base_cfg)
-    cons_out = run_v2_backtest_pipeline(cost_model=conservative, **base_cfg)
+    assert optimistic_out.folds and conservative_out.folds
+    optimistic_score = optimistic_out.folds[0].baseline_scores[1].cost_adjusted_accuracy
+    conservative_score = conservative_out.folds[0].baseline_scores[1].cost_adjusted_accuracy
+    assert conservative_score < optimistic_score
 
-    opt_avg = sum(
-        f.execution_adjusted_avg_return_bps for f in opt_out.folds
-    ) / len(opt_out.folds)
-    cons_avg = sum(
-        f.execution_adjusted_avg_return_bps for f in cons_out.folds
-    ) / len(cons_out.folds)
-    assert cons_avg < opt_avg
-
-    opt_trades = sum(f.execution_adjusted_trade_count for f in opt_out.folds)
-    cons_trades = sum(f.execution_adjusted_trade_count for f in cons_out.folds)
-    cons_rejected = sum(f.execution_rejected_count for f in cons_out.folds)
-    assert cons_trades <= opt_trades
-    assert cons_rejected >= 0
+    optimistic_avg_return = optimistic_out.folds[0].execution_adjusted_avg_return_bps
+    conservative_avg_return = conservative_out.folds[0].execution_adjusted_avg_return_bps
+    assert conservative_avg_return < optimistic_avg_return

tests/test_blackout_manager.py

@@ -35,6 +35,7 @@ def test_recovery_batch_only_after_blackout_exit() -> None:
     intent = QueuedOrderIntent(
         market_code="KR",
         exchange_code="KRX",
+        session_id="KRX_REG",
         stock_code="005930",
         order_type="BUY",
         quantity=1,
@@ -64,6 +65,7 @@ def test_requeued_intent_is_processed_next_non_blackout_cycle() -> None:
     intent = QueuedOrderIntent(
         market_code="KR",
         exchange_code="KRX",
+        session_id="KRX_REG",
         stock_code="005930",
         order_type="BUY",
         quantity=1,
@@ -79,3 +81,54 @@ def test_requeued_intent_is_processed_next_non_blackout_cycle() -> None:
     manager.requeue(first_batch[0])
     second_batch = manager.pop_recovery_batch(outside_blackout)
     assert len(second_batch) == 1
+
+
+def test_queue_overflow_drops_oldest_and_keeps_latest() -> None:
+    manager = BlackoutOrderManager(
+        enabled=True,
+        windows=parse_blackout_windows_kst("23:30-00:10"),
+        max_queue_size=2,
+    )
+    first = QueuedOrderIntent(
+        market_code="KR",
+        exchange_code="KRX",
+        session_id="KRX_REG",
+        stock_code="000001",
+        order_type="BUY",
+        quantity=1,
+        price=100.0,
+        source="first",
+        queued_at=datetime.now(UTC),
+    )
+    second = QueuedOrderIntent(
+        market_code="KR",
+        exchange_code="KRX",
+        session_id="KRX_REG",
+        stock_code="000002",
+        order_type="BUY",
+        quantity=1,
+        price=101.0,
+        source="second",
+        queued_at=datetime.now(UTC),
+    )
+    third = QueuedOrderIntent(
+        market_code="KR",
+        exchange_code="KRX",
+        session_id="KRX_REG",
+        stock_code="000003",
+        order_type="SELL",
+        quantity=2,
+        price=102.0,
+        source="third",
+        queued_at=datetime.now(UTC),
+    )
+
+    assert manager.enqueue(first)
+    assert manager.enqueue(second)
+    assert manager.enqueue(third)
+    assert manager.pending_count == 2
+    assert manager.overflow_drop_count == 1
+
+    outside_blackout = datetime(2026, 1, 1, 15, 20, tzinfo=UTC)
+    batch = manager.pop_recovery_batch(outside_blackout)
+    assert [intent.stock_code for intent in batch] == ["000002", "000003"]

tests/test_main.py

@@ -1,6 +1,7 @@
 """Tests for main trading loop integration."""
 
 from datetime import UTC, date, datetime
+from typing import Any
 from unittest.mock import ANY, AsyncMock, MagicMock, patch
 
 import pytest
@@ -9,6 +10,7 @@ import src.main as main_module
 from src.config import Settings
 from src.context.layer import ContextLayer
 from src.context.scheduler import ScheduleResult
+from src.core.blackout_manager import BlackoutOrderManager
 from src.core.order_policy import OrderPolicyRejected, get_session_info
 from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected
 from src.db import init_db, log_trade
@@ -33,6 +35,7 @@ from src.main import (
     _extract_held_qty_from_balance,
     _handle_market_close,
     _inject_staged_exit_features,
+    _maybe_queue_order_intent,
     _resolve_market_setting,
     _resolve_sell_qty_for_pnl,
     _retry_connection,
@@ -40,6 +43,7 @@ from src.main import (
     _run_evolution_loop,
     _should_block_overseas_buy_for_fx_buffer,
     _should_force_exit_for_overnight,
+    _split_trade_pnl_components,
     _start_dashboard_server,
     _stoploss_cooldown_minutes,
     _trigger_emergency_kill_switch,
@@ -102,22 +106,22 @@ def _make_sell_match(stock_code: str = "005930") -> ScenarioMatch:
 @pytest.fixture(autouse=True)
 def _reset_kill_switch_state() -> None:
     """Prevent cross-test leakage from global kill-switch state."""
+    def _reset_session_risk_globals() -> None:
+        _SESSION_RISK_LAST_BY_MARKET.clear()
+        _SESSION_RISK_OVERRIDES_BY_MARKET.clear()
+        _SESSION_RISK_PROFILES_MAP.clear()
+        main_module._SESSION_RISK_PROFILES_RAW = "{}"
+
     KILL_SWITCH.clear_block()
     _RUNTIME_EXIT_STATES.clear()
     _RUNTIME_EXIT_PEAKS.clear()
-    _SESSION_RISK_LAST_BY_MARKET.clear()
-    _SESSION_RISK_OVERRIDES_BY_MARKET.clear()
-    _SESSION_RISK_PROFILES_MAP.clear()
-    main_module._SESSION_RISK_PROFILES_RAW = "__reset__"
+    _reset_session_risk_globals()
     _STOPLOSS_REENTRY_COOLDOWN_UNTIL.clear()
     yield
     KILL_SWITCH.clear_block()
     _RUNTIME_EXIT_STATES.clear()
     _RUNTIME_EXIT_PEAKS.clear()
-    _SESSION_RISK_LAST_BY_MARKET.clear()
-    _SESSION_RISK_OVERRIDES_BY_MARKET.clear()
-    _SESSION_RISK_PROFILES_MAP.clear()
-    main_module._SESSION_RISK_PROFILES_RAW = "__reset__"
+    _reset_session_risk_globals()
     _STOPLOSS_REENTRY_COOLDOWN_UNTIL.clear()
 
 
@@ -3181,6 +3185,13 @@ async def test_sell_order_uses_broker_balance_qty_not_db() -> None:
     updated_buy = decision_logger.get_decision_by_id(buy_decision_id)
     assert updated_buy is not None
     assert updated_buy.outcome_pnl == -25.0
+    sell_row = db_conn.execute(
+        "SELECT pnl, strategy_pnl, fx_pnl FROM trades WHERE action='SELL' ORDER BY id DESC LIMIT 1"
+    ).fetchone()
+    assert sell_row is not None
+    assert sell_row[0] == -25.0
+    assert sell_row[1] == -25.0
+    assert sell_row[2] == 0.0
 
 
 @pytest.mark.asyncio
@@ -4598,6 +4609,23 @@ def test_fx_buffer_guard_applies_only_to_us_and_respects_boundary() -> None:
     assert required_jp == 0.0
 
 
+def test_split_trade_pnl_components_overseas_fx_split_preserves_total() -> None:
+    market = MagicMock()
+    market.is_domestic = False
+    strategy_pnl, fx_pnl = _split_trade_pnl_components(
+        market=market,
+        trade_pnl=20.0,
+        buy_price=100.0,
+        sell_price=110.0,
+        quantity=2,
+        buy_fx_rate=1200.0,
+        sell_fx_rate=1260.0,
+    )
+    assert strategy_pnl == 10.0
+    assert fx_pnl == 10.0
+    assert strategy_pnl + fx_pnl == pytest.approx(20.0)
+
+
 # run_daily_session — daily CB baseline (daily_start_eval) tests (issue #207)
 # ---------------------------------------------------------------------------
 
@@ -6351,6 +6379,225 @@ async def test_us_min_price_filter_not_applied_to_kr_market() -> None:
     broker.send_order.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_session_boundary_reloads_us_min_price_override_in_trading_cycle() -> None:
+    db_conn = init_db(":memory:")
+    decision_logger = DecisionLogger(db_conn)
+
+    broker = MagicMock()
+    broker.get_balance = AsyncMock(return_value={"output1": [], "output2": [{}]})
+
+    overseas_broker = MagicMock()
+    overseas_broker.get_overseas_price = AsyncMock(
+        return_value={"output": {"last": "7.0", "rate": "0.0"}}
+    )
+    overseas_broker.get_overseas_balance = AsyncMock(
+        return_value={
+            "output1": [],
+            "output2": [{"frcr_evlu_tota": "10000", "frcr_buy_amt_smtl": "0"}],
+        }
+    )
+    overseas_broker.get_overseas_buying_power = AsyncMock(
+        return_value={"output": {"ovrs_ord_psbl_amt": "10000"}}
+    )
+    overseas_broker.send_overseas_order = AsyncMock(return_value={"rt_cd": "0", "msg1": "OK"})
+
+    market = MagicMock()
+    market.name = "NASDAQ"
+    market.code = "US_NASDAQ"
+    market.exchange_code = "NASD"
+    market.is_domestic = False
+
+    telegram = MagicMock()
+    telegram.notify_trade_execution = AsyncMock()
+    telegram.notify_fat_finger = AsyncMock()
+    telegram.notify_circuit_breaker = AsyncMock()
+    telegram.notify_scenario_matched = AsyncMock()
+
+    settings = Settings(
+        KIS_APP_KEY="k",
+        KIS_APP_SECRET="s",
+        KIS_ACCOUNT_NO="12345678-01",
+        GEMINI_API_KEY="g",
+        MODE="paper",
+        PAPER_OVERSEAS_CASH=50000.0,
+        US_MIN_PRICE=5.0,
+        USD_BUFFER_MIN=1000.0,
+        SESSION_RISK_RELOAD_ENABLED=True,
+        SESSION_RISK_PROFILES_JSON=(
+            '{"US_PRE": {"US_MIN_PRICE": 8.0}, "US_DAY": {"US_MIN_PRICE": 5.0}}'
+        ),
+    )
+
+    current_session = {"id": "US_PRE"}
+
+    def _session_info(_: Any) -> MagicMock:
+        return MagicMock(session_id=current_session["id"])
+
+    with (
+        patch("src.main.get_open_position", return_value=None),
+        patch("src.main.get_session_info", side_effect=_session_info),
+    ):
+        await trading_cycle(
+            broker=broker,
+            overseas_broker=overseas_broker,
+            scenario_engine=MagicMock(evaluate=MagicMock(return_value=_make_buy_match("AAPL"))),
+            playbook=_make_playbook("US_NASDAQ"),
+            risk=MagicMock(validate_order=MagicMock(), check_circuit_breaker=MagicMock()),
+            db_conn=db_conn,
+            decision_logger=decision_logger,
+            context_store=MagicMock(
+                get_latest_timeframe=MagicMock(return_value=None),
+                set_context=MagicMock(),
+            ),
+            criticality_assessor=MagicMock(
+                assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
+                get_timeout=MagicMock(return_value=5.0),
+            ),
+            telegram=telegram,
+            market=market,
+            stock_code="AAPL",
+            scan_candidates={},
+            settings=settings,
+        )
+        assert overseas_broker.send_overseas_order.call_count == 0
+
+        current_session["id"] = "US_DAY"
+        await trading_cycle(
+            broker=broker,
+            overseas_broker=overseas_broker,
+            scenario_engine=MagicMock(evaluate=MagicMock(return_value=_make_buy_match("AAPL"))),
+            playbook=_make_playbook("US_NASDAQ"),
+            risk=MagicMock(validate_order=MagicMock(), check_circuit_breaker=MagicMock()),
+            db_conn=db_conn,
+            decision_logger=decision_logger,
+            context_store=MagicMock(
+                get_latest_timeframe=MagicMock(return_value=None),
+                set_context=MagicMock(),
+            ),
+            criticality_assessor=MagicMock(
+                assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
+                get_timeout=MagicMock(return_value=5.0),
+            ),
+            telegram=telegram,
+            market=market,
+            stock_code="AAPL",
+            scan_candidates={},
+            settings=settings,
+        )
+
+    assert overseas_broker.send_overseas_order.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_session_boundary_falls_back_when_profile_reload_fails() -> None:
+    db_conn = init_db(":memory:")
+    decision_logger = DecisionLogger(db_conn)
+
+    broker = MagicMock()
+    broker.get_balance = AsyncMock(return_value={"output1": [], "output2": [{}]})
+
+    overseas_broker = MagicMock()
+    overseas_broker.get_overseas_price = AsyncMock(
+        return_value={"output": {"last": "7.0", "rate": "0.0"}}
+    )
+    overseas_broker.get_overseas_balance = AsyncMock(
+        return_value={
+            "output1": [],
+            "output2": [{"frcr_evlu_tota": "10000", "frcr_buy_amt_smtl": "0"}],
+        }
+    )
+    overseas_broker.get_overseas_buying_power = AsyncMock(
+        return_value={"output": {"ovrs_ord_psbl_amt": "10000"}}
+    )
+    overseas_broker.send_overseas_order = AsyncMock(return_value={"rt_cd": "0", "msg1": "OK"})
+
+    market = MagicMock()
+    market.name = "NASDAQ"
+    market.code = "US_NASDAQ"
+    market.exchange_code = "NASD"
+    market.is_domestic = False
+
+    telegram = MagicMock()
+    telegram.notify_trade_execution = AsyncMock()
+    telegram.notify_fat_finger = AsyncMock()
+    telegram.notify_circuit_breaker = AsyncMock()
+    telegram.notify_scenario_matched = AsyncMock()
+
+    settings = Settings(
+        KIS_APP_KEY="k",
+        KIS_APP_SECRET="s",
+        KIS_ACCOUNT_NO="12345678-01",
+        GEMINI_API_KEY="g",
+        MODE="paper",
+        PAPER_OVERSEAS_CASH=50000.0,
+        US_MIN_PRICE=5.0,
+        USD_BUFFER_MIN=1000.0,
+        SESSION_RISK_RELOAD_ENABLED=True,
+        SESSION_RISK_PROFILES_JSON='{"US_PRE": {"US_MIN_PRICE": 8.0}}',
+    )
+
+    current_session = {"id": "US_PRE"}
+
+    def _session_info(_: Any) -> MagicMock:
+        return MagicMock(session_id=current_session["id"])
+
+    with (
+        patch("src.main.get_open_position", return_value=None),
+        patch("src.main.get_session_info", side_effect=_session_info),
+    ):
+        await trading_cycle(
+            broker=broker,
+            overseas_broker=overseas_broker,
+            scenario_engine=MagicMock(evaluate=MagicMock(return_value=_make_buy_match("AAPL"))),
+            playbook=_make_playbook("US_NASDAQ"),
+            risk=MagicMock(validate_order=MagicMock(), check_circuit_breaker=MagicMock()),
+            db_conn=db_conn,
+            decision_logger=decision_logger,
+            context_store=MagicMock(
+                get_latest_timeframe=MagicMock(return_value=None),
+                set_context=MagicMock(),
+            ),
+            criticality_assessor=MagicMock(
+                assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
+                get_timeout=MagicMock(return_value=5.0),
+            ),
+            telegram=telegram,
+            market=market,
+            stock_code="AAPL",
+            scan_candidates={},
+            settings=settings,
+        )
+        assert overseas_broker.send_overseas_order.call_count == 0
+
+        settings.SESSION_RISK_PROFILES_JSON = "{invalid-json"
+        current_session["id"] = "US_DAY"
+        await trading_cycle(
+            broker=broker,
+            overseas_broker=overseas_broker,
+            scenario_engine=MagicMock(evaluate=MagicMock(return_value=_make_buy_match("AAPL"))),
+            playbook=_make_playbook("US_NASDAQ"),
+            risk=MagicMock(validate_order=MagicMock(), check_circuit_breaker=MagicMock()),
+            db_conn=db_conn,
+            decision_logger=decision_logger,
+            context_store=MagicMock(
+                get_latest_timeframe=MagicMock(return_value=None),
+                set_context=MagicMock(),
+            ),
+            criticality_assessor=MagicMock(
+                assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
+                get_timeout=MagicMock(return_value=5.0),
+            ),
+            telegram=telegram,
+            market=market,
+            stock_code="AAPL",
+            scan_candidates={},
+            settings=settings,
+        )
+
+    assert overseas_broker.send_overseas_order.call_count == 1
+
+
 def test_overnight_policy_prioritizes_killswitch_over_exception() -> None:
     market = MagicMock()
     with patch("src.main.get_session_info", return_value=MagicMock(session_id="US_AFTER")):
@@ -6475,6 +6722,7 @@ async def test_blackout_queues_order_and_skips_submission() -> None:
     blackout_manager.in_blackout.return_value = True
     blackout_manager.enqueue.return_value = True
     blackout_manager.pending_count = 1
+    blackout_manager.overflow_drop_count = 0
 
     with patch("src.main.BLACKOUT_ORDER_MANAGER", blackout_manager):
         await trading_cycle(
@@ -6504,6 +6752,43 @@ async def test_blackout_queues_order_and_skips_submission() -> None:
     blackout_manager.enqueue.assert_called_once()
 
 
+def test_blackout_queue_overflow_keeps_latest_intent() -> None:
+    manager = BlackoutOrderManager(enabled=True, windows=[], max_queue_size=1)
+    manager.in_blackout = lambda now=None: True  # type: ignore[method-assign]
+
+    market = MagicMock()
+    market.code = "KR"
+    market.exchange_code = "KRX"
+
+    with patch("src.main.BLACKOUT_ORDER_MANAGER", manager):
+        assert _maybe_queue_order_intent(
+            market=market,
+            session_id="KRX_REG",
+            stock_code="005930",
+            order_type="BUY",
+            quantity=1,
+            price=100.0,
+            source="test-first",
+        )
+        assert _maybe_queue_order_intent(
+            market=market,
+            session_id="KRX_REG",
+            stock_code="000660",
+            order_type="BUY",
+            quantity=2,
+            price=200.0,
+            source="test-second",
+        )
+
+    assert manager.pending_count == 1
+    assert manager.overflow_drop_count == 1
+    manager.in_blackout = lambda now=None: False  # type: ignore[method-assign]
+    batch = manager.pop_recovery_batch()
+    assert len(batch) == 1
+    assert batch[0].stock_code == "000660"
+    assert batch[0].session_id == "KRX_REG"
+
+
 @pytest.mark.asyncio
 async def test_process_blackout_recovery_executes_valid_intents() -> None:
     """Recovery must execute queued intents that pass revalidation."""
@@ -6581,6 +6866,7 @@ async def test_process_blackout_recovery_drops_policy_rejected_intent() -> None:
     intent.quantity = 1
     intent.price = 100.0
     intent.source = "test"
+    intent.session_id = "KRX_REG"
     intent.attempts = 0
 
     blackout_manager = MagicMock()
@@ -6630,6 +6916,7 @@ async def test_process_blackout_recovery_drops_intent_on_excessive_price_drift()
     intent.quantity = 1
     intent.price = 100.0
     intent.source = "test"
+    intent.session_id = "US_PRE"
     intent.attempts = 0
 
     blackout_manager = MagicMock()
@@ -6680,6 +6967,7 @@ async def test_process_blackout_recovery_drops_overseas_intent_on_excessive_pric
     intent.quantity = 1
     intent.price = 100.0
     intent.source = "test"
+    intent.session_id = "KRX_REG"
     intent.attempts = 0
 
     blackout_manager = MagicMock()
@@ -6729,6 +7017,7 @@ async def test_process_blackout_recovery_requeues_intent_when_price_lookup_fails
     intent.quantity = 1
     intent.price = 100.0
     intent.source = "test"
+    intent.session_id = "KRX_REG"
     intent.attempts = 0
 
     blackout_manager = MagicMock()