"""스윙 도메인 4회차 검증 잔여 결함 회귀 테스트 (2026-10-02). L-4 현재가 조회 실패(예외·빈 문자열)여도 대체 가격으로 폴백 손절 판정 L-6 손절률·ATR 배수 합리 범위 검증(config 키) L-3 시간손절 원 진입일 탐색을 마지막 전량 청산 이후 BUY 로 제한 L-7 KRX_EXTRA_HOLIDAYS 파싱 캐시 + 형식 오류 warning 설정값당 1회 규약(CLAUDE.md §4.1): 주문 경로 목은 예외 발생형(_Reached, BaseException). 기대값은 settings.*. """ import logging from datetime import date, 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 class _Reached(BaseException): pass @pytest.fixture(autouse=True) def _reset(monkeypatch): monkeypatch.setattr(mr, "_previous_regime", None) from app.core import log_throttle log_throttle.reset_for_tests() import app.services.swing.exit_service as es monkeypatch.setattr(es, "_last_live_prices", {}) yield def _no_adaptive(monkeypatch): 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) # ═══════════════════════════════════════════════════════════════════════════ # L-4 # ═══════════════════════════════════════════════════════════════════════════ def _harness(monkeypatch, *, holding_price, avg=10000, price_resp=None, price_exc=None, holding_age_sec=60, entry_days_ago=90): import app.services.swing.exit_service as es import app.services.trade_executor as te from app.core.timeutil import now_kst # [H-1/M-1] Portfolio 대체가는 updated_at 기준 나이 상한 안이어야 쓰인다 holding = SimpleNamespace(stock_code="005930", stock_name="T", quantity=10, average_price=avg, current_price=holding_price, market="KRX", updated_at=now_kst() - timedelta(seconds=holding_age_sec)) 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=entry_days_ago), entry_reason=None) monkeypatch.setattr(es, "get_swing_active_trade", AsyncMock(return_value=at)) api = MagicMock() if price_exc is not None: api.get_current_price = AsyncMock(side_effect=price_exc) else: api.get_current_price = AsyncMock(return_value=price_resp) bars = [{"stck_clpr": str(avg), "stck_hgpr": str(avg + 100), "stck_lwpr": str(avg - 100)} for _ in range(30)] api.get_daily_chart = AsyncMock(return_value={"output2": bars}) # [C-1 6회차] 국내 매도 직전 KIS 실보유 확인 — 보유 10주 전부 스윙(DT 없음) api.get_balance = AsyncMock(return_value={"rt_cd": "0", "output1": [ {"pdno": "005930", "hldg_qty": "10", "ord_psbl_qty": "10"}]}) monkeypatch.setattr(es, "korea_investment_api", api) monkeypatch.setattr(es, "calculate_atr", lambda h, l, c: avg * 0.01) monkeypatch.setattr(es, "calculate_rsi", lambda c: 45.0) monkeypatch.setattr(es, "calculate_bollinger", lambda c: (avg * 1.2, avg * 1.1, avg * 0.9)) sig_calls = [] def _sig(**kw): sig_calls.append(kw) return mr.ExitSignal("SELL_PARTIAL", "익절", 0.5) monkeypatch.setattr(es, "check_exit_signal", _sig) placed = [] async def _spo(**kw): placed.append(kw) raise _Reached("sell") monkeypatch.setattr(te, "safe_place_order", _spo) from app.services.daytrading.position_manager import day_trading_state monkeypatch.setattr(day_trading_state, "positions", {}) es._recent_exits.clear() return es, api, placed, sig_calls async def _run(es): try: await es.check_and_execute_exit_v2(MagicMock()) except _Reached: pass @pytest.mark.asyncio @pytest.mark.parametrize("kw", [ dict(price_exc=RuntimeError("KIS down")), dict(price_resp={"output": {"stck_prpr": ""}}), dict(price_resp={"output": {}}), dict(price_resp=None), ]) async def test_l4_price_failure_still_runs_fallback_stop(monkeypatch, kw): _no_adaptive(monkeypatch) es, api, placed, sig_calls = _harness(monkeypatch, holding_price=8000, **kw) # -20% await _run(es) assert len(placed) == 1 and placed[0]["quantity"] == 10 and placed[0]["order_type"] == "SELL" assert sig_calls == [] # 낡은 가격으로 지표 판정(익절)은 하지 않는다 api.get_daily_chart.assert_not_called() @pytest.mark.asyncio async def test_l4_stale_price_never_takes_profit_or_time_stop(monkeypatch): _no_adaptive(monkeypatch) # 90일 보유 + 대체 가격 -1% → 실시세라면 시간손절, 낡은 가격이면 판정 안 함 es, _, placed, sig_calls = _harness(monkeypatch, holding_price=9900, price_exc=RuntimeError("down")) await _run(es) assert placed == [] and sig_calls == [] def _cache(es, price, age_sec): """실시세 캐시 항목을 age_sec 초 전에 기록된 것으로 심는다.""" import time as _t from app.core.timeutil import now_kst es._last_live_prices["005930"] = (float(price), _t.monotonic() - age_sec, now_kst() - timedelta(seconds=age_sec)) @pytest.mark.asyncio async def test_l4_newer_cache_beats_older_portfolio(monkeypatch): """[H-1/M-1 사양 교체] 캐시·Portfolio 중 **더 최신** 값을 쓴다 — 캐시가 최신.""" _no_adaptive(monkeypatch) es, _, placed, _ = _harness(monkeypatch, holding_price=10000, holding_age_sec=300, price_exc=RuntimeError("down")) _cache(es, 8000, 10) await _run(es) assert len(placed) == 1 @pytest.mark.asyncio async def test_l4_newer_portfolio_beats_older_cache(monkeypatch): _no_adaptive(monkeypatch) es, _, placed, _ = _harness(monkeypatch, holding_price=8000, holding_age_sec=10, price_exc=RuntimeError("down")) _cache(es, 10000, 300) # 더 오래된 캐시(보합) — 최신 Portfolio(-20%)를 가리면 안 된다 await _run(es) assert len(placed) == 1 @pytest.mark.asyncio async def test_l4_aged_out_prices_skip_with_error(monkeypatch, caplog): _no_adaptive(monkeypatch) too_old = float(settings.SWING_FALLBACK_PRICE_MAX_AGE_SEC) + 60 es, _, placed, _ = _harness(monkeypatch, holding_price=8000, holding_age_sec=too_old, price_exc=RuntimeError("down")) _cache(es, 8000, too_old) with caplog.at_level(logging.ERROR, logger=es.logger.name): await _run(es) assert placed == [] assert any(r.levelno >= logging.ERROR and "수동 확인" in r.getMessage() for r in caplog.records) @pytest.mark.asyncio async def test_l4_portfolio_without_updated_at_is_not_used(monkeypatch): _no_adaptive(monkeypatch) es, _, placed, _ = _harness(monkeypatch, holding_price=8000, price_exc=RuntimeError("down")) es.get_swing_holdings.return_value[0].updated_at = None await _run(es) assert placed == [] @pytest.mark.asyncio async def test_h1_rebuy_does_not_use_previous_position_price(monkeypatch): """[재현] 예전 보유분 실시세 9,300 이 캐시에 남은 채 같은 종목 재매수(진입 11,000). 시세 조회 1회 실패 → 이전: 9,300 으로 −15% 판정 → 시장가 전량 오손절 + loss_tracker 오염. 이제: 진입 이전 값은 거부, 재매수 이후 Portfolio 평가가(11,000)로 판정 → 손절 없음.""" _no_adaptive(monkeypatch) from app.core.timeutil import now_kst es, _, placed, _ = _harness(monkeypatch, holding_price=11000, avg=11000, holding_age_sec=30, price_exc=RuntimeError("down"), entry_days_ago=0) # 진입(ActiveTrade.created_at) = 2분 전, 캐시 = 5분 전(이전 보유분) import app.services.swing.exit_service as _es at = await _es.get_swing_active_trade(None, "005930") at.created_at = now_kst() - timedelta(seconds=120) _cache(es, 9300, 300) await _run(es) assert placed == [] @pytest.mark.asyncio async def test_h1_rebuy_only_stale_cache_skips(monkeypatch): """재매수 후 Portfolio 도 아직 갱신 전(진입 이전 updated_at)이면 판정 skip.""" _no_adaptive(monkeypatch) from app.core.timeutil import now_kst es, _, placed, _ = _harness(monkeypatch, holding_price=9300, avg=11000, holding_age_sec=300, price_exc=RuntimeError("down"), entry_days_ago=0) at = await es.get_swing_active_trade(None, "005930") at.created_at = now_kst() - timedelta(seconds=120) _cache(es, 9300, 300) await _run(es) assert placed == [] @pytest.mark.asyncio async def test_l4_no_price_at_all_logs_error(monkeypatch, caplog): _no_adaptive(monkeypatch) es, _, placed, _ = _harness(monkeypatch, holding_price=0, price_exc=RuntimeError("down")) with caplog.at_level(logging.ERROR, logger=es.logger.name): await _run(es) assert placed == [] assert any(r.levelno >= logging.ERROR and "005930" in r.getMessage() for r in caplog.records) @pytest.mark.asyncio async def test_l4_live_price_success_is_remembered(monkeypatch): _no_adaptive(monkeypatch) es, _, placed, sig_calls = _harness(monkeypatch, holding_price=1, price_resp={"output": {"stck_prpr": "10500"}}) await _run(es) assert es._last_live_prices["005930"][0] == 10500 # (가격, monotonic, wall) assert sig_calls and sig_calls[0]["close"] == 10500 # 실시세면 지표 판정 정상 수행 # ═══════════════════════════════════════════════════════════════════════════ # L-6 # ═══════════════════════════════════════════════════════════════════════════ def _cfg(monkeypatch, values): import app.services.adaptive_strategy as ad monkeypatch.setattr(mr.strategy_cfg, "get", lambda k, d=None: values.get(k)) monkeypatch.setattr(ad, "get_current_adaptive_params", lambda: SimpleNamespace(fallback_used=True)) @pytest.mark.parametrize("bad", [-0.5, 0.5, -35.0, 500.0]) def test_l6_stop_loss_out_of_range_falls_back(monkeypatch, bad): _cfg(monkeypatch, {"STOP_LOSS_PERCENT": bad}) assert mr.resolve_swing_stop_params_v2(100.0, 10000.0)[0] == -abs(settings.STOP_LOSS_PERCENT) @pytest.mark.parametrize("ok", [-3.0, -settings.SWING_STOP_LOSS_PCT_MIN, -settings.SWING_STOP_LOSS_PCT_MAX]) def test_l6_stop_loss_in_range_accepted(monkeypatch, ok): _cfg(monkeypatch, {"STOP_LOSS_PERCENT": ok}) assert mr.resolve_swing_stop_params_v2(100.0, 10000.0)[0] == -abs(ok) @pytest.mark.parametrize("bad", [0.4, 10.5, -2.0]) def test_l6_atr_mult_out_of_range_falls_back(monkeypatch, bad): _cfg(monkeypatch, {"ATR_STOP_MULTIPLIER": bad}) assert mr.resolve_swing_stop_params_v2(100.0, 10000.0)[1] == float(settings.ATR_STOP_MULTIPLIER) def test_l6_atr_mult_in_range_accepted(monkeypatch): _cfg(monkeypatch, {"ATR_STOP_MULTIPLIER": 1.5}) assert mr.resolve_swing_stop_params_v2(100.0, 10000.0)[1] == 1.5 def test_l6_bounds_come_from_settings(monkeypatch): monkeypatch.setattr(settings, "SWING_STOP_LOSS_PCT_MAX", 5.0) _cfg(monkeypatch, {"STOP_LOSS_PERCENT": -6.0}) assert mr.resolve_swing_stop_params_v2(100.0, 10000.0)[0] == -abs(settings.STOP_LOSS_PERCENT) def test_l6_adaptive_out_of_range_uses_lower_source(monkeypatch): import app.services.adaptive_strategy as ad monkeypatch.setattr(mr.strategy_cfg, "get", lambda k, d=None: None) monkeypatch.setattr(ad, "get_current_adaptive_params", lambda: SimpleNamespace( fallback_used=False, stop_loss_pct=-0.01, stop_loss_high_vol=-0.01, atr_stop_multiplier=50.0)) assert mr.resolve_swing_stop_params_v2(100.0, 10000.0) == ( -abs(settings.STOP_LOSS_PERCENT), float(settings.ATR_STOP_MULTIPLIER)) @pytest.mark.parametrize("mode", ["aggressive", "normal", "defensive", "crisis"]) def test_l6_all_adaptive_modes_pass_validation(monkeypatch, mode): """현행 _MODE_PARAMS 값이 범위 검증에 걸려 조용히 config 로 대체되지 않는다.""" import app.services.adaptive_strategy as ad p = ad._MODE_PARAMS[ad.TradingMode(mode)] if hasattr(ad, "TradingMode") else ad._MODE_PARAMS[mode] monkeypatch.setattr(mr.strategy_cfg, "get", lambda k, d=None: None) monkeypatch.setattr(ad, "get_current_adaptive_params", lambda: SimpleNamespace( fallback_used=False, stop_loss_pct=p["stop_loss_pct"], stop_loss_high_vol=p["stop_loss_high_vol"], atr_stop_multiplier=p["atr_stop_multiplier"])) assert mr.resolve_swing_stop_params_v2(100.0, 10000.0) == ( -abs(p["stop_loss_pct"]), float(p["atr_stop_multiplier"])) # ═══════════════════════════════════════════════════════════════════════════ # L-3 # ═══════════════════════════════════════════════════════════════════════════ class _SeqDB: """execute 호출 순서대로 scalar 값을 돌려주고 문장을 기록한다.""" def __init__(self, *values): self.values = list(values) self.statements = [] async def execute(self, stmt): self.statements.append(stmt) v = self.values.pop(0) if self.values else None return SimpleNamespace(scalar=lambda: v) def _sql(stmt) -> str: return str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True})) _SYNC = "동기화 유입 포지션 (부분매도 시 자동 생성)" @pytest.mark.asyncio async def test_l3_buy_search_bounded_by_last_full_close(): import app.services.swing.exit_service as es created = datetime(2026, 3, 2, 10, 0) last_close = datetime(2026, 2, 1, 15, 0) buy = datetime(2026, 2, 10, 9, 30) db = _SeqDB(last_close, buy) at = SimpleNamespace(created_at=created, entry_reason=_SYNC) basis = await es._resolve_time_stop_entry_v2(db, "005930", at) assert basis.created_at == buy buy_sql = _sql(db.statements[1]) assert "trade_history.created_at > '2026-02-01 15:00:00'" in buy_sql closed_sql = _sql(db.statements[0]) assert "active_trades.status = 'closed'" in closed_sql or "status = 'closed'" in closed_sql @pytest.mark.asyncio async def test_l3_no_buy_after_close_keeps_created_at(): import app.services.swing.exit_service as es at = SimpleNamespace(created_at=datetime(2026, 3, 2, 10, 0), entry_reason=_SYNC) db = _SeqDB(datetime(2026, 2, 1), None) # 전량 청산 이후 BUY 없음 assert await es._resolve_time_stop_entry_v2(db, "005930", at) is at @pytest.mark.asyncio async def test_l3_no_closed_row_has_no_lower_bound(): import app.services.swing.exit_service as es at = SimpleNamespace(created_at=datetime(2026, 3, 2, 10, 0), entry_reason=_SYNC) db = _SeqDB(None, datetime(2026, 1, 5)) basis = await es._resolve_time_stop_entry_v2(db, "005930", at) assert basis.created_at == datetime(2026, 1, 5) assert "trade_history.created_at >" not in _sql(db.statements[1]) @pytest.mark.asyncio async def test_l3_query_failure_keeps_created_at(): import app.services.swing.exit_service as es class _Boom: async def execute(self, stmt): raise RuntimeError("db down") at = SimpleNamespace(created_at=datetime(2026, 3, 2, 10, 0), entry_reason=_SYNC) assert await es._resolve_time_stop_entry_v2(_Boom(), "005930", at) is at # ═══════════════════════════════════════════════════════════════════════════ # L-7 # ═══════════════════════════════════════════════════════════════════════════ def test_l7_extra_holidays_parsed_once_and_warns_once(monkeypatch, caplog): from app.core import market_schedule as ms monkeypatch.setattr(ms, "_extra_holidays_cache", None) monkeypatch.setattr(settings, "KRX_EXTRA_HOLIDAYS", "2026-10-06,not-a-date", raising=False) with caplog.at_level(logging.WARNING, logger=ms.__name__): assert ms.is_krx_holiday(date(2026, 10, 6)) assert ms.is_krx_holiday(date(2026, 10, 6)) ms.count_krx_trading_days_between(date(2026, 9, 1), date(2026, 10, 30)) warns = [r for r in caplog.records if "KRX_EXTRA_HOLIDAYS" in r.getMessage()] assert len(warns) == 1 def test_l7_cache_follows_setting_changes(monkeypatch): from app.core import market_schedule as ms monkeypatch.setattr(ms, "_extra_holidays_cache", None) monkeypatch.setattr(settings, "KRX_EXTRA_HOLIDAYS", "2026-10-06", raising=False) assert ms.is_krx_holiday(date(2026, 10, 6)) monkeypatch.setattr(settings, "KRX_EXTRA_HOLIDAYS", "2026-10-07", raising=False) assert not ms.is_krx_holiday(date(2026, 10, 6)) assert ms.is_krx_holiday(date(2026, 10, 7))