auto-stock/backend/tests/test_swing_partial_exit_no_active_trade.py

"""ActiveTrade 행이 없는 스윙 포지션의 부분매도 중복 실행 회귀 테스트.

2026-09-22 두산에너빌리티(034020): 동기화로 유입된 포지션(ActiveTrade 없음)에
"1차 익절(50%)"이 2분 간격으로 두 번 실행돼 전량 청산됨.
원인: exit_service가 remaining_pct 갱신 upsert에 quantity/entry_price를 넣지 않아
db_worker의 신규 INSERT가 NOT NULL 위반 → remaining_pct 미저장 + Portfolio.quantity 미감소.
"""
import asyncio
from decimal import Decimal
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy import select

from app.models.models import Base, Portfolio, ActiveTrade
from app.services.mean_reversion_strategy import ExitSignal
from app.services import db_worker
from app.services.swing import exit_service


class _DBBlocked(RuntimeError):
    """테스트 중 운영 DB 세션 생성 시도. 연결 실패에 우연히 의존하지 않도록 즉시 예외."""


def _blocked_session_factory(*_a, **_k):
    raise _DBBlocked("테스트에서 app.core.database.AsyncSessionLocal 직접 사용 차단")


@pytest.fixture(autouse=True)
def _block_real_db_session(monkeypatch):
    """portfolio_cache_bridge._db_quantity 는 함수 내부에서 app.core.database.AsyncSessionLocal 을
    import 하므로 127.0.0.1:5432 실연결을 시도한다. 호출 즉시 예외로 치환한다
    (_db_quantity 는 Exception 을 삼켜 None 반환 = 기존 기대 동작과 동일, 실소켓은 없음)."""
    import app.core.database as database
    monkeypatch.setattr(database, "AsyncSessionLocal", _blocked_session_factory)


def _daily_chart(n=30, close=86200.0):
    return {"output2": [
        {"stck_clpr": str(close), "stck_hgpr": str(close * 1.01), "stck_lwpr": str(close * 0.99)}
        for _ in range(n)
    ]}


async def _drain(queue, session_factory):
    """db_worker 로직으로 큐를 처리 (실제 워커 코드 경로 사용)."""
    with patch.object(db_worker, "AsyncSessionLocal", session_factory), \
         patch.object(db_worker, "get_db_task_queue", return_value=queue):
        worker = asyncio.create_task(db_worker.db_writer_worker())
        await queue.join()
        worker.cancel()
        try:
            await worker
        except (asyncio.CancelledError, Exception):
            pass


@pytest.mark.asyncio
async def test_partial_exit_without_active_trade_runs_once():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(lambda c: Base.metadata.create_all(c, tables=[Portfolio.__table__, ActiveTrade.__table__]))
    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    async with Session() as s:
        s.add(Portfolio(
            stock_code="034020", stock_name="두산에너빌리티", quantity=18,
            average_price=Decimal("85589"), current_price=Decimal("86200"),
            strategy_type="swing",
        ))
        await s.commit()

    queue = asyncio.Queue()
    partial = ExitSignal(action="SELL_PARTIAL", reason="1차 익절(50%): RSI(55.2)>=55", sell_pct=0.5)
    order_ok = {"rt_cd": "0", "output": {"ODNO": "0000002153"}}

    api = SimpleNamespace(
        get_current_price=AsyncMock(return_value={"output": {"stck_prpr": "86200"}}),
        get_daily_chart=AsyncMock(return_value=_daily_chart()),
        # [C-1 2026-10-02 6회차] 국내 스윙 매도 직전 KIS 실보유 확인(상한 = min(주문가능, 보유) − DT)
        get_balance=AsyncMock(return_value={"rt_cd": "0", "output1": [
            {"pdno": "034020", "hldg_qty": "18", "ord_psbl_qty": "18"}]}),
    )
    api.get_balance_all_v2 = api.get_balance  # [R9] 연속조회 목 보정 — 같은 목을 공유
    place_order = AsyncMock(return_value=order_ok)

    with patch.object(exit_service, "korea_investment_api", api), \
         patch.object(exit_service, "check_market_regime", AsyncMock(return_value={"regime_200d": "bull"})), \
         patch.object(exit_service, "check_exit_signal", return_value=partial), \
         patch.object(exit_service, "calculate_atr", return_value=1500.0), \
         patch.object(exit_service, "calculate_rsi", return_value=55.2), \
         patch.object(exit_service, "calculate_bollinger", return_value=(90000.0, 85370.0, 80000.0)), \
         patch("app.services.trade_executor.safe_place_order", place_order), \
         patch.object(db_worker, "db_task_queue", queue), \
         patch.object(exit_service.loss_tracker, "record_trade_result_safe", AsyncMock()), \
         patch.object(exit_service.loss_tracker, "save_to_db", AsyncMock()), \
         patch.object(exit_service, "_schedule_swing_sell_fill_check_v2"):  # [2026-10-08] 체결 확인 백그라운드(KIS 조회) 차단

        # 1차 사이클: 18주 → 9주 매도
        async with Session() as s:
            exits = await exit_service.check_and_execute_exit_v2(s)
        assert len(exits) == 1 and exits[0]["sell_quantity"] == 9
        await _drain(queue, Session)

        async with Session() as s:
            at = (await s.execute(select(ActiveTrade).where(ActiveTrade.stock_code == "034020"))).scalars().all()
            pf = (await s.execute(select(Portfolio).where(Portfolio.stock_code == "034020"))).scalar_one()
        assert len(at) == 1, "부분매도 후 ActiveTrade 행이 생성돼야 함"
        assert float(at[0].remaining_pct) == 0.5
        assert at[0].quantity == 9 and float(at[0].entry_price) == 85589.0
        assert pf.quantity == 9, "부분매도 후 Portfolio 잔량이 즉시 감소해야 함"

        # 2차 사이클: remaining_pct=0.5 → check_exit_signal에 전달되는 값으로 재실행 차단 확인
        seen = {}
        def _capture(**kw):
            seen.update(kw)
            return None
        with patch.object(exit_service, "check_exit_signal", side_effect=_capture):
            async with Session() as s:
                exits2 = await exit_service.check_and_execute_exit_v2(s)
        assert exits2 == []
        assert seen["remaining_pct"] == 0.5
        assert place_order.await_count == 1, "매도 주문은 한 번만 나가야 함"

    await engine.dispose()


@pytest.mark.asyncio
async def test_db_worker_skips_active_trade_insert_without_required_keys():
    """필수 키 없는 신규 upsert는 IntegrityError 재시도 대신 스킵."""
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(lambda c: Base.metadata.create_all(c, tables=[Portfolio.__table__, ActiveTrade.__table__]))
    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    queue = asyncio.Queue()
    await queue.put({"type": "upsert_active_trade", "data": {"stock_code": "000000", "remaining_pct": 0.5}})
    await asyncio.wait_for(_drain(queue, Session), timeout=5)  # 재시도(1+2+4초) 없이 즉시 끝나야 함

    async with Session() as s:
        rows = (await s.execute(select(ActiveTrade))).scalars().all()
    assert rows == []
    await engine.dispose()


def test_real_db_session_is_blocked_in_this_module():
    """회귀: 이 모듈 테스트는 실 DB 세션을 만들 수 없어야 한다(예외 발생형 차단)."""
    from app.core import database
    with pytest.raises(_DBBlocked):
        database.AsyncSessionLocal()


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