auto-stock/backend/tests/test_mdd_reduction_retry_20261002.py

"""[C-1 2026-10-02] MDD 강제축소 재시도가 성공 종목을 누적 매도하던 결함 회귀 테스트.

재현(검증 에이전트 시나리오): 005930 100주 정상 + 000660 주문 실패.
기존 코드는 실패 1건으로 executed 플래그를 리셋 → swing/scheduler 가 60초마다
execute_reduction 을 재호출 → **모든 보유종목**을 다시 30% 매도.
5사이클에 005930 이 [30, 21, 14, 10, 7] = 82주(82%) 청산됐다.

수정 후 기대: 005930 은 30주 1회만, 000660 만 재시도(상한 MDD_REDUCTION_MAX_RETRIES).

목킹 지점: 축소 경로가 수렴하는 주문 종단 `trade_executor.safe_place_order`
(락 경유 함수 — mdd_manager 가 함수 안에서 import 하므로 모듈 속성을 덮으면 된다)와
잔고 조회 `korea_investment_api.get_balance`. 부기 함수는 예외 발생형/캡처형으로 덮는다.
"""
import json

import pytest

from app.core.config import settings
from app.services import mdd_manager as mm
from app.services import trade_executor as te
from app.services.korea_investment_api import korea_investment_api


class _FakeRedis:
    def __init__(self):
        self.store = {}

    async def set(self, k, v, **kw):
        self.store[k] = v
        return True

    async def get(self, k):
        return self.store.get(k)


class _Account:
    """잔고(보유수량)를 체결에 따라 실제로 줄여 주는 가짜 계좌."""

    def __init__(self, holdings, fail_codes=(), busy_codes=()):
        self.qty = dict(holdings)
        self.fail_codes = set(fail_codes)
        self.busy_codes = set(busy_codes)
        self.sells = {}      # code -> [qty,...] (접수 성공분)
        self.attempts = {}   # code -> 시도 횟수

    async def get_balance(self, *a, **k):
        return {
            "output1": [
                {"pdno": c, "prdt_name": c, "hldg_qty": str(q), "prpr": "10000",
                 "pchs_avg_pric": "11000"}
                for c, q in self.qty.items() if q > 0
            ]
        }

    async def safe_place_order(self, stock_code, order_type, quantity, price=None, **kw):
        assert order_type == "SELL"
        self.attempts[stock_code] = self.attempts.get(stock_code, 0) + 1
        if stock_code in self.busy_codes:
            return {"rt_cd": "-1", "msg1": "order_lock_busy"}
        if stock_code in self.fail_codes:
            return {"rt_cd": "1", "msg1": "거래정지 종목"}
        self.qty[stock_code] -= quantity
        self.sells.setdefault(stock_code, []).append(quantity)
        return {"rt_cd": "0", "output": {"ODNO": f"O{stock_code}{len(self.sells[stock_code])}"}}


@pytest.fixture
def env(monkeypatch):
    fake_redis = _FakeRedis()
    import app.core.redis as redis_mod
    monkeypatch.setattr(redis_mod, "get_redis", lambda: fake_redis)

    async def _nosleep(*a, **k):
        return None
    monkeypatch.setattr(mm.asyncio, "sleep", _nosleep)
    # [L-1 4회차] 강제축소는 장중에만 주문 — 테스트는 장중으로 고정
    import app.core.market_schedule as _ms
    monkeypatch.setattr(_ms, "is_market_open", lambda: True)

    async def _noop(*a, **k):
        return None
    monkeypatch.setattr(mm, "_sync_active_trade_after_reduction", _noop)
    monkeypatch.setattr(mm, "_reflect_reduction", _noop)
    import app.services.db_worker as dbw
    monkeypatch.setattr(dbw, "enqueue_trade_history", _noop)

    # 텔레그램 알림 경로 차단(ensure_future 로 실제 전송 시도 방지)
    try:
        from app.services import telegram_notifier as tn

        async def _no_notify(*a, **k):
            return None
        monkeypatch.setattr(tn.telegram_notifier, "notify_mdd_warning", _no_notify)
    except Exception:
        pass

    def _make(account):
        monkeypatch.setattr(korea_investment_api, "get_balance", account.get_balance)
        monkeypatch.setattr(korea_investment_api, "get_balance_all_v2", account.get_balance)  # [R9] 연속조회 목 보정 — 같은 목을 공유
        monkeypatch.setattr(te, "safe_place_order", account.safe_place_order)
        return account

    return fake_redis, _make


