"""swing/entry_service 하드닝 루프 라운드 2 회귀 테스트 (2026-10-09). H KIS get_daily_chart(FHKST01010400)는 30봉 상한 → 눌림목(60봉)·박스권(40봉)이 구조적 미발화. 듀얼·박스권 스캔은 네이버 → KIS 기간 조회(FHKST03010100) → 기존 경로 순으로 일봉을 확보한다. L 국내 가격 필터 경계 통일: MIN_PRICE_KRW '이하' 거부(§5.1 '>5,000원'). 규약(CLAUDE.md §4.1): 도달하면 안 되는 경로의 목은 예외 발생형(_Reached, BaseException), 기대값은 settings.* 참조, 네이버(외부 HTTP)는 conftest 차단 대상이 아니므로 전부 목킹. """ from datetime import timedelta from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest import app.services.mean_reversion_strategy as mr import app.services.naver_finance as nf from app.core.config import settings from app.core.timeutil import today_kst class _Reached(BaseException): """도달하면 안 되는(또는 도달 확인용) 관문 — 프로덕션 except Exception 에 삼켜지지 않는다.""" def _bars(n: int, end_today: bool = True, start_close: float = 10000.0) -> list[dict]: """오래된 → 최신 순 KIS 필드명 일봉 n개. 마지막 봉이 오늘(KST) 또는 어제.""" last = today_kst() if end_today else today_kst() - timedelta(days=1) out = [] for i in range(n): d = last - timedelta(days=n - 1 - i) c = start_close + i * 10 out.append({ "stck_bsop_date": d.strftime("%Y%m%d"), "stck_oprc": str(c - 5), "stck_hgpr": str(c + 10), "stck_lwpr": str(c - 10), "stck_clpr": str(c), "acml_vol": "100000", }) return out def _api(legacy_rows=None, range_rows=None, legacy_raises=False): api = MagicMock() api._quote_delegate = lambda: None if legacy_raises: api.get_daily_chart = AsyncMock(side_effect=_Reached("legacy get_daily_chart")) else: api.get_daily_chart = AsyncMock(return_value={"output2": legacy_rows or []}) api.get_daily_ohlcv_range = AsyncMock(return_value=range_rows or []) return api # ═══════════════════════════════════════════════════════════════════════════ # H — _fetch_scan_daily_bars_v2 # ═══════════════════════════════════════════════════════════════════════════ def test_kis_daily_price_30_rows_cannot_fire_pullback(): """[결함 재현] 30봉으로는 눌림목 v2 가 길이 검사에서 무조건 None.""" closes = [float(b["stck_clpr"]) for b in _bars(30)] assert mr.check_pullback_buy_signal_v2( closes=closes, current_close=closes[-1], current_volume=10**9, rsi=50.0, rsi_prev=40.0, volumes=[10**5] * 30, exclude_today_volume=True, ) is None @pytest.mark.asyncio async def test_fetch_prefers_naver_when_enough_bars(monkeypatch): from app.services.swing import entry_service as es nv = _bars(100) monkeypatch.setattr(nf, "get_naver_daily_chart", AsyncMock(return_value=nv)) api = _api(legacy_raises=True) api.get_daily_ohlcv_range = AsyncMock(side_effect=_Reached("kis range")) bars, src = await es._fetch_scan_daily_bars_v2(api, "005930", es.DUAL_SCAN_MIN_BARS) assert src == "naver" assert len(bars) >= es.DUAL_SCAN_MIN_BARS assert bars[-1]["stck_bsop_date"] == today_kst().strftime("%Y%m%d") @pytest.mark.asyncio async def test_fetch_falls_back_to_kis_range_sorted_oldest_first(monkeypatch): from app.services.swing import entry_service as es monkeypatch.setattr(nf, "get_naver_daily_chart", AsyncMock(return_value=None)) newest_first = list(reversed(_bars(90))) api = _api(range_rows=newest_first, legacy_raises=True) bars, src = await es._fetch_scan_daily_bars_v2(api, "005930", es.DUAL_SCAN_MIN_BARS) assert src == "kis_range" assert len(bars) == 90 dates = [b["stck_bsop_date"] for b in bars] assert dates == sorted(dates) assert dates[-1] == today_kst().strftime("%Y%m%d") @pytest.mark.asyncio async def test_fetch_rejects_source_without_today_bar(monkeypatch): """오늘 봉이 빠진 소스는 쓰지 않는다 — 어제 종가로 신호·주문가가 정해지는 것을 막는다.""" from app.services.swing import entry_service as es monkeypatch.setattr(nf, "get_naver_daily_chart", AsyncMock(return_value=_bars(100, end_today=False))) api = _api(range_rows=list(reversed(_bars(90))), legacy_raises=True) bars, src = await es._fetch_scan_daily_bars_v2(api, "005930", es.DUAL_SCAN_MIN_BARS) assert src == "kis_range" assert bars[-1]["stck_bsop_date"] == today_kst().strftime("%Y%m%d") @pytest.mark.asyncio async def test_fetch_uses_legacy_when_both_sources_fail(monkeypatch): from app.services.swing import entry_service as es monkeypatch.setattr(nf, "get_naver_daily_chart", AsyncMock(side_effect=RuntimeError("naver down"))) legacy_newest_first = list(reversed(_bars(30))) api = _api(legacy_rows=legacy_newest_first) api.get_daily_ohlcv_range = AsyncMock(side_effect=RuntimeError("kis down")) bars, src = await es._fetch_scan_daily_bars_v2(api, "005930", es.DUAL_SCAN_MIN_BARS) assert src == "kis_daily_price" assert len(bars) == 30 assert [b["stck_bsop_date"] for b in bars] == sorted(b["stck_bsop_date"] for b in bars) @pytest.mark.asyncio async def test_fetch_short_but_today_prefers_longer_source_over_legacy(monkeypatch): """신규 상장 등으로 두 소스 모두 min_bars 미달이면 긴 쪽 — 기존 30봉 경로로 내려가지 않는다.""" from app.services.swing import entry_service as es monkeypatch.setattr(nf, "get_naver_daily_chart", AsyncMock(return_value=_bars(45))) api = _api(range_rows=list(reversed(_bars(50))), legacy_raises=True) bars, src = await es._fetch_scan_daily_bars_v2(api, "005930", es.DUAL_SCAN_MIN_BARS) assert (src, len(bars)) == ("kis_range", 50) # ═══════════════════════════════════════════════════════════════════════════ # 듀얼 스캔 통합 — 실제 scan_dual_strategy_signals 호출 # ═══════════════════════════════════════════════════════════════════════════ class _DualDB: def __init__(self, candidates): self._candidates = candidates async def execute(self, stmt): r = MagicMock() r.fetchall.return_value = [] r.scalars.return_value.all.return_value = list(self._candidates) return r @pytest.fixture def dual_env(monkeypatch): import app.core.market_schedule as ms import app.services.korea_investment_api as kia import app.services.swing.entry_service as es import app.services.trade_executor as te import app.services.adaptive_strategy as ad monkeypatch.setattr(settings, "SWING_INDEX_BREAKER_ENABLED", False) monkeypatch.setattr(settings, "MOMENTUM_BREAKOUT_ENABLED", True) monkeypatch.setattr(ms, "is_market_open", lambda: True) monkeypatch.setattr(es, "is_entry_allowed_by_playbook", lambda: True) monkeypatch.setattr(es, "check_market_regime", AsyncMock(return_value={ "regime_200d": "bull", "drop_from_ma20": 1.0, "max_trades": 5, "position_multiplier": 1.0})) monkeypatch.setattr(es, "count_swing_positions", AsyncMock(return_value=0)) monkeypatch.setattr(es.mdd_manager, "can_open_new_position", AsyncMock(return_value=True)) monkeypatch.setattr(ad, "get_current_adaptive_params", lambda: SimpleNamespace(fallback_used=False, trading_mode="normal")) monkeypatch.setattr(te, "_count_swing_buys_today_v2", AsyncMock(return_value=0)) monkeypatch.setattr(es.asyncio, "sleep", AsyncMock(return_value=None)) api = _api(legacy_raises=True) monkeypatch.setattr(kia, "korea_investment_api", api) monkeypatch.setattr(nf, "get_naver_daily_chart", AsyncMock(return_value=_bars(100))) monkeypatch.setattr(mr, "check_momentum_breakout_signal_v2", lambda **kw: None) seen = {} def _pullback_spy(**kw): seen["n"] = len(kw["closes"]) return {"signal": "BUY", "strategy": "PULLBACK_BUY", "confidence": 0.6, "description": "t"} monkeypatch.setattr(mr, "check_pullback_buy_signal_v2", _pullback_spy) delegate = AsyncMock(side_effect=_Reached("delegated")) monkeypatch.setattr(es, "execute_auto_trade_background", delegate) stock = SimpleNamespace(stock_code="005930", stock_name="T", current_price=70000, score=5) return SimpleNamespace(es=es, api=api, delegate=delegate, seen=seen, stock=stock) @pytest.mark.asyncio async def test_dual_scan_feeds_pullback_at_least_min_bars(dual_env): """[결함 반전] 이전엔 get_daily_chart 30봉만 전달돼 눌림목이 매번 길이 검사에서 탈락했다.""" with pytest.raises(_Reached): await dual_env.es.scan_dual_strategy_signals(_DualDB([dual_env.stock])) assert dual_env.seen["n"] >= dual_env.es.DUAL_SCAN_MIN_BARS dual_env.api.get_daily_chart.assert_not_called() rec = dual_env.delegate.call_args.args[0] assert rec["recommendations_by_rule"][0]["stocks"][0]["strategy_name"] == "PULLBACK_BUY" @pytest.mark.asyncio async def test_dual_scan_warns_when_bars_short(monkeypatch, dual_env): """60봉 미달은 warning 으로 가시화(조용한 미발화 금지).""" monkeypatch.setattr(nf, "get_naver_daily_chart", AsyncMock(return_value=None)) dual_env.api.get_daily_chart = AsyncMock(return_value={"output2": list(reversed(_bars(30)))}) monkeypatch.setattr(mr, "check_pullback_buy_signal_v2", lambda **kw: None) calls = [] monkeypatch.setattr(dual_env.es, "warn_throttled", lambda lg, key, msg, *a, **k: calls.append((key, msg))) result = await dual_env.es.scan_dual_strategy_signals(_DualDB([dual_env.stock])) assert result == [] short = [m for k, m in calls if k.startswith("entry_service:short_bars:DualStrategy")] assert short and f"{dual_env.es.DUAL_SCAN_MIN_BARS}봉 미달" in short[0] assert "005930:30봉" in short[0] # ═══════════════════════════════════════════════════════════════════════════ # L — 가격 필터 경계 통일 # ═══════════════════════════════════════════════════════════════════════════ @pytest.mark.parametrize("price,blocked", [ (lambda: settings.MIN_PRICE_KRW, True), # 정확히 경계 → 거부('초과'만 허용) (lambda: settings.MIN_PRICE_KRW - 1, True), (lambda: settings.MIN_PRICE_KRW + 1, False), (lambda: None, True), (lambda: "abc", True), (lambda: float("nan"), True), (lambda: float("inf"), True), ]) def test_below_min_price_krw_v2_boundary(price, blocked): from app.services.swing.entry_service import _below_min_price_krw_v2 assert _below_min_price_krw_v2(price()) is blocked @pytest.mark.asyncio async def test_dual_scan_rejects_exact_min_price_before_chart(monkeypatch, dual_env): """[결함 반전] 듀얼 후보 단계가 `<` 라 정확히 MIN_PRICE_KRW 종목이 통과해 차트 조회·평가 슬롯을 썼다.""" stock = SimpleNamespace(stock_code="000001", stock_name="P", current_price=settings.MIN_PRICE_KRW, score=5) monkeypatch.setattr(nf, "get_naver_daily_chart", AsyncMock(side_effect=_Reached("chart fetched"))) result = await dual_env.es.scan_dual_strategy_signals(_DualDB([stock])) assert result == [] dual_env.delegate.assert_not_called() @pytest.mark.asyncio async def test_dual_scan_rejects_when_chart_close_at_min_price(monkeypatch, dual_env): """후보 가격은 통과해도 차트 현재 종가가 MIN_PRICE_KRW 이하이면 진입 위임하지 않는다.""" bars = _bars(100, start_close=float(settings.MIN_PRICE_KRW) - 99 * 10) assert float(bars[-1]["stck_clpr"]) == float(settings.MIN_PRICE_KRW) monkeypatch.setattr(nf, "get_naver_daily_chart", AsyncMock(return_value=bars)) result = await dual_env.es.scan_dual_strategy_signals(_DualDB([dual_env.stock])) assert result == [] dual_env.delegate.assert_not_called()