auto-stock/backend/tests/test_swing_safety_20261002.py

"""스윙 매수 안전장치 동작 회귀 테스트 (2026-10-02 3회차).

test_position_cap_hard_ceiling.py 가 "동작 검증은 test_swing_safety_20261002.py" 라고
참조하던 파일이다(이전엔 존재하지 않았다). 소스 문자열이 아니라 `_execute_buy` 를
실제로 호출해 다음을 확인한다.

1. 적응형 position_size 6~7%(Normal/Aggressive)는 MAX_POSITION_SIZE(4%)로 클램프된다.
2. 4% 이하 모드(Crisis 1.5% 등, 단 진입 허용 시)는 적응형 값이 그대로 쓰인다.
3. Crisis(max_daily_trades=0) → 주문 관문 도달 전 SKIP.
4. 적응형 조회 예외 / None → fail-closed SKIP.

규약(CLAUDE.md §4.1): 주문 경로 목은 예외 발생형(_Reached, BaseException),
공통 종단(_acquire_order_lock)에 건다. 기대값은 settings.* 참조.
"""
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest

from app.core.config import settings
from app.services import trade_executor


class _Reached(BaseException):
    """가드를 통과해 사이징 지점에 도달했다는 신호 (except Exception 에 삼켜지지 않음)."""


class _Result:
    def __init__(self, rows=None):
        self._rows = list(rows or [])

    def scalars(self):
        return self

    def first(self):
        return self._rows[0] if self._rows else None

    def all(self):
        return list(self._rows)


class _FakeDB:
    async def execute(self, stmt):
        return _Result([])

    async def commit(self):
        raise AssertionError("trade_executor 가 db.commit() 을 직접 호출했다 (§3.2 위반)")


TOTAL = 100_000_000.0


@pytest.fixture
def _pipeline(monkeypatch):
    """_execute_buy 의 사이징 직전까지 통과시키고, Kelly 호출 인자를 기록한 뒤 중단한다."""
    monkeypatch.setattr(trade_executor, "check_opening_range_volatility", AsyncMock(return_value=False))
    monkeypatch.setattr(trade_executor, "check_surge_prevention", AsyncMock(return_value={}))
    monkeypatch.setattr(trade_executor, "check_trend_filter", AsyncMock(return_value={}))
    monkeypatch.setattr(trade_executor, "_get_cached_account", lambda: (TOTAL, TOTAL))
    monkeypatch.setattr(trade_executor, "_get_cached_account_v2", lambda: (TOTAL, TOTAL))
    monkeypatch.setattr(trade_executor, "validate_correlation_filter", AsyncMock(return_value=(True, "ok")))
    monkeypatch.setattr(trade_executor, "calculate_cooldown_minutes", lambda *a, **k: 0)
    gate = AsyncMock(side_effect=_Reached("order gate reached"))
    monkeypatch.setattr(trade_executor, "_acquire_order_lock", gate)
    from app.services.daytrading.position_manager import day_trading_state
    monkeypatch.setattr(day_trading_state, "positions", {})

    seen = {}

    async def _kelly(db, stock_code, total_assets, available_cash, max_per_position=None, **kw):
        seen["max_per_position"] = max_per_position
        raise _Reached("sizing reached")

    monkeypatch.setattr(trade_executor, "_calculate_kelly_position", _kelly)
    return SimpleNamespace(seen=seen, gate=gate)


def _set_adaptive(monkeypatch, factory):
    import app.services.adaptive_strategy as _ad
    monkeypatch.setattr(_ad, "get_current_adaptive_params", factory)


async def _call_buy():
    return await trade_executor._execute_buy(
        stock={"_regime_str": "bull"}, db=_FakeDB(), holding=None,
        current_price=10_000.0, signals=["x"], score_val=5.0, is_domestic=True,
        exchange_code="KRX", stock_code="005930", stock_name="테스트",
    )


@pytest.mark.asyncio
@pytest.mark.parametrize("mode,size", [("normal", 0.06), ("aggressive", 0.07)])
async def test_adaptive_position_size_is_clamped_to_hard_ceiling(monkeypatch, _pipeline, mode, size):
    _set_adaptive(monkeypatch, lambda: SimpleNamespace(
        fallback_used=False, trading_mode=mode, max_daily_trades=10, position_size=size,
    ))
    with pytest.raises(_Reached):
        await _call_buy()
    assert _pipeline.seen["max_per_position"] == pytest.approx(settings.MAX_POSITION_SIZE)
    assert _pipeline.seen["max_per_position"] < size


@pytest.mark.asyncio
async def test_adaptive_position_size_below_ceiling_is_kept(monkeypatch, _pipeline):
    small = settings.MAX_POSITION_SIZE / 2
    _set_adaptive(monkeypatch, lambda: SimpleNamespace(
        fallback_used=False, trading_mode="defensive", max_daily_trades=5, position_size=small,
    ))
    with pytest.raises(_Reached):
        await _call_buy()
    assert _pipeline.seen["max_per_position"] == pytest.approx(small)


@pytest.mark.asyncio
async def test_adaptive_position_size_garbage_falls_back_to_ceiling(monkeypatch, _pipeline):
    _set_adaptive(monkeypatch, lambda: SimpleNamespace(
        fallback_used=False, trading_mode="normal", max_daily_trades=5, position_size="abc",
    ))
    with pytest.raises(_Reached):
        await _call_buy()
    assert _pipeline.seen["max_per_position"] == pytest.approx(settings.MAX_POSITION_SIZE)


@pytest.mark.asyncio
async def test_crisis_mode_blocks_before_sizing(monkeypatch, _pipeline):
    _set_adaptive(monkeypatch, lambda: SimpleNamespace(
        fallback_used=False, trading_mode="crisis", max_daily_trades=0, position_size=0.015,
    ))
    res = await _call_buy()
    assert res["action"] == "SKIP" and res["reason"] == "adaptive_crisis_mode"
    assert "max_per_position" not in _pipeline.seen
    _pipeline.gate.assert_not_called()


@pytest.mark.asyncio
async def test_adaptive_lookup_failure_is_fail_closed(monkeypatch, _pipeline):
    def _boom():
        raise RuntimeError("adaptive down")
    _set_adaptive(monkeypatch, _boom)
    res = await _call_buy()
    assert res["action"] == "SKIP" and res["reason"] == "adaptive_params_unavailable"
    _pipeline.gate.assert_not_called()


@pytest.mark.asyncio
async def test_adaptive_none_is_fail_closed(monkeypatch, _pipeline):
    _set_adaptive(monkeypatch, lambda: None)
    res = await _call_buy()
    assert res["action"] == "SKIP" and res["reason"] == "adaptive_params_unavailable"


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