async def _arm(m: "mm.MDDManager", level_balance: float = 89.0):
    await m.update(100.0)
    r = await m.update(level_balance)
    return r


def _action(m):
    """swing/scheduler.py 가 조립하는 것과 동일한 mdd_action."""
    lv = m.reduction_applied
    return {
        "action": "FORCE_REDUCE" if lv == "20pct" else "REDUCE_AND_PAUSE",
        "reduce_pct": 0.50 if lv == "20pct" else 0.30,
        "prioritize_leverage": lv == "20pct",
    }


@pytest.mark.asyncio
async def test_c1_repro_success_stock_sold_once_only(env):
    _, make = env
    acct = make(_Account({"005930": 100, "000660": 50}, fail_codes={"000660"}))
    m = mm.MDDManager()
    r = await _arm(m)
    assert r["action"] == "REDUCE_AND_PAUSE"

    for _ in range(5):  # 스케줄러 5사이클
        await m.execute_reduction(_action(m))

    assert acct.sells.get("005930") == [30], (
        f"정상 종목이 재시도마다 누적 매도됨: {acct.sells.get('005930')}"
    )
    assert acct.qty["005930"] == 70
    # 실패 종목만 재시도된다
    assert acct.attempts["005930"] == 1
    assert acct.attempts["000660"] == min(5, settings.MDD_REDUCTION_MAX_RETRIES)


@pytest.mark.asyncio
async def test_c1_retry_cap_then_stop(env):
    _, make = env
    acct = make(_Account({"005930": 100, "000660": 50}, fail_codes={"000660"}))
    m = mm.MDDManager()
    await _arm(m)
    cap = settings.MDD_REDUCTION_MAX_RETRIES
    for _ in range(cap + 3):
        await m.execute_reduction(_action(m))
    assert acct.attempts["000660"] == cap, "재시도 상한 미적용"
    # [3회차] 포기 종목이 남으면 executed 는 False(다음 날 재시도 가능) — 당일 주문은 없다
    assert m._reduction_executed_10pct is False
    assert "000660" in m._reduction_gave_up["10pct"]
    assert acct.sells["005930"] == [30]


@pytest.mark.asyncio
async def test_c1_order_lock_busy_is_deferred_not_failure(env):
    _, make = env
    acct = make(_Account({"005930": 100, "000660": 50}, busy_codes={"000660"}))
    m = mm.MDDManager()
    await _arm(m)
    n = settings.MDD_REDUCTION_MAX_RETRIES + 2
    for _ in range(n):
        await m.execute_reduction(_action(m))
    assert acct.attempts["000660"] == n, "busy 를 실패로 계수해 포기했다"
    assert "000660" not in m._reduction_gave_up.get("10pct", set())
    assert m._reduction_attempts.get("10pct", {}).get("000660", 0) == 0
    # 락 해제 후 1회만 축소
    acct.busy_codes.clear()
    await m.execute_reduction(_action(m))
    await m.execute_reduction(_action(m))
    assert acct.sells["000660"] == [15]
    assert acct.sells["005930"] == [30]


