auto-stock/backend/tests/test_swing_exit_hardening_r1_20261009.py

"""스윙 청산 하드닝 R1 (2026-10-09) — swing/exit_service 회귀 테스트.

#1/#3 실시세 정상 + 일봉/지표 조회 실패 시 폴백 손절이 ATR=None → 일반 손절(-7%)로 고변동 종목을
      오손절하던 것 → 대체가 경로와 같은 규칙(캐시 ATR → 느슨한 쪽).
#2    'synced'/'외부 유입' 시딩 행의 시간손절 진입일 보정 + TradeHistory 기반 전량청산 하한.
#4    전량 청산 loss_tracker 승패 = 포지션 누적 순손익(이전 분할 레그 + 이번 레그).
#6    스윙 SELL TradeHistory 의 commission·net_profit_loss(비용 차감).

규약(CLAUDE.md §4.1): 주문 목은 공통 종단(safe_place_order), 기대값은 settings.* 참조.
"""
import time
from datetime import datetime, timedelta
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock

import pytest
from sqlalchemy.dialects import postgresql

import app.services.mean_reversion_strategy as mr
from app.core.config import settings
from app.services.mean_reversion_strategy import ExitSignal

OK = {"rt_cd": "0", "output": {"ODNO": "9"}}


@pytest.fixture(autouse=True)
def _reset(monkeypatch):
    import app.services.swing.exit_service as es
    for name in ("_pending_unknown_sells", "_last_live_prices", "_last_atr",
                 "_presell_check_failures", "_no_price_skips", "_recent_exits"):
        monkeypatch.setattr(es, name, {})
    from app.core import log_throttle
    log_throttle.reset_for_tests()
    import app.services.adaptive_strategy as ad
    monkeypatch.setattr(ad, "get_current_adaptive_params", lambda: SimpleNamespace(fallback_used=True))
    monkeypatch.setattr(mr.strategy_cfg, "get", lambda k, d=None: None)
    import app.services.trade_executor as te
    monkeypatch.setattr(te, "_delayed_sync_portfolio", AsyncMock())


class _RowsDB:
    """execute 마다 미리 준비한 행 목록을 .all()/.scalar() 로 돌려주고 문장을 기록한다."""

    def __init__(self, *row_sets):
        self.row_sets = list(row_sets)
        self.statements = []

    async def execute(self, stmt):
        self.statements.append(stmt)
        rows = self.row_sets.pop(0) if self.row_sets else []
        return SimpleNamespace(all=lambda: list(rows), scalar=lambda: (rows[0] if rows else None))


class _BoomDB:
    async def execute(self, stmt):
        raise RuntimeError("db down")


def _sql(stmt) -> str:
    return str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))


