auto-stock/backend/tests/test_fill_sweep_target_20261009.py

"""[하드닝 R1 2026-10-09] swing/fill_sweep 회귀.

1) BUY 진입가 보정 대상: '최신 비-DT active 행'(synced·타 전략 오보정)이 아니라
   entry_reason 으로 확정한 이 매수의 전략명(strategy_name 동등). 확정 불가면 보정 안 함.
2) 미확인 판정: executed_price NULL + 레거시(체결가=주문가 & slippage NULL) 포함, 일일 지표.
"""
import asyncio
import logging
import sys
from decimal import Decimal
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest
from sqlalchemy.dialects import postgresql

import app.services.db_worker as dbw
import app.services.trade_executor as te
from app.services.swing import fill_sweep as fs


def _run(coro):
    return asyncio.run(coro)


class _FakeDB:
    """1번째 execute = TradeHistory 후보, 이후 = ActiveTrade 조회. 쿼리를 기록한다."""

    def __init__(self, rows, active_rows=None, active_exc=None):
        self.rows = rows
        self.active_rows = list(active_rows or [])
        self.active_exc = active_exc
        self.queries = []

    async def execute(self, q):
        self.queries.append(q)
        if len(self.queries) == 1:
            rows = self.rows
        else:
            if self.active_exc is not None:
                raise self.active_exc
            rows = self.active_rows
        return SimpleNamespace(scalars=lambda: SimpleNamespace(all=lambda: rows))


def _th_row(**kw):
    base = dict(id=433, stock_code="005380", order_type="BUY", order_id="0000901",
                order_quantity=3, order_price=Decimal("337177"), total_amount=Decimal("1011531"),
                net_profit_loss=None, commission=None, strategy_type="swing",
                entry_reason="평균회귀 RSI 과매도")
    base.update(kw)
    return SimpleNamespace(**base)


def _at(id_, name, reason):
    return SimpleNamespace(id=id_, strategy_name=name, entry_reason=reason)


def _patch_exec(monkeypatch, result):
    m = AsyncMock(return_value=result)
    stub = SimpleNamespace(get_order_execution=m)
    monkeypatch.setattr(te, "korea_investment_api", stub)
    mod = sys.modules.get("app.services.korea_investment_api")
    if mod is not None:
        monkeypatch.setattr(mod, "korea_investment_api", stub, raising=False)
    return m


_FILLED = {"ok": True, "found": True, "filled_qty": 3, "filled_price": 335500.0, "remaining_qty": 0}


@pytest.fixture
def env(monkeypatch):
    calls = {"th": [], "at": []}

    async def _th(tid, fields):
        calls["th"].append((tid, dict(fields)))

    async def _at_up(data):
        calls["at"].append(dict(data))

    monkeypatch.setattr(dbw, "enqueue_update_trade_history", _th)
    monkeypatch.setattr(dbw, "enqueue_active_trade_upsert", _at_up)
    fs._attempts.clear()
    fs._queued_ids.clear()
    fs._last_stats.update(day="", candidates=0, exhausted=0, queued=0, exhausted_warned=0)
    te._entry_fill_adjusted_orders.clear()
    return calls


# ── 1. 대상 행 확정 (순수 함수) ───────────────────────────────────────────────
def test_resolve_skips_synced_and_picks_matching_swing():
    rows = [_at(217, "synced", "외부 유입 포지션 — 동기화 시 자동 등록"),
            _at(216, "swing", "평균회귀 RSI 과매도")]
    assert fs.resolve_entry_fill_target_v2(_th_row(), rows) == ("swing", "ok")


def test_resolve_only_synced_active_returns_none():
    # 운영 005380: 스윙 행 closed, synced 행만 active — 구 경로는 synced 행을 보정했다
    rows = [_at(217, "synced", "외부 유입 포지션 — 동기화 시 자동 등록")]
    name, why = fs.resolve_entry_fill_target_v2(_th_row(), rows)
    assert name is None and "없음" in why


def test_resolve_legacy_seeded_swing_name_excluded_by_reason():
    # 7abbb34 이전 시딩은 strategy_name='swing' — entry_reason 접두사로 걸러야 한다
    row = _th_row(entry_reason="외부 유입 포지션 — 동기화 시 자동 등록")
    rows = [_at(10, "swing", "외부 유입 포지션 — 동기화 시 자동 등록")]
    assert fs.resolve_entry_fill_target_v2(row, rows)[0] is None


def test_resolve_dual_strategy_newer_row_does_not_steal():
    rows = [_at(300, "swing", "평균회귀 RSI 과매도"),
            _at(301, "MOMENTUM_BREAKOUT", "20일 신고가 돌파")]
    assert fs.resolve_entry_fill_target_v2(_th_row(), rows)[0] == "swing"
    row_m = _th_row(entry_reason="20일 신고가 돌파")
    assert fs.resolve_entry_fill_target_v2(row_m, rows)[0] == "MOMENTUM_BREAKOUT"


def test_resolve_same_name_newer_unmatched_row_returns_none():
    # db_worker 는 같은 전략명 중 id 최대 행을 고른다 — 그 행이 이 매수 행이 아니면 보정 금지
    rows = [_at(300, "swing", "평균회귀 RSI 과매도"), _at(305, "swing", "다른 사유")]
    name, why = fs.resolve_entry_fill_target_v2(_th_row(), rows)
    assert name is None and "최신" in why


