auto-stock/backend/tests/test_swing_entry_hardening_r2_20261009.py

"""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()


from _swing_redis_guard import _swing_redis_guard  # noqa: E402,F401  (R9: 단독 실행 시 Redis 실접속 방지)