def _env(monkeypatch, *, avg=10000, price, high_vol=True, chart=None, signal=None,
         order=None, at_kwargs=None):
    """chart: None=정상 30봉, Exception=조회 실패, list=그 봉들."""
    import app.services.swing.exit_service as es
    import app.services.trade_executor as te
    import app.services.db_worker as dbw
    import app.services.portfolio_cache_bridge as pcb
    from app.core.timeutil import now_kst
    from app.services.daytrading.position_manager import day_trading_state

    holding = SimpleNamespace(stock_code="005930", stock_name="T", quantity=10, average_price=avg,
                              current_price=price, market="KRX", updated_at=now_kst())
    monkeypatch.setattr(es, "get_swing_holdings", AsyncMock(return_value=[holding]))
    monkeypatch.setattr(es, "check_market_regime", AsyncMock(return_value={"regime_200d": "bull"}))
    at = SimpleNamespace(strategy_name="swing", remaining_pct=1.0, highest_since_entry=None,
                         created_at=now_kst() - timedelta(days=2), entry_reason=None,
                         entry_price=avg)
    for k, v in (at_kwargs or {}).items():
        setattr(at, k, v)
    monkeypatch.setattr(es, "get_swing_active_trade", AsyncMock(return_value=at))
    api = MagicMock()
    rng = 250 if high_vol else 40  # 고변동: ATR≈500(>3%)
    bars = [{"stck_clpr": str(price + 100 + (i % 3) * 50), "stck_hgpr": str(price + 100 + rng + (i % 3) * 50),
             "stck_lwpr": str(price + 100 - rng + (i % 3) * 50)} for i in range(30)]
    if isinstance(chart, Exception):
        api.get_daily_chart = AsyncMock(side_effect=chart)
    else:
        api.get_daily_chart = AsyncMock(return_value={"output2": bars if chart is None else chart})
    api.get_current_price = AsyncMock(return_value={"output": {"stck_prpr": str(price)}})
    api.get_balance = AsyncMock(return_value={"rt_cd": "0", "output1": [
        {"pdno": "005930", "hldg_qty": "10", "ord_psbl_qty": "10"}]})
    api.get_balance_all_v2 = api.get_balance  # [R9] 연속조회 목 보정 — 같은 목을 공유
    monkeypatch.setattr(es, "korea_investment_api", api)
    if signal is not None:
        monkeypatch.setattr(es, "check_exit_signal", lambda **kw: signal)
    monkeypatch.setattr(day_trading_state, "positions", {})
    monkeypatch.setattr(day_trading_state, "save_daily_state", AsyncMock())
    sent = []

    async def _spo(**kw):
        sent.append(kw["quantity"])
        return order if order is not None else {"rt_cd": "1", "msg_cd": "X", "msg1": "stop here"}
    monkeypatch.setattr(te, "safe_place_order", _spo)
    history = []
    monkeypatch.setattr(es, "enqueue_trade_history", AsyncMock(side_effect=lambda p: history.append(p)))
    monkeypatch.setattr(es, "enqueue_close_active_trade", AsyncMock())
    monkeypatch.setattr(dbw, "enqueue_active_trade_upsert", AsyncMock())
    monkeypatch.setattr(pcb, "reflect_sell", AsyncMock())
    monkeypatch.setattr(es, "_schedule_swing_sell_fill_check_v2", MagicMock())
    # [R1-b] 보유 집계 SELECT(휴면·페이지 감시)는 _RowsDB 의 순차 결과를 소비하므로 제외 —
    # 집계 자체는 test_swing_exit_hardening_r1b_20261009 에서 검증한다.
    monkeypatch.setattr(es, "_portfolio_census_v2", AsyncMock(return_value=None))
    loss = AsyncMock()
    monkeypatch.setattr(es.loss_tracker, "record_trade_result_safe", loss)
    monkeypatch.setattr(es.loss_tracker, "save_to_db", AsyncMock())
    return SimpleNamespace(es=es, sent=sent, history=history, loss=loss, at=at)


async def _cycle(es, db=None):
    es._recent_exits.clear()
    await es.check_and_execute_exit_v2(db if db is not None else MagicMock())


# ═══════════════════════════════════════════════════════════════════════════
# #1 / #3 — 실시세 정상 + 일봉 조회 실패 폴백 손절
# ═══════════════════════════════════════════════════════════════════════════

@pytest.mark.asyncio
async def test_nonstale_chart_failure_uses_cached_high_vol_atr(monkeypatch):
    """[재현 반전] 고변동 -8%: 정상 주기 보유 → 차트 실패 주기에도 보유(이전: -7% 시장가 오손절)."""
    env = _env(monkeypatch, price=9200)
    await _cycle(env.es)
    assert env.sent == [] and "005930" in env.es._last_atr
    env2 = _env(monkeypatch, price=9200, chart=RuntimeError("EGW00201"))
    await _cycle(env2.es)
    assert env2.sent == []


@pytest.mark.asyncio
async def test_nonstale_short_chart_without_cache_uses_loosest(monkeypatch):
    loosest = -max(abs(settings.STOP_LOSS_PERCENT), abs(settings.STOP_LOSS_HIGH_VOL))
    tight = -min(abs(settings.STOP_LOSS_PERCENT), abs(settings.STOP_LOSS_HIGH_VOL))
    between = int(10000 * (1 + (tight + loosest) / 2 / 100))
    env = _env(monkeypatch, price=between, chart=[])  # 20봉 미만
    await _cycle(env.es)
    assert env.sent == []
    beyond = int(10000 * (1 + (loosest - 1) / 100))
    env2 = _env(monkeypatch, price=beyond, chart=RuntimeError("down"))
    await _cycle(env2.es)
    assert env2.sent == [10]  # 손절 자체는 계속 동작(try 밖 판정)


