Merge pull request 'process: automate backtest gate for PR/push/schedule (#314)' (#315) from feature/issue-314-backtest-gate-automation into feature/v3-session-policy-stream

.gitea/PULL_REQUEST_TEMPLATE.md

@@ -35,6 +35,12 @@
 - [ ] `docs/commands.md`와 `docs/workflow.md` 트러블슈팅 선확인
 - [ ] `tea` 사용 (`gh` 미사용)
 
+## Session Handover Gate
+
+- [ ] `python3 scripts/session_handover_check.py --strict` 통과
+- [ ] `workflow/session-handover.md` 최신 엔트리가 현재 브랜치/당일(UTC) 기준으로 갱신됨
+- 최신 handover 엔트리 heading:
+
 ## Runtime Evidence
 
 - 시스템 실제 구동 커맨드:

.gitea/workflows/ci.yml

@@ -0,0 +1,38 @@
+name: Gitea CI
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+      - feature/**
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Install dependencies
+        run: pip install ".[dev]"
+
+      - name: Session handover gate
+        run: python3 scripts/session_handover_check.py --strict
+
+      - name: Validate governance assets
+        run: python3 scripts/validate_governance_assets.py
+
+      - name: Validate Ouroboros docs
+        run: python3 scripts/validate_ouroboros_docs.py
+
+      - name: Lint
+        run: ruff check src/ tests/
+
+      - name: Run tests with coverage
+        run: pytest -v --cov=src --cov-report=term-missing --cov-fail-under=80

.github/workflows/backtest-gate.yml

@@ -0,0 +1,66 @@
+name: Backtest Gate
+
+on:
+  pull_request:
+    branches: ["**"]
+  push:
+    branches:
+      - "feature/**"
+  schedule:
+    # Daily scheduled gate (KST 01:20)
+    - cron: "20 16 * * *"
+  workflow_dispatch:
+    inputs:
+      mode:
+        description: "backtest mode (auto|smoke|full)"
+        required: false
+        default: "auto"
+      base_ref:
+        description: "git base ref for changed-file diff"
+        required: false
+        default: "origin/main"
+
+jobs:
+  backtest-gate:
+    runs-on: ubuntu-latest
+    concurrency:
+      group: backtest-gate-${{ github.ref }}
+      cancel-in-progress: true
+
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Set up Python 3.11
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Install dependencies
+        run: pip install ".[dev]"
+
+      - name: Resolve base ref
+        id: base
+        run: |
+          if [ "${{ github.event_name }}" = "pull_request" ]; then
+            echo "ref=origin/${{ github.base_ref }}" >> "$GITHUB_OUTPUT"
+          elif [ "${{ github.event_name }}" = "workflow_dispatch" ] && [ -n "${{ github.event.inputs.base_ref }}" ]; then
+            echo "ref=${{ github.event.inputs.base_ref }}" >> "$GITHUB_OUTPUT"
+          else
+            echo "ref=origin/main" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Run backtest gate
+        env:
+          BASE_REF: ${{ steps.base.outputs.ref }}
+          BACKTEST_MODE: ${{ github.event_name == 'workflow_dispatch' && github.event.inputs.mode || 'auto' }}
+          FORCE_FULL_BACKTEST: ${{ github.event_name == 'schedule' && 'true' || 'false' }}
+        run: bash scripts/backtest_gate.sh
+
+      - name: Upload backtest logs
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: backtest-gate-logs
+          path: data/backtest-gate/*.log

.github/workflows/ci.yml

@@ -21,6 +21,9 @@ jobs:
       - name: Install dependencies
         run: pip install ".[dev]"
 
+      - name: Session handover gate
+        run: python3 scripts/session_handover_check.py --strict
+
       - name: Validate governance assets
         run: python3 scripts/validate_governance_assets.py
 

docs/agent-constraints.md

@@ -32,6 +32,16 @@ It is distinct from `docs/requirements-log.md`, which records **project/product
      (or in a dedicated policy doc) and reference it when working.
    - Keep entries short and concrete, with dates.
 
+5. **Session start handover gate**
+   - Before implementation/verification work, run `python3 scripts/session_handover_check.py --strict`.
+   - Keep `workflow/session-handover.md` updated with a same-day entry for the active branch.
+   - If the check fails, stop and fix handover artifacts first.
+
+6. **Process-change-first execution gate**
+   - If process/governance change is required, merge the process ticket to the feature branch first.
+   - Do not start code/test edits for implementation tickets until process merge evidence is confirmed.
+   - Subagents must be constrained to read-only exploration until the process gate is satisfied.
+
 ## Change Control
 
 - Changes to this file follow the same workflow as code changes.
@@ -50,3 +60,10 @@ It is distinct from `docs/requirements-log.md`, which records **project/product
 
 - All agents must pre-read `docs/commands.md` and `docs/workflow.md` troubleshooting before running Gitea issue/PR/comment commands.
 - `gh` CLI is prohibited for repository ticket/PR operations; use `tea` (or documented Gitea API fallback only).
+- Session start must pass `python3 scripts/session_handover_check.py --strict`, with branch-matched entry in `workflow/session-handover.md`.
+
+### 2026-02-27
+
+- Apply process-change-first as an execution gate: process ticket must be merged before implementation ticket coding.
+- Handover entry must record concrete `next_ticket` and `process_gate_checked`; placeholders are not allowed in strict gate.
+- Before process merge confirmation, all subagent tasks must remain read-only (analysis only).

docs/commands.md

@@ -11,6 +11,16 @@
 - 기본 도구는 `tea`이며, `tea` 미지원 케이스만 Gitea API를 fallback으로 사용한다.
 - 실행 전 `docs/workflow.md`의 `Gitea CLI Formatting Troubleshooting`을 반드시 확인한다.
 
+## Session Handover Preflight (Mandatory)
+
+- 세션 시작 직후(코드 변경 전) 아래 명령을 먼저 실행한다.
+
+```bash
+python3 scripts/session_handover_check.py --strict
+```
+
+- 실패 시 `workflow/session-handover.md` 최신 엔트리를 보강한 뒤 재실행한다.
+
 ### tea CLI (Gitea Command Line Tool)
 
 #### ❌ TTY Error - Interactive Confirmation Fails
@@ -150,6 +160,9 @@ python -m src.main --mode=paper --dashboard
 # Runtime verification monitor (NOT_OBSERVED detection)
 bash scripts/runtime_verify_monitor.sh
 
+# Session handover gate (must pass before implementation)
+python3 scripts/session_handover_check.py --strict
+
 # Follow runtime verification log
 tail -f data/overnight/runtime_verify_*.log
 

docs/ouroboros/60_repo_enforcement_checklist.md

@@ -3,7 +3,7 @@ Doc-ID: DOC-OPS-002
 Version: 1.0.0
 Status: active
 Owner: tpm
-Updated: 2026-02-26
+Updated: 2026-02-27
 -->
 
 # 저장소 강제 설정 체크리스트
@@ -58,6 +58,7 @@ Updated: 2026-02-26
 자동 점검:
 - 문서 검증 스크립트 통과
 - 테스트 통과
+- `python3 scripts/session_handover_check.py --strict` 통과
 - 개발 완료 시 시스템 구동/모니터링 증적 코멘트 존재
 - 이슈/PR 조작 전에 `docs/commands.md` 및 `docs/workflow.md` 트러블슈팅 확인 코멘트 존재
 - `gh` CLI 미사용, `tea` 사용 증적 존재

docs/requirements-log.md

@@ -355,3 +355,36 @@ Order result: 모의투자 매수주문이 완료 되었습니다.  ✓
    - `TestOverseasGhostPositionClose` 2개: ghost-close 로그 확인, 일반 오류 무시
 
 **이슈/PR:** #235, PR #236
+
+---
+
+## 2026-02-27
+
+### v2 백테스트 파이프라인 통합 (#305)
+
+**배경:**
+- `TripleBarrier`, `WalkForward`, `BacktestCostGuard`는 개별 모듈로 존재했으나,
+  하나의 실행 경로로 연결된 파이프라인이 없어 통합 검증이 불가능했다.
+
+**구현 내용:**
+
+1. `src/analysis/backtest_pipeline.py`
+   - `run_v2_backtest_pipeline()` 추가:
+     - `validate_backtest_cost_model()` 선검증(fail-fast)
+     - `label_with_triple_barrier()`로 entry 라벨 생성
+     - `generate_walk_forward_splits()`로 fold 생성
+     - fold별 baseline(`B0`, `B1`, `M1`) score 산출
+   - 결과 아티팩트 계약 구조(`BacktestPipelineResult`) 정의
+   - leakage 검사 유틸 `fold_has_leakage()` 제공
+
+2. `tests/test_backtest_pipeline_integration.py` 신규
+   - happy path 통합 검증
+   - cost guard 실패 fail-fast 검증
+   - purge/embargo 기반 누수 방지 검증
+   - 동일 입력 재실행 결정성 검증
+
+**검증:**
+- `pytest -q tests/test_backtest_pipeline_integration.py tests/test_triple_barrier.py tests/test_walk_forward_split.py tests/test_backtest_cost_guard.py tests/test_backtest_execution_model.py`
+- `ruff check src/analysis/backtest_pipeline.py tests/test_backtest_pipeline_integration.py`
+
+**이슈/PR:** #305

docs/testing.md

@@ -181,6 +181,29 @@ pytest -v --cov=src --cov-report=term-missing
 
 **Note:** `main.py` has lower coverage as it contains the main loop which is tested via integration/manual testing.
 
+## Backtest Automation Gate
+
+백테스트 관련 검증은 `scripts/backtest_gate.sh`와 `.github/workflows/backtest-gate.yml`로 자동 실행된다.
+
+- PR: 변경 파일 기준 `auto` 모드
+- `feature/**` push: 변경 파일 기준 `auto` 모드
+- Daily schedule: `full` 강제 실행
+- Manual dispatch: `mode`(`auto|smoke|full`) 지정 가능
+
+실행 기준:
+- `src/analysis/`, `src/strategy/`, `src/strategies/`, `src/main.py`, `src/markets/`, `src/broker/`
+- 백테스트 핵심 테스트 파일 변경
+- `docs/ouroboros/` 변경
+
+`auto` 모드에서 백테스트 민감 영역 변경이 없으면 게이트는 `skip` 처리되며 실패로 간주하지 않는다.
+
+로컬 수동 실행:
+```bash
+bash scripts/backtest_gate.sh
+BACKTEST_MODE=full bash scripts/backtest_gate.sh
+BASE_REF=origin/feature/v3-session-policy-stream BACKTEST_MODE=auto bash scripts/backtest_gate.sh
+```
+
 ## Test Configuration
 
 ### `pyproject.toml`

docs/workflow.md

@@ -30,6 +30,19 @@ Gitea 이슈/PR/코멘트 작업 전에 모든 에이전트는 아래를 먼저
 - `tea` 실패 시 동일 명령 재시도 전에 원인/수정사항을 PR 코멘트에 남긴다.
 - 필요한 경우에만 Gitea API(`localhost:3000`)를 fallback으로 사용한다.
 
+## Session Handover Gate (Mandatory)
+
+새 세션에서 구현/검증을 시작하기 전에 아래를 선행해야 한다.
+
+1. `docs/workflow.md`, `docs/commands.md`, `docs/agent-constraints.md` 재확인
+2. `workflow/session-handover.md`에 최신 세션 엔트리 추가
+3. `python3 scripts/session_handover_check.py --strict` 통과 확인
+
+강제 규칙:
+- handover check 실패 상태에서 코드 수정/이슈 상태 전이/PR 생성 금지
+- 최신 handover 엔트리는 현재 작업 브랜치를 명시해야 한다
+- 최신 handover 엔트리는 당일(UTC) 날짜를 포함해야 한다
+
 ## Branch Strategy (Mandatory)
 
 - Team operation default branch is the **program feature branch**, not `main`.
@@ -38,6 +51,21 @@ Gitea 이슈/PR/코멘트 작업 전에 모든 에이전트는 아래를 먼저
 - Until final user sign-off, `main` merge is prohibited.
 - 각 에이전트는 주요 의사결정(리뷰 지적, 수정 방향, 검증 승인)마다 PR 코멘트를 적극 작성해 의사결정 과정을 남긴다.
 
+## Backtest Gate Policy (Mandatory)
+
+사람 의존도를 줄이기 위해 백테스트 검증은 자동 게이트를 기본으로 한다.
+
+- 워크플로우: `.github/workflows/backtest-gate.yml`
+- 실행 스크립트: `scripts/backtest_gate.sh`
+- 기본 모드: `auto` (변경 파일 기반 실행/skip 판정)
+- 정기 스케줄: daily `full` 강제 실행
+- 수동 재실행: workflow dispatch + `mode` 지정
+
+강제 규칙:
+- 백테스트 민감 변경(PR/feature push)에서 게이트 실패 시 머지 금지
+- 스케줄 게이트 실패 시 이슈 등록 후 원인/복구 계획 기록
+- `python` 대신 `python3` 기준으로 실행한다
+
 ## Gitea CLI Formatting Troubleshooting
 
 Issue/PR 본문 작성 시 줄바꿈(`\n`)이 문자열 그대로 저장되는 문제가 반복될 수 있다. 원인은 `-d "...\n..."` 형태에서 쉘/CLI가 이스케이프를 실제 개행으로 해석하지 않기 때문이다.
@@ -175,6 +203,19 @@ Use `run_in_background=True` for independent tasks that don't block subsequent w
 - process ticket 미반영 상태에서 기능 티켓 코딩/머지 금지
 - 세션 전환 시에도 동일 규칙 유지
 
+### Implementation Start Gate (Mandatory)
+
+구현 티켓을 시작하기 전에 아래 3개를 모두 만족해야 한다.
+
+1. `process ticket merge` 증적 확인 (feature branch 반영 커밋/PR)
+2. `workflow/session-handover.md` 최신 엔트리에 `next_ticket`과 `process_gate_checked` 기록
+3. `python3 scripts/session_handover_check.py --strict` 통과
+
+강제 규칙:
+- 위 3개 중 하나라도 불충족이면 코드/테스트 수정 금지
+- 서브에이전트 지시도 동일하게 제한한다 (`process merged 확인 전 read-only 탐색만 허용`)
+- 성급 착수 발견 시 구현 작업을 즉시 중단하고 handover/proces gate부터 복구한다
+
 ### Ticket Maturity Stages (Mandatory)
 
 모든 티켓은 아래 4단계를 순서대로 통과해야 한다.

scripts/backtest_gate.sh

@@ -0,0 +1,106 @@
+#!/usr/bin/env bash
+# Backtest gate for PR/push/scheduled verification.
+
+set -euo pipefail
+
+MODE="${BACKTEST_MODE:-auto}"          # auto | smoke | full
+BASE_REF="${BASE_REF:-origin/main}"    # used when MODE=auto
+FORCE_FULL="${FORCE_FULL_BACKTEST:-false}"
+LOG_DIR="${LOG_DIR:-data/backtest-gate}"
+
+mkdir -p "$LOG_DIR"
+STAMP="$(date -u +%Y%m%d_%H%M%S)"
+LOG_FILE="$LOG_DIR/backtest_gate_${STAMP}.log"
+
+log() {
+  printf '%s %s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$1" | tee -a "$LOG_FILE"
+}
+
+run_cmd() {
+  log "[RUN] $*"
+  "$@" 2>&1 | tee -a "$LOG_FILE"
+}
+
+resolve_mode_from_changes() {
+  if [ "$FORCE_FULL" = "true" ]; then
+    echo "full"
+    return
+  fi
+
+  if ! git rev-parse --verify "$BASE_REF" >/dev/null 2>&1; then
+    log "[WARN] BASE_REF not found: $BASE_REF; fallback to full"
+    echo "full"
+    return
+  fi
+
+  changed_files="$(git diff --name-only "$BASE_REF"...HEAD || true)"
+  if [ -z "$changed_files" ]; then
+    log "[INFO] no changed files between $BASE_REF...HEAD; skip backtest gate"
+    echo "skip"
+    return
+  fi
+
+  log "[INFO] changed files from $BASE_REF...HEAD:"
+  while IFS= read -r line; do
+    [ -n "$line" ] && log "  - $line"
+  done <<< "$changed_files"
+
+  # Backtest-sensitive areas: analysis/strategy/runtime execution semantics.
+  if printf '%s\n' "$changed_files" | rg -q \
+    '^(src/analysis/|src/strategy/|src/strategies/|src/main.py|src/markets/|src/broker/|tests/test_backtest_|tests/test_triple_barrier.py|tests/test_walk_forward_split.py|tests/test_main.py|docs/ouroboros/)'
+  then
+    echo "full"
+  else
+    echo "skip"
+  fi
+}
+
+SMOKE_TESTS=(
+  tests/test_backtest_pipeline_integration.py
+  tests/test_triple_barrier.py
+  tests/test_walk_forward_split.py
+  tests/test_backtest_cost_guard.py
+  tests/test_backtest_execution_model.py
+)
+
+FULL_TESTS=(
+  tests/test_backtest_pipeline_integration.py
+  tests/test_triple_barrier.py
+  tests/test_walk_forward_split.py
+  tests/test_backtest_cost_guard.py
+  tests/test_backtest_execution_model.py
+  tests/test_main.py
+)
+
+main() {
+  log "[INFO] backtest gate started mode=$MODE base_ref=$BASE_REF force_full=$FORCE_FULL"
+
+  selected_mode="$MODE"
+  if [ "$MODE" = "auto" ]; then
+    selected_mode="$(resolve_mode_from_changes)"
+  fi
+
+  case "$selected_mode" in
+    skip)
+      log "[PASS] backtest gate skipped (no backtest-sensitive changes)"
+      exit 0
+      ;;
+    smoke)
+      run_cmd python3 -m pytest -q "${SMOKE_TESTS[@]}"
+      log "[PASS] smoke backtest gate passed"
+      ;;
+    full)
+      run_cmd python3 -m pytest -q "${SMOKE_TESTS[@]}"
+      # Runtime semantics tied to v2 staged-exit must remain covered in full gate.
+      run_cmd python3 -m pytest -q tests/test_main.py -k \
+        "staged_exit_override or runtime_exit_cache_cleared or run_daily_session_applies_staged_exit_override_on_hold"
+      log "[PASS] full backtest gate passed"
+      ;;
+    *)
+      log "[FAIL] invalid BACKTEST_MODE=$selected_mode (expected auto|smoke|full)"
+      exit 2
+      ;;
+  esac
+}
+
+main "$@"

scripts/session_handover_check.py

@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""Session handover preflight gate.
+
+This script enforces a minimal handover record per working branch so that
+new sessions cannot start implementation without reading the required docs
+and recording current intent.
+"""
+
+from __future__ import annotations
+
+import argparse
+import subprocess
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+
+REQUIRED_DOCS = (
+    Path("docs/workflow.md"),
+    Path("docs/commands.md"),
+    Path("docs/agent-constraints.md"),
+)
+HANDOVER_LOG = Path("workflow/session-handover.md")
+
+
+def _run_git(*args: str) -> str:
+    try:
+        return (
+            subprocess.check_output(["git", *args], stderr=subprocess.DEVNULL)
+            .decode("utf-8")
+            .strip()
+        )
+    except Exception:
+        return ""
+
+
+def _current_branch() -> str:
+    branch = _run_git("branch", "--show-current")
+    if branch:
+        return branch
+    return _run_git("rev-parse", "--abbrev-ref", "HEAD")
+
+
+def _latest_entry(text: str) -> str:
+    chunks = text.split("\n### ")
+    if not chunks:
+        return ""
+    if chunks[0].startswith("### "):
+        chunks[0] = chunks[0][4:]
+    latest = chunks[-1].strip()
+    if not latest:
+        return ""
+    if not latest.startswith("### "):
+        latest = f"### {latest}"
+    return latest
+
+
+def _check_required_files(errors: list[str]) -> None:
+    for path in REQUIRED_DOCS:
+        if not path.exists():
+            errors.append(f"missing required document: {path}")
+    if not HANDOVER_LOG.exists():
+        errors.append(f"missing handover log: {HANDOVER_LOG}")
+
+
+def _check_handover_entry(
+    *,
+    branch: str,
+    strict: bool,
+    errors: list[str],
+) -> None:
+    if not HANDOVER_LOG.exists():
+        return
+    text = HANDOVER_LOG.read_text(encoding="utf-8")
+    latest = _latest_entry(text)
+    if not latest:
+        errors.append("handover log has no session entry")
+        return
+
+    required_tokens = (
+        "- branch:",
+        "- docs_checked:",
+        "- open_issues_reviewed:",
+        "- next_ticket:",
+        "- process_gate_checked:",
+    )
+    for token in required_tokens:
+        if token not in latest:
+            errors.append(f"latest handover entry missing token: {token}")
+
+    if strict:
+        today_utc = datetime.now(UTC).date().isoformat()
+        if today_utc not in latest:
+            errors.append(
+                f"latest handover entry must contain today's UTC date ({today_utc})"
+            )
+        branch_token = f"- branch: {branch}"
+        if branch_token not in latest:
+            errors.append(
+                "latest handover entry must target current branch "
+                f"({branch_token})"
+            )
+        if "- next_ticket: #TBD" in latest:
+            errors.append("latest handover entry must not use placeholder next_ticket (#TBD)")
+        if "merged_to_feature_branch=no" in latest:
+            errors.append(
+                "process gate indicates not merged; implementation must stay blocked "
+                "(merged_to_feature_branch=no)"
+            )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="Validate session handover gate requirements."
+    )
+    parser.add_argument(
+        "--strict",
+        action="store_true",
+        help="Enforce today-date and current-branch match on latest handover entry.",
+    )
+    args = parser.parse_args()
+
+    errors: list[str] = []
+    _check_required_files(errors)
+
+    branch = _current_branch()
+    if not branch:
+        errors.append("cannot resolve current git branch")
+    elif branch in {"main", "master"}:
+        errors.append(f"working branch must not be {branch}")
+
+    _check_handover_entry(branch=branch, strict=args.strict, errors=errors)
+
+    if errors:
+        print("[FAIL] session handover check failed")
+        for err in errors:
+            print(f"- {err}")
+        return 1
+
+    print("[OK] session handover check passed")
+    print(f"[OK] branch={branch}")
+    print(f"[OK] handover_log={HANDOVER_LOG}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/validate_governance_assets.py

@@ -22,6 +22,10 @@ def main() -> int:
 
     pr_template = Path(".gitea/PULL_REQUEST_TEMPLATE.md")
     issue_template = Path(".gitea/ISSUE_TEMPLATE/runtime_verification.md")
+    workflow_doc = Path("docs/workflow.md")
+    commands_doc = Path("docs/commands.md")
+    handover_script = Path("scripts/session_handover_check.py")
+    handover_log = Path("workflow/session-handover.md")
 
     must_contain(
         pr_template,
@@ -32,6 +36,8 @@ def main() -> int:
             "NOT_OBSERVED",
             "tea",
             "gh",
+            "Session Handover Gate",
+            "session_handover_check.py --strict",
         ],
         errors,
     )
@@ -45,6 +51,35 @@ def main() -> int:
         ],
         errors,
     )
+    must_contain(
+        workflow_doc,
+        [
+            "Session Handover Gate (Mandatory)",
+            "session_handover_check.py --strict",
+        ],
+        errors,
+    )
+    must_contain(
+        commands_doc,
+        [
+            "Session Handover Preflight (Mandatory)",
+            "session_handover_check.py --strict",
+        ],
+        errors,
+    )
+    must_contain(
+        handover_log,
+        [
+            "Session Handover Log",
+            "- branch:",
+            "- docs_checked:",
+            "- open_issues_reviewed:",
+            "- next_ticket:",
+        ],
+        errors,
+    )
+    if not handover_script.exists():
+        errors.append(f"missing file: {handover_script}")
 
     if errors:
         print("[FAIL] governance asset validation failed")
@@ -58,4 +93,3 @@ def main() -> int:
 
 if __name__ == "__main__":
     sys.exit(main())
-

src/analysis/backtest_pipeline.py

@@ -0,0 +1,187 @@
+"""Integrated v2 backtest pipeline.
+
+Wires TripleBarrier labeling + WalkForward split + CostGuard validation
+into a single deterministic orchestration path.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from statistics import mean
+from typing import Literal
+
+from src.analysis.backtest_cost_guard import BacktestCostModel, validate_backtest_cost_model
+from src.analysis.triple_barrier import TripleBarrierSpec, label_with_triple_barrier
+from src.analysis.walk_forward_split import WalkForwardFold, generate_walk_forward_splits
+
+
+@dataclass(frozen=True)
+class BacktestBar:
+    high: float
+    low: float
+    close: float
+    session_id: str
+
+
+@dataclass(frozen=True)
+class WalkForwardConfig:
+    train_size: int
+    test_size: int
+    step_size: int | None = None
+    purge_size: int = 0
+    embargo_size: int = 0
+    min_train_size: int = 1
+
+
+@dataclass(frozen=True)
+class BaselineScore:
+    name: Literal["B0", "B1", "M1"]
+    accuracy: float
+
+
+@dataclass(frozen=True)
+class BacktestFoldResult:
+    fold_index: int
+    train_indices: list[int]
+    test_indices: list[int]
+    train_label_distribution: dict[int, int]
+    test_label_distribution: dict[int, int]
+    baseline_scores: list[BaselineScore]
+
+
+@dataclass(frozen=True)
+class BacktestPipelineResult:
+    run_id: str
+    n_bars: int
+    n_entries: int
+    required_sessions: list[str]
+    label_distribution: dict[int, int]
+    folds: list[BacktestFoldResult]
+
+
+def run_v2_backtest_pipeline(
+    *,
+    bars: Sequence[BacktestBar],
+    entry_indices: Sequence[int],
+    side: int,
+    triple_barrier_spec: TripleBarrierSpec,
+    walk_forward: WalkForwardConfig,
+    cost_model: BacktestCostModel,
+    required_sessions: list[str] | None = None,
+) -> BacktestPipelineResult:
+    """Run v2 integrated pipeline (cost guard -> labels -> walk-forward baselines)."""
+    if not bars:
+        raise ValueError("bars must not be empty")
+    if not entry_indices:
+        raise ValueError("entry_indices must not be empty")
+
+    resolved_sessions = (
+        sorted(set(required_sessions))
+        if required_sessions is not None
+        else sorted({bar.session_id for bar in bars})
+    )
+    validate_backtest_cost_model(model=cost_model, required_sessions=resolved_sessions)
+
+    highs = [float(bar.high) for bar in bars]
+    lows = [float(bar.low) for bar in bars]
+    closes = [float(bar.close) for bar in bars]
+    normalized_entries = sorted(set(int(i) for i in entry_indices))
+    if normalized_entries[0] < 0 or normalized_entries[-1] >= len(bars):
+        raise IndexError("entry index out of range")
+
+    labels_by_bar_index: dict[int, int] = {}
+    for idx in normalized_entries:
+        labels_by_bar_index[idx] = label_with_triple_barrier(
+            highs=highs,
+            lows=lows,
+            closes=closes,
+            entry_index=idx,
+            side=side,
+            spec=triple_barrier_spec,
+        ).label
+
+    ordered_labels = [labels_by_bar_index[idx] for idx in normalized_entries]
+    folds = generate_walk_forward_splits(
+        n_samples=len(normalized_entries),
+        train_size=walk_forward.train_size,
+        test_size=walk_forward.test_size,
+        step_size=walk_forward.step_size,
+        purge_size=walk_forward.purge_size,
+        embargo_size=walk_forward.embargo_size,
+        min_train_size=walk_forward.min_train_size,
+    )
+
+    fold_results: list[BacktestFoldResult] = []
+    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]
+        if not test_labels:
+            continue
+        fold_results.append(
+            BacktestFoldResult(
+                fold_index=fold_idx,
+                train_indices=fold.train_indices,
+                test_indices=fold.test_indices,
+                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="M1",
+                        accuracy=_score_constant(_m1_pred(train_labels), test_labels),
+                    ),
+                ],
+            )
+        )
+
+    return BacktestPipelineResult(
+        run_id=_build_run_id(
+            n_entries=len(normalized_entries),
+            n_folds=len(fold_results),
+            sessions=resolved_sessions,
+        ),
+        n_bars=len(bars),
+        n_entries=len(normalized_entries),
+        required_sessions=resolved_sessions,
+        label_distribution=_label_dist(ordered_labels),
+        folds=fold_results,
+    )
+
+
+def _label_dist(labels: Sequence[int]) -> dict[int, int]:
+    dist: dict[int, int] = {-1: 0, 0: 0, 1: 0}
+    for val in labels:
+        if val in dist:
+            dist[val] += 1
+    return dist
+
+
+def _score_constant(pred: int, actual: Sequence[int]) -> float:
+    return mean(1.0 if pred == label else 0.0 for label in actual)
+
+
+def _baseline_b0(train_labels: Sequence[int], test_labels: Sequence[int]) -> float:
+    if not train_labels:
+        return _score_constant(0, test_labels)
+    # 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)
+
+
+def _m1_pred(train_labels: Sequence[int]) -> int:
+    if not train_labels:
+        return 0
+    return train_labels[-1]
+
+
+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}"
+
+
+def fold_has_leakage(fold: WalkForwardFold) -> bool:
+    """Utility for tests/verification: True when train/test overlap exists."""
+    return bool(set(fold.train_indices).intersection(fold.test_indices))

src/main.py

@@ -68,6 +68,8 @@ BLACKOUT_ORDER_MANAGER = BlackoutOrderManager(
     max_queue_size=500,
 )
 _SESSION_CLOSE_WINDOWS = {"NXT_AFTER", "US_AFTER"}
+_RUNTIME_EXIT_STATES: dict[str, PositionState] = {}
+_RUNTIME_EXIT_PEAKS: dict[str, float] = {}
 
 
 def safe_float(value: str | float | None, default: float = 0.0) -> float:
@@ -469,6 +471,118 @@ def _should_force_exit_for_overnight(
     return not settings.OVERNIGHT_EXCEPTION_ENABLED
 
 
+def _build_runtime_position_key(
+    *,
+    market_code: str,
+    stock_code: str,
+    open_position: dict[str, Any],
+) -> str:
+    decision_id = str(open_position.get("decision_id") or "")
+    timestamp = str(open_position.get("timestamp") or "")
+    return f"{market_code}:{stock_code}:{decision_id}:{timestamp}"
+
+
+def _clear_runtime_exit_cache_for_symbol(*, market_code: str, stock_code: str) -> None:
+    prefix = f"{market_code}:{stock_code}:"
+    stale_keys = [key for key in _RUNTIME_EXIT_STATES if key.startswith(prefix)]
+    for key in stale_keys:
+        _RUNTIME_EXIT_STATES.pop(key, None)
+        _RUNTIME_EXIT_PEAKS.pop(key, None)
+
+
+def _apply_staged_exit_override_for_hold(
+    *,
+    decision: TradeDecision,
+    market: MarketInfo,
+    stock_code: str,
+    open_position: dict[str, Any] | None,
+    market_data: dict[str, Any],
+    stock_playbook: Any | None,
+) -> TradeDecision:
+    """Apply v2 staged exit semantics for HOLD positions using runtime state."""
+    if decision.action != "HOLD" or not open_position:
+        return decision
+
+    entry_price = safe_float(open_position.get("price"), 0.0)
+    current_price = safe_float(market_data.get("current_price"), 0.0)
+    if entry_price <= 0 or current_price <= 0:
+        return decision
+
+    stop_loss_threshold = -2.0
+    take_profit_threshold = 3.0
+    if stock_playbook and stock_playbook.scenarios:
+        stop_loss_threshold = stock_playbook.scenarios[0].stop_loss_pct
+        take_profit_threshold = stock_playbook.scenarios[0].take_profit_pct
+
+    runtime_key = _build_runtime_position_key(
+        market_code=market.code,
+        stock_code=stock_code,
+        open_position=open_position,
+    )
+    current_state = _RUNTIME_EXIT_STATES.get(runtime_key, PositionState.HOLDING)
+    prev_peak = _RUNTIME_EXIT_PEAKS.get(runtime_key, 0.0)
+    peak_hint = max(
+        safe_float(market_data.get("peak_price"), 0.0),
+        safe_float(market_data.get("session_high_price"), 0.0),
+    )
+    peak_price = max(entry_price, current_price, prev_peak, peak_hint)
+
+    exit_eval = evaluate_exit(
+        current_state=current_state,
+        config=ExitRuleConfig(
+            hard_stop_pct=stop_loss_threshold,
+            be_arm_pct=max(0.5, take_profit_threshold * 0.4),
+            arm_pct=take_profit_threshold,
+        ),
+        inp=ExitRuleInput(
+            current_price=current_price,
+            entry_price=entry_price,
+            peak_price=peak_price,
+            atr_value=safe_float(market_data.get("atr_value"), 0.0),
+            pred_down_prob=safe_float(market_data.get("pred_down_prob"), 0.0),
+            liquidity_weak=safe_float(market_data.get("volume_ratio"), 1.0) < 1.0,
+        ),
+    )
+    _RUNTIME_EXIT_STATES[runtime_key] = exit_eval.state
+    _RUNTIME_EXIT_PEAKS[runtime_key] = peak_price
+
+    if not exit_eval.should_exit:
+        return decision
+
+    pnl_pct = (current_price - entry_price) / entry_price * 100.0
+    if exit_eval.reason == "hard_stop":
+        rationale = (
+            f"Stop-loss triggered ({pnl_pct:.2f}% <= "
+            f"{stop_loss_threshold:.2f}%)"
+        )
+    elif exit_eval.reason == "arm_take_profit":
+        rationale = (
+            f"Take-profit triggered ({pnl_pct:.2f}% >= "
+            f"{take_profit_threshold:.2f}%)"
+        )
+    elif exit_eval.reason == "atr_trailing_stop":
+        rationale = "ATR trailing-stop triggered"
+    elif exit_eval.reason == "be_lock_threat":
+        rationale = "Break-even lock threat detected"
+    elif exit_eval.reason == "model_liquidity_exit":
+        rationale = "Model/liquidity exit triggered"
+    else:
+        rationale = f"Exit rule triggered ({exit_eval.reason})"
+
+    logger.info(
+        "Staged exit override for %s (%s): HOLD -> SELL (reason=%s, state=%s)",
+        stock_code,
+        market.name,
+        exit_eval.reason,
+        exit_eval.state.value,
+    )
+    return TradeDecision(
+        action="SELL",
+        confidence=max(decision.confidence, 90),
+        rationale=rationale,
+    )
+
+
 async def build_overseas_symbol_universe(
     db_conn: Any,
     overseas_broker: OverseasBroker,
@@ -977,6 +1091,11 @@ async def trading_cycle(
         "foreigner_net": foreigner_net,
         "price_change_pct": price_change_pct,
     }
+    session_high_price = safe_float(
+        price_output.get("high") or price_output.get("ovrs_hgpr") or price_output.get("stck_hgpr")
+    )
+    if session_high_price > 0:
+        market_data["session_high_price"] = session_high_price
 
     # Enrich market_data with scanner metrics for scenario engine
     market_candidates = scan_candidates.get(market.code, {})
@@ -1175,82 +1294,36 @@ async def trading_cycle(
 
     if decision.action == "HOLD":
         open_position = get_open_position(db_conn, stock_code, market.code)
-        if open_position:
-            entry_price = safe_float(open_position.get("price"), 0.0)
-            if entry_price > 0 and current_price > 0:
-                loss_pct = (current_price - entry_price) / entry_price * 100
-                stop_loss_threshold = -2.0
-                take_profit_threshold = 3.0
-                if stock_playbook and stock_playbook.scenarios:
-                    stop_loss_threshold = stock_playbook.scenarios[0].stop_loss_pct
-                    take_profit_threshold = stock_playbook.scenarios[0].take_profit_pct
-
-                exit_eval = evaluate_exit(
-                    current_state=PositionState.HOLDING,
-                    config=ExitRuleConfig(
-                        hard_stop_pct=stop_loss_threshold,
-                        be_arm_pct=max(0.5, take_profit_threshold * 0.4),
-                        arm_pct=take_profit_threshold,
-                    ),
-                    inp=ExitRuleInput(
-                        current_price=current_price,
-                        entry_price=entry_price,
-                        peak_price=max(entry_price, current_price),
-                        atr_value=0.0,
-                        pred_down_prob=0.0,
-                        liquidity_weak=market_data.get("volume_ratio", 1.0) < 1.0,
-                    ),
-                )
-
-                if exit_eval.reason == "hard_stop":
-                    decision = TradeDecision(
-                        action="SELL",
-                        confidence=95,
-                        rationale=(
-                            f"Stop-loss triggered ({loss_pct:.2f}% <= "
-                            f"{stop_loss_threshold:.2f}%)"
-                        ),
-                    )
-                    logger.info(
-                        "Stop-loss override for %s (%s): %.2f%% <= %.2f%%",
-                        stock_code,
-                        market.name,
-                        loss_pct,
-                        stop_loss_threshold,
-                    )
-                elif exit_eval.reason == "arm_take_profit":
-                    decision = TradeDecision(
-                        action="SELL",
-                        confidence=90,
-                        rationale=(
-                            f"Take-profit triggered ({loss_pct:.2f}% >= "
-                            f"{take_profit_threshold:.2f}%)"
-                        ),
-                    )
-                    logger.info(
-                        "Take-profit override for %s (%s): %.2f%% >= %.2f%%",
-                        stock_code,
-                        market.name,
-                        loss_pct,
-                        take_profit_threshold,
-                    )
-            if decision.action == "HOLD" and _should_force_exit_for_overnight(
+        if not open_position:
+            _clear_runtime_exit_cache_for_symbol(
+                market_code=market.code,
+                stock_code=stock_code,
+            )
+        decision = _apply_staged_exit_override_for_hold(
+            decision=decision,
+            market=market,
+            stock_code=stock_code,
+            open_position=open_position,
+            market_data=market_data,
+            stock_playbook=stock_playbook,
+        )
+        if open_position and decision.action == "HOLD" and _should_force_exit_for_overnight(
                 market=market,
                 settings=settings,
-            ):
-                decision = TradeDecision(
-                    action="SELL",
-                    confidence=max(decision.confidence, 85),
-                    rationale=(
-                        "Forced exit by overnight policy"
-                        " (session close window / kill switch priority)"
-                    ),
-                )
-                logger.info(
-                    "Overnight policy override for %s (%s): HOLD -> SELL",
-                    stock_code,
-                    market.name,
-                )
+        ):
+            decision = TradeDecision(
+                action="SELL",
+                confidence=max(decision.confidence, 85),
+                rationale=(
+                    "Forced exit by overnight policy"
+                    " (session close window / kill switch priority)"
+                ),
+            )
+            logger.info(
+                "Overnight policy override for %s (%s): HOLD -> SELL",
+                stock_code,
+                market.name,
+            )
     logger.info(
         "Decision for %s (%s): %s (confidence=%d)",
         stock_code,
@@ -2190,6 +2263,14 @@ async def run_daily_session(
                     "foreigner_net": foreigner_net,
                     "price_change_pct": price_change_pct,
                 }
+                if not market.is_domestic:
+                    session_high_price = safe_float(
+                        price_data.get("output", {}).get("high")
+                        or price_data.get("output", {}).get("ovrs_hgpr")
+                        or price_data.get("output", {}).get("stck_hgpr")
+                    )
+                    if session_high_price > 0:
+                        stock_data["session_high_price"] = session_high_price
                 # Enrich with scanner metrics
                 cand = candidate_map.get(stock_code)
                 if cand:
@@ -2317,6 +2398,7 @@ async def run_daily_session(
         )
         for stock_data in stocks_data:
             stock_code = stock_data["stock_code"]
+            stock_playbook = playbook.get_stock_playbook(stock_code)
             match = scenario_engine.evaluate(
                 playbook, stock_code, stock_data, portfolio_data,
             )
@@ -2362,7 +2444,20 @@ async def run_daily_session(
                     )
             if decision.action == "HOLD":
                 daily_open = get_open_position(db_conn, stock_code, market.code)
-                if daily_open and _should_force_exit_for_overnight(
+                if not daily_open:
+                    _clear_runtime_exit_cache_for_symbol(
+                        market_code=market.code,
+                        stock_code=stock_code,
+                    )
+                decision = _apply_staged_exit_override_for_hold(
+                    decision=decision,
+                    market=market,
+                    stock_code=stock_code,
+                    open_position=daily_open,
+                    market_data=stock_data,
+                    stock_playbook=stock_playbook,
+                )
+                if daily_open and decision.action == "HOLD" and _should_force_exit_for_overnight(
                     market=market,
                     settings=settings,
                 ):

tests/test_backtest_pipeline_integration.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+from src.analysis.backtest_cost_guard import BacktestCostModel
+from src.analysis.backtest_pipeline import (
+    BacktestBar,
+    WalkForwardConfig,
+    fold_has_leakage,
+    run_v2_backtest_pipeline,
+)
+from src.analysis.triple_barrier import TripleBarrierSpec
+from src.analysis.walk_forward_split import generate_walk_forward_splits
+
+
+def _bars() -> list[BacktestBar]:
+    closes = [100.0, 101.0, 102.0, 101.5, 103.0, 102.5, 104.0, 103.5, 105.0, 104.5, 106.0, 105.5]
+    bars: list[BacktestBar] = []
+    for i, close in enumerate(closes):
+        bars.append(
+            BacktestBar(
+                high=close + 1.0,
+                low=close - 1.0,
+                close=close,
+                session_id="KRX_REG" if i % 2 == 0 else "US_PRE",
+            )
+        )
+    return bars
+
+
+def _cost_model() -> BacktestCostModel:
+    return 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},
+        unfavorable_fill_required=True,
+    )
+
+
+def test_pipeline_happy_path_returns_fold_and_artifact_contract() -> None:
+    out = run_v2_backtest_pipeline(
+        bars=_bars(),
+        entry_indices=[0, 1, 2, 3, 4, 5, 6, 7],
+        side=1,
+        triple_barrier_spec=TripleBarrierSpec(
+            take_profit_pct=0.02,
+            stop_loss_pct=0.01,
+            max_holding_bars=3,
+        ),
+        walk_forward=WalkForwardConfig(
+            train_size=4,
+            test_size=2,
+            step_size=2,
+            purge_size=1,
+            embargo_size=1,
+            min_train_size=3,
+        ),
+        cost_model=_cost_model(),
+    )
+
+    assert out.run_id.startswith("v2p-e8-f")
+    assert out.n_bars == 12
+    assert out.n_entries == 8
+    assert out.required_sessions == ["KRX_REG", "US_PRE"]
+    assert len(out.folds) > 0
+    assert set(out.label_distribution) == {-1, 0, 1}
+    for fold in out.folds:
+        names = {score.name for score in fold.baseline_scores}
+        assert names == {"B0", "B1", "M1"}
+        for score in fold.baseline_scores:
+            assert 0.0 <= score.accuracy <= 1.0
+
+
+def test_pipeline_cost_guard_fail_fast() -> None:
+    bad = BacktestCostModel(
+        commission_bps=3.0,
+        slippage_bps_by_session={"KRX_REG": 10.0},
+        failure_rate_by_session={"KRX_REG": 0.01},
+        unfavorable_fill_required=True,
+    )
+    try:
+        run_v2_backtest_pipeline(
+            bars=_bars(),
+            entry_indices=[0, 1, 2, 3],
+            side=1,
+            triple_barrier_spec=TripleBarrierSpec(
+                take_profit_pct=0.02,
+                stop_loss_pct=0.01,
+                max_holding_bars=3,
+            ),
+            walk_forward=WalkForwardConfig(train_size=2, test_size=1),
+            cost_model=bad,
+            required_sessions=["KRX_REG", "US_PRE"],
+        )
+    except ValueError as exc:
+        assert "missing slippage_bps_by_session" in str(exc)
+    else:
+        raise AssertionError("expected cost guard validation error")
+
+
+def test_pipeline_fold_leakage_guard() -> None:
+    folds = generate_walk_forward_splits(
+        n_samples=12,
+        train_size=6,
+        test_size=2,
+        step_size=2,
+        purge_size=1,
+        embargo_size=1,
+        min_train_size=5,
+    )
+    assert folds
+    for fold in folds:
+        assert not fold_has_leakage(fold)
+
+
+def test_pipeline_deterministic_seed_free_deterministic_result() -> None:
+    cfg = dict(
+        bars=_bars(),
+        entry_indices=[0, 1, 2, 3, 4, 5, 6, 7],
+        side=1,
+        triple_barrier_spec=TripleBarrierSpec(
+            take_profit_pct=0.02,
+            stop_loss_pct=0.01,
+            max_holding_bars=3,
+        ),
+        walk_forward=WalkForwardConfig(
+            train_size=4,
+            test_size=2,
+            step_size=2,
+            purge_size=1,
+            embargo_size=1,
+            min_train_size=3,
+        ),
+        cost_model=_cost_model(),
+    )
+    out1 = run_v2_backtest_pipeline(**cfg)
+    out2 = run_v2_backtest_pipeline(**cfg)
+    assert out1 == out2

tests/test_main.py

@@ -15,6 +15,8 @@ from src.evolution.scorecard import DailyScorecard
 from src.logging.decision_logger import DecisionLogger
 from src.main import (
     KILL_SWITCH,
+    _RUNTIME_EXIT_PEAKS,
+    _RUNTIME_EXIT_STATES,
     _should_force_exit_for_overnight,
     _should_block_overseas_buy_for_fx_buffer,
     _trigger_emergency_kill_switch,
@@ -42,6 +44,7 @@ from src.strategy.models import (
     StockCondition,
     StockScenario,
 )
+from src.strategy.position_state_machine import PositionState
 from src.strategy.scenario_engine import ScenarioEngine, ScenarioMatch
 
 
@@ -87,8 +90,12 @@ def _make_sell_match(stock_code: str = "005930") -> ScenarioMatch:
 def _reset_kill_switch_state() -> None:
     """Prevent cross-test leakage from global kill-switch state."""
     KILL_SWITCH.clear_block()
+    _RUNTIME_EXIT_STATES.clear()
+    _RUNTIME_EXIT_PEAKS.clear()
     yield
     KILL_SWITCH.clear_block()
+    _RUNTIME_EXIT_STATES.clear()
+    _RUNTIME_EXIT_PEAKS.clear()
 
 
 class TestExtractAvgPriceFromBalance:
@@ -2337,6 +2344,218 @@ async def test_hold_not_overridden_when_between_stop_loss_and_take_profit() -> N
     broker.send_order.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_hold_overridden_to_sell_on_be_lock_threat_after_state_arms() -> None:
+    """Staged exit must use runtime state (BE_LOCK -> be_lock_threat -> SELL)."""
+    db_conn = init_db(":memory:")
+    decision_logger = DecisionLogger(db_conn)
+
+    buy_decision_id = decision_logger.log_decision(
+        stock_code="005930",
+        market="KR",
+        exchange_code="KRX",
+        action="BUY",
+        confidence=90,
+        rationale="entry",
+        context_snapshot={},
+        input_data={},
+    )
+    log_trade(
+        conn=db_conn,
+        stock_code="005930",
+        action="BUY",
+        confidence=90,
+        rationale="entry",
+        quantity=1,
+        price=100.0,
+        market="KR",
+        exchange_code="KRX",
+        decision_id=buy_decision_id,
+    )
+
+    broker = MagicMock()
+    broker.get_current_price = AsyncMock(side_effect=[(102.0, 2.0, 0.0), (99.0, -1.0, 0.0)])
+    broker.get_balance = AsyncMock(
+        return_value={
+            "output1": [{"pdno": "005930", "ord_psbl_qty": "1"}],
+            "output2": [
+                {
+                    "tot_evlu_amt": "100000",
+                    "dnca_tot_amt": "10000",
+                    "pchs_amt_smtl_amt": "90000",
+                }
+            ],
+        }
+    )
+    broker.send_order = AsyncMock(return_value={"msg1": "OK"})
+
+    scenario = StockScenario(
+        condition=StockCondition(rsi_below=30),
+        action=ScenarioAction.BUY,
+        confidence=88,
+        stop_loss_pct=-5.0,
+        take_profit_pct=3.0,
+        rationale="staged exit policy",
+    )
+    playbook = DayPlaybook(
+        date=date(2026, 2, 8),
+        market="KR",
+        stock_playbooks=[
+            {"stock_code": "005930", "stock_name": "Samsung", "scenarios": [scenario]}
+        ],
+    )
+    engine = MagicMock(spec=ScenarioEngine)
+    engine.evaluate = MagicMock(return_value=_make_hold_match())
+
+    market = MagicMock()
+    market.name = "Korea"
+    market.code = "KR"
+    market.exchange_code = "KRX"
+    market.is_domestic = True
+
+    telegram = MagicMock()
+    telegram.notify_trade_execution = AsyncMock()
+    telegram.notify_fat_finger = AsyncMock()
+    telegram.notify_circuit_breaker = AsyncMock()
+    telegram.notify_scenario_matched = AsyncMock()
+
+    for _ in range(2):
+        await trading_cycle(
+            broker=broker,
+            overseas_broker=MagicMock(),
+            scenario_engine=engine,
+            playbook=playbook,
+            risk=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="005930",
+            scan_candidates={},
+        )
+
+    broker.send_order.assert_called_once()
+    assert broker.send_order.call_args.kwargs["order_type"] == "SELL"
+
+
+@pytest.mark.asyncio
+async def test_runtime_exit_cache_cleared_when_position_closed() -> None:
+    """Runtime staged-exit cache must be cleared when no open position exists."""
+    db_conn = init_db(":memory:")
+    decision_logger = DecisionLogger(db_conn)
+
+    buy_decision_id = decision_logger.log_decision(
+        stock_code="005930",
+        market="KR",
+        exchange_code="KRX",
+        action="BUY",
+        confidence=90,
+        rationale="entry",
+        context_snapshot={},
+        input_data={},
+    )
+    log_trade(
+        conn=db_conn,
+        stock_code="005930",
+        action="BUY",
+        confidence=90,
+        rationale="entry",
+        quantity=1,
+        price=100.0,
+        market="KR",
+        exchange_code="KRX",
+        decision_id=buy_decision_id,
+    )
+
+    broker = MagicMock()
+    broker.get_current_price = AsyncMock(return_value=(100.0, 0.0, 0.0))
+    broker.get_balance = AsyncMock(
+        return_value={
+            "output1": [{"pdno": "005930", "ord_psbl_qty": "1"}],
+            "output2": [
+                {
+                    "tot_evlu_amt": "100000",
+                    "dnca_tot_amt": "10000",
+                    "pchs_amt_smtl_amt": "90000",
+                }
+            ],
+        }
+    )
+    broker.send_order = AsyncMock(return_value={"msg1": "OK"})
+
+    market = MagicMock()
+    market.name = "Korea"
+    market.code = "KR"
+    market.exchange_code = "KRX"
+    market.is_domestic = True
+
+    telegram = MagicMock()
+    telegram.notify_trade_execution = AsyncMock()
+    telegram.notify_fat_finger = AsyncMock()
+    telegram.notify_circuit_breaker = AsyncMock()
+    telegram.notify_scenario_matched = AsyncMock()
+
+    _RUNTIME_EXIT_STATES[f"{market.code}:005930:{buy_decision_id}:dummy-ts"] = PositionState.BE_LOCK
+    _RUNTIME_EXIT_PEAKS[f"{market.code}:005930:{buy_decision_id}:dummy-ts"] = 120.0
+
+    # Close position first so trading_cycle observes no open position.
+    sell_decision_id = decision_logger.log_decision(
+        stock_code="005930",
+        market="KR",
+        exchange_code="KRX",
+        action="SELL",
+        confidence=90,
+        rationale="manual close",
+        context_snapshot={},
+        input_data={},
+    )
+    log_trade(
+        conn=db_conn,
+        stock_code="005930",
+        action="SELL",
+        confidence=90,
+        rationale="manual close",
+        quantity=1,
+        price=100.0,
+        market="KR",
+        exchange_code="KRX",
+        decision_id=sell_decision_id,
+    )
+
+    await trading_cycle(
+        broker=broker,
+        overseas_broker=MagicMock(),
+        scenario_engine=MagicMock(evaluate=MagicMock(return_value=_make_hold_match())),
+        playbook=_make_playbook(),
+        risk=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="005930",
+        scan_candidates={},
+    )
+
+    assert not [k for k in _RUNTIME_EXIT_STATES if k.startswith("KR:005930:")]
+    assert not [k for k in _RUNTIME_EXIT_PEAKS if k.startswith("KR:005930:")]
+
+
 @pytest.mark.asyncio
 async def test_stop_loss_not_triggered_when_current_price_is_zero() -> None:
     """HOLD must stay HOLD when current_price=0 even if entry_price is set (issue #251).
@@ -4135,6 +4354,130 @@ class TestDailyCBBaseline:
         assert result == 55000.0
 
 
+@pytest.mark.asyncio
+async def test_run_daily_session_applies_staged_exit_override_on_hold() -> None:
+    """run_daily_session must apply HOLD staged exit semantics (issue #304)."""
+    from src.analysis.smart_scanner import ScanCandidate
+
+    db_conn = init_db(":memory:")
+    log_trade(
+        conn=db_conn,
+        stock_code="005930",
+        action="BUY",
+        confidence=90,
+        rationale="entry",
+        quantity=1,
+        price=100.0,
+        market="KR",
+        exchange_code="KRX",
+        decision_id="buy-d1",
+    )
+
+    settings = Settings(
+        KIS_APP_KEY="k",
+        KIS_APP_SECRET="s",
+        KIS_ACCOUNT_NO="12345678-01",
+        GEMINI_API_KEY="g",
+        MODE="paper",
+    )
+
+    broker = MagicMock()
+    broker.get_balance = AsyncMock(
+        return_value={
+            "output1": [{"pdno": "005930", "ord_psbl_qty": "1"}],
+            "output2": [
+                {
+                    "tot_evlu_amt": "100000",
+                    "dnca_tot_amt": "10000",
+                    "pchs_amt_smtl_amt": "90000",
+                }
+            ],
+        }
+    )
+    broker.get_current_price = AsyncMock(return_value=(95.0, -5.0, 0.0))
+    broker.send_order = AsyncMock(return_value={"msg1": "OK"})
+
+    market = MagicMock()
+    market.name = "Korea"
+    market.code = "KR"
+    market.exchange_code = "KRX"
+    market.is_domestic = True
+    market.timezone = __import__("zoneinfo").ZoneInfo("Asia/Seoul")
+
+    scenario = StockScenario(
+        condition=StockCondition(rsi_below=30),
+        action=ScenarioAction.BUY,
+        confidence=88,
+        stop_loss_pct=-2.0,
+        take_profit_pct=3.0,
+        rationale="stop loss policy",
+    )
+    playbook = DayPlaybook(
+        date=date(2026, 2, 8),
+        market="KR",
+        stock_playbooks=[
+            {"stock_code": "005930", "stock_name": "Samsung", "scenarios": [scenario]}
+        ],
+    )
+    playbook_store = MagicMock()
+    playbook_store.load = MagicMock(return_value=playbook)
+
+    smart_scanner = MagicMock()
+    smart_scanner.scan = AsyncMock(
+        return_value=[
+            ScanCandidate(
+                stock_code="005930",
+                name="Samsung",
+                price=95.0,
+                volume=1_000_000.0,
+                volume_ratio=2.0,
+                rsi=42.0,
+                signal="momentum",
+                score=80.0,
+            )
+        ]
+    )
+
+    scenario_engine = MagicMock(spec=ScenarioEngine)
+    scenario_engine.evaluate = MagicMock(return_value=_make_hold_match("005930"))
+
+    risk = MagicMock()
+    risk.check_circuit_breaker = MagicMock()
+    risk.validate_order = MagicMock()
+
+    decision_logger = MagicMock()
+    decision_logger.log_decision = MagicMock(return_value="d1")
+
+    telegram = MagicMock()
+    telegram.notify_trade_execution = AsyncMock()
+    telegram.notify_scenario_matched = AsyncMock()
+
+    async def _passthrough(fn, *a, label: str = "", **kw):  # type: ignore[override]
+        return await fn(*a, **kw)
+
+    with patch("src.main.get_open_markets", return_value=[market]), \
+         patch("src.main._retry_connection", new=_passthrough):
+        await run_daily_session(
+            broker=broker,
+            overseas_broker=MagicMock(),
+            scenario_engine=scenario_engine,
+            playbook_store=playbook_store,
+            pre_market_planner=MagicMock(),
+            risk=risk,
+            db_conn=db_conn,
+            decision_logger=decision_logger,
+            context_store=MagicMock(),
+            criticality_assessor=MagicMock(),
+            telegram=telegram,
+            settings=settings,
+            smart_scanner=smart_scanner,
+            daily_start_eval=0.0,
+        )
+
+    broker.send_order.assert_called_once()
+    assert broker.send_order.call_args.kwargs["order_type"] == "SELL"
+
+
 # ---------------------------------------------------------------------------
 # sync_positions_from_broker — startup DB sync tests (issue #206)
 # ---------------------------------------------------------------------------

workflow/session-handover.md

@@ -0,0 +1,83 @@
+# Session Handover Log
+
+목적: 세션 시작 시 인수인계 확인을 기록하고, 구현/검증 작업 시작 전에 공통 컨텍스트를 강제한다.
+
+작성 규칙:
+- 세션 시작마다 최신 엔트리를 맨 아래에 추가한다.
+- `docs/workflow.md`, `docs/commands.md`, `docs/agent-constraints.md`를 먼저 확인한 뒤 기록한다.
+- 각 엔트리는 현재 작업 브랜치 기준으로 작성한다.
+
+템플릿:
+
+```md
+### YYYY-MM-DD | session=<id or short label>
+- branch: <current-branch>
+- docs_checked: docs/workflow.md, docs/commands.md, docs/agent-constraints.md
+- open_issues_reviewed: #...
+- next_ticket: #...
+- process_gate_checked: process_ticket=#..., merged_to_feature_branch=yes|no|n/a
+- risks_or_notes: ...
+```
+
+### 2026-02-27 | session=handover-gate-bootstrap
+- branch: feature/v3-session-policy-stream
+- docs_checked: docs/workflow.md, docs/commands.md, docs/agent-constraints.md
+- open_issues_reviewed: #304, #305, #306
+- next_ticket: #304
+- risks_or_notes: 세션 시작 게이트를 문서/스크립트/CI로 강제 적용
+
+### 2026-02-27 | session=codex-handover-start
+- branch: feature/v3-session-policy-stream
+- docs_checked: docs/workflow.md, docs/commands.md, docs/agent-constraints.md
+- open_issues_reviewed: #306, #308, #309
+- next_ticket: #304
+- process_gate_checked: process_ticket=#306,#308 merged_to_feature_branch=yes
+- risks_or_notes: 미추적 로컬 파일 존재(문서/DB/lock)로 커밋 범위 분리 필요
+
+### 2026-02-27 | session=codex-process-gate-hardening
+- branch: feature/issue-304-runtime-staged-exit-semantics
+- docs_checked: docs/workflow.md, docs/commands.md, docs/agent-constraints.md
+- open_issues_reviewed: #304, #305
+- next_ticket: #304
+- process_gate_checked: process_ticket=#306,#308 merged_to_feature_branch=yes
+- risks_or_notes: process-change-first 실행 게이트를 문서+스크립트로 강화
+
+### 2026-02-27 | session=codex-handover-start-2
+- branch: feature/issue-304-runtime-staged-exit-semantics
+- docs_checked: docs/workflow.md, docs/commands.md, docs/agent-constraints.md
+- open_issues_reviewed: #304, #305
+- next_ticket: #304
+- process_gate_checked: process_ticket=#306,#308 merged_to_feature_branch=yes
+- risks_or_notes: handover 재시작 요청으로 세션 엔트리 추가, 미추적 산출물(AMS/NAS/NYS, DB, lock, xlsx) 커밋 분리 필요
+
+### 2026-02-27 | session=codex-issue305-start
+- branch: feature/v3-session-policy-stream
+- docs_checked: docs/workflow.md, docs/commands.md, docs/agent-constraints.md
+- open_issues_reviewed: #305
+- next_ticket: #305
+- process_gate_checked: process_ticket=#306,#308 merged_to_feature_branch=yes
+- risks_or_notes: #305 구현을 위해 분석/백테스트 모듈 통합 경로 점검 시작
+
+### 2026-02-27 | session=codex-issue305-ticket-branch
+- branch: feature/issue-305-backtest-pipeline-integration
+- docs_checked: docs/workflow.md, docs/commands.md, docs/agent-constraints.md
+- open_issues_reviewed: #305
+- next_ticket: #305
+- process_gate_checked: process_ticket=#306,#308 merged_to_feature_branch=yes
+- risks_or_notes: 티켓 브랜치 분기 후 strict gate 재통과를 위한 엔트리 추가
+
+### 2026-02-27 | session=codex-backtest-gate-automation
+- branch: feature/v3-session-policy-stream
+- docs_checked: docs/workflow.md, docs/commands.md, docs/agent-constraints.md
+- open_issues_reviewed: #304, #305
+- next_ticket: (create) backtest automation gate
+- process_gate_checked: process_ticket=#306,#308 merged_to_feature_branch=yes
+- risks_or_notes: 백테스트 자동화 누락 재발 방지 위해 이슈/티켓 브랜치/PR 절차로 즉시 정규화
+
+### 2026-02-27 | session=codex-issue314-ticket-branch
+- branch: feature/issue-314-backtest-gate-automation
+- docs_checked: docs/workflow.md, docs/commands.md, docs/agent-constraints.md
+- open_issues_reviewed: #314
+- next_ticket: #314
+- process_gate_checked: process_ticket=#306,#308 merged_to_feature_branch=yes
+- risks_or_notes: 백테스트 자동 게이트 도입 티켓 브랜치 strict gate 통과용 엔트리