@pytest.mark.asyncio
async def test_c1_progress_survives_restart_via_redis(env):
    _, make = env
    acct = make(_Account({"005930": 100, "000660": 50}, fail_codes={"000660"}))
    m1 = mm.MDDManager()
    await _arm(m1)
    await m1.execute_reduction(_action(m1))
    assert acct.sells["005930"] == [30]

    # 재기동: 새 인스턴스가 Redis 에서 복원 후 재시도
    m2 = mm.MDDManager()
    assert await m2.load_from_redis() is True
    assert m2._reduction_executed_10pct is False  # 실패 종목이 남아 재시도 대상
    acct.fail_codes.clear()
    await m2.execute_reduction(_action(m2))
    assert acct.sells["005930"] == [30], "재기동 후 완료 종목을 재축소했다"
    assert acct.sells["000660"] == [15]


@pytest.mark.asyncio
async def test_c1_peak_reset_clears_progress(env):
    _, make = env
    acct = make(_Account({"005930": 100}))
    m = mm.MDDManager()
    await _arm(m)
    await m.execute_reduction(_action(m))
    assert m._reduced_codes["10pct"] == {"005930"}
    await m.update(200.0)  # 신고점 → 모든 MDD 플래그·진행상태 리셋(§5.3)
    assert m._reduced_codes == {}
    await m.update(170.0)  # 새 -15% 이벤트 → 다시 축소 대상
    await m.execute_reduction(_action(m))
    assert acct.sells["005930"] == [30, 21]


@pytest.mark.asyncio
async def test_c1_day_roll_resets_attempts_not_done(env, monkeypatch):
    _, make = env
    acct = make(_Account({"005930": 100, "000660": 50}, fail_codes={"000660"}))
    m = mm.MDDManager()
    await _arm(m)
    day = {"d": "2026-10-01"}
    monkeypatch.setattr(mm.MDDManager, "_today_kst", staticmethod(lambda: day["d"]))
    for _ in range(settings.MDD_REDUCTION_MAX_RETRIES):
        await m.execute_reduction(_action(m))
    assert "000660" in m._reduction_gave_up["10pct"]  # 당일 포기
    # 날짜 변경: 시도 횟수·포기 집합 초기화 → 별도 조작 없이 실패 종목만 재시도
    day["d"] = "2026-10-02"
    acct.fail_codes.clear()
    await m.execute_reduction(_action(m))
    assert acct.sells["005930"] == [30], "날짜 변경으로 완료 종목이 재축소됐다"
    assert acct.sells["000660"] == [15]


@pytest.mark.asyncio
async def test_c1_bookkeeping_error_after_fill_does_not_resell(env, monkeypatch):
    _, make = env
    acct = make(_Account({"005930": 100, "000660": 50}, fail_codes={"000660"}))

    async def _boom(*a, **k):
        raise RuntimeError("enqueue 실패")
    import app.services.db_worker as dbw
    monkeypatch.setattr(dbw, "enqueue_trade_history", _boom)

    m = mm.MDDManager()
    await _arm(m)
    for _ in range(3):
        await m.execute_reduction(_action(m))
    assert acct.sells["005930"] == [30], "주문 접수 후 부기 예외로 재매도됐다"


@pytest.mark.asyncio
async def test_c1_one_share_policy_and_20pct_level(env):
    _, make = env
    acct = make(_Account({"005930": 100, "069500": 1}))
    m = mm.MDDManager()
    await _arm(m)
    await m.execute_reduction(_action(m))
    # 1주 보유: max(int(1*0.3),1)=1 → 전량 매도 (문서화된 정책)
    assert acct.sells["069500"] == [1]
    assert acct.sells["005930"] == [30]
    # -20% 진입: 별도 레벨 → 남은 보유분을 50% 1회
    r = await m.update(79.0)
    assert r["action"] == "FORCE_REDUCE"
    for _ in range(3):
        await m.execute_reduction(_action(m))
    assert acct.sells["005930"] == [30, 35]


def test_c1_persisted_state_is_json(env):
    m = mm.MDDManager()
    m._reduced_codes = {"10pct": {"005930"}}
    st = m._reduction_progress_state()
    json.loads(json.dumps(st))  # set 이 남아 있으면 직렬화 실패
    m2 = mm.MDDManager()
    m2._restore_reduction_progress(st)
    assert m2._reduced_codes == {"10pct": {"005930"}}