@pytest.mark.asyncio
async def test_nonstale_chart_failure_cached_low_vol_keeps_normal_stop(monkeypatch):
    floor = -abs(settings.STOP_LOSS_PERCENT)
    below = int(10000 * (1 + (floor - 1) / 100))
    env = _env(monkeypatch, price=below, chart=RuntimeError("down"))
    from app.core.timeutil import now_kst
    env.es._last_atr["005930"] = (below * 0.01, time.monotonic(), now_kst())  # ATR 1% 저변동
    await _cycle(env.es)
    assert env.sent == [10]


def test_nonstale_helper_prefers_current_atr_and_signs_negative(monkeypatch):
    import app.services.swing.exit_service as es
    # 이번 ATR 이 있으면 기존 _fallback_stop_loss_pct_v2 경로(저변동 → 일반 손절)
    assert es._nonstale_fallback_stop_pct_v2("005930", 10.0, 10000.0, None) == -abs(settings.STOP_LOSS_PERCENT)
    # 없으면 대체가 규칙(캐시 없음 → 느슨한 쪽)
    assert es._nonstale_fallback_stop_pct_v2("005930", None, 10000.0, None) == -max(
        abs(settings.STOP_LOSS_PERCENT), abs(settings.STOP_LOSS_HIGH_VOL))
    assert es._nonstale_fallback_stop_pct_v2("005930", "bad", 10000.0, None) < 0


# ═══════════════════════════════════════════════════════════════════════════
# #2 — 시딩 행 시간손절 진입일
# ═══════════════════════════════════════════════════════════════════════════

_EXT = "외부 유입 포지션 — 동기화 시 자동 등록(진입가=평균단가)"


@pytest.mark.asyncio
async def test_v3_synced_row_uses_original_buy_despite_race_closed_row():
    """181710 재현: BUY 10 (9/23) → 분할 SELL 5 (10/6 09:12) → 레이스로 swing 행 closed →
    synced 재시딩(10/6 09:18). 이전: created_at(10/6). 이제: 9/23."""
    import app.services.swing.exit_service as es
    created = datetime(2026, 10, 6, 9, 18)
    buy = datetime(2026, 9, 23, 12, 53)
    db = _RowsDB([("BUY", 10, 10, buy), ("SELL", 5, 5, datetime(2026, 10, 6, 9, 12))])
    at = SimpleNamespace(created_at=created, entry_reason=_EXT, strategy_name="synced")
    basis = await es._resolve_time_stop_entry_v3(db, "181710", at)
    assert basis.created_at == buy
    sql = _sql(db.statements[0])
    assert "trade_history.status = 'COMPLETED'" in sql
    assert "trade_history.created_at <= '2026-10-06 09:18:00'" in sql
    assert "day_trade" in sql  # 비-DT 정본 절


@pytest.mark.asyncio
async def test_v3_recognizes_strategy_name_or_either_prefix():
    import app.services.swing.exit_service as es
    buy = datetime(2026, 9, 1, 10, 0)
    for kw in ({"strategy_name": "synced", "entry_reason": None},
               {"strategy_name": None, "entry_reason": _EXT},
               {"strategy_name": "swing", "entry_reason": "동기화 유입 포지션 (부분매도 시 자동 생성)"}):
        at = SimpleNamespace(created_at=datetime(2026, 10, 6), **kw)
        basis = await es._resolve_time_stop_entry_v3(_RowsDB([("BUY", 3, 3, buy)]), "X", at)
        assert basis.created_at == buy


@pytest.mark.asyncio
async def test_v3_lower_bound_is_last_flat_sell():
    """이전 포지션이 TradeHistory 상 전량 청산된 뒤의 BUY 만 진입으로 본다."""
    import app.services.swing.exit_service as es
    rows = [("BUY", 5, 5, datetime(2026, 7, 1)), ("SELL", 5, 5, datetime(2026, 7, 20)),
            ("BUY", 4, 4, datetime(2026, 9, 2))]
    at = SimpleNamespace(created_at=datetime(2026, 10, 6), entry_reason=_EXT, strategy_name="synced")
    basis = await es._resolve_time_stop_entry_v3(_RowsDB(rows), "X", at)
    assert basis.created_at == datetime(2026, 9, 2)


