shuim
d140055e64
- 8회차 HIGH 수정 16건 레드팀 재검증: 12 확인·3 부분·1 미수정 → 전부 보강 - trade_executor/fill_sweep: 늦게 확정된 0체결 BUY → CANCELLED + ActiveTrade close, 부분체결 → 수량 축소(멱등, 재기동 후 TH 재구성) - korea_investment_api: get_balance_all_v2(tr_cont 연속조회, _partial), get_todays_orders_v2(실패 None), 브레이커 청산용 조회 면제(OPEN 중 미집계) - swing/exit_service·mdd_manager: 매도 직전 상한·MDD 축소가 전체 잔고 페이지 사용(§8.5-37), 불명 대기 중 부분익절 skip·쿨다운·pre_hldg 차감, 대기 TTL 15:30+재매수 폐기(대조 1회 선행)(§8.5-38 해소) - swing/scheduler: 청산 TIMED_OUT/SKIPPED/in-flight 시 MDD 축소·진입 보류(예외는 진입만), 3사이클 연속 시 error+텔레그램 - daytrading: 결과 불명 매도 재판정 v3(주문조회 v2 None→보류, 등록 이후 일치 행→accepted, 원장≤0→보류), 재기동 복원 min(주문가능,보유), 잔고 조회 후 일일손실한도 재확인, 단독 실행 Redis 접속 제거 - main: 스트림 무수신 경보 v4(희망 구독 기준), 첫 하트비트 없음 stale(HB_FIRST_BEAT_GRACE_SEC) - 운영: Jenkinsfile health_watch.py 디렉터리 잔존 복구, prod_smoke 사이드카 확인, config 키 12개 선언(동작 무변경) - 진입가 v3(max(AT,PA))는 KIS 이동평균 평단 전제 오류로 레드팀 반증 → 폐기 - CLAUDE.md §3.2 ③-d, §8.5-37/38 갱신, §8.5-39~48 신규(Crisis 손절 소급 §8.5-39 는 사용자 결정 대기) - 유닛 스위트 2781 passed / 0 failed (소켓 차단), 네트워크 시도 0건 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
332 lines
15 KiB
Python
332 lines
15 KiB
Python
"""하드닝 R3 — 스윙 스케줄러 회귀 테스트 (swing/scheduler.py).
|
|
|
|
결함:
|
|
1. 비취소형 단계가 무기한 정체되면 이후 모든 사이클이 _STEP_SKIPPED 로 영구 생략(하드 데드라인 없음)
|
|
2. 텔레그램 경보 전송 실패가 '전송됨'으로 처리되어 재경보 간격 동안 억제, 실패 누계·/health 노출 없음
|
|
3. MDD 단계 전이 예측이 settings 임계를 써서 mdd_manager(-10/-20 리터럴)와 어긋남 + DB 손잡이 무효
|
|
|
|
목은 공통 종단(_send_system_warning_checked_v2 / telegram_notifier.send_message /
|
|
strategy_cfg.aget / MDDManager Redis 쓰기)에 건다. 텔레그램 실전송은 예외 발생형 목으로 막는다.
|
|
"""
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from app.core.config import settings
|
|
|
|
_real_sleep = asyncio.sleep
|
|
|
|
import app.services.swing.scheduler as _sch_mod # noqa: E402
|
|
|
|
_REAL_CHECKED_SEND = _sch_mod._send_system_warning_checked_v2
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def env(monkeypatch):
|
|
import app.services.swing.scheduler as sch
|
|
import app.services.telegram_notifier as tn
|
|
|
|
monkeypatch.setattr(sch, "_inflight_steps", {})
|
|
monkeypatch.setattr(sch, "_inflight_meta", {})
|
|
monkeypatch.setattr(sch, "_alert_sent_at", {})
|
|
monkeypatch.setattr(sch, "_step_timeouts", {})
|
|
monkeypatch.setattr(sch, "_hard_deadline_stats", {"cancels": 0, "exceeded": {}})
|
|
monkeypatch.setattr(sch, "_telegram_send_health", {
|
|
"sent": 0, "failures": 0, "consecutive_failures": 0,
|
|
"last_success_at": None, "last_failure_at": None, "last_error": None,
|
|
})
|
|
monkeypatch.setattr(sch, "_mdd_knob_state", {"mismatches": []})
|
|
monkeypatch.setattr(sch, "_telegram_state", {"enabled": True})
|
|
# 텔레그램 실전송 차단 — 예외 발생형(목킹 누락이 거짓 통과가 되지 않게)
|
|
monkeypatch.setattr(tn.telegram_notifier, "send_message",
|
|
AsyncMock(side_effect=AssertionError("텔레그램 실전송 금지")))
|
|
monkeypatch.setattr(tn.telegram_notifier, "notify_system_warning",
|
|
AsyncMock(side_effect=AssertionError("텔레그램 실전송 금지")))
|
|
notify = AsyncMock(return_value=True)
|
|
monkeypatch.setattr(sch, "_send_system_warning_checked_v2", notify)
|
|
holder = SimpleNamespace(sch=sch, tn=tn, notify=notify, tasks=[])
|
|
yield holder
|
|
for t in list(sch._inflight_steps.values()) + holder.tasks:
|
|
if not t.done():
|
|
t.cancel()
|
|
|
|
|
|
async def _drain(n=10):
|
|
for _ in range(n):
|
|
await _real_sleep(0.002)
|
|
|
|
|
|
def _messages(notify):
|
|
return [c.kwargs.get("message", "") for c in notify.await_args_list]
|
|
|
|
|
|
def _age_out(sch, step):
|
|
"""하드 데드라인을 넘긴 것으로 시작 시각을 과거로 민다."""
|
|
meta = sch._inflight_meta[step]
|
|
meta["started"] -= meta["hard"] + 1.0
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 결함 1 — 하드 데드라인
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stuck_exit_step_is_cancelled_after_hard_deadline_and_restarted(env):
|
|
sch = env.sch
|
|
never = asyncio.Event()
|
|
calls = {"n": 0}
|
|
|
|
async def _factory():
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
await never.wait() # 락 대기처럼 영원히 멈춤
|
|
return ["ok"]
|
|
|
|
assert await sch._guard_detached_v2(_factory, "청산 판정", 0.02) is sch._STEP_TIMED_OUT
|
|
stuck = sch._inflight_steps["청산 판정"]
|
|
# 하드 데드라인 전에는 생략(취소 안 함)
|
|
assert await sch._guard_detached_v2(_factory, "청산 판정", 0.02) is sch._STEP_SKIPPED
|
|
assert not stuck.done()
|
|
|
|
_age_out(sch, "청산 판정")
|
|
res = await sch._guard_detached_v2(_factory, "청산 판정", 0.5)
|
|
assert res == ["ok"] # 같은 사이클에 재시작되어 완료
|
|
assert stuck.cancelled()
|
|
assert sch._hard_deadline_stats["cancels"] == 1
|
|
await _drain()
|
|
assert any("강제 취소" in m for m in _messages(env.notify))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exit_step_not_cancelled_while_shielded_order_in_flight(env):
|
|
"""shield 주문(_sell_and_record_v2)이 살아 있으면 재시작하지 않는다 — 이중 분할매도 방지."""
|
|
sch = env.sch
|
|
never = asyncio.Event()
|
|
|
|
async def _sell_and_record_v2(): # swing/exit_service 의 shield 내부 코루틴 이름
|
|
await never.wait()
|
|
|
|
order_task = asyncio.ensure_future(_sell_and_record_v2())
|
|
env.tasks.append(order_task)
|
|
|
|
async def _factory():
|
|
await asyncio.shield(order_task)
|
|
|
|
assert await sch._guard_detached_v2(_factory, "청산 판정", 0.02) is sch._STEP_TIMED_OUT
|
|
stuck = sch._inflight_steps["청산 판정"]
|
|
_age_out(sch, "청산 판정")
|
|
assert await sch._guard_detached_v2(_factory, "청산 판정", 0.02) is sch._STEP_SKIPPED
|
|
assert not stuck.done() and not order_task.done()
|
|
assert sch._hard_deadline_stats["cancels"] == 0
|
|
await _drain()
|
|
assert any("shield 주문" in m for m in _messages(env.notify))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_order_path_steps_never_cancelled_but_reported_degraded(env):
|
|
sch = env.sch
|
|
never = asyncio.Event()
|
|
|
|
async def _factory():
|
|
await never.wait()
|
|
|
|
for step in ("MDD 축소", "진입"):
|
|
assert await sch._guard_detached_v2(_factory, step, 0.02) is sch._STEP_TIMED_OUT
|
|
_age_out(sch, step)
|
|
assert await sch._guard_detached_v2(_factory, step, 0.02) is sch._STEP_SKIPPED
|
|
assert not sch._inflight_steps[step].done()
|
|
await _drain()
|
|
assert sum("중복 주문 방지" in m for m in _messages(env.notify)) == 2
|
|
|
|
h = sch.get_swing_scheduler_health_v2()
|
|
assert h["status"] == "degraded"
|
|
assert "stuck_steps" in h["degraded_reasons"]
|
|
assert h["stuck_steps"] == sorted(["MDD 축소", "진입"])
|
|
assert h["inflight_step_ages"]["진입"] > 0
|
|
assert h["hard_deadline_exceeded"] == {"MDD 축소": 1, "진입": 1}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_ignoring_task_is_not_restarted(env, monkeypatch):
|
|
sch = env.sch
|
|
monkeypatch.setattr(sch, "_DEFAULT_STEP_CANCEL_GRACE_SEC", 0.02)
|
|
release = asyncio.Event()
|
|
|
|
async def _stubborn():
|
|
while not release.is_set():
|
|
try:
|
|
await _real_sleep(0.005)
|
|
except asyncio.CancelledError:
|
|
continue # 취소 무시
|
|
|
|
assert await sch._guard_detached_v2(_stubborn, "청산 판정", 0.01) is sch._STEP_TIMED_OUT
|
|
_age_out(sch, "청산 판정")
|
|
assert await sch._guard_detached_v2(_stubborn, "청산 판정", 0.01) is sch._STEP_SKIPPED
|
|
assert sch._hard_deadline_stats["cancels"] == 0
|
|
release.set()
|
|
await _drain()
|
|
|
|
|
|
def test_hard_deadline_multiplier_from_settings_floor(env, monkeypatch):
|
|
sch = env.sch
|
|
assert sch._hard_deadline_sec_v2(10.0) == pytest.approx(10.0 * sch._DEFAULT_STEP_HARD_DEADLINE_MULT)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 결함 2 — 텔레그램 전송 실패 추적
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failed_alert_is_retried_counted_and_throttle_rolled_back(env, monkeypatch):
|
|
sch = env.sch
|
|
clock = [1000.0]
|
|
monkeypatch.setattr(sch, "_now_mono_v2", lambda: clock[0])
|
|
monkeypatch.setattr(sch, "_DEFAULT_ALERT_RETRY_BACKOFF_SEC", 0.001)
|
|
env.notify.return_value = False # 비200 응답 등
|
|
assert sch._notify_throttled_v2("exit_failures", "SWING_EXIT", "청산 정지") is True
|
|
await _drain()
|
|
attempts = sch._DEFAULT_ALERT_SEND_RETRIES + 1
|
|
assert env.notify.await_count == attempts
|
|
h = sch.get_swing_scheduler_health_v2()
|
|
assert h["telegram_send_failures"] == 1
|
|
assert h["telegram_consecutive_failures"] == 1
|
|
assert "telegram_send_failing" in h["degraded_reasons"]
|
|
|
|
# 원래 재경보 간격(SWING_ALERT_REPEAT_SEC)이 아니라 짧은 재시도 간격 뒤 다시 보낸다
|
|
clock[0] += sch._DEFAULT_ALERT_FAIL_RETRY_SEC + 1
|
|
env.notify.return_value = True
|
|
assert sch._notify_throttled_v2("exit_failures", "SWING_EXIT", "청산 정지") is True
|
|
await _drain()
|
|
h = sch.get_swing_scheduler_health_v2()
|
|
assert h["telegram_consecutive_failures"] == 0 and h["telegram_sent"] == 1
|
|
assert h["telegram_last_success_age"] == 0.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_successful_alert_keeps_throttle(env, monkeypatch):
|
|
sch = env.sch
|
|
clock = [1000.0]
|
|
monkeypatch.setattr(sch, "_now_mono_v2", lambda: clock[0])
|
|
assert sch._notify_throttled_v2("k", "C", "m") is True
|
|
await _drain()
|
|
clock[0] += sch._DEFAULT_ALERT_FAIL_RETRY_SEC + 1
|
|
assert sch._notify_throttled_v2("k", "C", "m") is False # 성공 후엔 repeat_sec 동안 억제
|
|
assert sch._telegram_send_health["failures"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unconfigured_telegram_is_not_counted_as_failure(env):
|
|
sch = env.sch
|
|
env.notify.return_value = None
|
|
assert sch._notify_throttled_v2("k", "C", "m") is True
|
|
await _drain()
|
|
assert env.notify.await_count == 1
|
|
assert sch._telegram_send_health["failures"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_checked_send_returns_transport_result(env, monkeypatch):
|
|
"""공통 종단: send_message(bool) 결과를 그대로 돌려준다. 미설정이면 전송하지 않고 None."""
|
|
tn = env.tn
|
|
fn = _REAL_CHECKED_SEND
|
|
|
|
monkeypatch.setattr(tn, "_is_enabled", lambda: False)
|
|
assert await fn(category="C", message="m") is None # send_message 목은 예외 발생형 → 미호출 증명
|
|
|
|
monkeypatch.setattr(tn, "_is_enabled", lambda: True)
|
|
send = AsyncMock(return_value=False)
|
|
monkeypatch.setattr(tn.telegram_notifier, "send_message", send)
|
|
assert await fn(category="SWING_EXIT", message="청산 정지") is False
|
|
assert "SWING" in send.await_args.args[0]
|
|
send.return_value = True
|
|
assert await fn(category="C", message="m") is True
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 결함 3 — MDD 임계 정합
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
@pytest.fixture
|
|
def mdd_env(env, monkeypatch):
|
|
from app.services.mdd_manager import MDDManager
|
|
sch = env.sch
|
|
mgr = MDDManager()
|
|
|
|
async def _no_write(key, state):
|
|
return None
|
|
|
|
monkeypatch.setattr(mgr, "_write_state_v2", _no_write)
|
|
monkeypatch.setattr(mgr, "save_to_redis", AsyncMock())
|
|
monkeypatch.setattr(sch, "mdd_manager", mgr)
|
|
monkeypatch.setattr(sch, "_mdd_gate_state", {"prev": None, "pending": 0, "suspect": 0, "last_flow": None})
|
|
monkeypatch.setattr(env.tn.telegram_notifier, "notify_mdd_warning", AsyncMock())
|
|
return sch, mgr
|
|
|
|
|
|
def _resp(total, scts):
|
|
return {"output1": [], "output2": [{"tot_evlu_amt": str(total), "scts_evlu_amt": str(scts)}]}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shallower_setting_does_not_create_pending_without_transition(mdd_env, monkeypatch):
|
|
"""settings 임계가 -8 이어도 mdd_manager 는 -10 에서만 전이 → -9% 구간에서 pending(진입 차단)만
|
|
반복되던 결함. 예측은 실제 전이와 같아야 한다."""
|
|
sch, mgr = mdd_env
|
|
monkeypatch.setattr(settings, "MDD_PAUSE_THRESHOLD", -8.0)
|
|
await sch._mdd_update_guarded_v2(10_000_000.0, _resp(10_000_000, 6_000_000))
|
|
pause = sch._mdd_stage_thresholds_v2()[0]
|
|
total = 10_000_000 * (1 + (pause + 1.0) / 100.0) # 실효 임계보다 1%p 얕은 낙폭(-9%)
|
|
need = sch._DEFAULT_MDD_STAGE_CONFIRM_CYCLES
|
|
for _ in range(need + 2):
|
|
r = await sch._mdd_update_guarded_v2(total, _resp(total, total - 4_000_000))
|
|
assert r["status"] == "applied"
|
|
assert sch._mdd_gate_state["pending"] == 0
|
|
assert mgr.reduction_applied is None and not mgr.is_paused
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prediction_matches_mdd_manager_transition(mdd_env):
|
|
sch, mgr = mdd_env
|
|
await sch._mdd_update_guarded_v2(10_000_000.0, _resp(10_000_000, 6_000_000))
|
|
pause = sch._mdd_stage_thresholds_v2()[0]
|
|
total = 10_000_000 * (1 + (pause - 0.5) / 100.0)
|
|
need = sch._DEFAULT_MDD_STAGE_CONFIRM_CYCLES
|
|
for i in range(need):
|
|
r = await sch._mdd_update_guarded_v2(total, _resp(total, total - 4_000_000))
|
|
assert r["status"] == "applied"
|
|
assert mgr.reduction_applied == "10pct" # 예측한 단계로 실제 전이
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mdd_threshold_knob_mismatch_alerts_and_degrades(env, monkeypatch):
|
|
sch = env.sch
|
|
from app.services.strategy_config_service import strategy_cfg
|
|
vals = {"MDD_PAUSE_THRESHOLD": -8.0, "MDD_FORCE_THRESHOLD": -20.0}
|
|
monkeypatch.setattr(strategy_cfg, "aget", AsyncMock(side_effect=lambda k, d=None: vals[k]))
|
|
monkeypatch.setattr(settings, "MDD_PAUSE_THRESHOLD", sch._MDD_MANAGER_PAUSE_PCT)
|
|
monkeypatch.setattr(settings, "MDD_FORCE_THRESHOLD", sch._MDD_MANAGER_FORCE_PCT)
|
|
mism = await sch._check_mdd_threshold_knobs_v2()
|
|
assert [(m["source"], m["key"]) for m in mism] == [("strategy_cfg", "MDD_PAUSE_THRESHOLD")]
|
|
await _drain()
|
|
assert any("효과 없음" in m for m in _messages(env.notify))
|
|
assert "mdd_threshold_knob_mismatch" in sch.get_swing_scheduler_health_v2()["degraded_reasons"]
|
|
|
|
vals["MDD_PAUSE_THRESHOLD"] = sch._MDD_MANAGER_PAUSE_PCT
|
|
assert await sch._check_mdd_threshold_knobs_v2() == []
|
|
assert sch.get_swing_scheduler_health_v2()["mdd_threshold_knob_mismatch"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mdd_knob_db_failure_does_not_break(env, monkeypatch):
|
|
sch = env.sch
|
|
from app.services.strategy_config_service import strategy_cfg
|
|
monkeypatch.setattr(strategy_cfg, "aget", AsyncMock(side_effect=RuntimeError("db down")))
|
|
monkeypatch.setattr(settings, "MDD_PAUSE_THRESHOLD", -8.0)
|
|
mism = await sch._check_mdd_threshold_knobs_v2()
|
|
assert [(m["source"], m["key"]) for m in mism] == [("settings", "MDD_PAUSE_THRESHOLD")]
|
|
|
|
|
|
from _swing_redis_guard import _swing_redis_guard # noqa: E402,F401 (R9: 단독 실행 시 Redis 실접속 방지)
|