shuim
9fab4018d4
하네스 규칙 검사 전역 위반 117→0, 회차별 검증(code-verifier opus 4도메인)으로 Critical/High 해소. 주문·리스크 - MDD 강제축소 반복 매도(종목별 완료 집합·재시도 상한·날짜 롤오버·잔고 오류 보류·epoch·장외 미주문) - KIS 거부 FAILED 기록, 결과 불명(전송 후) 재전송 금지·PENDING, 미전송 실패만 재시도(msg_cd 분류) - 국내 전량·손실 매도 시장가, 부분 익절 직전 현재가, 지정가만 미체결 취소(cancel_order_v2: TR 0013U·조직번호) + 0체결 롤백 - 누적 4%·현금·레버리지(신규분 포함)·사이클 내 매수 누적, 수동매수 공용 가드(buy_guards), 수동매도 DT 차감·락 - 가드 우회 엔드포인트 차단(/recommendations/auto-trade 위임, US /execute 매수 410·매도 가드) - 매수 슬롯=성공 매수, 평가 상한·회전, 업무거부/계좌 쿨다운, 프로세스 내 in-flight 주문 중복 차단 데이트레이딩 - 진입·청산 shield, 청산 진행 중 표식(소유 토큰), 2단계 청산 체크, 결과 불명 → 보유 감소 판정·비DT 차감 상한 - 전일 포지션 재시도(일시 실패 미산정), 분할매도 영속, ORB 구간 손절, /cycle 락, 사이클 내 현금 차감, 체결 수량 보정 스윙 - 손절 공백(판정 예외 시 하드 바닥), 단일 손절 소스·범위 검증, 매도 직전 KIS 잔고 상한(min(주문가능,보유)−DT) - 결과 불명 체결 대조·장부 반영, 대체가(나이·진입 이후·ATR 캐시), 분할 단계 기록, 국면 실패 fail-closed - 듀얼 플레이북·서킷브레이커 게이트, 평균회귀 우선·듀얼 독립, 시간손절 거래일 - 거래 확대(사용자 결정, 백테스트 미검증): RSI 46/35 정본 경로, 모멘텀 v2 ON, SIDEWAYS ADX AND, 트레일링 1차 익절 후, 20일 시간손절 ON 인프라·보안 - 계좌·손익 GET 인증, 세션 is_active·관리자 경로, 서명키 파생, realip, WS 토큰 마스킹 - KIS 토큰 즉시 실패(sleep 제거)·무효 시 캐시 삭제, 주문 5xx+JSON 본문 반환 - db_worker 종료 drain(멱등 재실행·커밋 불명 격리·예산·밴 커밋), stop_grace 40s, Redis 타임아웃·volatile-lru, /health DLQ·캐시 - KRX 휴장일, py3.11 비호환 f-string 수정 + Jenkins 이미지 내 compileall·헬스 실패 시 빌드 실패 하네스: rules_check 에 PEP 701 f-string 검사, 마커(allow-*), deploy 스모크 401 기준 CLAUDE.md: 실제 동작·플래그·백로그(§8.5-27~38) 갱신 backend 1451 passed / hooks 112 passed / tsc ok / py3.9 compile 0 errors Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
629 lines
27 KiB
Python
629 lines
27 KiB
Python
"""
|
|
자동매매 실행 엔진 (trade_executor.py) 유닛 테스트
|
|
|
|
테스트 시나리오:
|
|
1. 세마포어 관리 — _cleanup_semaphores(), _get_order_semaphore()
|
|
2. Redis 분산락 — _acquire_order_lock(), _release_order_lock()
|
|
3. 주문 재시도 — _place_order_with_retry() (성공·실패·최대재시도 소진)
|
|
4. 신호 분류 — BUY/SELL/STRONG_BUY/BLOCK 키워드 매칭
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
import asyncio
|
|
import pytest
|
|
from unittest.mock import AsyncMock, Mock, patch, MagicMock
|
|
|
|
from app.services.signal_types import (
|
|
BuySignal,
|
|
SellSignal,
|
|
StrongBuySignal,
|
|
BlockSignal,
|
|
BUY_KEYWORDS,
|
|
SELL_KEYWORDS,
|
|
STRONG_BUY_KEYWORDS,
|
|
BLOCK_KEYWORDS,
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 1. 세마포어 관리 테스트
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestCleanupSemaphores:
|
|
"""_cleanup_semaphores() 테스트"""
|
|
|
|
def setup_method(self):
|
|
"""각 테스트 전 세마포어 딕셔너리 초기화"""
|
|
from app.services import trade_executor
|
|
|
|
trade_executor._order_semaphores.clear()
|
|
|
|
def test_no_cleanup_under_limit(self):
|
|
"""임계값 이하일 때 정리하지 않음"""
|
|
from app.services.trade_executor import _cleanup_semaphores, _order_semaphores, _SEMAPHORE_MAX_SIZE
|
|
|
|
# _SEMAPHORE_MAX_SIZE(200) 이하 → 아무 일도 안 함
|
|
for i in range(50):
|
|
_order_semaphores[f"CODE_{i:04d}"] = asyncio.Semaphore(1)
|
|
|
|
_cleanup_semaphores()
|
|
assert len(_order_semaphores) == 50
|
|
|
|
def test_cleanup_removes_unlocked_semaphores(self):
|
|
"""임계값 초과 시 잠기지 않은(미사용) 세마포어 제거"""
|
|
from app.services.trade_executor import _cleanup_semaphores, _order_semaphores, _SEMAPHORE_MAX_SIZE
|
|
|
|
# 201개 세마포어 생성 (모두 unlocked — _value == 1)
|
|
for i in range(_SEMAPHORE_MAX_SIZE + 1):
|
|
_order_semaphores[f"CODE_{i:04d}"] = asyncio.Semaphore(1)
|
|
|
|
_cleanup_semaphores()
|
|
# 전부 unlocked 이므로 전부 제거됨
|
|
assert len(_order_semaphores) == 0
|
|
|
|
def test_cleanup_preserves_locked_semaphores(self):
|
|
"""사용 중(locked)인 세마포어는 유지"""
|
|
from app.services.trade_executor import _cleanup_semaphores, _order_semaphores, _SEMAPHORE_MAX_SIZE
|
|
|
|
for i in range(_SEMAPHORE_MAX_SIZE + 1):
|
|
sem = asyncio.Semaphore(1)
|
|
_order_semaphores[f"CODE_{i:04d}"] = sem
|
|
|
|
# 3개의 세마포어를 locked 상태로 만듦 (_value == 0)
|
|
locked_codes = ["CODE_0010", "CODE_0050", "CODE_0100"]
|
|
for code in locked_codes:
|
|
_order_semaphores[code]._value = 0 # acquire 된 상태 시뮬레이션
|
|
|
|
_cleanup_semaphores()
|
|
# locked 3개만 남아야 함
|
|
assert len(_order_semaphores) == 3
|
|
for code in locked_codes:
|
|
assert code in _order_semaphores
|
|
|
|
def test_cleanup_at_exact_limit(self):
|
|
"""딱 임계값일 때 정리하지 않음"""
|
|
from app.services.trade_executor import _cleanup_semaphores, _order_semaphores, _SEMAPHORE_MAX_SIZE
|
|
|
|
for i in range(_SEMAPHORE_MAX_SIZE):
|
|
_order_semaphores[f"CODE_{i:04d}"] = asyncio.Semaphore(1)
|
|
|
|
_cleanup_semaphores()
|
|
assert len(_order_semaphores) == _SEMAPHORE_MAX_SIZE
|
|
|
|
|
|
class TestGetOrderSemaphore:
|
|
"""_get_order_semaphore() 테스트"""
|
|
|
|
def setup_method(self):
|
|
from app.services import trade_executor
|
|
|
|
trade_executor._order_semaphores.clear()
|
|
|
|
def test_creates_new_semaphore(self):
|
|
"""새로운 종목 코드에 대해 세마포어 생성"""
|
|
from app.services.trade_executor import _get_order_semaphore, _order_semaphores
|
|
|
|
sem = _get_order_semaphore("005930")
|
|
assert isinstance(sem, asyncio.Semaphore)
|
|
assert "005930" in _order_semaphores
|
|
|
|
def test_reuses_existing_semaphore(self):
|
|
"""동일 종목 코드에 대해 기존 세마포어 반환"""
|
|
from app.services.trade_executor import _get_order_semaphore
|
|
|
|
sem1 = _get_order_semaphore("005930")
|
|
sem2 = _get_order_semaphore("005930")
|
|
assert sem1 is sem2
|
|
|
|
def test_different_codes_get_different_semaphores(self):
|
|
"""서로 다른 종목 코드는 서로 다른 세마포어"""
|
|
from app.services.trade_executor import _get_order_semaphore
|
|
|
|
sem_samsung = _get_order_semaphore("005930")
|
|
sem_sk = _get_order_semaphore("000660")
|
|
assert sem_samsung is not sem_sk
|
|
|
|
def test_triggers_cleanup_when_over_limit(self):
|
|
"""한도 초과 시 _cleanup_semaphores 호출 확인"""
|
|
from app.services.trade_executor import _get_order_semaphore, _order_semaphores, _SEMAPHORE_MAX_SIZE
|
|
|
|
# 201개까지 채운 뒤 새 코드 요청 → cleanup 트리거
|
|
for i in range(_SEMAPHORE_MAX_SIZE + 1):
|
|
_order_semaphores[f"CODE_{i:04d}"] = asyncio.Semaphore(1)
|
|
|
|
with patch("app.services.trade_executor._cleanup_semaphores") as mock_cleanup:
|
|
_get_order_semaphore("NEW_CODE")
|
|
# 이미 _order_semaphores에 없으므로 cleanup 호출
|
|
mock_cleanup.assert_called_once()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 2. Redis 분산락 테스트
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestAcquireOrderLock:
|
|
"""_acquire_order_lock() 테스트"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lock_success(self):
|
|
"""Redis SET NX 성공 시 True 반환"""
|
|
mock_client = AsyncMock()
|
|
mock_client.set = AsyncMock(return_value=True)
|
|
|
|
with patch("app.core.redis.get_redis", return_value=mock_client):
|
|
from app.services.trade_executor import _acquire_order_lock
|
|
|
|
result = await _acquire_order_lock("005930", "buy")
|
|
assert result is True
|
|
mock_client.set.assert_called_once_with(
|
|
"order_lock:005930:buy", "1", nx=True, ex=60
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lock_failure_key_exists(self):
|
|
"""이미 락이 존재하면 False 반환"""
|
|
mock_client = AsyncMock()
|
|
mock_client.set = AsyncMock(return_value=None) # NX 실패 → None
|
|
|
|
with patch("app.core.redis.get_redis", return_value=mock_client):
|
|
from app.services.trade_executor import _acquire_order_lock
|
|
|
|
result = await _acquire_order_lock("005930", "buy")
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lock_redis_unavailable(self):
|
|
"""Redis 미연결 시 True 반환 (락 없이 진행)"""
|
|
with patch("app.core.redis.get_redis", return_value=None):
|
|
from app.services.trade_executor import _acquire_order_lock
|
|
|
|
result = await _acquire_order_lock("005930", "buy")
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lock_redis_exception(self):
|
|
"""Redis 연결 오류 시 True 반환 (세마포어가 대신 보호)"""
|
|
with patch(
|
|
"app.core.redis.get_redis",
|
|
side_effect=ConnectionError("Redis connection refused"),
|
|
):
|
|
from app.services.trade_executor import _acquire_order_lock
|
|
|
|
result = await _acquire_order_lock("005930", "sell")
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lock_key_format(self):
|
|
"""락 키 형식이 order_lock:{stock_code}:{order_type} 인지 확인"""
|
|
mock_client = AsyncMock()
|
|
mock_client.set = AsyncMock(return_value=True)
|
|
|
|
with patch("app.core.redis.get_redis", return_value=mock_client):
|
|
from app.services.trade_executor import _acquire_order_lock
|
|
|
|
await _acquire_order_lock("AAPL", "sell")
|
|
call_args = mock_client.set.call_args
|
|
assert call_args[0][0] == "order_lock:AAPL:sell"
|
|
|
|
|
|
class TestReleaseOrderLock:
|
|
"""_release_order_lock() 테스트"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_release_success(self):
|
|
"""정상적으로 락 해제"""
|
|
mock_delete = AsyncMock()
|
|
with patch("app.core.redis.delete_key", mock_delete):
|
|
from app.services.trade_executor import _release_order_lock
|
|
|
|
await _release_order_lock("005930", "buy")
|
|
mock_delete.assert_called_once_with("order_lock:005930:buy")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_release_exception_silenced(self):
|
|
"""락 해제 실패 시 예외가 무시됨 (TTL 만료로 자동 해제)"""
|
|
mock_delete = AsyncMock(side_effect=ConnectionError("Redis down"))
|
|
with patch("app.core.redis.delete_key", mock_delete):
|
|
from app.services.trade_executor import _release_order_lock
|
|
|
|
# 예외가 전파되지 않아야 함
|
|
await _release_order_lock("005930", "buy")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_release_key_format(self):
|
|
"""해제 키 형식 확인"""
|
|
mock_delete = AsyncMock()
|
|
with patch("app.core.redis.delete_key", mock_delete):
|
|
from app.services.trade_executor import _release_order_lock
|
|
|
|
await _release_order_lock("TSLA", "sell")
|
|
mock_delete.assert_called_once_with("order_lock:TSLA:sell")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 3. 주문 재시도 테스트
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestPlaceOrderWithRetry:
|
|
"""_place_order_with_retry() 테스트"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_domestic_success_first_try(self):
|
|
"""국내 주문 — 첫 번째 시도에 성공"""
|
|
mock_api = AsyncMock()
|
|
mock_api.place_order = AsyncMock(
|
|
return_value={"rt_cd": "0", "odno": "12345", "msg1": "정상 처리"}
|
|
)
|
|
|
|
with patch("app.services.trade_executor.korea_investment_api", mock_api):
|
|
from app.services.trade_executor import _place_order_with_retry
|
|
|
|
result = await _place_order_with_retry(
|
|
is_domestic=True,
|
|
stock_code="005930",
|
|
order_type="buy",
|
|
quantity=10,
|
|
price=72000.0,
|
|
)
|
|
assert result["rt_cd"] == "0"
|
|
assert result["odno"] == "12345"
|
|
mock_api.place_order.assert_called_once_with("005930", "buy", 10, 72000.0)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_overseas_success_first_try(self):
|
|
"""해외 주문 — 첫 번째 시도에 성공"""
|
|
mock_api = AsyncMock()
|
|
mock_api.place_overseas_order = AsyncMock(
|
|
return_value={"rt_cd": "0", "odno": "OVS001", "msg1": "success"}
|
|
)
|
|
|
|
with patch("app.services.trade_executor.korea_investment_api", mock_api):
|
|
from app.services.trade_executor import _place_order_with_retry
|
|
|
|
result = await _place_order_with_retry(
|
|
is_domestic=False,
|
|
stock_code="AAPL",
|
|
order_type="buy",
|
|
quantity=5,
|
|
price=185.50,
|
|
exchange_code="NASD",
|
|
)
|
|
assert result["rt_cd"] == "0"
|
|
mock_api.place_overseas_order.assert_called_once_with(
|
|
"AAPL", "buy", 5, 185.50, "NASD"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pending_status_no_retry(self):
|
|
"""rt_cd "1"(PENDING) — 재시도 없이 즉시 반환"""
|
|
mock_api = AsyncMock()
|
|
mock_api.place_order = AsyncMock(
|
|
return_value={"rt_cd": "1", "odno": "PND01", "msg1": "접수 처리 중"}
|
|
)
|
|
|
|
with patch("app.services.trade_executor.korea_investment_api", mock_api):
|
|
from app.services.trade_executor import _place_order_with_retry
|
|
|
|
result = await _place_order_with_retry(
|
|
is_domestic=True,
|
|
stock_code="005930",
|
|
order_type="buy",
|
|
quantity=10,
|
|
price=72000.0,
|
|
)
|
|
assert result["rt_cd"] == "1"
|
|
assert mock_api.place_order.call_count == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_on_error_then_success(self):
|
|
"""레이트리밋(msg_cd EGW00201) 후 재시도 → 두 번째 시도에 성공
|
|
|
|
[2026-10-02 H-1 사양 갱신] 재시도는 시스템 오류(rt_cd "-1")·레이트리밋만.
|
|
"""
|
|
mock_api = AsyncMock()
|
|
mock_api.place_order = AsyncMock(
|
|
side_effect=[
|
|
{"rt_cd": "1", "msg_cd": "EGW00201", "msg1": "초당 거래건수를 초과하였습니다."}, # 1차 실패
|
|
{"rt_cd": "0", "odno": "12345", "msg1": "정상"}, # 2차 성공
|
|
]
|
|
)
|
|
|
|
with patch("app.services.trade_executor.korea_investment_api", mock_api):
|
|
with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
|
from app.services.trade_executor import _place_order_with_retry
|
|
|
|
result = await _place_order_with_retry(
|
|
is_domestic=True,
|
|
stock_code="005930",
|
|
order_type="buy",
|
|
quantity=10,
|
|
price=72000.0,
|
|
max_retries=3,
|
|
)
|
|
assert result["rt_cd"] == "0"
|
|
assert mock_api.place_order.call_count == 2
|
|
# 첫 재시도 대기: 1초
|
|
mock_sleep.assert_called_once_with(1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_max_retries_exhausted(self):
|
|
"""[H-3 2026-10-02 5회차 사양 갱신] msg_cd 없는 rt_cd "-1" = 전송 후 결과 불명 →
|
|
재전송하지 않는다(이중 체결 방지). 주문내역에 없어도(조회 실패와 구분 불가) 1회로 끝."""
|
|
mock_api = AsyncMock()
|
|
mock_api.place_order = AsyncMock(
|
|
return_value={"rt_cd": "-1", "msg1": "시스템 오류", "odno": None}
|
|
)
|
|
mock_api.get_todays_orders = AsyncMock(return_value=[]) # 중복 접수 없음
|
|
|
|
with patch("app.services.trade_executor.korea_investment_api", mock_api):
|
|
with patch("asyncio.sleep", new_callable=AsyncMock):
|
|
from app.services.trade_executor import _place_order_with_retry
|
|
|
|
result = await _place_order_with_retry(
|
|
is_domestic=True,
|
|
stock_code="005930",
|
|
order_type="buy",
|
|
quantity=10,
|
|
price=72000.0,
|
|
max_retries=3,
|
|
)
|
|
assert result["rt_cd"] == "-1"
|
|
assert mock_api.place_order.call_count == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_presend_failure_is_retried(self):
|
|
"""[H-3 5회차] 전송 전 실패(KIS_PRESEND_CONNECT)는 재전송 안전 → 재시도 후 성공"""
|
|
mock_api = AsyncMock()
|
|
mock_api.place_order = AsyncMock(side_effect=[
|
|
{"rt_cd": "-1", "msg_cd": "KIS_PRESEND_CONNECT", "msg1": "connect error"},
|
|
{"rt_cd": "0", "odno": "1", "msg1": "정상"},
|
|
])
|
|
with patch("app.services.trade_executor.korea_investment_api", mock_api):
|
|
with patch("asyncio.sleep", new_callable=AsyncMock):
|
|
from app.services.trade_executor import _place_order_with_retry
|
|
result = await _place_order_with_retry(True, "005930", "buy", 10, 72000.0)
|
|
assert result["rt_cd"] == "0"
|
|
assert mock_api.place_order.call_count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_network_exception_retry(self):
|
|
"""[H-3 5회차 사양 갱신] place_order 호출 중 예외 = 전송 여부 불명 → 재전송 안 함"""
|
|
mock_api = AsyncMock()
|
|
mock_api.place_order = AsyncMock(
|
|
side_effect=[
|
|
ConnectionError("Connection refused"), # 1차 예외
|
|
{"rt_cd": "0", "odno": "12345", "msg1": "정상"}, # 2차 성공
|
|
]
|
|
)
|
|
|
|
with patch("app.services.trade_executor.korea_investment_api", mock_api):
|
|
with patch("asyncio.sleep", new_callable=AsyncMock):
|
|
from app.services.trade_executor import _place_order_with_retry
|
|
|
|
result = await _place_order_with_retry(
|
|
is_domestic=True,
|
|
stock_code="005930",
|
|
order_type="buy",
|
|
quantity=10,
|
|
price=72000.0,
|
|
max_retries=3,
|
|
)
|
|
assert result["rt_cd"] == "-1"
|
|
assert result["msg_cd"] == "KIS_TRANSPORT_UNKNOWN"
|
|
assert mock_api.place_order.call_count == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_all_retries_exception(self):
|
|
"""모든 재시도에서 예외 발생 시 오류 응답 반환"""
|
|
mock_api = AsyncMock()
|
|
mock_api.place_order = AsyncMock(
|
|
side_effect=TimeoutError("Request timed out")
|
|
)
|
|
|
|
with patch("app.services.trade_executor.korea_investment_api", mock_api):
|
|
with patch("asyncio.sleep", new_callable=AsyncMock):
|
|
from app.services.trade_executor import _place_order_with_retry
|
|
|
|
result = await _place_order_with_retry(
|
|
is_domestic=True,
|
|
stock_code="005930",
|
|
order_type="buy",
|
|
quantity=10,
|
|
price=72000.0,
|
|
max_retries=3,
|
|
)
|
|
assert result["rt_cd"] == "-1"
|
|
assert "timed out" in result["msg1"]
|
|
assert result["odno"] is None
|
|
assert mock_api.place_order.call_count == 1 # [H-3 5회차] 결과 불명 → 재전송 금지
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exponential_backoff_timing(self):
|
|
"""재시도 대기 시간이 exponential backoff (1→2→4초)를 따르는지 확인"""
|
|
mock_api = AsyncMock()
|
|
mock_api.place_order = AsyncMock(
|
|
return_value={"rt_cd": "1", "msg_cd": "EGW00201", "msg1": "레이트리밋"}
|
|
)
|
|
|
|
with patch("app.services.trade_executor.korea_investment_api", mock_api):
|
|
with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
|
from app.services.trade_executor import _place_order_with_retry
|
|
|
|
await _place_order_with_retry(
|
|
is_domestic=True,
|
|
stock_code="005930",
|
|
order_type="buy",
|
|
quantity=10,
|
|
price=72000.0,
|
|
max_retries=4,
|
|
)
|
|
# 4번 시도 → sleep 3번 (1차 실패 후 1초, 2차 실패 후 2초, 3차 실패 후 4초)
|
|
sleep_calls = [call.args[0] for call in mock_sleep.call_args_list]
|
|
assert sleep_calls == [1, 2, 4]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_single_retry_max(self):
|
|
"""max_retries=1 이면 재시도 없이 한 번만 시도"""
|
|
mock_api = AsyncMock()
|
|
mock_api.place_order = AsyncMock(
|
|
return_value={"rt_cd": "99", "msg1": "오류"}
|
|
)
|
|
|
|
with patch("app.services.trade_executor.korea_investment_api", mock_api):
|
|
with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
|
from app.services.trade_executor import _place_order_with_retry
|
|
|
|
result = await _place_order_with_retry(
|
|
is_domestic=True,
|
|
stock_code="005930",
|
|
order_type="buy",
|
|
quantity=1,
|
|
price=72000.0,
|
|
max_retries=1,
|
|
)
|
|
assert result["rt_cd"] == "99"
|
|
assert mock_api.place_order.call_count == 1
|
|
mock_sleep.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_partial_fill_result(self):
|
|
"""부분체결 응답 (rt_cd "0") — 재시도 없이 반환"""
|
|
mock_api = AsyncMock()
|
|
mock_api.place_order = AsyncMock(
|
|
return_value={
|
|
"rt_cd": "0",
|
|
"odno": "PF001",
|
|
"msg1": "부분 체결",
|
|
"filled_qty": 5,
|
|
"total_qty": 10,
|
|
}
|
|
)
|
|
|
|
with patch("app.services.trade_executor.korea_investment_api", mock_api):
|
|
from app.services.trade_executor import _place_order_with_retry
|
|
|
|
result = await _place_order_with_retry(
|
|
is_domestic=True,
|
|
stock_code="005930",
|
|
order_type="buy",
|
|
quantity=10,
|
|
price=72000.0,
|
|
)
|
|
assert result["rt_cd"] == "0"
|
|
assert result["filled_qty"] == 5
|
|
assert mock_api.place_order.call_count == 1
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 4. 신호 분류 (signal_types.py) 테스트
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestSignalKeywords:
|
|
"""BUY_KEYWORDS, SELL_KEYWORDS, STRONG_BUY_KEYWORDS, BLOCK_KEYWORDS 매칭 테스트"""
|
|
|
|
def test_buy_keywords_contain_all_buy_signals(self):
|
|
"""BUY_KEYWORDS가 모든 BuySignal enum 값을 포함하는지 확인"""
|
|
for signal in BuySignal:
|
|
assert signal.value in BUY_KEYWORDS
|
|
|
|
def test_sell_keywords_contain_all_sell_signals(self):
|
|
"""SELL_KEYWORDS가 모든 SellSignal enum 값을 포함하는지 확인"""
|
|
for signal in SellSignal:
|
|
assert signal.value in SELL_KEYWORDS
|
|
|
|
def test_strong_buy_keywords_contain_all_strong_buy_signals(self):
|
|
"""STRONG_BUY_KEYWORDS가 모든 StrongBuySignal enum 값을 포함하는지 확인"""
|
|
for signal in StrongBuySignal:
|
|
assert signal.value in STRONG_BUY_KEYWORDS
|
|
|
|
def test_block_keywords_contain_all_block_signals(self):
|
|
"""BLOCK_KEYWORDS가 모든 BlockSignal enum 값을 포함하는지 확인"""
|
|
for signal in BlockSignal:
|
|
assert signal.value in BLOCK_KEYWORDS
|
|
|
|
def test_strong_buy_is_subset_of_buy(self):
|
|
"""STRONG_BUY는 BUY의 부분집합이어야 함"""
|
|
for kw in STRONG_BUY_KEYWORDS:
|
|
assert kw in BUY_KEYWORDS, f"강한 매수 신호 '{kw}'가 매수 키워드에 없음"
|
|
|
|
def test_buy_sell_no_overlap(self):
|
|
"""매수·매도 키워드 간 중복이 없어야 함"""
|
|
overlap = set(BUY_KEYWORDS) & set(SELL_KEYWORDS)
|
|
assert len(overlap) == 0, f"매수·매도 키워드 중복: {overlap}"
|
|
|
|
def test_block_buy_no_overlap(self):
|
|
"""차단·매수 키워드 간 중복이 없어야 함"""
|
|
overlap = set(BLOCK_KEYWORDS) & set(BUY_KEYWORDS)
|
|
assert len(overlap) == 0, f"차단·매수 키워드 중복: {overlap}"
|
|
|
|
def test_buy_keywords_count(self):
|
|
"""BUY_KEYWORDS 개수가 BuySignal enum 멤버 수와 일치"""
|
|
assert len(BUY_KEYWORDS) == len(BuySignal)
|
|
|
|
def test_sell_keywords_count(self):
|
|
"""SELL_KEYWORDS 개수가 SellSignal enum 멤버 수와 일치"""
|
|
assert len(SELL_KEYWORDS) == len(SellSignal)
|
|
|
|
def test_specific_buy_keyword_matching(self):
|
|
"""특정 매수 키워드 매칭 테스트"""
|
|
test_description = "RSI 바닥 탈출 반등"
|
|
assert test_description in BUY_KEYWORDS
|
|
|
|
def test_specific_sell_keyword_matching(self):
|
|
"""특정 매도 키워드 매칭 테스트"""
|
|
test_description = "RSI 과매수"
|
|
assert test_description in SELL_KEYWORDS
|
|
|
|
def test_specific_block_keyword_matching(self):
|
|
"""특정 차단 키워드 매칭 테스트"""
|
|
test_descriptions = ["이격도 과열", "매수금지", "급등 종목 추격 금지", "급등과열"]
|
|
for desc in test_descriptions:
|
|
assert desc in BLOCK_KEYWORDS, f"'{desc}'가 BLOCK_KEYWORDS에 없음"
|
|
|
|
def test_specific_strong_buy_keyword_matching(self):
|
|
"""특정 강한 매수 키워드 매칭 테스트"""
|
|
strong_signals = ["RSI 바닥 탈출 반등", "볼린저 하한선 터치", "눌림목 진입 매수타점", "정배열 눌림목 매수"]
|
|
for desc in strong_signals:
|
|
assert desc in STRONG_BUY_KEYWORDS, f"'{desc}'가 STRONG_BUY_KEYWORDS에 없음"
|
|
|
|
def test_unknown_signal_not_in_keywords(self):
|
|
"""알 수 없는 신호 문자열은 어떤 키워드 목록에도 포함되지 않음"""
|
|
unknown = "존재하지 않는 신호"
|
|
assert unknown not in BUY_KEYWORDS
|
|
assert unknown not in SELL_KEYWORDS
|
|
assert unknown not in STRONG_BUY_KEYWORDS
|
|
assert unknown not in BLOCK_KEYWORDS
|
|
|
|
def test_keyword_signal_classification(self):
|
|
"""신호 문자열로 매수/매도/차단 분류가 올바르게 동작하는지 확인"""
|
|
test_cases = [
|
|
("정배열 눌림목 매수", "buy"),
|
|
("RSI 과매수", "sell"),
|
|
("이격도 과열", "block"),
|
|
("골든크로스", "buy"),
|
|
("데드크로스", "sell"),
|
|
("매수금지", "block"),
|
|
]
|
|
for signal, expected_type in test_cases:
|
|
if expected_type == "buy":
|
|
assert signal in BUY_KEYWORDS, f"'{signal}'이 BUY_KEYWORDS에 없음"
|
|
elif expected_type == "sell":
|
|
assert signal in SELL_KEYWORDS, f"'{signal}'이 SELL_KEYWORDS에 없음"
|
|
elif expected_type == "block":
|
|
assert signal in BLOCK_KEYWORDS, f"'{signal}'이 BLOCK_KEYWORDS에 없음"
|
|
|
|
def test_enum_string_comparison(self):
|
|
"""Enum이 str 서브클래스이므로 직접 문자열 비교 가능"""
|
|
assert BuySignal.골든크로스 == "골든크로스"
|
|
assert SellSignal.데드크로스 == "데드크로스"
|
|
assert BlockSignal.매수금지 == "매수금지"
|
|
assert StrongBuySignal.RSI바닥탈출 == "RSI 바닥 탈출 반등"
|