@pytest.mark.asyncio
async def test_v3_fully_sold_history_keeps_created_at():
    import app.services.swing.exit_service as es
    rows = [("BUY", 5, 5, datetime(2026, 7, 1)), ("SELL", 5, 5, datetime(2026, 7, 20))]
    at = SimpleNamespace(created_at=datetime(2026, 10, 6), entry_reason=_EXT, strategy_name="synced")
    assert await es._resolve_time_stop_entry_v3(_RowsDB(rows), "X", at) is at


@pytest.mark.asyncio
async def test_v3_non_seeded_row_does_not_query_and_failure_keeps_created_at():
    import app.services.swing.exit_service as es
    at = SimpleNamespace(created_at=datetime(2026, 10, 6), entry_reason="기술: RSI", strategy_name="swing")
    assert await es._resolve_time_stop_entry_v3(_BoomDB(), "X", at) is at  # DB 미조회
    seeded = SimpleNamespace(created_at=datetime(2026, 10, 6), entry_reason=_EXT, strategy_name="synced")
    assert await es._resolve_time_stop_entry_v3(_BoomDB(), "X", seeded) is seeded


@pytest.mark.asyncio
async def test_v3_wired_into_time_stop(monkeypatch):
    """synced 행 + 원 BUY 가 SWING_TIME_STOP_DAYS 거래일 이전 → 시간손절 발동(이전: 재시딩일 기준 미발동)."""
    from app.core.timeutil import now_kst
    monkeypatch.setattr(settings, "SWING_TIME_STOP_ENABLED", True, raising=False)
    env = _env(monkeypatch, price=10050, high_vol=False, signal=None, order=OK,
               at_kwargs={"strategy_name": "synced", "entry_reason": _EXT,
                          "created_at": now_kst().replace(tzinfo=None) - timedelta(days=1)})
    monkeypatch.setattr(env.es, "check_exit_signal", lambda **kw: None)
    old_buy = now_kst().replace(tzinfo=None) - timedelta(days=int(settings.SWING_TIME_STOP_DAYS) * 2)
    db = _RowsDB([("BUY", 10, 10, old_buy)], [])  # ① 시간손절 진입일 ② 누적 실현손익(없음)
    await _cycle(env.es, db)
    assert env.sent == [10]
    assert env.history[0]["exit_reason"].startswith("시간손절")


# ═══════════════════════════════════════════════════════════════════════════
# #4 — 포지션 단위 순손익 승패 / #6 — commission·net 기록
# ═══════════════════════════════════════════════════════════════════════════

def _cost(entry, exit_, q):
    from app.services.trade_helpers import calculate_trade_costs
    return calculate_trade_costs(entry, exit_, q, True, "KOSPI")["total_cost"]


def test_position_is_loss_v2():
    import app.services.swing.exit_service as es
    assert es._position_is_loss_v2(8000.0, -410.0) is False
    assert es._position_is_loss_v2(None, -410.0) is True
    assert es._position_is_loss_v2(-1.0, 0.5) is True


@pytest.mark.asyncio
async def test_prior_realized_net_converts_gross_rows_and_uses_net_rows():
    import app.services.swing.exit_service as es
    # 행: (net_profit_loss, commission, total_amount, executed_quantity, order_quantity)
    # ① 과거 gross 기록(commission NULL): 진입 10,000 → 10,500 × 10 = gross 5,000
    # ② 비용 기록 행: net 그대로
    rows = [(5000.0, None, 105000.0, 10, 10), (1234.0, 50.0, 99999.0, 5, 5), (None, None, 1, 1, 1)]
    at = SimpleNamespace(created_at=datetime(2026, 10, 1), entry_reason=None, strategy_name="swing")
    db = _RowsDB(rows)
    got = await es._prior_position_realized_net_v2(db, "005930", at, "KRX")
    assert got == pytest.approx(5000.0 - _cost(10000.0, 10500.0, 10) + 1234.0)
    sql = _sql(db.statements[0])
    assert "trade_history.order_type = 'SELL'" in sql and "trade_history.created_at >= '2026-10-01" in sql