# ── [3회차 2026-10-02] 검증 에이전트 재현 시나리오 ────────────────────────────

@pytest.mark.asyncio
async def test_c1r3_gave_up_only_stock_retried_next_day(env, monkeypatch):
    """repro_mdd.py: MAX_RETRIES=2, 거래정지 1종목뿐 → day1 2건, day2 에 다시 시도돼야 한다.

    기존: 포기 종목만 남으면 pending=[] → executed=True 고정 → 롤오버 전에 반환 → 영구 생략.
    """
    _, make = env
    monkeypatch.setattr(settings, "MDD_REDUCTION_MAX_RETRIES", 2)
    acct = make(_Account({"005930": 10}, fail_codes={"005930"}))
    m = mm.MDDManager()
    await _arm(m)
    day = {"d": "2026-10-01"}
    monkeypatch.setattr(mm.MDDManager, "_today_kst", staticmethod(lambda: day["d"]))
    for _ in range(3):
        await m.execute_reduction(_action(m))
    assert acct.attempts["005930"] == 2
    day["d"] = "2026-10-02"
    for _ in range(3):
        await m.execute_reduction(_action(m))
    assert acct.attempts["005930"] == 4, "다음 날 재시도가 일어나지 않았다"
    acct.fail_codes.clear()
    day["d"] = "2026-10-03"
    await m.execute_reduction(_action(m))
    assert acct.sells["005930"] == [3]
    assert m._reduction_executed_10pct is True


@pytest.mark.asyncio
@pytest.mark.parametrize("bad", [
    {"rt_cd": "1", "msg_cd": "EGW00201", "msg1": "초당 거래건수 초과"},
    {"rt_cd": "1", "msg_cd": "40910000", "msg1": "모의투자 계좌 오류"},
    {"rt_cd": "0"},  # output1 없음
])
async def test_c2_balance_error_is_not_empty_holdings(env, bad):
    _, make = env
    acct = make(_Account({"005930": 100}))
    good_balance = acct.get_balance

    async def _bad(*a, **k):
        return bad
    korea = __import__("app.services.korea_investment_api", fromlist=["x"]).korea_investment_api
    m = mm.MDDManager()
    await _arm(m)
    import pytest as _pt
    mp = _pt.MonkeyPatch()
    mp.setattr(korea, "get_balance", _bad)
    mp.setattr(korea, "get_balance_all_v2", _bad)  # [R9] 연속조회 목 보정 — 같은 목을 공유
    try:
        assert await m.execute_reduction(_action(m)) == []
        assert m._reduction_executed_10pct is False, "오류 응답을 '보유 없음'으로 처리해 레벨 축소가 생략됐다"
    finally:
        mp.setattr(korea, "get_balance", good_balance)
        mp.setattr(korea, "get_balance_all_v2", good_balance)  # [R9] 연속조회 목 보정 — 같은 목을 공유
    await m.execute_reduction(_action(m))
    assert acct.sells["005930"] == [30]


@pytest.mark.asyncio
async def test_c2_genuinely_empty_holdings_marks_done(env):
    _, make = env
    acct = make(_Account({}))
    m = mm.MDDManager()
    await _arm(m)
    await m.execute_reduction(_action(m))
    assert m._reduction_executed_10pct is True