def test_resolve_ambiguous_names_or_missing_reason_returns_none():
    rows = [_at(300, "swing", "평균회귀 RSI 과매도"),
            _at(301, "PULLBACK_BUY", "평균회귀 RSI 과매도")]
    assert fs.resolve_entry_fill_target_v2(_th_row(), rows)[0] is None
    assert fs.resolve_entry_fill_target_v2(_th_row(entry_reason=None), rows)[0] is None
    # strategy_name NULL 행은 동등 필터로 지정할 수 없다
    assert fs.resolve_entry_fill_target_v2(_th_row(), [_at(1, None, "평균회귀 RSI 과매도")])[0] is None
    # DT 행은 대상이 아니다
    assert fs.resolve_entry_fill_target_v2(
        _th_row(), [_at(1, dbw.DT_STRATEGY_NAMES[0], "평균회귀 RSI 과매도")])[0] is None


# ── 1-b. sweep 종단 ──────────────────────────────────────────────────────────
def test_sweep_adjusts_exact_strategy_name_not_latest_non_dt(monkeypatch, env):
    _patch_exec(monkeypatch, _FILLED)
    db = _FakeDB([_th_row()], [_at(217, "synced", "외부 유입 포지션 — 동기화 시 자동 등록"),
                               _at(216, "swing", "평균회귀 RSI 과매도")])
    assert _run(fs.sweep_unconfirmed_fills_v2(db)) == 1
    (at,) = env["at"]
    assert at["strategy_name"] == "swing"
    assert "strategy_name_not" not in at and "strategy_names" not in at
    assert at["entry_fill_adjust"] == {"order_price": 337177.0, "filled_price": 335500.0,
                                       "filled_qty": 3}
    # ActiveTrade 조회는 비-DT + active 조건
    sql = str(db.queries[1].compile(dialect=postgresql.dialect()))
    assert "active_trades" in sql and "NOT IN" in sql.upper()


def test_sweep_no_target_updates_history_only_and_warns(monkeypatch, env, caplog):
    _patch_exec(monkeypatch, _FILLED)
    db = _FakeDB([_th_row()], [_at(217, "synced", "외부 유입 포지션 — 동기화 시 자동 등록")])
    with caplog.at_level(logging.WARNING, logger=fs.logger.name):
        assert _run(fs.sweep_unconfirmed_fills_v2(db)) == 1
    assert len(env["th"]) == 1 and env["at"] == []
    assert any("진입가 보정 생략" in r.getMessage() for r in caplog.records)


def test_sweep_active_lookup_failure_no_adjust(monkeypatch, env, caplog):
    _patch_exec(monkeypatch, _FILLED)
    db = _FakeDB([_th_row()], active_exc=RuntimeError("db down"))
    with caplog.at_level(logging.WARNING, logger=fs.logger.name):
        assert _run(fs.sweep_unconfirmed_fills_v2(db)) == 1
    assert len(env["th"]) == 1 and env["at"] == []
    assert any("조회 실패" in r.getMessage() and r.levelno >= logging.WARNING
               for r in caplog.records)


def test_enqueue_without_db_never_adjusts(env):
    ok = _run(fs._enqueue_buy_entry_adjust_v2(_th_row(), 3, 335500.0))
    assert ok is False and env["at"] == []


# ── 2. 미확인 판정 + 지표 ─────────────────────────────────────────────────────
def test_unconfirmed_clause_includes_legacy_order_price_rows():
    sql = str(fs.unconfirmed_fill_clause_v2().compile(dialect=postgresql.dialect())).lower()
    assert "trade_history.executed_price is null" in sql
    assert "trade_history.slippage is null" in sql
    assert "trade_history.executed_price = trade_history.order_price" in sql


def test_sweep_query_uses_unconfirmed_clause(monkeypatch, env):
    _patch_exec(monkeypatch, _FILLED)
    db = _FakeDB([])
    _run(fs.sweep_unconfirmed_fills_v2(db))
    sql = str(db.queries[0].compile(dialect=postgresql.dialect())).lower()
    assert "trade_history.slippage is null" in sql
    assert "trade_history.executed_price = trade_history.order_price" in sql


def test_sweep_legacy_sell_row_recomputed(monkeypatch, env):
    # 레거시: 스윙 SELL 이 체결가=주문가로 기록(slippage NULL) — 체결가로 손익 재계산
    row = _th_row(id=440, order_type="SELL", order_quantity=10, order_price=Decimal("11000"),
                  total_amount=Decimal("110000"), net_profit_loss=Decimal("10000"),
                  commission=None, entry_reason=None)
    _patch_exec(monkeypatch, {"ok": True, "found": True, "filled_qty": 10,
                              "filled_price": 10900.0, "remaining_qty": 0})
    assert _run(fs.sweep_unconfirmed_fills_v2(_FakeDB([row]))) == 1
    (tid, f), = env["th"]
    assert tid == 440 and f["executed_price"] == Decimal("10900.0")
    assert f["slippage"] is not None
    assert float(f["profit_loss_rate"]) == pytest.approx(9.0)
    h = fs.get_fill_sweep_health_v2()
    assert h["candidates"] == 1 and h["queued"] == 1


def test_exhausted_metric_and_warning_throttled(monkeypatch, env, caplog):
    _patch_exec(monkeypatch, {"ok": False})
    monkeypatch.setattr(fs, "FILL_SWEEP_RETRY_INTERVAL_SEC", 0)
    for _ in range(fs.FILL_SWEEP_MAX_ATTEMPTS):
        _run(fs.sweep_unconfirmed_fills_v2(_FakeDB([_th_row()])))
    with caplog.at_level(logging.WARNING, logger=fs.logger.name):
        for _ in range(3):
            _run(fs.sweep_unconfirmed_fills_v2(_FakeDB([])))
    assert fs.get_fill_sweep_health_v2()["exhausted"] == 1
    msgs = [r for r in caplog.records if "시도 소진" in r.getMessage()]
    assert len(msgs) == 1