@pytest.mark.asyncio
async def test_prior_realized_net_failure_returns_none():
    import app.services.swing.exit_service as es
    at = SimpleNamespace(created_at=datetime(2026, 10, 1), entry_reason=None, strategy_name="swing")
    assert await es._prior_position_realized_net_v2(_BoomDB(), "005930", at, "KRX") is None
    assert await es._prior_position_realized_net_v2(_RowsDB(), "005930", None, "KRX") is None


@pytest.mark.asyncio
async def test_full_exit_after_profitable_partials_is_not_loss(monkeypatch):
    """449450 재현: 1차 익절 2회(+3,900/+4,125) 후 잔량 트레일링 소폭 손실 → 포지션은 이익 → 패 아님."""
    env = _env(monkeypatch, price=9990, high_vol=False, order=OK,
               signal=ExitSignal("SELL_ALL", "전량청산: 트레일링", 1.0))
    monkeypatch.setattr(settings, "SWING_TIME_STOP_ENABLED", False, raising=False)
    db = _RowsDB([(3900.0, None, 0, 0, 0), (4125.0, None, 0, 0, 0)])
    await _cycle(env.es, db)
    assert env.sent == [10]
    env.loss.assert_awaited_once()
    assert env.loss.await_args.kwargs["is_loss"] is False
    # #6: commission·순손익 기록
    h = env.history[0]
    assert h["commission"] == _cost(10000.0, 9990.0, 10)
    assert h["net_profit_loss"] == pytest.approx((9990 - 10000) * 10 - _cost(10000.0, 9990.0, 10))


@pytest.mark.asyncio
async def test_full_exit_tiny_gross_profit_within_costs_is_loss(monkeypatch):
    """비용 이내 미세 이익(+0.05%) 단일 레그 → 순손익 음수 → 패(§5.1)."""
    env = _env(monkeypatch, price=10005, high_vol=False, order=OK,
               signal=ExitSignal("SELL_ALL", "BB상단", 1.0))
    monkeypatch.setattr(settings, "SWING_TIME_STOP_ENABLED", False, raising=False)
    await _cycle(env.es, _RowsDB([]))
    assert env.loss.await_args.kwargs["is_loss"] is True


@pytest.mark.asyncio
async def test_full_exit_prior_lookup_failure_still_sells_and_records(monkeypatch):
    env = _env(monkeypatch, price=9000, high_vol=False, order=OK,
               signal=ExitSignal("SELL_ALL", "손절", 1.0))
    monkeypatch.setattr(settings, "SWING_TIME_STOP_ENABLED", False, raising=False)
    await _cycle(env.es, _BoomDB())
    assert env.sent == [10]
    assert env.loss.await_args.kwargs["is_loss"] is True


@pytest.mark.asyncio
async def test_unknown_reconcile_uses_prior_realized_net(monkeypatch):
    import app.services.swing.exit_service as es
    import app.services.portfolio_cache_bridge as pcb
    monkeypatch.setattr(es, "enqueue_trade_history", AsyncMock())
    monkeypatch.setattr(es, "enqueue_close_active_trade", AsyncMock())
    monkeypatch.setattr(pcb, "reflect_sell", AsyncMock())
    loss = AsyncMock()
    monkeypatch.setattr(es.loss_tracker, "record_trade_result_safe", loss)
    monkeypatch.setattr(es.loss_tracker, "save_to_db", AsyncMock())
    es._pending_unknown_sells["005930"] = {
        "qty": 10, "pre_hldg": 10, "since": time.time(), "price": 9990.0, "avg_price": 10000.0,
        "swing_qty": 10, "sell_pct": 1.0, "remaining_pct": 0.5, "original_qty": 20,
        "strategy": "swing", "filter_not_dt": False, "had_active_trade": True,
        "stock_name": "T", "reason": "트레일링", "market": "KRX", "prior_realized_net": 8000.0,
    }
    assert await es._reconcile_unknown_sell_v2("005930", 0, 0) is True
    assert loss.await_args.kwargs["is_loss"] is False


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