@pytest.mark.asyncio
async def test_low_executed_not_persisted_mid_loop(env):
    """루프 도중 영속된 상태에 executed=True 가 실리면 중간 재기동 시 남은 종목이 영구 미축소."""
    fake_redis, make = env
    acct = make(_Account({"005930": 100, "000660": 50}))
    snapshots = []
    orig = acct.safe_place_order

    async def _snap(stock_code, order_type, quantity, price=None, **kw):
        if stock_code == "000660":
            snapshots.append(json.loads(fake_redis.store["mdd:state"]))
        return await orig(stock_code, order_type, quantity, price, **kw)
    from app.services import trade_executor as _te
    import pytest as _pt
    mp = _pt.MonkeyPatch()
    mp.setattr(_te, "safe_place_order", _snap)
    try:
        m = mm.MDDManager()
        await _arm(m)
        await m.execute_reduction(_action(m))
    finally:
        mp.undo()
    assert snapshots, "두 번째 종목 주문 전 상태 스냅샷이 없다"
    assert snapshots[0]["_reduction_executed_10pct"] is False
    assert snapshots[0]["reduced_codes"] == {"10pct": ["005930"]}
    # 재기동 시뮬레이션: 스냅샷 상태로 복원 → 남은 종목만 축소
    m2 = mm.MDDManager()
    m2.peak_balance = snapshots[0]["peak_balance"]
    m2.reduction_applied = snapshots[0]["reduction_applied"]
    m2._restore_reduction_progress(snapshots[0])
    assert "005930" in m2._reduced_codes["10pct"]


@pytest.mark.asyncio
async def test_concurrent_reduction_runs_once(env):
    import asyncio as _aio
    _, make = env
    acct = make(_Account({"005930": 100}))
    m = mm.MDDManager()
    await _arm(m)
    await _aio.gather(m.execute_reduction(_action(m)), m.execute_reduction(_action(m)))
    assert acct.sells["005930"] == [30]


# ── [4회차 2026-10-02] reset 경합 / 장외 ─────────────────────────────────────

@pytest.mark.asyncio
async def test_m1r4_reset_mid_run_does_not_leave_stale_executed(env):
    """test_r4_verify.py 재현 뒤집기: 루프 도중 reset() 후에도 다음 -10% 축소가 실행된다."""
    _, make = env
    m = mm.MDDManager()
    acct = make(_Account({"005930": 100, "000660": 50}))
    orig = acct.safe_place_order
    calls = {"n": 0}

    async def _order(stock_code, order_type, quantity, price=None, **kw):
        calls["n"] += 1
        if calls["n"] == 1:
            await m.reset()  # 운영자 수동 리셋이 루프 도중 들어옴
        return await orig(stock_code, order_type, quantity, price, **kw)
    from app.services import trade_executor as _te
    import pytest as _pt
    mp = _pt.MonkeyPatch()
    mp.setattr(_te, "safe_place_order", _order)
    try:
        await _arm(m)
        await m.execute_reduction(_action(m))
        assert m._reduction_executed_10pct is False, "reset 이전 세대 결과로 executed 가 섰다"
        assert acct.attempts == {"005930": 1}, "reset 이후 남은 종목 매도를 중단해야 한다"
        # [R1] reset 은 peak 를 현재 잔고(89)로 재설정 — 새 -10% 이벤트는 추가 하락 시 발생
        assert (await m.update(89.0))["action"] == "NONE"
        r = await m.update(79.0)
        assert r["action"] == "REDUCE_AND_PAUSE"
        await m.execute_reduction(_action(m))
    finally:
        mp.undo()
    assert acct.attempts == {"005930": 2, "000660": 1}, acct.attempts


@pytest.mark.asyncio
async def test_l1r4_no_reduction_orders_when_market_closed(env, monkeypatch):
    _, make = env
    acct = make(_Account({"005930": 100}))
    import app.core.market_schedule as _ms
    monkeypatch.setattr(_ms, "is_market_open", lambda: False)
    m = mm.MDDManager()
    await _arm(m)
    assert await m.execute_reduction(_action(m)) == []
    assert acct.attempts == {} and m.reduction_applied == "10pct"
    assert m._reduction_executed_10pct is False  # 판정·상태 유지 → 개장 후 실행
    monkeypatch.setattr(_ms, "is_market_open", lambda: True)
    await m.execute_reduction(_action(m))
    assert acct.sells["005930"] == [30]


# ── [R1 2026-10-09] reset peak 재설정 / 갱신 신선도 ─────────────────────────────
async def _noop_async(*a, **k):
    return None


