Compare commits

...

35 Commits

Author SHA1 Message Date
agentson
694d73b212 fix: lazy session resolver and one-time session_id backfill 2026-02-27 08:51:00 +09:00
agentson
b2b02b6f57 feat: enforce session_id persistence in trade ledger (TASK-CODE-007) 2026-02-27 08:49:04 +09:00
2dbe98615d Merge pull request '[FX-ACCOUNTING] TKT-P1-006 전략/환율 PnL 분리 회계' (#296) from feature/issue-tkt-p1-006-fx-pnl-separation into feature/v3-session-policy-stream 2026-02-27 08:46:56 +09:00
agentson
34cf081c96 fix: backfill split pnl migration and harden partial pnl inputs 2026-02-27 08:46:22 +09:00
agentson
7bc4e88335 feat: separate strategy and fx pnl fields in trade logs (TASK-CODE-011) 2026-02-27 08:44:05 +09:00
386e039ff6 Merge pull request '[BACKTEST-MODEL] TKT-P1-005 보수적 체결 모델 구현' (#294) from feature/issue-tkt-p1-005-conservative-fill-model into feature/v3-session-policy-stream 2026-02-27 08:42:22 +09:00
agentson
13ba9e8081 fix: validate execution assumption ranges in backtest model 2026-02-27 08:41:56 +09:00
agentson
5b52f593a8 feat: add conservative backtest execution simulator (TASK-CODE-010) 2026-02-27 08:40:23 +09:00
2798558bf3 Merge pull request '[BACKTEST-MODEL] TKT-P1-002 백테스트 비용/슬리피지 옵션 필수화' (#292) from feature/issue-tkt-p1-002-backtest-cost-mandatory into feature/v3-session-policy-stream 2026-02-27 08:37:15 +09:00
agentson
2331d80915 fix: reject non-finite backtest cost assumptions 2026-02-27 08:36:38 +09:00
agentson
7d72669cb8 feat: enforce mandatory backtest cost assumptions (TASK-CODE-006) 2026-02-27 08:34:44 +09:00
74a4784b7a Merge pull request '[BACKTEST-MODEL] TKT-P1-004 Walk-forward + Purge/Embargo 분할 유틸' (#290) from feature/issue-tkt-p1-004-walkforward-purge-embargo into feature/v3-session-policy-stream 2026-02-27 08:33:01 +09:00
agentson
dc70311aed fix: keep embargo tied to accepted folds and enforce PR-comment decision logs 2026-02-27 08:32:09 +09:00
agentson
e56819e9e2 feat: add walk-forward splitter with purge and embargo controls (TASK-CODE-005) 2026-02-27 08:28:11 +09:00
cfd5351b58 Merge pull request '[FX-ACCOUNTING] TKT-P1-001 USD/KRW 버퍼 진입 제한' (#288) from feature/issue-tkt-p1-001-fx-buffer-guard into feature/v3-session-policy-stream 2026-02-27 00:53:21 +09:00
agentson
b206c23fc9 fix: scope USD buffer guard to US markets and add boundary tests 2026-02-27 00:52:44 +09:00
agentson
4d9f3e2cfc feat: enforce overseas buy guard with USD buffer threshold (TASK-V3-014) 2026-02-27 00:50:12 +09:00
a93a5c616b Merge pull request '[BACKTEST-MODEL] TKT-P1-003 Triple Barrier 라벨러 구현' (#286) from feature/issue-tkt-p1-003-triple-barrier-labeler into feature/v3-session-policy-stream 2026-02-27 00:47:37 +09:00
agentson
9f64c9944a fix: correct short-side tie-break semantics in triple barrier 2026-02-27 00:47:09 +09:00
agentson
bb391d502c feat: add triple barrier labeler with first-touch logic (TASK-CODE-004) 2026-02-27 00:45:18 +09:00
b0100fde10 Merge pull request '[RISK-EMERGENCY][SCN-FAIL-003] TKT-P0-002 Kill Switch 순서 강제 검증 자동화' (#284) from feature/issue-tkt-p0-002-killswitch-ordering into feature/v3-session-policy-stream 2026-02-27 00:42:16 +09:00
agentson
0a4e69d40c fix: record kill switch cancel failures and add failure-path tests 2026-02-27 00:41:13 +09:00
agentson
25401ac132 feat: enforce operational kill switch callbacks in runtime flow (TASK-CODE-003) 2026-02-27 00:38:26 +09:00
1381b140ab Merge pull request '[EXEC-POLICY][SCN-FAIL-001] TKT-P0-001 블랙아웃 차단/큐/복구 재검증' (#282) from feature/issue-tkt-p0-001-blackout-queue-revalidate into feature/v3-session-policy-stream 2026-02-27 00:32:59 +09:00
agentson
356d085ab0 feat: implement blackout queue and recovery revalidation (TASK-CODE-008) 2026-02-27 00:31:29 +09:00
54d6cc3d7c Merge pull request 'docs: feature-branch 팀 운영 규칙 및 모니터링 검증 게이트 반영 (#279)' (#280) from feature/issue-279-session-order-policy-guard into feature/v3-session-policy-stream 2026-02-27 00:19:55 +09:00
agentson
3ffad58d57 docs: allow ticket->feature merges without user approval; keep main gated (#279) 2026-02-27 00:19:51 +09:00
agentson
df6baee7f1 feat: add session-aware order policy guard for low-liquidity market-order rejection (#279) 2026-02-27 00:13:47 +09:00
agentson
c31a6a569d docs: enforce feature-branch team flow and mandatory runtime monitoring validation (#279) 2026-02-27 00:05:01 +09:00
990f9696ab Merge pull request 'docs: TPM 티켓 우선순위/메인 아이디에이션/무머지 세션 규칙 반영 (#277)' (#278) from feature/issue-277-tpm-priority-main-ideation-no-merge-session into main
Some checks are pending
CI / test (push) Waiting to run
2026-02-26 23:58:03 +09:00
agentson
9bf72c63ec docs: clarify no-merge-by-default server reflection rule (#277)
Some checks are pending
CI / test (pull_request) Waiting to run
2026-02-26 23:57:58 +09:00
agentson
1399fa4d09 docs: enforce TPM ticket ownership and add main-agent ideation backlog (#277)
Some checks are pending
CI / test (pull_request) Waiting to run
2026-02-26 23:56:25 +09:00
f63fb53289 Merge pull request 'feat: phase1 상태기계/청산엔진/kill-switch 구현 (#275)' (#276) from feature/issue-275-phase1-state-exit-killswitch into main
Some checks failed
CI / test (push) Has been cancelled
2026-02-26 23:46:11 +09:00
agentson
5050a4cf84 fix: address reviewer feedback for kill-switch enforcement and observability (#275)
Some checks are pending
CI / test (pull_request) Waiting to run
2026-02-26 23:46:02 +09:00
agentson
4987b6393a feat: implement phase1 state machine, composite exits, and kill-switch orchestration (#275)
Some checks are pending
CI / test (pull_request) Waiting to run
2026-02-26 23:22:58 +09:00
29 changed files with 2938 additions and 21 deletions

View File

@@ -43,6 +43,11 @@ Updated: 2026-02-26
- 기존 `tests/` 스위트 전량 실행 - 기존 `tests/` 스위트 전량 실행
- 신규 기능 플래그 ON/OFF 비교 - 신규 기능 플래그 ON/OFF 비교
4. 구동/모니터링 검증 (필수)
- 개발 완료 후 시스템을 실제 구동해 핵심 경로를 관찰
- 필수 관찰 항목: 주문 차단 정책, Kill Switch 동작, 경보/예외 로그, 세션 전환 로그
- Runtime Verifier 코멘트로 증적(실행 명령/요약 로그) 첨부
## 실행 명령 ## 실행 명령
```bash ```bash
@@ -55,3 +60,4 @@ python3 scripts/validate_ouroboros_docs.py
- 문서 검증 실패 시 구현 PR 병합 금지 - 문서 검증 실패 시 구현 PR 병합 금지
- `REQ-*` 변경 후 테스트 매핑 누락 시 병합 금지 - `REQ-*` 변경 후 테스트 매핑 누락 시 병합 금지
- 회귀 실패 시 원인 모듈 분리 후 재검증 - 회귀 실패 시 원인 모듈 분리 후 재검증
- 구동/모니터링 증적 누락 시 검증 승인 금지

View File

@@ -17,11 +17,16 @@ Updated: 2026-02-26
- Main Agent: 최종 취합/우선순위/승인 게이트 오너 - Main Agent: 최종 취합/우선순위/승인 게이트 오너
- PM Agent: 시나리오/요구사항/티켓 관리 - PM Agent: 시나리오/요구사항/티켓 관리
- TPM Agent: PM-Dev-검증 간 구현 가능성/달성률 통제 - TPM Agent: PM-Dev-검증 간 구현 가능성/달성률 통제, 티켓 등록 및 구현 우선순위 지정 오너
- Dev Agent: 구현 수행, 블로커 발생 시 재계획 요청 - Dev Agent: 구현 수행, 블로커 발생 시 재계획 요청
- Verifier Agent: 문서/코드/테스트 산출물 검증 - Verifier Agent: 문서/코드/테스트 산출물 검증
- Runtime Verifier Agent: 실제 동작 모니터링, 이상 징후 이슈 발행, 수정 후 이슈 클로즈 판정 - Runtime Verifier Agent: 실제 동작 모니터링, 이상 징후 이슈 발행, 수정 후 이슈 클로즈 판정
Main Agent 아이디에이션 책임:
- 진행 중 신규 구현 아이디어를 별도 문서에 누적 기록한다.
- 기록 위치: [70_main_agent_ideation.md](./70_main_agent_ideation.md)
- 각 항목은 `IDEA-*` 식별자, 배경, 기대효과, 리스크, 후속 티켓 후보를 포함해야 한다.
## Main Decision Checkpoints (Mandatory) ## Main Decision Checkpoints (Mandatory)
- DCP-01 범위 확정: Phase 0 종료 전 Main Agent 승인 필수 - DCP-01 범위 확정: Phase 0 종료 전 Main Agent 승인 필수
@@ -140,6 +145,16 @@ Control checks:
- Main Agent 승인 없는 재계획은 실행 금지 - Main Agent 승인 없는 재계획은 실행 금지
- 재계획 반영 시 문서(`REQ/TASK/TEST`) 동시 갱신 필수 - 재계획 반영 시 문서(`REQ/TASK/TEST`) 동시 갱신 필수
TPM 티켓 운영 규칙:
- TPM은 합의된 변경을 이슈로 등록하고 우선순위(`P0/P1/P2`)를 지정한다.
- PR 본문에는 TPM이 지정한 우선순위와 범위가 그대로 반영되어야 한다.
- 우선순위 변경은 TPM 제안 + Main Agent 승인으로만 가능하다.
- PM/TPM/Dev/Reviewer/Verifier/Runtime Verifier는 주요 의사결정 시점마다 PR 코멘트를 남겨 결정 근거를 추적 가능 상태로 유지한다.
브랜치 운영 규칙:
- TPM은 각 티켓에 대해 `ticket temp branch -> program feature branch` PR 경로를 지정한다.
- 티켓 머지 대상은 항상 program feature branch이며, `main`은 최종 통합 단계에서만 사용한다.
## Runtime Verification Protocol ## Runtime Verification Protocol
- Runtime Verifier는 테스트 통과 이후 실제 동작(스테이징/실운영)을 모니터링한다. - Runtime Verifier는 테스트 통과 이후 실제 동작(스테이징/실운영)을 모니터링한다.
@@ -149,6 +164,16 @@ Control checks:
- 이슈 클로즈 규칙: - 이슈 클로즈 규칙:
- Dev 수정 완료 + Verifier 재검증 통과 + Runtime Verifier 재관측 정상 - Dev 수정 완료 + Verifier 재검증 통과 + Runtime Verifier 재관측 정상
- 최종 클로즈 승인자는 Main Agent - 최종 클로즈 승인자는 Main Agent
- 개발 완료 필수 절차:
- 시스템 실제 구동(스테이징/로컬 실운영 모드) 실행
- 모니터링 체크리스트(핵심 경보/주문 경로/예외 로그) 수행
- 결과를 티켓/PR 코멘트에 증적으로 첨부하지 않으면 완료로 간주하지 않음
## Server Reflection Rule
- `ticket temp branch -> program feature branch` 머지는 검증 승인 후 자동/수동 진행 가능하다.
- `program feature branch -> main` 머지는 사용자 명시 승인 시에만 허용한다.
- Main 병합 시 Main Agent가 승인 근거를 PR 코멘트에 기록한다.
## Acceptance Matrix (PM Scenario -> Dev Tasks -> Verifier Checks) ## Acceptance Matrix (PM Scenario -> Dev Tasks -> Verifier Checks)

View File

@@ -50,10 +50,13 @@ Updated: 2026-02-26
- PR 본문에 `REQ-*`, `TASK-*`, `TEST-*` 매핑 표 존재 - PR 본문에 `REQ-*`, `TASK-*`, `TEST-*` 매핑 표 존재
- `src/core/risk_manager.py` 변경 없음 - `src/core/risk_manager.py` 변경 없음
- 주요 의사결정 체크포인트(DCP-01~04) 중 해당 단계 Main Agent 확인 기록 존재 - 주요 의사결정 체크포인트(DCP-01~04) 중 해당 단계 Main Agent 확인 기록 존재
- 주요 의사결정(리뷰 지적/수정 합의/검증 승인)에 대한 에이전트 PR 코멘트 존재
- 티켓 PR의 base가 `main`이 아닌 program feature branch인지 확인
자동 점검: 자동 점검:
- 문서 검증 스크립트 통과 - 문서 검증 스크립트 통과
- 테스트 통과 - 테스트 통과
- 개발 완료 시 시스템 구동/모니터링 증적 코멘트 존재
## 5) 감사 추적 ## 5) 감사 추적
@@ -86,3 +89,15 @@ Updated: 2026-02-26
- Dev가 `REPLAN-REQUEST` 발행 시 TPM 심사 없이는 스코프/일정 변경 금지 - Dev가 `REPLAN-REQUEST` 발행 시 TPM 심사 없이는 스코프/일정 변경 금지
- `REPLAN-REQUEST`는 Main Agent 승인 전 \"제안\" 상태로 유지 - `REPLAN-REQUEST`는 Main Agent 승인 전 \"제안\" 상태로 유지
- 승인된 재계획은 `REQ/TASK/TEST` 문서를 동시 갱신해야 유효 - 승인된 재계획은 `REQ/TASK/TEST` 문서를 동시 갱신해야 유효
## 9) 서버 반영 규칙
- 티켓 PR(`feature/issue-* -> feature/{stream}`)은 검증 승인 후 머지 가능하다.
- 최종 통합 PR(`feature/{stream} -> main`)은 사용자 명시 승인 전 `tea pulls merge` 실행 금지.
- Main 병합 시 승인 근거 코멘트 필수.
## 10) 최종 main 병합 조건
- 모든 티켓이 program feature branch로 병합 완료
- Runtime Verifier의 구동/모니터링 검증 완료
- 사용자 최종 승인 코멘트 확인 후에만 `feature -> main` PR 머지 허용

View File

@@ -0,0 +1,48 @@
<!--
Doc-ID: DOC-IDEA-001
Version: 1.0.0
Status: active
Owner: main-agent
Updated: 2026-02-26
-->
# 메인 에이전트 아이디에이션 백로그
목적:
- 구현 진행 중 떠오른 신규 구현 아이디어를 계획 반영 전 임시 저장한다.
- 본 문서는 사용자 검토 후 다음 계획 포함 여부를 결정하기 위한 검토 큐다.
운영 규칙:
- 각 아이디어는 `IDEA-*` 식별자를 사용한다.
- 필수 필드: 배경, 기대효과, 리스크, 후속 티켓 후보.
- 상태는 `proposed`, `under-review`, `accepted`, `rejected` 중 하나를 사용한다.
## 아이디어 목록
- `IDEA-001` (status: proposed)
- 제목: Kill-Switch 전역 상태를 프로세스 단일 전역에서 시장/세션 단위 상태로 분리
- 배경: 현재는 전역 block 플래그 기반이라 시장별 분리 제어가 제한될 수 있음
- 기대효과: KR/US 병행 운용 시 한 시장 장애가 다른 시장 주문을 불필요하게 막는 리스크 축소
- 리스크: 상태 동기화 복잡도 증가, 테스트 케이스 확장 필요
- 후속 티켓 후보: `TKT-P1-KS-SCOPE-SPLIT`
- `IDEA-002` (status: proposed)
- 제목: Exit Engine 입력 계약(ATR/peak/model_prob/liquidity) 표준 DTO를 데이터 파이프라인에 고정
- 배경: 현재 ATR/모델확률 일부가 fallback 기반이라 운영 일관성이 약함
- 기대효과: 백테스트-실거래 입력 동형성 강화, 회귀 분석 용이
- 리스크: 기존 스캐너/시나리오 엔진 연동 작업량 증가
- 후속 티켓 후보: `TKT-P1-EXIT-CONTRACT`
- `IDEA-003` (status: proposed)
- 제목: Runtime Verifier 자동 이슈 생성기(로그 패턴 -> 이슈 템플릿 자동화)
- 배경: 런타임 이상 리포트가 수동 작성 중심이라 누락 가능성 존재
- 기대효과: 이상 탐지 후 이슈 등록 리드타임 단축, 증적 표준화
- 리스크: 오탐 이슈 폭증 가능성, 필터링 룰 필요
- 후속 티켓 후보: `TKT-P1-RUNTIME-AUTO-ISSUE`
- `IDEA-004` (status: proposed)
- 제목: PR 코멘트 워크플로우 자동 점검(리뷰어->개발논의->검증승인 누락 차단)
- 배경: 현재 절차는 강력하지만 수행 확인이 수동
- 기대효과: 절차 누락 방지, 감사 추적 자동화
- 리스크: CLI/API 연동 유지보수 비용
- 후속 티켓 후보: `TKT-P0-WORKFLOW-GUARD`

View File

@@ -21,6 +21,7 @@ Updated: 2026-02-26
7. PM 시나리오/이슈 분류: [50_scenario_matrix_and_issue_taxonomy.md](./50_scenario_matrix_and_issue_taxonomy.md) 7. PM 시나리오/이슈 분류: [50_scenario_matrix_and_issue_taxonomy.md](./50_scenario_matrix_and_issue_taxonomy.md)
8. TPM 제어 프로토콜/수용 매트릭스: [50_tpm_control_protocol.md](./50_tpm_control_protocol.md) 8. TPM 제어 프로토콜/수용 매트릭스: [50_tpm_control_protocol.md](./50_tpm_control_protocol.md)
9. 저장소 강제 설정 체크리스트: [60_repo_enforcement_checklist.md](./60_repo_enforcement_checklist.md) 9. 저장소 강제 설정 체크리스트: [60_repo_enforcement_checklist.md](./60_repo_enforcement_checklist.md)
10. 메인 에이전트 아이디에이션 백로그: [70_main_agent_ideation.md](./70_main_agent_ideation.md)
## 운영 규칙 ## 운영 규칙

View File

@@ -5,14 +5,25 @@
**CRITICAL: All code changes MUST follow this workflow. Direct pushes to `main` are ABSOLUTELY PROHIBITED.** **CRITICAL: All code changes MUST follow this workflow. Direct pushes to `main` are ABSOLUTELY PROHIBITED.**
1. **Create Gitea Issue First** — All features, bug fixes, and policy changes require a Gitea issue before any code is written 1. **Create Gitea Issue First** — All features, bug fixes, and policy changes require a Gitea issue before any code is written
2. **Create Feature Branch** — Branch from `main` using format `feature/issue-{N}-{short-description}` 2. **Create Program Feature Branch** — Branch from `main` for the whole development stream
- After creating the branch, run `git pull origin main` and rebase to ensure the branch is up to date - Format: `feature/{epic-or-stream-name}`
3. **Implement Changes** — Write code, tests, and documentation on the feature branch 3. **Create Ticket Temp Branch** — Branch from the program feature branch per ticket
4. **Create Pull Request** — Submit PR to `main` branch referencing the issue number - Format: `feature/issue-{N}-{short-description}`
5. **Review & Merge** — After approval, merge via PR (squash or merge commit) 4. **Implement Per Ticket** — Write code, tests, and documentation on the ticket temp branch
5. **Create Pull Request to Program Feature Branch**`feature/issue-N-* -> feature/{stream}`
6. **Review/Verify and Merge into Program Feature Branch** — user approval not required
7. **Final Integration PR to main** — Only after all ticket stages complete and explicit user approval
**Never commit directly to `main`.** This policy applies to all changes, no exceptions. **Never commit directly to `main`.** This policy applies to all changes, no exceptions.
## Branch Strategy (Mandatory)
- Team operation default branch is the **program feature branch**, not `main`.
- Ticket-level development happens only on **ticket temp branches** cut from the program feature branch.
- Ticket PR merges into program feature branch are allowed after verifier approval.
- Until final user sign-off, `main` merge is prohibited.
- 각 에이전트는 주요 의사결정(리뷰 지적, 수정 방향, 검증 승인)마다 PR 코멘트를 적극 작성해 의사결정 과정을 남긴다.
## Gitea CLI Formatting Troubleshooting ## Gitea CLI Formatting Troubleshooting
Issue/PR 본문 작성 시 줄바꿈(`\n`)이 문자열 그대로 저장되는 문제가 반복될 수 있다. 원인은 `-d "...\n..."` 형태에서 쉘/CLI가 이스케이프를 실제 개행으로 해석하지 않기 때문이다. Issue/PR 본문 작성 시 줄바꿈(`\n`)이 문자열 그대로 저장되는 문제가 반복될 수 있다. 원인은 `-d "...\n..."` 형태에서 쉘/CLI가 이스케이프를 실제 개행으로 해석하지 않기 때문이다.

View File

@@ -0,0 +1,52 @@
"""Backtest cost/slippage/failure validation guard."""
from __future__ import annotations
from dataclasses import dataclass
import math
@dataclass(frozen=True)
class BacktestCostModel:
commission_bps: float | None = None
slippage_bps_by_session: dict[str, float] | None = None
failure_rate_by_session: dict[str, float] | None = None
unfavorable_fill_required: bool = True
def validate_backtest_cost_model(
*,
model: BacktestCostModel,
required_sessions: list[str],
) -> None:
"""Raise ValueError when required cost assumptions are missing/invalid."""
if (
model.commission_bps is None
or not math.isfinite(model.commission_bps)
or model.commission_bps < 0
):
raise ValueError("commission_bps must be provided and >= 0")
if not model.unfavorable_fill_required:
raise ValueError("unfavorable_fill_required must be True")
slippage = model.slippage_bps_by_session or {}
failure = model.failure_rate_by_session or {}
missing_slippage = [s for s in required_sessions if s not in slippage]
if missing_slippage:
raise ValueError(
f"missing slippage_bps_by_session for sessions: {', '.join(missing_slippage)}"
)
missing_failure = [s for s in required_sessions if s not in failure]
if missing_failure:
raise ValueError(
f"missing failure_rate_by_session for sessions: {', '.join(missing_failure)}"
)
for sess, bps in slippage.items():
if not math.isfinite(bps) or bps < 0:
raise ValueError(f"slippage bps must be >= 0 for session={sess}")
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}")

View File

@@ -0,0 +1,103 @@
"""Conservative backtest execution model."""
from __future__ import annotations
from dataclasses import dataclass
import math
from random import Random
from typing import Literal
OrderSide = Literal["BUY", "SELL"]
@dataclass(frozen=True)
class ExecutionRequest:
side: OrderSide
session_id: str
qty: int
reference_price: float
@dataclass(frozen=True)
class ExecutionAssumptions:
slippage_bps_by_session: dict[str, float]
failure_rate_by_session: dict[str, float]
partial_fill_rate_by_session: dict[str, float]
partial_fill_min_ratio: float = 0.3
partial_fill_max_ratio: float = 0.8
seed: int = 0
@dataclass(frozen=True)
class ExecutionResult:
status: Literal["FILLED", "PARTIAL", "REJECTED"]
filled_qty: int
avg_price: float
slippage_bps: float
reason: str
class BacktestExecutionModel:
"""Execution simulator with conservative unfavorable fill assumptions."""
def __init__(self, assumptions: ExecutionAssumptions) -> None:
self.assumptions = assumptions
self._rng = Random(assumptions.seed)
if assumptions.partial_fill_min_ratio <= 0 or assumptions.partial_fill_max_ratio > 1:
raise ValueError("partial fill ratios must be within (0,1]")
if assumptions.partial_fill_min_ratio > assumptions.partial_fill_max_ratio:
raise ValueError("partial_fill_min_ratio must be <= partial_fill_max_ratio")
for sess, bps in assumptions.slippage_bps_by_session.items():
if not math.isfinite(bps) or bps < 0:
raise ValueError(f"slippage_bps must be finite and >= 0 for session={sess}")
for sess, rate in assumptions.failure_rate_by_session.items():
if not math.isfinite(rate) or rate < 0 or rate > 1:
raise ValueError(f"failure_rate must be in [0,1] for session={sess}")
for sess, rate in assumptions.partial_fill_rate_by_session.items():
if not math.isfinite(rate) or rate < 0 or rate > 1:
raise ValueError(f"partial_fill_rate must be in [0,1] for session={sess}")
def simulate(self, request: ExecutionRequest) -> ExecutionResult:
if request.qty <= 0:
raise ValueError("qty must be positive")
if request.reference_price <= 0:
raise ValueError("reference_price must be positive")
slippage_bps = self.assumptions.slippage_bps_by_session.get(request.session_id, 0.0)
failure_rate = self.assumptions.failure_rate_by_session.get(request.session_id, 0.0)
partial_rate = self.assumptions.partial_fill_rate_by_session.get(request.session_id, 0.0)
if self._rng.random() < failure_rate:
return ExecutionResult(
status="REJECTED",
filled_qty=0,
avg_price=0.0,
slippage_bps=slippage_bps,
reason="execution_failure",
)
slip_mult = 1.0 + (slippage_bps / 10000.0 if request.side == "BUY" else -slippage_bps / 10000.0)
exec_price = request.reference_price * slip_mult
if self._rng.random() < partial_rate:
ratio = self._rng.uniform(
self.assumptions.partial_fill_min_ratio,
self.assumptions.partial_fill_max_ratio,
)
filled = max(1, min(request.qty - 1, int(request.qty * ratio)))
return ExecutionResult(
status="PARTIAL",
filled_qty=filled,
avg_price=exec_price,
slippage_bps=slippage_bps,
reason="partial_fill",
)
return ExecutionResult(
status="FILLED",
filled_qty=request.qty,
avg_price=exec_price,
slippage_bps=slippage_bps,
reason="filled",
)

View File

@@ -0,0 +1,111 @@
"""Triple barrier labeler utilities.
Implements first-touch labeling with upper/lower/time barriers.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal, Sequence
TieBreakMode = Literal["stop_first", "take_first"]
@dataclass(frozen=True)
class TripleBarrierSpec:
take_profit_pct: float
stop_loss_pct: float
max_holding_bars: int
tie_break: TieBreakMode = "stop_first"
@dataclass(frozen=True)
class TripleBarrierLabel:
label: int # +1 take-profit first, -1 stop-loss first, 0 timeout
touched: Literal["take_profit", "stop_loss", "time"]
touch_bar: int
entry_price: float
upper_barrier: float
lower_barrier: float
def label_with_triple_barrier(
*,
highs: Sequence[float],
lows: Sequence[float],
closes: Sequence[float],
entry_index: int,
side: int,
spec: TripleBarrierSpec,
) -> TripleBarrierLabel:
"""Label one entry using triple-barrier first-touch rule.
Args:
highs/lows/closes: OHLC components with identical length.
entry_index: Entry bar index in the sequences.
side: +1 for long, -1 for short.
spec: Barrier specification.
"""
if side not in {1, -1}:
raise ValueError("side must be +1 or -1")
if len(highs) != len(lows) or len(highs) != len(closes):
raise ValueError("highs, lows, closes lengths must match")
if entry_index < 0 or entry_index >= len(closes):
raise IndexError("entry_index out of range")
if spec.max_holding_bars <= 0:
raise ValueError("max_holding_bars must be positive")
entry_price = float(closes[entry_index])
if entry_price <= 0:
raise ValueError("entry price must be positive")
if side == 1:
upper = entry_price * (1.0 + spec.take_profit_pct)
lower = entry_price * (1.0 - spec.stop_loss_pct)
else:
# For short side, favorable move is down.
upper = entry_price * (1.0 + spec.stop_loss_pct)
lower = entry_price * (1.0 - spec.take_profit_pct)
last_index = min(len(closes) - 1, entry_index + spec.max_holding_bars)
for idx in range(entry_index + 1, last_index + 1):
h = float(highs[idx])
l = float(lows[idx])
up_touch = h >= upper
down_touch = l <= lower
if not up_touch and not down_touch:
continue
if up_touch and down_touch:
if spec.tie_break == "stop_first":
touched = "stop_loss"
label = -1
else:
touched = "take_profit"
label = 1
elif up_touch:
touched = "take_profit" if side == 1 else "stop_loss"
label = 1 if side == 1 else -1
else:
touched = "stop_loss" if side == 1 else "take_profit"
label = -1 if side == 1 else 1
return TripleBarrierLabel(
label=label,
touched=touched,
touch_bar=idx,
entry_price=entry_price,
upper_barrier=upper,
lower_barrier=lower,
)
return TripleBarrierLabel(
label=0,
touched="time",
touch_bar=last_index,
entry_price=entry_price,
upper_barrier=upper,
lower_barrier=lower,
)

View File

@@ -0,0 +1,74 @@
"""Walk-forward splitter with purge/embargo controls."""
from __future__ import annotations
from dataclasses import dataclass
@dataclass(frozen=True)
class WalkForwardFold:
train_indices: list[int]
test_indices: list[int]
@property
def train_size(self) -> int:
return len(self.train_indices)
@property
def test_size(self) -> int:
return len(self.test_indices)
def generate_walk_forward_splits(
*,
n_samples: int,
train_size: int,
test_size: int,
step_size: int | None = None,
purge_size: int = 0,
embargo_size: int = 0,
min_train_size: int = 1,
) -> list[WalkForwardFold]:
"""Generate chronological folds with purge/embargo leakage controls."""
if n_samples <= 0:
raise ValueError("n_samples must be positive")
if train_size <= 0 or test_size <= 0:
raise ValueError("train_size and test_size must be positive")
if purge_size < 0 or embargo_size < 0:
raise ValueError("purge_size and embargo_size must be >= 0")
if min_train_size <= 0:
raise ValueError("min_train_size must be positive")
step = step_size if step_size is not None else test_size
if step <= 0:
raise ValueError("step_size must be positive")
folds: list[WalkForwardFold] = []
prev_test_end: int | None = None
test_start = train_size + purge_size
while test_start + test_size <= n_samples:
test_end = test_start + test_size - 1
train_end = test_start - purge_size - 1
if train_end < 0:
break
train_start = max(0, train_end - train_size + 1)
train_indices = list(range(train_start, train_end + 1))
if prev_test_end is not None and embargo_size > 0:
emb_from = prev_test_end + 1
emb_to = prev_test_end + embargo_size
train_indices = [i for i in train_indices if i < emb_from or i > emb_to]
if len(train_indices) >= min_train_size:
folds.append(
WalkForwardFold(
train_indices=train_indices,
test_indices=list(range(test_start, test_end + 1)),
)
)
prev_test_end = test_end
test_start += step
return folds

View File

@@ -59,11 +59,15 @@ class Settings(BaseSettings):
# KIS VTS overseas balance API returns errors for most accounts. # KIS VTS overseas balance API returns errors for most accounts.
# This value is used as a fallback when the balance API returns 0 in paper mode. # This value is used as a fallback when the balance API returns 0 in paper mode.
PAPER_OVERSEAS_CASH: float = Field(default=50000.0, ge=0.0) PAPER_OVERSEAS_CASH: float = Field(default=50000.0, ge=0.0)
USD_BUFFER_MIN: float = Field(default=1000.0, ge=0.0)
# Trading frequency mode (daily = batch API calls, realtime = per-stock calls) # Trading frequency mode (daily = batch API calls, realtime = per-stock calls)
TRADE_MODE: str = Field(default="daily", pattern="^(daily|realtime)$") TRADE_MODE: str = Field(default="daily", pattern="^(daily|realtime)$")
DAILY_SESSIONS: int = Field(default=4, ge=1, le=10) DAILY_SESSIONS: int = Field(default=4, ge=1, le=10)
SESSION_INTERVAL_HOURS: int = Field(default=6, ge=1, le=24) SESSION_INTERVAL_HOURS: int = Field(default=6, ge=1, le=24)
ORDER_BLACKOUT_ENABLED: bool = True
ORDER_BLACKOUT_WINDOWS_KST: str = "23:30-00:10"
ORDER_BLACKOUT_QUEUE_MAX: int = Field(default=500, ge=10, le=5000)
# Pre-Market Planner # Pre-Market Planner
PRE_MARKET_MINUTES: int = Field(default=30, ge=10, le=120) PRE_MARKET_MINUTES: int = Field(default=30, ge=10, le=120)

View File

@@ -0,0 +1,105 @@
"""Blackout policy and queued order-intent manager."""
from __future__ import annotations
from collections import deque
from dataclasses import dataclass
from datetime import UTC, datetime, time
from zoneinfo import ZoneInfo
@dataclass(frozen=True)
class BlackoutWindow:
start: time
end: time
def contains(self, kst_time: time) -> bool:
if self.start <= self.end:
return self.start <= kst_time < self.end
return kst_time >= self.start or kst_time < self.end
@dataclass
class QueuedOrderIntent:
market_code: str
exchange_code: str
stock_code: str
order_type: str
quantity: int
price: float
source: str
queued_at: datetime
attempts: int = 0
def parse_blackout_windows_kst(raw: str) -> list[BlackoutWindow]:
"""Parse comma-separated KST windows like '23:30-00:10,11:20-11:30'."""
windows: list[BlackoutWindow] = []
for token in raw.split(","):
span = token.strip()
if not span or "-" not in span:
continue
start_raw, end_raw = [part.strip() for part in span.split("-", 1)]
try:
start_h, start_m = [int(v) for v in start_raw.split(":", 1)]
end_h, end_m = [int(v) for v in end_raw.split(":", 1)]
except (ValueError, TypeError):
continue
if not (0 <= start_h <= 23 and 0 <= end_h <= 23):
continue
if not (0 <= start_m <= 59 and 0 <= end_m <= 59):
continue
windows.append(BlackoutWindow(start=time(start_h, start_m), end=time(end_h, end_m)))
return windows
class BlackoutOrderManager:
"""Tracks blackout mode and queues order intents until recovery."""
def __init__(
self,
*,
enabled: bool,
windows: list[BlackoutWindow],
max_queue_size: int = 500,
) -> None:
self.enabled = enabled
self._windows = windows
self._queue: deque[QueuedOrderIntent] = deque()
self._was_blackout = False
self._max_queue_size = max_queue_size
@property
def pending_count(self) -> int:
return len(self._queue)
def in_blackout(self, now: datetime | None = None) -> bool:
if not self.enabled or not self._windows:
return False
now = now or datetime.now(UTC)
kst_now = now.astimezone(ZoneInfo("Asia/Seoul")).timetz().replace(tzinfo=None)
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:
return False
self._queue.append(intent)
return True
def pop_recovery_batch(self, now: datetime | None = None) -> list[QueuedOrderIntent]:
in_blackout_now = self.in_blackout(now)
batch: list[QueuedOrderIntent] = []
if not in_blackout_now and self._queue:
while self._queue:
batch.append(self._queue.popleft())
self._was_blackout = in_blackout_now
return batch
def requeue(self, intent: QueuedOrderIntent) -> None:
if len(self._queue) < self._max_queue_size:
self._queue.append(intent)
def clear(self) -> int:
count = len(self._queue)
self._queue.clear()
return count

71
src/core/kill_switch.py Normal file
View File

@@ -0,0 +1,71 @@
"""Kill switch orchestration for emergency risk actions.
Order is fixed:
1) block new orders
2) cancel pending orders
3) refresh order state
4) reduce risk
5) snapshot and notify
"""
from __future__ import annotations
import inspect
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable
StepCallable = Callable[[], Any | Awaitable[Any]]
@dataclass
class KillSwitchReport:
reason: str
steps: list[str] = field(default_factory=list)
errors: list[str] = field(default_factory=list)
class KillSwitchOrchestrator:
def __init__(self) -> None:
self.new_orders_blocked = False
async def _run_step(
self,
report: KillSwitchReport,
name: str,
fn: StepCallable | None,
) -> None:
report.steps.append(name)
if fn is None:
return
try:
result = fn()
if inspect.isawaitable(result):
await result
except Exception as exc: # pragma: no cover - intentionally resilient
report.errors.append(f"{name}: {exc}")
async def trigger(
self,
*,
reason: str,
cancel_pending_orders: StepCallable | None = None,
refresh_order_state: StepCallable | None = None,
reduce_risk: StepCallable | None = None,
snapshot_state: StepCallable | None = None,
notify: StepCallable | None = None,
) -> KillSwitchReport:
report = KillSwitchReport(reason=reason)
self.new_orders_blocked = True
report.steps.append("block_new_orders")
await self._run_step(report, "cancel_pending_orders", cancel_pending_orders)
await self._run_step(report, "refresh_order_state", refresh_order_state)
await self._run_step(report, "reduce_risk", reduce_risk)
await self._run_step(report, "snapshot_state", snapshot_state)
await self._run_step(report, "notify", notify)
return report
def clear_block(self) -> None:
self.new_orders_blocked = False

93
src/core/order_policy.py Normal file
View File

@@ -0,0 +1,93 @@
"""Session-aware order policy guards.
Default policy:
- Low-liquidity sessions must reject market orders (price <= 0).
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, datetime, time
from zoneinfo import ZoneInfo
from src.markets.schedule import MarketInfo
_LOW_LIQUIDITY_SESSIONS = {"NXT_AFTER", "US_PRE", "US_DAY", "US_AFTER"}
class OrderPolicyRejected(Exception):
"""Raised when an order violates session policy."""
def __init__(self, message: str, *, session_id: str, market_code: str) -> None:
super().__init__(message)
self.session_id = session_id
self.market_code = market_code
@dataclass(frozen=True)
class SessionInfo:
session_id: str
is_low_liquidity: bool
def classify_session_id(market: MarketInfo, now: datetime | None = None) -> str:
"""Classify current session by KST schedule used in v3 docs."""
now = now or datetime.now(UTC)
# v3 session tables are explicitly defined in KST perspective.
kst_time = now.astimezone(ZoneInfo("Asia/Seoul")).timetz().replace(tzinfo=None)
if market.code == "KR":
if time(8, 0) <= kst_time < time(8, 50):
return "NXT_PRE"
if time(9, 0) <= kst_time < time(15, 30):
return "KRX_REG"
if time(15, 30) <= kst_time < time(20, 0):
return "NXT_AFTER"
return "KR_OFF"
if market.code.startswith("US"):
if time(10, 0) <= kst_time < time(18, 0):
return "US_DAY"
if time(18, 0) <= kst_time < time(23, 30):
return "US_PRE"
if time(23, 30) <= kst_time or kst_time < time(6, 0):
return "US_REG"
if time(6, 0) <= kst_time < time(7, 0):
return "US_AFTER"
return "US_OFF"
return "GENERIC_REG"
def get_session_info(market: MarketInfo, now: datetime | None = None) -> SessionInfo:
session_id = classify_session_id(market, now)
return SessionInfo(session_id=session_id, is_low_liquidity=session_id in _LOW_LIQUIDITY_SESSIONS)
def validate_order_policy(
*,
market: MarketInfo,
order_type: str,
price: float,
now: datetime | None = None,
) -> SessionInfo:
"""Validate order against session policy and return resolved session info."""
info = get_session_info(market, now)
is_market_order = price <= 0
if info.is_low_liquidity and is_market_order:
raise OrderPolicyRejected(
f"Market order is forbidden in low-liquidity session ({info.session_id})",
session_id=info.session_id,
market_code=market.code,
)
# Guard against accidental unsupported actions.
if order_type not in {"BUY", "SELL"}:
raise OrderPolicyRejected(
f"Unsupported order_type={order_type}",
session_id=info.session_id,
market_code=market.code,
)
return info

View File

@@ -31,8 +31,12 @@ def init_db(db_path: str) -> sqlite3.Connection:
quantity INTEGER, quantity INTEGER,
price REAL, price REAL,
pnl REAL DEFAULT 0.0, pnl REAL DEFAULT 0.0,
strategy_pnl REAL DEFAULT 0.0,
fx_pnl REAL DEFAULT 0.0,
market TEXT DEFAULT 'KR', market TEXT DEFAULT 'KR',
exchange_code TEXT DEFAULT 'KRX', exchange_code TEXT DEFAULT 'KRX',
session_id TEXT DEFAULT 'UNKNOWN',
selection_context TEXT,
decision_id TEXT, decision_id TEXT,
mode TEXT DEFAULT 'paper' mode TEXT DEFAULT 'paper'
) )
@@ -53,6 +57,32 @@ def init_db(db_path: str) -> sqlite3.Connection:
conn.execute("ALTER TABLE trades ADD COLUMN decision_id TEXT") conn.execute("ALTER TABLE trades ADD COLUMN decision_id TEXT")
if "mode" not in columns: if "mode" not in columns:
conn.execute("ALTER TABLE trades ADD COLUMN mode TEXT DEFAULT 'paper'") conn.execute("ALTER TABLE trades ADD COLUMN mode TEXT DEFAULT 'paper'")
session_id_added = False
if "session_id" not in columns:
conn.execute("ALTER TABLE trades ADD COLUMN session_id TEXT DEFAULT 'UNKNOWN'")
session_id_added = True
if "strategy_pnl" not in columns:
conn.execute("ALTER TABLE trades ADD COLUMN strategy_pnl REAL DEFAULT 0.0")
if "fx_pnl" not in columns:
conn.execute("ALTER TABLE trades ADD COLUMN fx_pnl REAL DEFAULT 0.0")
# Backfill legacy rows where only pnl existed before split accounting columns.
conn.execute(
"""
UPDATE trades
SET strategy_pnl = pnl, fx_pnl = 0.0
WHERE pnl != 0.0
AND strategy_pnl = 0.0
AND fx_pnl = 0.0
"""
)
if session_id_added:
conn.execute(
"""
UPDATE trades
SET session_id = 'UNKNOWN'
WHERE session_id IS NULL OR session_id = ''
"""
)
# Context tree tables for multi-layered memory management # Context tree tables for multi-layered memory management
conn.execute( conn.execute(
@@ -171,8 +201,11 @@ def log_trade(
quantity: int = 0, quantity: int = 0,
price: float = 0.0, price: float = 0.0,
pnl: float = 0.0, pnl: float = 0.0,
strategy_pnl: float | None = None,
fx_pnl: float | None = None,
market: str = "KR", market: str = "KR",
exchange_code: str = "KRX", exchange_code: str = "KRX",
session_id: str | None = None,
selection_context: dict[str, any] | None = None, selection_context: dict[str, any] | None = None,
decision_id: str | None = None, decision_id: str | None = None,
mode: str = "paper", mode: str = "paper",
@@ -187,24 +220,37 @@ def log_trade(
rationale: AI decision rationale rationale: AI decision rationale
quantity: Number of shares quantity: Number of shares
price: Trade price price: Trade price
pnl: Profit/loss pnl: Total profit/loss (backward compatibility)
strategy_pnl: Strategy PnL component
fx_pnl: FX PnL component
market: Market code market: Market code
exchange_code: Exchange code exchange_code: Exchange code
session_id: Session identifier (if omitted, auto-derived from market)
selection_context: Scanner selection data (RSI, volume_ratio, signal, score) selection_context: Scanner selection data (RSI, volume_ratio, signal, score)
decision_id: Unique decision identifier for audit linking decision_id: Unique decision identifier for audit linking
mode: Trading mode ('paper' or 'live') for data separation mode: Trading mode ('paper' or 'live') for data separation
""" """
# Serialize selection context to JSON # Serialize selection context to JSON
context_json = json.dumps(selection_context) if selection_context else None context_json = json.dumps(selection_context) if selection_context else None
resolved_session_id = _resolve_session_id(market=market, session_id=session_id)
if strategy_pnl is None and fx_pnl is None:
strategy_pnl = pnl
fx_pnl = 0.0
elif strategy_pnl is None:
strategy_pnl = pnl - float(fx_pnl or 0.0) if pnl != 0.0 else 0.0
elif fx_pnl is None:
fx_pnl = pnl - float(strategy_pnl) if pnl != 0.0 else 0.0
if pnl == 0.0 and (strategy_pnl or fx_pnl):
pnl = float(strategy_pnl) + float(fx_pnl)
conn.execute( conn.execute(
""" """
INSERT INTO trades ( INSERT INTO trades (
timestamp, stock_code, action, confidence, rationale, timestamp, stock_code, action, confidence, rationale,
quantity, price, pnl, market, exchange_code, selection_context, decision_id, quantity, price, pnl, strategy_pnl, fx_pnl,
mode market, exchange_code, session_id, selection_context, decision_id, mode
) )
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
datetime.now(UTC).isoformat(), datetime.now(UTC).isoformat(),
@@ -215,8 +261,11 @@ def log_trade(
quantity, quantity,
price, price,
pnl, pnl,
strategy_pnl,
fx_pnl,
market, market,
exchange_code, exchange_code,
resolved_session_id,
context_json, context_json,
decision_id, decision_id,
mode, mode,
@@ -225,6 +274,21 @@ def log_trade(
conn.commit() conn.commit()
def _resolve_session_id(*, market: str, session_id: str | None) -> str:
if session_id:
return session_id
try:
from src.core.order_policy import classify_session_id
from src.markets.schedule import MARKETS
market_info = MARKETS.get(market)
if market_info is not None:
return classify_session_id(market_info)
except Exception:
pass
return "UNKNOWN"
def get_latest_buy_trade( def get_latest_buy_trade(
conn: sqlite3.Connection, stock_code: str, market: str conn: sqlite3.Connection, stock_code: str, market: str
) -> dict[str, Any] | None: ) -> dict[str, Any] | None:

View File

@@ -27,6 +27,13 @@ from src.context.layer import ContextLayer
from src.context.scheduler import ContextScheduler from src.context.scheduler import ContextScheduler
from src.context.store import ContextStore from src.context.store import ContextStore
from src.core.criticality import CriticalityAssessor from src.core.criticality import CriticalityAssessor
from src.core.blackout_manager import (
BlackoutOrderManager,
QueuedOrderIntent,
parse_blackout_windows_kst,
)
from src.core.kill_switch import KillSwitchOrchestrator
from src.core.order_policy import OrderPolicyRejected, validate_order_policy
from src.core.priority_queue import PriorityTaskQueue from src.core.priority_queue import PriorityTaskQueue
from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected, RiskManager from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected, RiskManager
from src.db import ( from src.db import (
@@ -43,11 +50,19 @@ from src.logging_config import setup_logging
from src.markets.schedule import MARKETS, MarketInfo, get_next_market_open, get_open_markets from src.markets.schedule import MARKETS, MarketInfo, get_next_market_open, get_open_markets
from src.notifications.telegram_client import NotificationFilter, TelegramClient, TelegramCommandHandler from src.notifications.telegram_client import NotificationFilter, TelegramClient, TelegramCommandHandler
from src.strategy.models import DayPlaybook, MarketOutlook from src.strategy.models import DayPlaybook, MarketOutlook
from src.strategy.exit_rules import ExitRuleConfig, ExitRuleInput, evaluate_exit
from src.strategy.playbook_store import PlaybookStore from src.strategy.playbook_store import PlaybookStore
from src.strategy.pre_market_planner import PreMarketPlanner from src.strategy.pre_market_planner import PreMarketPlanner
from src.strategy.position_state_machine import PositionState
from src.strategy.scenario_engine import ScenarioEngine from src.strategy.scenario_engine import ScenarioEngine
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
KILL_SWITCH = KillSwitchOrchestrator()
BLACKOUT_ORDER_MANAGER = BlackoutOrderManager(
enabled=False,
windows=[],
max_queue_size=500,
)
def safe_float(value: str | float | None, default: float = 0.0) -> float: def safe_float(value: str | float | None, default: float = 0.0) -> float:
@@ -414,6 +429,26 @@ def _determine_order_quantity(
return quantity return quantity
def _should_block_overseas_buy_for_fx_buffer(
*,
market: MarketInfo,
action: str,
total_cash: float,
order_amount: float,
settings: Settings | None,
) -> tuple[bool, float, float]:
if (
market.is_domestic
or not market.code.startswith("US")
or action != "BUY"
or settings is None
):
return False, total_cash - order_amount, 0.0
remaining = total_cash - order_amount
required = settings.USD_BUFFER_MIN
return remaining < required, remaining, required
async def build_overseas_symbol_universe( async def build_overseas_symbol_universe(
db_conn: Any, db_conn: Any,
overseas_broker: OverseasBroker, overseas_broker: OverseasBroker,
@@ -456,6 +491,352 @@ async def build_overseas_symbol_universe(
return ordered_unique return ordered_unique
def _build_queued_order_intent(
*,
market: MarketInfo,
stock_code: str,
order_type: str,
quantity: int,
price: float,
source: str,
) -> QueuedOrderIntent:
return QueuedOrderIntent(
market_code=market.code,
exchange_code=market.exchange_code,
stock_code=stock_code,
order_type=order_type,
quantity=quantity,
price=price,
source=source,
queued_at=datetime.now(UTC),
)
def _maybe_queue_order_intent(
*,
market: MarketInfo,
stock_code: str,
order_type: str,
quantity: int,
price: float,
source: str,
) -> bool:
if not BLACKOUT_ORDER_MANAGER.in_blackout():
return False
queued = BLACKOUT_ORDER_MANAGER.enqueue(
_build_queued_order_intent(
market=market,
stock_code=stock_code,
order_type=order_type,
quantity=quantity,
price=price,
source=source,
)
)
if queued:
logger.warning(
"Blackout active: queued order intent %s %s (%s) qty=%d price=%.4f source=%s pending=%d",
order_type,
stock_code,
market.code,
quantity,
price,
source,
BLACKOUT_ORDER_MANAGER.pending_count,
)
else:
logger.error(
"Blackout queue full: dropped order intent %s %s (%s) qty=%d source=%s",
order_type,
stock_code,
market.code,
quantity,
source,
)
return True
async def process_blackout_recovery_orders(
*,
broker: KISBroker,
overseas_broker: OverseasBroker,
db_conn: Any,
) -> None:
intents = BLACKOUT_ORDER_MANAGER.pop_recovery_batch()
if not intents:
return
logger.info(
"Blackout recovery started: processing %d queued intents",
len(intents),
)
for intent in intents:
market = MARKETS.get(intent.market_code)
if market is None:
continue
open_position = get_open_position(db_conn, intent.stock_code, market.code)
if intent.order_type == "BUY" and open_position is not None:
logger.info(
"Drop stale queued BUY %s (%s): position already open",
intent.stock_code,
market.code,
)
continue
if intent.order_type == "SELL" and open_position is None:
logger.info(
"Drop stale queued SELL %s (%s): no open position",
intent.stock_code,
market.code,
)
continue
try:
validate_order_policy(
market=market,
order_type=intent.order_type,
price=float(intent.price),
)
if market.is_domestic:
result = await broker.send_order(
stock_code=intent.stock_code,
order_type=intent.order_type,
quantity=intent.quantity,
price=intent.price,
)
else:
result = await overseas_broker.send_overseas_order(
exchange_code=market.exchange_code,
stock_code=intent.stock_code,
order_type=intent.order_type,
quantity=intent.quantity,
price=intent.price,
)
accepted = result.get("rt_cd", "0") == "0"
if accepted:
logger.info(
"Recovered queued order executed: %s %s (%s) qty=%d price=%.4f source=%s",
intent.order_type,
intent.stock_code,
market.code,
intent.quantity,
intent.price,
intent.source,
)
continue
logger.warning(
"Recovered queued order rejected: %s %s (%s) qty=%d msg=%s",
intent.order_type,
intent.stock_code,
market.code,
intent.quantity,
result.get("msg1"),
)
except Exception as exc:
if isinstance(exc, OrderPolicyRejected):
logger.info(
"Drop queued intent by policy: %s %s (%s): %s",
intent.order_type,
intent.stock_code,
market.code,
exc,
)
continue
logger.warning(
"Recovered queued order failed: %s %s (%s): %s",
intent.order_type,
intent.stock_code,
market.code,
exc,
)
if intent.attempts < 2:
intent.attempts += 1
BLACKOUT_ORDER_MANAGER.requeue(intent)
def _resolve_kill_switch_markets(
*,
settings: Settings | None,
current_market: MarketInfo | None,
) -> list[MarketInfo]:
if settings is not None:
markets: list[MarketInfo] = []
seen: set[str] = set()
for market_code in settings.enabled_market_list:
market = MARKETS.get(market_code)
if market is None or market.code in seen:
continue
markets.append(market)
seen.add(market.code)
if markets:
return markets
if current_market is not None:
return [current_market]
return []
async def _cancel_pending_orders_for_kill_switch(
*,
broker: KISBroker,
overseas_broker: OverseasBroker,
markets: list[MarketInfo],
) -> None:
failures: list[str] = []
domestic = [m for m in markets if m.is_domestic]
overseas = [m for m in markets if not m.is_domestic]
if domestic:
try:
orders = await broker.get_domestic_pending_orders()
except Exception as exc:
logger.warning("KillSwitch: failed to fetch domestic pending orders: %s", exc)
orders = []
for order in orders:
stock_code = str(order.get("pdno", ""))
try:
orgn_odno = order.get("orgn_odno", "")
krx_fwdg_ord_orgno = order.get("ord_gno_brno", "")
psbl_qty = int(order.get("psbl_qty", "0") or "0")
if not stock_code or not orgn_odno or psbl_qty <= 0:
continue
cancel_result = await broker.cancel_domestic_order(
stock_code=stock_code,
orgn_odno=orgn_odno,
krx_fwdg_ord_orgno=krx_fwdg_ord_orgno,
qty=psbl_qty,
)
if cancel_result.get("rt_cd") != "0":
failures.append(
"domestic cancel failed for"
f" {stock_code}: rt_cd={cancel_result.get('rt_cd')}"
f" msg={cancel_result.get('msg1')}"
)
except Exception as exc:
logger.warning("KillSwitch: domestic cancel failed: %s", exc)
failures.append(f"domestic cancel exception for {stock_code}: {exc}")
us_exchanges = frozenset({"NASD", "NYSE", "AMEX"})
exchange_codes: list[str] = []
seen_us = False
for market in overseas:
exc_code = market.exchange_code
if exc_code in us_exchanges:
if not seen_us:
exchange_codes.append("NASD")
seen_us = True
elif exc_code not in exchange_codes:
exchange_codes.append(exc_code)
for exchange_code in exchange_codes:
try:
orders = await overseas_broker.get_overseas_pending_orders(exchange_code)
except Exception as exc:
logger.warning(
"KillSwitch: failed to fetch overseas pending orders for %s: %s",
exchange_code,
exc,
)
continue
for order in orders:
stock_code = str(order.get("pdno", ""))
order_exchange = str(order.get("ovrs_excg_cd") or exchange_code)
try:
odno = order.get("odno", "")
nccs_qty = int(order.get("nccs_qty", "0") or "0")
if not stock_code or not odno or nccs_qty <= 0:
continue
cancel_result = await overseas_broker.cancel_overseas_order(
exchange_code=order_exchange,
stock_code=stock_code,
odno=odno,
qty=nccs_qty,
)
if cancel_result.get("rt_cd") != "0":
failures.append(
"overseas cancel failed for"
f" {order_exchange}/{stock_code}: rt_cd={cancel_result.get('rt_cd')}"
f" msg={cancel_result.get('msg1')}"
)
except Exception as exc:
logger.warning("KillSwitch: overseas cancel failed: %s", exc)
failures.append(
f"overseas cancel exception for {order_exchange}/{stock_code}: {exc}"
)
if failures:
raise RuntimeError("; ".join(failures[:3]))
async def _refresh_order_state_for_kill_switch(
*,
broker: KISBroker,
overseas_broker: OverseasBroker,
markets: list[MarketInfo],
) -> None:
seen_overseas: set[str] = set()
for market in markets:
try:
if market.is_domestic:
await broker.get_balance()
elif market.exchange_code not in seen_overseas:
seen_overseas.add(market.exchange_code)
await overseas_broker.get_overseas_balance(market.exchange_code)
except Exception as exc:
logger.warning(
"KillSwitch: refresh state failed for %s/%s: %s",
market.code,
market.exchange_code,
exc,
)
def _reduce_risk_for_kill_switch() -> None:
dropped = BLACKOUT_ORDER_MANAGER.clear()
logger.critical("KillSwitch: reduced queued order risk by clearing %d queued intents", dropped)
async def _trigger_emergency_kill_switch(
*,
reason: str,
broker: KISBroker,
overseas_broker: OverseasBroker,
telegram: TelegramClient,
settings: Settings | None,
current_market: MarketInfo | None,
stock_code: str,
pnl_pct: float,
threshold: float,
) -> Any:
markets = _resolve_kill_switch_markets(settings=settings, current_market=current_market)
return await KILL_SWITCH.trigger(
reason=reason,
cancel_pending_orders=lambda: _cancel_pending_orders_for_kill_switch(
broker=broker,
overseas_broker=overseas_broker,
markets=markets,
),
refresh_order_state=lambda: _refresh_order_state_for_kill_switch(
broker=broker,
overseas_broker=overseas_broker,
markets=markets,
),
reduce_risk=_reduce_risk_for_kill_switch,
snapshot_state=lambda: logger.critical(
"KillSwitch snapshot %s/%s pnl=%.2f threshold=%.2f",
current_market.code if current_market else "UNKNOWN",
stock_code,
pnl_pct,
threshold,
),
notify=lambda: telegram.notify_circuit_breaker(
pnl_pct=pnl_pct,
threshold=threshold,
),
)
async def trading_cycle( async def trading_cycle(
broker: KISBroker, broker: KISBroker,
overseas_broker: OverseasBroker, overseas_broker: OverseasBroker,
@@ -784,7 +1165,24 @@ async def trading_cycle(
stop_loss_threshold = stock_playbook.scenarios[0].stop_loss_pct stop_loss_threshold = stock_playbook.scenarios[0].stop_loss_pct
take_profit_threshold = stock_playbook.scenarios[0].take_profit_pct take_profit_threshold = stock_playbook.scenarios[0].take_profit_pct
if loss_pct <= stop_loss_threshold: 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( decision = TradeDecision(
action="SELL", action="SELL",
confidence=95, confidence=95,
@@ -800,7 +1198,7 @@ async def trading_cycle(
loss_pct, loss_pct,
stop_loss_threshold, stop_loss_threshold,
) )
elif loss_pct >= take_profit_threshold: elif exit_eval.reason == "arm_take_profit":
decision = TradeDecision( decision = TradeDecision(
action="SELL", action="SELL",
confidence=90, confidence=90,
@@ -876,6 +1274,15 @@ async def trading_cycle(
trade_price = current_price trade_price = current_price
trade_pnl = 0.0 trade_pnl = 0.0
if decision.action in ("BUY", "SELL"): if decision.action in ("BUY", "SELL"):
if KILL_SWITCH.new_orders_blocked:
logger.critical(
"KillSwitch block active: skip %s order for %s (%s)",
decision.action,
stock_code,
market.name,
)
return
broker_held_qty = ( broker_held_qty = (
_extract_held_qty_from_balance( _extract_held_qty_from_balance(
balance_data, stock_code, is_domestic=market.is_domestic balance_data, stock_code, is_domestic=market.is_domestic
@@ -905,6 +1312,24 @@ async def trading_cycle(
) )
return return
order_amount = current_price * quantity order_amount = current_price * quantity
fx_blocked, remaining_cash, required_buffer = _should_block_overseas_buy_for_fx_buffer(
market=market,
action=decision.action,
total_cash=total_cash,
order_amount=order_amount,
settings=settings,
)
if fx_blocked:
logger.warning(
"Skip BUY %s (%s): FX buffer guard (remaining=%.2f, required=%.2f, cash=%.2f, order=%.2f)",
stock_code,
market.name,
remaining_cash,
required_buffer,
total_cash,
order_amount,
)
return
# 4. Check BUY cooldown (set when a prior BUY failed due to insufficient balance) # 4. Check BUY cooldown (set when a prior BUY failed due to insufficient balance)
if decision.action == "BUY" and buy_cooldown is not None: if decision.action == "BUY" and buy_cooldown is not None:
@@ -944,6 +1369,26 @@ async def trading_cycle(
except Exception as notify_exc: except Exception as notify_exc:
logger.warning("Fat finger notification failed: %s", notify_exc) logger.warning("Fat finger notification failed: %s", notify_exc)
raise # Re-raise to prevent trade raise # Re-raise to prevent trade
except CircuitBreakerTripped as exc:
ks_report = await _trigger_emergency_kill_switch(
reason=f"circuit_breaker:{market.code}:{stock_code}:{exc.pnl_pct:.2f}",
broker=broker,
overseas_broker=overseas_broker,
telegram=telegram,
settings=settings,
current_market=market,
stock_code=stock_code,
pnl_pct=exc.pnl_pct,
threshold=exc.threshold,
)
if ks_report.errors:
logger.critical(
"KillSwitch step errors for %s/%s: %s",
market.code,
stock_code,
"; ".join(ks_report.errors),
)
raise
# 5. Send order # 5. Send order
order_succeeded = True order_succeeded = True
@@ -956,6 +1401,31 @@ async def trading_cycle(
order_price = kr_round_down(current_price * 1.002) order_price = kr_round_down(current_price * 1.002)
else: else:
order_price = kr_round_down(current_price * 0.998) order_price = kr_round_down(current_price * 0.998)
try:
validate_order_policy(
market=market,
order_type=decision.action,
price=float(order_price),
)
except OrderPolicyRejected as exc:
logger.warning(
"Order policy rejected %s %s (%s): %s [session=%s]",
decision.action,
stock_code,
market.name,
exc,
exc.session_id,
)
return
if _maybe_queue_order_intent(
market=market,
stock_code=stock_code,
order_type=decision.action,
quantity=quantity,
price=float(order_price),
source="trading_cycle",
):
return
result = await broker.send_order( result = await broker.send_order(
stock_code=stock_code, stock_code=stock_code,
order_type=decision.action, order_type=decision.action,
@@ -978,6 +1448,31 @@ async def trading_cycle(
overseas_price = round(current_price * 1.002, _price_decimals) overseas_price = round(current_price * 1.002, _price_decimals)
else: else:
overseas_price = round(current_price * 0.998, _price_decimals) overseas_price = round(current_price * 0.998, _price_decimals)
try:
validate_order_policy(
market=market,
order_type=decision.action,
price=float(overseas_price),
)
except OrderPolicyRejected as exc:
logger.warning(
"Order policy rejected %s %s (%s): %s [session=%s]",
decision.action,
stock_code,
market.name,
exc,
exc.session_id,
)
return
if _maybe_queue_order_intent(
market=market,
stock_code=stock_code,
order_type=decision.action,
quantity=quantity,
price=float(overseas_price),
source="trading_cycle",
):
return
result = await overseas_broker.send_overseas_order( result = await overseas_broker.send_overseas_order(
exchange_code=market.exchange_code, exchange_code=market.exchange_code,
stock_code=stock_code, stock_code=stock_code,
@@ -1222,6 +1717,11 @@ async def handle_domestic_pending_orders(
f"Invalid price ({last_price}) for {stock_code}" f"Invalid price ({last_price}) for {stock_code}"
) )
new_price = kr_round_down(last_price * 0.996) new_price = kr_round_down(last_price * 0.996)
validate_order_policy(
market=MARKETS["KR"],
order_type="SELL",
price=float(new_price),
)
await broker.send_order( await broker.send_order(
stock_code=stock_code, stock_code=stock_code,
order_type="SELL", order_type="SELL",
@@ -1395,6 +1895,19 @@ async def handle_overseas_pending_orders(
f"Invalid price ({last_price}) for {stock_code}" f"Invalid price ({last_price}) for {stock_code}"
) )
new_price = round(last_price * 0.996, 4) new_price = round(last_price * 0.996, 4)
market_info = next(
(
m for m in MARKETS.values()
if m.exchange_code == order_exchange and not m.is_domestic
),
None,
)
if market_info is not None:
validate_order_policy(
market=market_info,
order_type="SELL",
price=float(new_price),
)
await overseas_broker.send_overseas_order( await overseas_broker.send_overseas_order(
exchange_code=order_exchange, exchange_code=order_exchange,
stock_code=stock_code, stock_code=stock_code,
@@ -1483,6 +1996,11 @@ async def run_daily_session(
# Process each open market # Process each open market
for market in open_markets: for market in open_markets:
await process_blackout_recovery_orders(
broker=broker,
overseas_broker=overseas_broker,
db_conn=db_conn,
)
# Use market-local date for playbook keying # Use market-local date for playbook keying
market_today = datetime.now(market.timezone).date() market_today = datetime.now(market.timezone).date()
@@ -1845,6 +2363,15 @@ async def run_daily_session(
trade_pnl = 0.0 trade_pnl = 0.0
order_succeeded = True order_succeeded = True
if decision.action in ("BUY", "SELL"): if decision.action in ("BUY", "SELL"):
if KILL_SWITCH.new_orders_blocked:
logger.critical(
"KillSwitch block active: skip %s order for %s (%s)",
decision.action,
stock_code,
market.name,
)
continue
daily_broker_held_qty = ( daily_broker_held_qty = (
_extract_held_qty_from_balance( _extract_held_qty_from_balance(
balance_data, stock_code, is_domestic=market.is_domestic balance_data, stock_code, is_domestic=market.is_domestic
@@ -1871,6 +2398,24 @@ async def run_daily_session(
) )
continue continue
order_amount = stock_data["current_price"] * quantity order_amount = stock_data["current_price"] * quantity
fx_blocked, remaining_cash, required_buffer = _should_block_overseas_buy_for_fx_buffer(
market=market,
action=decision.action,
total_cash=total_cash,
order_amount=order_amount,
settings=settings,
)
if fx_blocked:
logger.warning(
"Skip BUY %s (%s): FX buffer guard (remaining=%.2f, required=%.2f, cash=%.2f, order=%.2f)",
stock_code,
market.name,
remaining_cash,
required_buffer,
total_cash,
order_amount,
)
continue
# Check BUY cooldown (insufficient balance) # Check BUY cooldown (insufficient balance)
if decision.action == "BUY": if decision.action == "BUY":
@@ -1911,15 +2456,24 @@ async def run_daily_session(
logger.warning("Fat finger notification failed: %s", notify_exc) logger.warning("Fat finger notification failed: %s", notify_exc)
continue # Skip this order continue # Skip this order
except CircuitBreakerTripped as exc: except CircuitBreakerTripped as exc:
logger.critical("Circuit breaker tripped — stopping session") ks_report = await _trigger_emergency_kill_switch(
try: reason=f"daily_circuit_breaker:{market.code}:{stock_code}:{exc.pnl_pct:.2f}",
await telegram.notify_circuit_breaker( broker=broker,
overseas_broker=overseas_broker,
telegram=telegram,
settings=settings,
current_market=market,
stock_code=stock_code,
pnl_pct=exc.pnl_pct, pnl_pct=exc.pnl_pct,
threshold=exc.threshold, threshold=exc.threshold,
) )
except Exception as notify_exc: logger.critical("Circuit breaker tripped — stopping session")
logger.warning( if ks_report.errors:
"Circuit breaker notification failed: %s", notify_exc logger.critical(
"Daily KillSwitch step errors for %s/%s: %s",
market.code,
stock_code,
"; ".join(ks_report.errors),
) )
raise raise
@@ -1937,6 +2491,31 @@ async def run_daily_session(
order_price = kr_round_down( order_price = kr_round_down(
stock_data["current_price"] * 0.998 stock_data["current_price"] * 0.998
) )
try:
validate_order_policy(
market=market,
order_type=decision.action,
price=float(order_price),
)
except OrderPolicyRejected as exc:
logger.warning(
"Order policy rejected %s %s (%s): %s [session=%s]",
decision.action,
stock_code,
market.name,
exc,
exc.session_id,
)
continue
if _maybe_queue_order_intent(
market=market,
stock_code=stock_code,
order_type=decision.action,
quantity=quantity,
price=float(order_price),
source="run_daily_session",
):
continue
result = await broker.send_order( result = await broker.send_order(
stock_code=stock_code, stock_code=stock_code,
order_type=decision.action, order_type=decision.action,
@@ -1949,6 +2528,31 @@ async def run_daily_session(
order_price = round(stock_data["current_price"] * 1.005, 4) order_price = round(stock_data["current_price"] * 1.005, 4)
else: else:
order_price = stock_data["current_price"] order_price = stock_data["current_price"]
try:
validate_order_policy(
market=market,
order_type=decision.action,
price=float(order_price),
)
except OrderPolicyRejected as exc:
logger.warning(
"Order policy rejected %s %s (%s): %s [session=%s]",
decision.action,
stock_code,
market.name,
exc,
exc.session_id,
)
continue
if _maybe_queue_order_intent(
market=market,
stock_code=stock_code,
order_type=decision.action,
quantity=quantity,
price=float(order_price),
source="run_daily_session",
):
continue
result = await overseas_broker.send_overseas_order( result = await overseas_broker.send_overseas_order(
exchange_code=market.exchange_code, exchange_code=market.exchange_code,
stock_code=stock_code, stock_code=stock_code,
@@ -2187,6 +2791,19 @@ def _apply_dashboard_flag(settings: Settings, dashboard_flag: bool) -> Settings:
async def run(settings: Settings) -> None: async def run(settings: Settings) -> None:
"""Main async loop — iterate over open markets on a timer.""" """Main async loop — iterate over open markets on a timer."""
global BLACKOUT_ORDER_MANAGER
BLACKOUT_ORDER_MANAGER = BlackoutOrderManager(
enabled=settings.ORDER_BLACKOUT_ENABLED,
windows=parse_blackout_windows_kst(settings.ORDER_BLACKOUT_WINDOWS_KST),
max_queue_size=settings.ORDER_BLACKOUT_QUEUE_MAX,
)
logger.info(
"Blackout manager initialized: enabled=%s windows=%s queue_max=%d",
settings.ORDER_BLACKOUT_ENABLED,
settings.ORDER_BLACKOUT_WINDOWS_KST,
settings.ORDER_BLACKOUT_QUEUE_MAX,
)
broker = KISBroker(settings) broker = KISBroker(settings)
overseas_broker = OverseasBroker(broker) overseas_broker = OverseasBroker(broker)
brain = GeminiClient(settings) brain = GeminiClient(settings)
@@ -2786,6 +3403,12 @@ async def run(settings: Settings) -> None:
if shutdown.is_set(): if shutdown.is_set():
break break
await process_blackout_recovery_orders(
broker=broker,
overseas_broker=overseas_broker,
db_conn=db_conn,
)
# Notify market open if it just opened # Notify market open if it just opened
if not _market_states.get(market.code, False): if not _market_states.get(market.code, False):
try: try:

104
src/strategy/exit_rules.py Normal file
View File

@@ -0,0 +1,104 @@
"""Composite exit rules: hard stop, break-even lock, ATR trailing, model assist."""
from __future__ import annotations
from dataclasses import dataclass
from src.strategy.position_state_machine import PositionState, StateTransitionInput, promote_state
@dataclass(frozen=True)
class ExitRuleConfig:
hard_stop_pct: float = -2.0
be_arm_pct: float = 1.2
arm_pct: float = 3.0
atr_multiplier_k: float = 2.2
model_prob_threshold: float = 0.62
@dataclass(frozen=True)
class ExitRuleInput:
current_price: float
entry_price: float
peak_price: float
atr_value: float = 0.0
pred_down_prob: float = 0.0
liquidity_weak: bool = False
@dataclass(frozen=True)
class ExitEvaluation:
state: PositionState
should_exit: bool
reason: str
unrealized_pnl_pct: float
trailing_stop_price: float | None
def evaluate_exit(
*,
current_state: PositionState,
config: ExitRuleConfig,
inp: ExitRuleInput,
) -> ExitEvaluation:
"""Evaluate composite exit logic and return updated state."""
if inp.entry_price <= 0 or inp.current_price <= 0:
return ExitEvaluation(
state=current_state,
should_exit=False,
reason="invalid_price",
unrealized_pnl_pct=0.0,
trailing_stop_price=None,
)
unrealized = (inp.current_price - inp.entry_price) / inp.entry_price * 100.0
hard_stop_hit = unrealized <= config.hard_stop_pct
take_profit_hit = unrealized >= config.arm_pct
trailing_stop_price: float | None = None
trailing_stop_hit = False
if inp.atr_value > 0 and inp.peak_price > 0:
trailing_stop_price = inp.peak_price - (config.atr_multiplier_k * inp.atr_value)
trailing_stop_hit = inp.current_price <= trailing_stop_price
be_lock_threat = current_state in (PositionState.BE_LOCK, PositionState.ARMED) and (
inp.current_price <= inp.entry_price
)
model_exit_signal = inp.pred_down_prob >= config.model_prob_threshold and inp.liquidity_weak
next_state = promote_state(
current=current_state,
inp=StateTransitionInput(
unrealized_pnl_pct=unrealized,
be_arm_pct=config.be_arm_pct,
arm_pct=config.arm_pct,
hard_stop_hit=hard_stop_hit,
trailing_stop_hit=trailing_stop_hit,
model_exit_signal=model_exit_signal,
be_lock_threat=be_lock_threat,
),
)
if hard_stop_hit:
reason = "hard_stop"
elif trailing_stop_hit:
reason = "atr_trailing_stop"
elif be_lock_threat:
reason = "be_lock_threat"
elif model_exit_signal:
reason = "model_liquidity_exit"
elif take_profit_hit:
# Backward-compatible immediate profit-taking path.
reason = "arm_take_profit"
else:
reason = "hold"
should_exit = next_state == PositionState.EXITED or take_profit_hit
return ExitEvaluation(
state=next_state,
should_exit=should_exit,
reason=reason,
unrealized_pnl_pct=unrealized,
trailing_stop_price=trailing_stop_price,
)

View File

@@ -0,0 +1,70 @@
"""Position state machine for staged exit control.
State progression is monotonic (promotion-only) except terminal EXITED.
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
class PositionState(str, Enum):
HOLDING = "HOLDING"
BE_LOCK = "BE_LOCK"
ARMED = "ARMED"
EXITED = "EXITED"
_STATE_RANK: dict[PositionState, int] = {
PositionState.HOLDING: 0,
PositionState.BE_LOCK: 1,
PositionState.ARMED: 2,
PositionState.EXITED: 3,
}
@dataclass(frozen=True)
class StateTransitionInput:
unrealized_pnl_pct: float
be_arm_pct: float
arm_pct: float
hard_stop_hit: bool = False
trailing_stop_hit: bool = False
model_exit_signal: bool = False
be_lock_threat: bool = False
def evaluate_exit_first(inp: StateTransitionInput) -> bool:
"""Return True when terminal exit conditions are met.
EXITED must be evaluated before any promotion.
"""
return (
inp.hard_stop_hit
or inp.trailing_stop_hit
or inp.model_exit_signal
or inp.be_lock_threat
)
def promote_state(current: PositionState, inp: StateTransitionInput) -> PositionState:
"""Promote to highest admissible state for current tick/bar.
Rules:
- EXITED has highest precedence and is terminal.
- Promotions are monotonic (no downgrade).
"""
if current == PositionState.EXITED:
return PositionState.EXITED
if evaluate_exit_first(inp):
return PositionState.EXITED
target = PositionState.HOLDING
if inp.unrealized_pnl_pct >= inp.arm_pct:
target = PositionState.ARMED
elif inp.unrealized_pnl_pct >= inp.be_arm_pct:
target = PositionState.BE_LOCK
return target if _STATE_RANK[target] > _STATE_RANK[current] else current

View File

@@ -0,0 +1,83 @@
from __future__ import annotations
import pytest
from src.analysis.backtest_cost_guard import BacktestCostModel, validate_backtest_cost_model
def test_valid_backtest_cost_model_passes() -> None:
model = BacktestCostModel(
commission_bps=5.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,
)
validate_backtest_cost_model(model=model, required_sessions=["KRX_REG", "US_PRE"])
def test_missing_required_slippage_session_raises() -> None:
model = BacktestCostModel(
commission_bps=5.0,
slippage_bps_by_session={"KRX_REG": 10.0},
failure_rate_by_session={"KRX_REG": 0.01, "US_PRE": 0.08},
unfavorable_fill_required=True,
)
with pytest.raises(ValueError, match="missing slippage_bps_by_session.*US_PRE"):
validate_backtest_cost_model(model=model, required_sessions=["KRX_REG", "US_PRE"])
def test_missing_required_failure_rate_session_raises() -> None:
model = BacktestCostModel(
commission_bps=5.0,
slippage_bps_by_session={"KRX_REG": 10.0, "US_PRE": 50.0},
failure_rate_by_session={"KRX_REG": 0.01},
unfavorable_fill_required=True,
)
with pytest.raises(ValueError, match="missing failure_rate_by_session.*US_PRE"):
validate_backtest_cost_model(model=model, required_sessions=["KRX_REG", "US_PRE"])
def test_invalid_failure_rate_range_raises() -> None:
model = BacktestCostModel(
commission_bps=5.0,
slippage_bps_by_session={"KRX_REG": 10.0},
failure_rate_by_session={"KRX_REG": 1.2},
unfavorable_fill_required=True,
)
with pytest.raises(ValueError, match="failure rate must be within"):
validate_backtest_cost_model(model=model, required_sessions=["KRX_REG"])
def test_unfavorable_fill_requirement_cannot_be_disabled() -> None:
model = BacktestCostModel(
commission_bps=5.0,
slippage_bps_by_session={"KRX_REG": 10.0},
failure_rate_by_session={"KRX_REG": 0.02},
unfavorable_fill_required=False,
)
with pytest.raises(ValueError, match="unfavorable_fill_required must be True"):
validate_backtest_cost_model(model=model, required_sessions=["KRX_REG"])
@pytest.mark.parametrize("bad_commission", [float("nan"), float("inf"), float("-inf")])
def test_non_finite_commission_rejected(bad_commission: float) -> None:
model = BacktestCostModel(
commission_bps=bad_commission,
slippage_bps_by_session={"KRX_REG": 10.0},
failure_rate_by_session={"KRX_REG": 0.02},
unfavorable_fill_required=True,
)
with pytest.raises(ValueError, match="commission_bps"):
validate_backtest_cost_model(model=model, required_sessions=["KRX_REG"])
@pytest.mark.parametrize("bad_slippage", [float("nan"), float("inf"), float("-inf")])
def test_non_finite_slippage_rejected(bad_slippage: float) -> None:
model = BacktestCostModel(
commission_bps=5.0,
slippage_bps_by_session={"KRX_REG": bad_slippage},
failure_rate_by_session={"KRX_REG": 0.02},
unfavorable_fill_required=True,
)
with pytest.raises(ValueError, match="slippage bps"):
validate_backtest_cost_model(model=model, required_sessions=["KRX_REG"])

View File

@@ -0,0 +1,108 @@
from __future__ import annotations
import pytest
from src.analysis.backtest_execution_model import (
BacktestExecutionModel,
ExecutionAssumptions,
ExecutionRequest,
)
def test_buy_uses_unfavorable_slippage_direction() -> None:
model = BacktestExecutionModel(
ExecutionAssumptions(
slippage_bps_by_session={"US_PRE": 50.0},
failure_rate_by_session={"US_PRE": 0.0},
partial_fill_rate_by_session={"US_PRE": 0.0},
seed=1,
)
)
out = model.simulate(
ExecutionRequest(side="BUY", session_id="US_PRE", qty=10, reference_price=100.0)
)
assert out.status == "FILLED"
assert out.avg_price == pytest.approx(100.5)
def test_sell_uses_unfavorable_slippage_direction() -> None:
model = BacktestExecutionModel(
ExecutionAssumptions(
slippage_bps_by_session={"US_PRE": 50.0},
failure_rate_by_session={"US_PRE": 0.0},
partial_fill_rate_by_session={"US_PRE": 0.0},
seed=1,
)
)
out = model.simulate(
ExecutionRequest(side="SELL", session_id="US_PRE", qty=10, reference_price=100.0)
)
assert out.status == "FILLED"
assert out.avg_price == pytest.approx(99.5)
def test_failure_rate_can_reject_order() -> None:
model = BacktestExecutionModel(
ExecutionAssumptions(
slippage_bps_by_session={"KRX_REG": 10.0},
failure_rate_by_session={"KRX_REG": 1.0},
partial_fill_rate_by_session={"KRX_REG": 0.0},
seed=42,
)
)
out = model.simulate(
ExecutionRequest(side="BUY", session_id="KRX_REG", qty=10, reference_price=100.0)
)
assert out.status == "REJECTED"
assert out.filled_qty == 0
def test_partial_fill_applies_when_rate_is_one() -> None:
model = BacktestExecutionModel(
ExecutionAssumptions(
slippage_bps_by_session={"KRX_REG": 0.0},
failure_rate_by_session={"KRX_REG": 0.0},
partial_fill_rate_by_session={"KRX_REG": 1.0},
partial_fill_min_ratio=0.4,
partial_fill_max_ratio=0.4,
seed=0,
)
)
out = model.simulate(
ExecutionRequest(side="BUY", session_id="KRX_REG", qty=10, reference_price=100.0)
)
assert out.status == "PARTIAL"
assert out.filled_qty == 4
assert out.avg_price == 100.0
@pytest.mark.parametrize("bad_slip", [-1.0, float("nan"), float("inf")])
def test_invalid_slippage_is_rejected(bad_slip: float) -> None:
with pytest.raises(ValueError, match="slippage_bps"):
BacktestExecutionModel(
ExecutionAssumptions(
slippage_bps_by_session={"US_PRE": bad_slip},
failure_rate_by_session={"US_PRE": 0.0},
partial_fill_rate_by_session={"US_PRE": 0.0},
)
)
@pytest.mark.parametrize("bad_rate", [-0.1, 1.1, float("nan")])
def test_invalid_failure_or_partial_rates_are_rejected(bad_rate: float) -> None:
with pytest.raises(ValueError, match="failure_rate"):
BacktestExecutionModel(
ExecutionAssumptions(
slippage_bps_by_session={"US_PRE": 10.0},
failure_rate_by_session={"US_PRE": bad_rate},
partial_fill_rate_by_session={"US_PRE": 0.0},
)
)
with pytest.raises(ValueError, match="partial_fill_rate"):
BacktestExecutionModel(
ExecutionAssumptions(
slippage_bps_by_session={"US_PRE": 10.0},
failure_rate_by_session={"US_PRE": 0.0},
partial_fill_rate_by_session={"US_PRE": bad_rate},
)
)

View File

@@ -0,0 +1,81 @@
from __future__ import annotations
from datetime import UTC, datetime
from src.core.blackout_manager import (
BlackoutOrderManager,
QueuedOrderIntent,
parse_blackout_windows_kst,
)
def test_parse_blackout_windows_kst() -> None:
windows = parse_blackout_windows_kst("23:30-00:10,11:20-11:30,invalid")
assert len(windows) == 2
def test_blackout_manager_handles_cross_midnight_window() -> None:
manager = BlackoutOrderManager(
enabled=True,
windows=parse_blackout_windows_kst("23:30-00:10"),
max_queue_size=10,
)
# 2026-01-01 23:40 KST = 2026-01-01 14:40 UTC
assert manager.in_blackout(datetime(2026, 1, 1, 14, 40, tzinfo=UTC))
# 2026-01-02 00:20 KST = 2026-01-01 15:20 UTC
assert not manager.in_blackout(datetime(2026, 1, 1, 15, 20, tzinfo=UTC))
def test_recovery_batch_only_after_blackout_exit() -> None:
manager = BlackoutOrderManager(
enabled=True,
windows=parse_blackout_windows_kst("23:30-00:10"),
max_queue_size=10,
)
intent = QueuedOrderIntent(
market_code="KR",
exchange_code="KRX",
stock_code="005930",
order_type="BUY",
quantity=1,
price=100.0,
source="test",
queued_at=datetime.now(UTC),
)
assert manager.enqueue(intent)
# Inside blackout: no pop yet
inside_blackout = datetime(2026, 1, 1, 14, 40, tzinfo=UTC)
assert manager.pop_recovery_batch(inside_blackout) == []
# Outside blackout: pop full batch once
outside_blackout = datetime(2026, 1, 1, 15, 20, tzinfo=UTC)
batch = manager.pop_recovery_batch(outside_blackout)
assert len(batch) == 1
assert manager.pending_count == 0
def test_requeued_intent_is_processed_next_non_blackout_cycle() -> None:
manager = BlackoutOrderManager(
enabled=True,
windows=parse_blackout_windows_kst("23:30-00:10"),
max_queue_size=10,
)
intent = QueuedOrderIntent(
market_code="KR",
exchange_code="KRX",
stock_code="005930",
order_type="BUY",
quantity=1,
price=100.0,
source="test",
queued_at=datetime.now(UTC),
)
manager.enqueue(intent)
outside_blackout = datetime(2026, 1, 1, 15, 20, tzinfo=UTC)
first_batch = manager.pop_recovery_batch(outside_blackout)
assert len(first_batch) == 1
manager.requeue(first_batch[0])
second_batch = manager.pop_recovery_batch(outside_blackout)
assert len(second_batch) == 1

View File

@@ -155,6 +155,9 @@ def test_mode_column_exists_in_schema() -> None:
cursor = conn.execute("PRAGMA table_info(trades)") cursor = conn.execute("PRAGMA table_info(trades)")
columns = {row[1] for row in cursor.fetchall()} columns = {row[1] for row in cursor.fetchall()}
assert "mode" in columns assert "mode" in columns
assert "session_id" in columns
assert "strategy_pnl" in columns
assert "fx_pnl" in columns
def test_mode_migration_adds_column_to_existing_db() -> None: def test_mode_migration_adds_column_to_existing_db() -> None:
@@ -182,6 +185,13 @@ def test_mode_migration_adds_column_to_existing_db() -> None:
decision_id TEXT decision_id TEXT
)""" )"""
) )
old_conn.execute(
"""
INSERT INTO trades (
timestamp, stock_code, action, confidence, rationale, quantity, price, pnl
) VALUES ('2026-01-01T00:00:00+00:00', 'AAPL', 'SELL', 90, 'legacy', 1, 100.0, 123.45)
"""
)
old_conn.commit() old_conn.commit()
old_conn.close() old_conn.close()
@@ -190,6 +200,132 @@ def test_mode_migration_adds_column_to_existing_db() -> None:
cursor = conn.execute("PRAGMA table_info(trades)") cursor = conn.execute("PRAGMA table_info(trades)")
columns = {row[1] for row in cursor.fetchall()} columns = {row[1] for row in cursor.fetchall()}
assert "mode" in columns assert "mode" in columns
assert "session_id" in columns
assert "strategy_pnl" in columns
assert "fx_pnl" in columns
migrated = conn.execute(
"SELECT pnl, strategy_pnl, fx_pnl, session_id FROM trades WHERE stock_code='AAPL' LIMIT 1"
).fetchone()
assert migrated is not None
assert migrated[0] == 123.45
assert migrated[1] == 123.45
assert migrated[2] == 0.0
assert migrated[3] == "UNKNOWN"
conn.close() conn.close()
finally: finally:
os.unlink(db_path) os.unlink(db_path)
def test_log_trade_stores_strategy_and_fx_pnl_separately() -> None:
conn = init_db(":memory:")
log_trade(
conn=conn,
stock_code="AAPL",
action="SELL",
confidence=90,
rationale="fx split",
pnl=120.0,
strategy_pnl=100.0,
fx_pnl=20.0,
market="US_NASDAQ",
exchange_code="NASD",
)
row = conn.execute(
"SELECT pnl, strategy_pnl, fx_pnl FROM trades ORDER BY id DESC LIMIT 1"
).fetchone()
assert row is not None
assert row[0] == 120.0
assert row[1] == 100.0
assert row[2] == 20.0
def test_log_trade_backward_compat_sets_strategy_pnl_from_pnl() -> None:
conn = init_db(":memory:")
log_trade(
conn=conn,
stock_code="005930",
action="SELL",
confidence=80,
rationale="legacy",
pnl=50.0,
market="KR",
exchange_code="KRX",
)
row = conn.execute(
"SELECT pnl, strategy_pnl, fx_pnl FROM trades ORDER BY id DESC LIMIT 1"
).fetchone()
assert row is not None
assert row[0] == 50.0
assert row[1] == 50.0
assert row[2] == 0.0
def test_log_trade_partial_fx_input_does_not_infer_negative_strategy_pnl() -> None:
conn = init_db(":memory:")
log_trade(
conn=conn,
stock_code="AAPL",
action="SELL",
confidence=70,
rationale="fx only",
pnl=0.0,
fx_pnl=10.0,
market="US_NASDAQ",
exchange_code="NASD",
)
row = conn.execute(
"SELECT pnl, strategy_pnl, fx_pnl FROM trades ORDER BY id DESC LIMIT 1"
).fetchone()
assert row is not None
assert row[0] == 10.0
assert row[1] == 0.0
assert row[2] == 10.0
def test_log_trade_persists_explicit_session_id() -> None:
conn = init_db(":memory:")
log_trade(
conn=conn,
stock_code="AAPL",
action="BUY",
confidence=70,
rationale="session test",
market="US_NASDAQ",
exchange_code="NASD",
session_id="US_PRE",
)
row = conn.execute("SELECT session_id FROM trades ORDER BY id DESC LIMIT 1").fetchone()
assert row is not None
assert row[0] == "US_PRE"
def test_log_trade_auto_derives_session_id_when_not_provided() -> None:
conn = init_db(":memory:")
log_trade(
conn=conn,
stock_code="005930",
action="BUY",
confidence=70,
rationale="auto session",
market="KR",
exchange_code="KRX",
)
row = conn.execute("SELECT session_id FROM trades ORDER BY id DESC LIMIT 1").fetchone()
assert row is not None
assert row[0] != "UNKNOWN"
def test_log_trade_unknown_market_falls_back_to_unknown_session() -> None:
conn = init_db(":memory:")
log_trade(
conn=conn,
stock_code="X",
action="BUY",
confidence=70,
rationale="unknown market",
market="MARS",
exchange_code="MARS",
)
row = conn.execute("SELECT session_id FROM trades ORDER BY id DESC LIMIT 1").fetchone()
assert row is not None
assert row[0] == "UNKNOWN"

55
tests/test_kill_switch.py Normal file
View File

@@ -0,0 +1,55 @@
import pytest
from src.core.kill_switch import KillSwitchOrchestrator
@pytest.mark.asyncio
async def test_kill_switch_executes_steps_in_order() -> None:
ks = KillSwitchOrchestrator()
calls: list[str] = []
async def _cancel() -> None:
calls.append("cancel")
def _refresh() -> None:
calls.append("refresh")
def _reduce() -> None:
calls.append("reduce")
def _snapshot() -> None:
calls.append("snapshot")
def _notify() -> None:
calls.append("notify")
report = await ks.trigger(
reason="test",
cancel_pending_orders=_cancel,
refresh_order_state=_refresh,
reduce_risk=_reduce,
snapshot_state=_snapshot,
notify=_notify,
)
assert report.steps == [
"block_new_orders",
"cancel_pending_orders",
"refresh_order_state",
"reduce_risk",
"snapshot_state",
"notify",
]
assert calls == ["cancel", "refresh", "reduce", "snapshot", "notify"]
assert report.errors == []
@pytest.mark.asyncio
async def test_kill_switch_collects_step_errors() -> None:
ks = KillSwitchOrchestrator()
def _boom() -> None:
raise RuntimeError("boom")
report = await ks.trigger(reason="test", cancel_pending_orders=_boom)
assert any(err.startswith("cancel_pending_orders:") for err in report.errors)

View File

@@ -8,11 +8,15 @@ import pytest
from src.config import Settings from src.config import Settings
from src.context.layer import ContextLayer from src.context.layer import ContextLayer
from src.context.scheduler import ScheduleResult from src.context.scheduler import ScheduleResult
from src.core.order_policy import OrderPolicyRejected
from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected
from src.db import init_db, log_trade from src.db import init_db, log_trade
from src.evolution.scorecard import DailyScorecard from src.evolution.scorecard import DailyScorecard
from src.logging.decision_logger import DecisionLogger from src.logging.decision_logger import DecisionLogger
from src.main import ( from src.main import (
KILL_SWITCH,
_should_block_overseas_buy_for_fx_buffer,
_trigger_emergency_kill_switch,
_apply_dashboard_flag, _apply_dashboard_flag,
_determine_order_quantity, _determine_order_quantity,
_extract_avg_price_from_balance, _extract_avg_price_from_balance,
@@ -25,6 +29,7 @@ from src.main import (
_start_dashboard_server, _start_dashboard_server,
handle_domestic_pending_orders, handle_domestic_pending_orders,
handle_overseas_pending_orders, handle_overseas_pending_orders,
process_blackout_recovery_orders,
run_daily_session, run_daily_session,
safe_float, safe_float,
sync_positions_from_broker, sync_positions_from_broker,
@@ -77,6 +82,14 @@ 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."""
KILL_SWITCH.clear_block()
yield
KILL_SWITCH.clear_block()
class TestExtractAvgPriceFromBalance: class TestExtractAvgPriceFromBalance:
"""Tests for _extract_avg_price_from_balance() (issue #249).""" """Tests for _extract_avg_price_from_balance() (issue #249)."""
@@ -3678,6 +3691,81 @@ class TestOverseasBrokerIntegration:
# DB도 브로커도 보유 없음 → BUY 주문이 실행되어야 함 (회귀 테스트) # DB도 브로커도 보유 없음 → BUY 주문이 실행되어야 함 (회귀 테스트)
overseas_broker.send_overseas_order.assert_called_once() overseas_broker.send_overseas_order.assert_called_once()
@pytest.mark.asyncio
async def test_overseas_buy_blocked_by_usd_buffer_guard(self) -> None:
"""Overseas BUY must be blocked when USD buffer would be breached."""
db_conn = init_db(":memory:")
overseas_broker = MagicMock()
overseas_broker.get_overseas_price = AsyncMock(
return_value={"output": {"last": "182.50"}}
)
overseas_broker.get_overseas_balance = AsyncMock(
return_value={
"output1": [],
"output2": [
{
"frcr_evlu_tota": "50000.00",
"frcr_buy_amt_smtl": "0.00",
}
],
}
)
overseas_broker.get_overseas_buying_power = AsyncMock(
return_value={"output": {"ovrs_ord_psbl_amt": "50000.00"}}
)
overseas_broker.send_overseas_order = AsyncMock(return_value={"msg1": "주문접수"})
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_buy_match("AAPL"))
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()
decision_logger = MagicMock()
decision_logger.log_decision = MagicMock(return_value="decision-id")
settings = MagicMock()
settings.POSITION_SIZING_ENABLED = False
settings.CONFIDENCE_THRESHOLD = 80
settings.USD_BUFFER_MIN = 49900.0
settings.MODE = "paper"
settings.PAPER_OVERSEAS_CASH = 50000.0
await trading_cycle(
broker=MagicMock(),
overseas_broker=overseas_broker,
scenario_engine=engine,
playbook=_make_playbook(market="US"),
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="AAPL",
scan_candidates={},
settings=settings,
)
overseas_broker.send_overseas_order.assert_not_called()
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# _retry_connection — unit tests (issue #209) # _retry_connection — unit tests (issue #209)
@@ -3711,7 +3799,6 @@ class TestRetryConnection:
with patch("src.main.asyncio.sleep") as mock_sleep: with patch("src.main.asyncio.sleep") as mock_sleep:
mock_sleep.return_value = None mock_sleep.return_value = None
result = await _retry_connection(flaky, label="flaky") result = await _retry_connection(flaky, label="flaky")
assert result == "ok" assert result == "ok"
assert call_count == 2 assert call_count == 2
mock_sleep.assert_called_once() mock_sleep.assert_called_once()
@@ -3766,6 +3853,48 @@ class TestRetryConnection:
assert call_count == 1 # No retry for non-ConnectionError assert call_count == 1 # No retry for non-ConnectionError
def test_fx_buffer_guard_applies_only_to_us_and_respects_boundary() -> None:
settings = MagicMock()
settings.USD_BUFFER_MIN = 1000.0
us_market = MagicMock()
us_market.is_domestic = False
us_market.code = "US_NASDAQ"
blocked, remaining, required = _should_block_overseas_buy_for_fx_buffer(
market=us_market,
action="BUY",
total_cash=5000.0,
order_amount=4001.0,
settings=settings,
)
assert blocked
assert remaining == 999.0
assert required == 1000.0
blocked_eq, _, _ = _should_block_overseas_buy_for_fx_buffer(
market=us_market,
action="BUY",
total_cash=5000.0,
order_amount=4000.0,
settings=settings,
)
assert not blocked_eq
jp_market = MagicMock()
jp_market.is_domestic = False
jp_market.code = "JP"
blocked_jp, _, required_jp = _should_block_overseas_buy_for_fx_buffer(
market=jp_market,
action="BUY",
total_cash=5000.0,
order_amount=4500.0,
settings=settings,
)
assert not blocked_jp
assert required_jp == 0.0
# run_daily_session — daily CB baseline (daily_start_eval) tests (issue #207) # run_daily_session — daily CB baseline (daily_start_eval) tests (issue #207)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -5039,3 +5168,417 @@ class TestOverseasGhostPositionClose:
and "[ghost-close]" in (c.kwargs.get("rationale") or "") and "[ghost-close]" in (c.kwargs.get("rationale") or "")
] ]
assert not ghost_close_calls, "Ghost-close must NOT be triggered for non-잔고없음 errors" assert not ghost_close_calls, "Ghost-close must NOT be triggered for non-잔고없음 errors"
@pytest.mark.asyncio
async def test_kill_switch_block_skips_actionable_order_execution() -> None:
"""Active kill-switch must prevent actionable order execution."""
db_conn = init_db(":memory:")
decision_logger = DecisionLogger(db_conn)
broker = MagicMock()
broker.get_current_price = AsyncMock(return_value=(100.0, 0.5, 0.0))
broker.get_balance = AsyncMock(
return_value={
"output1": [],
"output2": [
{
"tot_evlu_amt": "100000",
"dnca_tot_amt": "50000",
"pchs_amt_smtl_amt": "50000",
}
],
}
)
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()
settings = MagicMock()
settings.POSITION_SIZING_ENABLED = False
settings.CONFIDENCE_THRESHOLD = 80
try:
KILL_SWITCH.new_orders_blocked = True
await trading_cycle(
broker=broker,
overseas_broker=MagicMock(),
scenario_engine=MagicMock(evaluate=MagicMock(return_value=_make_buy_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={},
settings=settings,
)
finally:
KILL_SWITCH.clear_block()
broker.send_order.assert_not_called()
@pytest.mark.asyncio
async def test_order_policy_rejection_skips_order_execution() -> None:
"""Order policy rejection must prevent order submission."""
db_conn = init_db(":memory:")
decision_logger = DecisionLogger(db_conn)
broker = MagicMock()
broker.get_current_price = AsyncMock(return_value=(100.0, 0.5, 0.0))
broker.get_balance = AsyncMock(
return_value={
"output1": [],
"output2": [
{
"tot_evlu_amt": "100000",
"dnca_tot_amt": "50000",
"pchs_amt_smtl_amt": "50000",
}
],
}
)
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()
settings = MagicMock()
settings.POSITION_SIZING_ENABLED = False
settings.CONFIDENCE_THRESHOLD = 80
with patch(
"src.main.validate_order_policy",
side_effect=OrderPolicyRejected(
"rejected",
session_id="NXT_AFTER",
market_code="KR",
),
):
await trading_cycle(
broker=broker,
overseas_broker=MagicMock(),
scenario_engine=MagicMock(evaluate=MagicMock(return_value=_make_buy_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={},
settings=settings,
)
broker.send_order.assert_not_called()
@pytest.mark.asyncio
async def test_blackout_queues_order_and_skips_submission() -> None:
"""When blackout is active, order submission is replaced by queueing."""
db_conn = init_db(":memory:")
decision_logger = DecisionLogger(db_conn)
broker = MagicMock()
broker.get_current_price = AsyncMock(return_value=(100.0, 0.5, 0.0))
broker.get_balance = AsyncMock(
return_value={
"output1": [],
"output2": [
{
"tot_evlu_amt": "100000",
"dnca_tot_amt": "50000",
"pchs_amt_smtl_amt": "50000",
}
],
}
)
broker.send_order = AsyncMock(return_value={"msg1": "OK"})
market = MagicMock()
market.name = "Korea"
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
settings = MagicMock()
settings.POSITION_SIZING_ENABLED = False
settings.CONFIDENCE_THRESHOLD = 80
telegram = MagicMock()
telegram.notify_trade_execution = AsyncMock()
telegram.notify_fat_finger = AsyncMock()
telegram.notify_circuit_breaker = AsyncMock()
telegram.notify_scenario_matched = AsyncMock()
blackout_manager = MagicMock()
blackout_manager.in_blackout.return_value = True
blackout_manager.enqueue.return_value = True
blackout_manager.pending_count = 1
with patch("src.main.BLACKOUT_ORDER_MANAGER", blackout_manager):
await trading_cycle(
broker=broker,
overseas_broker=MagicMock(),
scenario_engine=MagicMock(evaluate=MagicMock(return_value=_make_buy_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={},
settings=settings,
)
broker.send_order.assert_not_called()
blackout_manager.enqueue.assert_called_once()
@pytest.mark.asyncio
async def test_process_blackout_recovery_executes_valid_intents() -> None:
"""Recovery must execute queued intents that pass revalidation."""
db_conn = init_db(":memory:")
broker = MagicMock()
broker.send_order = AsyncMock(return_value={"rt_cd": "0", "msg1": "OK"})
overseas_broker = MagicMock()
market = MagicMock()
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
intent = MagicMock()
intent.market_code = "KR"
intent.stock_code = "005930"
intent.order_type = "BUY"
intent.quantity = 1
intent.price = 100.0
intent.source = "test"
intent.attempts = 0
blackout_manager = MagicMock()
blackout_manager.pop_recovery_batch.return_value = [intent]
with (
patch("src.main.BLACKOUT_ORDER_MANAGER", blackout_manager),
patch("src.main.MARKETS", {"KR": market}),
patch("src.main.get_open_position", return_value=None),
patch("src.main.validate_order_policy"),
):
await process_blackout_recovery_orders(
broker=broker,
overseas_broker=overseas_broker,
db_conn=db_conn,
)
broker.send_order.assert_called_once()
@pytest.mark.asyncio
async def test_process_blackout_recovery_drops_policy_rejected_intent() -> None:
"""Policy-rejected queued intents must not be requeued."""
db_conn = init_db(":memory:")
broker = MagicMock()
broker.send_order = AsyncMock(return_value={"rt_cd": "0", "msg1": "OK"})
overseas_broker = MagicMock()
market = MagicMock()
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
intent = MagicMock()
intent.market_code = "KR"
intent.stock_code = "005930"
intent.order_type = "BUY"
intent.quantity = 1
intent.price = 100.0
intent.source = "test"
intent.attempts = 0
blackout_manager = MagicMock()
blackout_manager.pop_recovery_batch.return_value = [intent]
with (
patch("src.main.BLACKOUT_ORDER_MANAGER", blackout_manager),
patch("src.main.MARKETS", {"KR": market}),
patch("src.main.get_open_position", return_value=None),
patch(
"src.main.validate_order_policy",
side_effect=OrderPolicyRejected(
"blocked",
session_id="NXT_AFTER",
market_code="KR",
),
),
):
await process_blackout_recovery_orders(
broker=broker,
overseas_broker=overseas_broker,
db_conn=db_conn,
)
broker.send_order.assert_not_called()
blackout_manager.requeue.assert_not_called()
@pytest.mark.asyncio
async def test_trigger_emergency_kill_switch_executes_operational_steps() -> None:
"""Emergency kill switch should execute cancel/refresh/reduce/notify callbacks."""
broker = MagicMock()
broker.get_domestic_pending_orders = AsyncMock(
return_value=[
{
"pdno": "005930",
"orgn_odno": "1",
"ord_gno_brno": "01",
"psbl_qty": "3",
}
]
)
broker.cancel_domestic_order = AsyncMock(return_value={"rt_cd": "0"})
broker.get_balance = AsyncMock(return_value={"output1": [], "output2": []})
overseas_broker = MagicMock()
overseas_broker.get_overseas_pending_orders = AsyncMock(return_value=[])
overseas_broker.get_overseas_balance = AsyncMock(return_value={"output1": [], "output2": []})
telegram = MagicMock()
telegram.notify_circuit_breaker = AsyncMock()
settings = MagicMock()
settings.enabled_market_list = ["KR"]
market = MagicMock()
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
with (
patch("src.main.MARKETS", {"KR": market}),
patch("src.main.BLACKOUT_ORDER_MANAGER.clear", return_value=2),
):
report = await _trigger_emergency_kill_switch(
reason="test",
broker=broker,
overseas_broker=overseas_broker,
telegram=telegram,
settings=settings,
current_market=market,
stock_code="005930",
pnl_pct=-3.2,
threshold=-3.0,
)
assert report.steps == [
"block_new_orders",
"cancel_pending_orders",
"refresh_order_state",
"reduce_risk",
"snapshot_state",
"notify",
]
broker.cancel_domestic_order.assert_called_once()
broker.get_balance.assert_called_once()
telegram.notify_circuit_breaker.assert_called_once_with(
pnl_pct=-3.2,
threshold=-3.0,
)
@pytest.mark.asyncio
async def test_trigger_emergency_kill_switch_records_cancel_failure() -> None:
"""Cancel API rejection should be captured in kill switch errors."""
broker = MagicMock()
broker.get_domestic_pending_orders = AsyncMock(
return_value=[
{
"pdno": "005930",
"orgn_odno": "1",
"ord_gno_brno": "01",
"psbl_qty": "3",
}
]
)
broker.cancel_domestic_order = AsyncMock(return_value={"rt_cd": "1", "msg1": "fail"})
broker.get_balance = AsyncMock(return_value={"output1": [], "output2": []})
overseas_broker = MagicMock()
overseas_broker.get_overseas_pending_orders = AsyncMock(return_value=[])
overseas_broker.get_overseas_balance = AsyncMock(return_value={"output1": [], "output2": []})
telegram = MagicMock()
telegram.notify_circuit_breaker = AsyncMock()
settings = MagicMock()
settings.enabled_market_list = ["KR"]
market = MagicMock()
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
with (
patch("src.main.MARKETS", {"KR": market}),
patch("src.main.BLACKOUT_ORDER_MANAGER.clear", return_value=0),
):
report = await _trigger_emergency_kill_switch(
reason="test-fail",
broker=broker,
overseas_broker=overseas_broker,
telegram=telegram,
settings=settings,
current_market=market,
stock_code="005930",
pnl_pct=-3.2,
threshold=-3.0,
)
assert any(err.startswith("cancel_pending_orders:") for err in report.errors)

View File

@@ -0,0 +1,40 @@
from datetime import UTC, datetime
import pytest
from src.core.order_policy import OrderPolicyRejected, classify_session_id, validate_order_policy
from src.markets.schedule import MARKETS
def test_classify_kr_nxt_after() -> None:
# 2026-02-26 16:00 KST == 07:00 UTC
now = datetime(2026, 2, 26, 7, 0, tzinfo=UTC)
assert classify_session_id(MARKETS["KR"], now) == "NXT_AFTER"
def test_classify_us_pre() -> None:
# 2026-02-26 19:00 KST == 10:00 UTC
now = datetime(2026, 2, 26, 10, 0, tzinfo=UTC)
assert classify_session_id(MARKETS["US_NASDAQ"], now) == "US_PRE"
def test_reject_market_order_in_low_liquidity_session() -> None:
now = datetime(2026, 2, 26, 10, 0, tzinfo=UTC) # 19:00 KST -> US_PRE
with pytest.raises(OrderPolicyRejected):
validate_order_policy(
market=MARKETS["US_NASDAQ"],
order_type="BUY",
price=0.0,
now=now,
)
def test_allow_limit_order_in_low_liquidity_session() -> None:
now = datetime(2026, 2, 26, 10, 0, tzinfo=UTC) # 19:00 KST -> US_PRE
info = validate_order_policy(
market=MARKETS["US_NASDAQ"],
order_type="BUY",
price=100.0,
now=now,
)
assert info.session_id == "US_PRE"

View File

@@ -0,0 +1,38 @@
from src.strategy.exit_rules import ExitRuleConfig, ExitRuleInput, evaluate_exit
from src.strategy.position_state_machine import PositionState
def test_hard_stop_exit() -> None:
out = evaluate_exit(
current_state=PositionState.HOLDING,
config=ExitRuleConfig(hard_stop_pct=-2.0, arm_pct=3.0),
inp=ExitRuleInput(current_price=97.0, entry_price=100.0, peak_price=100.0),
)
assert out.should_exit is True
assert out.reason == "hard_stop"
def test_take_profit_exit_for_backward_compatibility() -> None:
out = evaluate_exit(
current_state=PositionState.HOLDING,
config=ExitRuleConfig(hard_stop_pct=-2.0, arm_pct=3.0),
inp=ExitRuleInput(current_price=104.0, entry_price=100.0, peak_price=104.0),
)
assert out.should_exit is True
assert out.reason == "arm_take_profit"
def test_model_assist_exit_signal() -> None:
out = evaluate_exit(
current_state=PositionState.ARMED,
config=ExitRuleConfig(model_prob_threshold=0.62, arm_pct=10.0),
inp=ExitRuleInput(
current_price=101.0,
entry_price=100.0,
peak_price=105.0,
pred_down_prob=0.8,
liquidity_weak=True,
),
)
assert out.should_exit is True
assert out.reason == "model_liquidity_exit"

View File

@@ -0,0 +1,30 @@
from src.strategy.position_state_machine import (
PositionState,
StateTransitionInput,
promote_state,
)
def test_gap_jump_promotes_to_armed_directly() -> None:
state = promote_state(
PositionState.HOLDING,
StateTransitionInput(
unrealized_pnl_pct=4.0,
be_arm_pct=1.2,
arm_pct=2.8,
),
)
assert state == PositionState.ARMED
def test_exited_has_priority_over_promotion() -> None:
state = promote_state(
PositionState.HOLDING,
StateTransitionInput(
unrealized_pnl_pct=5.0,
be_arm_pct=1.2,
arm_pct=2.8,
hard_stop_hit=True,
),
)
assert state == PositionState.EXITED

View File

@@ -0,0 +1,131 @@
from __future__ import annotations
from src.analysis.triple_barrier import TripleBarrierSpec, label_with_triple_barrier
def test_long_take_profit_first() -> None:
highs = [100, 101, 103]
lows = [100, 99.6, 100]
closes = [100, 100, 102]
spec = TripleBarrierSpec(take_profit_pct=0.02, stop_loss_pct=0.01, max_holding_bars=3)
out = label_with_triple_barrier(
highs=highs,
lows=lows,
closes=closes,
entry_index=0,
side=1,
spec=spec,
)
assert out.label == 1
assert out.touched == "take_profit"
assert out.touch_bar == 2
def test_long_stop_loss_first() -> None:
highs = [100, 100.5, 101]
lows = [100, 98.8, 99]
closes = [100, 99.5, 100]
spec = TripleBarrierSpec(take_profit_pct=0.02, stop_loss_pct=0.01, max_holding_bars=3)
out = label_with_triple_barrier(
highs=highs,
lows=lows,
closes=closes,
entry_index=0,
side=1,
spec=spec,
)
assert out.label == -1
assert out.touched == "stop_loss"
assert out.touch_bar == 1
def test_time_barrier_timeout() -> None:
highs = [100, 100.8, 100.7]
lows = [100, 99.3, 99.4]
closes = [100, 100, 100]
spec = TripleBarrierSpec(take_profit_pct=0.02, stop_loss_pct=0.02, max_holding_bars=2)
out = label_with_triple_barrier(
highs=highs,
lows=lows,
closes=closes,
entry_index=0,
side=1,
spec=spec,
)
assert out.label == 0
assert out.touched == "time"
assert out.touch_bar == 2
def test_tie_break_stop_first_default() -> None:
highs = [100, 102.1]
lows = [100, 98.9]
closes = [100, 100]
spec = TripleBarrierSpec(take_profit_pct=0.02, stop_loss_pct=0.01, max_holding_bars=1)
out = label_with_triple_barrier(
highs=highs,
lows=lows,
closes=closes,
entry_index=0,
side=1,
spec=spec,
)
assert out.label == -1
assert out.touched == "stop_loss"
def test_short_side_inverts_barrier_semantics() -> None:
highs = [100, 100.5, 101.2]
lows = [100, 97.8, 98.0]
closes = [100, 99, 99]
spec = TripleBarrierSpec(take_profit_pct=0.02, stop_loss_pct=0.01, max_holding_bars=3)
out = label_with_triple_barrier(
highs=highs,
lows=lows,
closes=closes,
entry_index=0,
side=-1,
spec=spec,
)
assert out.label == 1
assert out.touched == "take_profit"
def test_short_tie_break_modes() -> None:
highs = [100, 101.1]
lows = [100, 97.9]
closes = [100, 100]
stop_first = TripleBarrierSpec(
take_profit_pct=0.02,
stop_loss_pct=0.01,
max_holding_bars=1,
tie_break="stop_first",
)
out_stop = label_with_triple_barrier(
highs=highs,
lows=lows,
closes=closes,
entry_index=0,
side=-1,
spec=stop_first,
)
assert out_stop.label == -1
assert out_stop.touched == "stop_loss"
take_first = TripleBarrierSpec(
take_profit_pct=0.02,
stop_loss_pct=0.01,
max_holding_bars=1,
tie_break="take_first",
)
out_take = label_with_triple_barrier(
highs=highs,
lows=lows,
closes=closes,
entry_index=0,
side=-1,
spec=take_first,
)
assert out_take.label == 1
assert out_take.touched == "take_profit"

View File

@@ -0,0 +1,92 @@
from __future__ import annotations
import pytest
from src.analysis.walk_forward_split import generate_walk_forward_splits
def test_generates_sequential_folds() -> None:
folds = generate_walk_forward_splits(
n_samples=30,
train_size=10,
test_size=5,
)
assert len(folds) == 4
assert folds[0].train_indices == list(range(0, 10))
assert folds[0].test_indices == list(range(10, 15))
assert folds[1].train_indices == list(range(5, 15))
assert folds[1].test_indices == list(range(15, 20))
def test_purge_removes_boundary_samples_before_test() -> None:
folds = generate_walk_forward_splits(
n_samples=25,
train_size=8,
test_size=4,
purge_size=2,
)
first = folds[0]
# test starts at 10, purge=2 => train end must be 7
assert first.train_indices == list(range(0, 8))
assert first.test_indices == list(range(10, 14))
def test_embargo_excludes_post_test_samples_from_next_train() -> None:
folds = generate_walk_forward_splits(
n_samples=45,
train_size=15,
test_size=5,
step_size=10,
embargo_size=3,
)
assert len(folds) >= 2
# Fold1 test: 15..19, next fold train window: 10..24.
# embargo_size=3 should remove 20,21,22 from fold2 train.
second_train = folds[1].train_indices
assert 20 not in second_train
assert 21 not in second_train
assert 22 not in second_train
assert 23 in second_train
def test_respects_min_train_size_and_returns_empty_when_impossible() -> None:
folds = generate_walk_forward_splits(
n_samples=15,
train_size=5,
test_size=5,
min_train_size=6,
)
assert folds == []
def test_embargo_uses_last_accepted_fold_when_intermediate_fold_skips() -> None:
folds = generate_walk_forward_splits(
n_samples=30,
train_size=5,
test_size=3,
step_size=5,
embargo_size=1,
min_train_size=5,
)
# 1st fold accepted, 2nd skipped by min_train_size, subsequent folds still generated.
assert len(folds) == 3
assert folds[0].test_indices == [5, 6, 7]
assert folds[1].test_indices == [15, 16, 17]
assert folds[2].test_indices == [25, 26, 27]
@pytest.mark.parametrize(
("n_samples", "train_size", "test_size"),
[
(0, 10, 2),
(10, 0, 2),
(10, 5, 0),
],
)
def test_invalid_args_raise(n_samples: int, train_size: int, test_size: int) -> None:
with pytest.raises(ValueError):
generate_walk_forward_splits(
n_samples=n_samples,
train_size=train_size,
test_size=test_size,
)