@pytest.mark.asyncio
async def test_reset_rebases_peak_so_reduction_not_retriggered(monkeypatch):
    m = mm.MDDManager()
    monkeypatch.setattr(m, "_write_state_v2", _noop_async)
    assert (await m.update(1_000_000))["action"] == "NONE"
    assert (await m.update(880_000))["action"] == "REDUCE_AND_PAUSE"
    await m.reset()
    assert m.peak_balance == 880_000
    assert (await m.update(880_000))["action"] == "NONE"
    assert await m.can_open_new_position() is True
    assert (await m.update(790_000))["action"] == "REDUCE_AND_PAUSE"


@pytest.mark.asyncio
async def test_reset_without_known_balance_rebases_on_next_update(monkeypatch):
    m = mm.MDDManager()
    monkeypatch.setattr(m, "_write_state_v2", _noop_async)
    m.peak_balance = 1_000_000.0
    await m.reset()
    assert (await m.update(850_000))["action"] == "NONE"
    assert m.peak_balance == 850_000


@pytest.mark.asyncio
async def test_reset_fail_safe_keeps_pause_and_peak(monkeypatch):
    m = mm.MDDManager()
    monkeypatch.setattr(m, "_write_state_v2", _noop_async)
    await m.update(1_000_000)
    await m.update(880_000)
    m._update_failures = 99
    await m.reset_fail_safe_v2()
    assert m.peak_balance == 1_000_000 and m.is_paused is True
    assert m._update_failures == 0


@pytest.mark.asyncio
async def test_stale_update_blocks_entry_during_market(monkeypatch):
    from datetime import datetime, timedelta
    from app.core import market_schedule as ms
    m = mm.MDDManager()
    base = datetime(2026, 10, 7, 10, 0, tzinfo=ms.KST)
    clock = {"t": base}
    monkeypatch.setattr(mm.MDDManager, "_now_kst_v2", staticmethod(lambda: clock["t"]))
    monkeypatch.setattr(ms, "is_market_open", lambda: True)
    m._update_last_ok_at = base
    m._created_at = base
    stale = mm._stale_sec_v2()
    clock["t"] = base + timedelta(seconds=stale - 1)
    assert await m.can_open_new_position() is True
    clock["t"] = base + timedelta(seconds=stale + 1)
    assert await m.can_open_new_position() is False
    await m.record_update_success()
    assert await m.can_open_new_position() is True
    monkeypatch.setattr(ms, "is_market_open", lambda: False)
    clock["t"] = base + timedelta(hours=20)
    assert await m.can_open_new_position() is True


# ── 하드닝 라운드 2 ──────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_r2_page_full_balance_does_not_mark_executed(env):
    """잔고 응답이 페이지 한도에 도달하면 보이는 종목만 축소하고 executed 는 세우지 않는다."""
    _, make = env
    n = int(settings.SWING_BALANCE_FULL_PAGE_ROWS)
    acct = make(_Account({f"{i:06d}": 100 for i in range(n)}))
    m = mm.MDDManager()
    await _arm(m)
    await m.execute_reduction(_action(m))
    assert len(acct.sells) == n
    assert not getattr(m, f"_reduction_executed_{m.reduction_applied}")


@pytest.mark.asyncio
async def test_r2_below_page_still_marks_executed(env):
    _, make = env
    n = int(settings.SWING_BALANCE_FULL_PAGE_ROWS)
    make(_Account({f"{i:06d}": 100 for i in range(n - 1)}))
    m = mm.MDDManager()
    await _arm(m)
    await m.execute_reduction(_action(m))
    assert getattr(m, f"_reduction_executed_{m.reduction_applied}")


def test_r2_stale_threshold_exceeds_swing_cycle_budget():
    from app.services.swing import scheduler as sw
    budget = (sw._DEFAULT_ENTRY_PHASE_TIMEOUT_SEC + sw._DEFAULT_EXIT_STEP_TIMEOUT_SEC
              + sw._DEFAULT_BALANCE_TIMEOUT_SEC + 60.0)
    assert mm._stale_sec_v2() > budget