fix: 스윙 자동편입 차단 + DT 처리량 복구 + ATR/양봉 세션 경계 + DLQ 안전

사용자 요구 "스윙으로 편입하는 건 없어야 해" — 시스템이 매수하지 않은 KIS
계좌 잔고가 strategy_type='swing' 이 되는 경로를 전부 닫았다. 2026-09-30
09:26 에 유한양행 8주가 -6.49%/-40,000원 손절된 것이 직접 계기다.

## 스윙 자동편입 차단
- 강등 허용(swing→external). 기존엔 승격만 가능해 유입 잔고가 영구히
  자동 손절 대상이었다. 3중 방어로 정상 포지션 강등을 막는다:
  system_buy_registry(주문 즉시 인메모리) / 연속2회+120초 / db_worker 큐 적체 보류
  ※ 레이스는 이론이 아니다 — _verify_and_sync_execution 이 매수 직후
    _delayed_sync_portfolio(0) 을 지연 0초로 부른다
- 수량 인식 분류(_classify_holding_strategy_type_v2):
  _sys<=0<qty → external / 0<_sys<qty AND _sys==_dt_held → external
  == 을 쓴 이유: 과소·과대 어느 쪽으로 틀려도 swing(관리 유지)으로 실패한다
- _system_qty 를 과대 집계 쪽으로 편향 — 오판이 안전한 방향으로 나게
- external 행에 스윙 추가매수 차단(혼재 발생 자체를 봉쇄, 청산은 통과)
- 근거를 두 벌로 분리: 신규태깅·강등방어는 registry 포함("있다고 보는 게 안전"),
  승격·ActiveTrade 교정은 durable("없다고 보는 게 안전"). 안전 방향이 반대다
- 정합 루프가 external/혼재 행을 교정하지 않게(전제 "active 1행 = 전량 대표"를
  직접 검사). 미교정 시 ActiveTrade 가 부풀어 스윙 청산이 유입분까지 판다(§3.5)
- status 허용목록 → 거부목록(FAILED 만 제외). 진짜 구멍은 FILLED 가 아니라
  PENDING 이었다(체결확인 실패 시 영구 PENDING, 주식은 보유 중)
- SELL 중복제거 키에 nullif+trim — order_id 는 NULL 이 아니라 '' 로 저장된다
  (build_kis_order_id 가 미발견 시 "" 반환). coalesce 만으로는 그룹이 붕괴해
  SELL 이 과소 집계되고 유입 행이 영구히 swing 이 된다

## DT 처리량
- 감시종목 20 → 30 (DAY_TRADE_WATCHLIST_SIZE, 예산 상한 32에서 마진)
- 선조회 동시성 4→8, per-stock 45초, 단계 데드라인 120초 신설
  (120 < 150 _SCAN_TIME_BUDGET < 180 _guard). 오후는 종목당 32초가
  타임아웃 20초를 넘겨 4~7종목만 스캔되던 포화 상태였다 — 약 9배 복구
- 리스크 노출 불변: 동시 5종목·종목당 6%·DT 30% 무변경. 기회만 증가

## 세션 경계 (ATR·양봉)
- _atr() 이 내부에서 세션을 끊는다. 워밍업 시계열의 당일 첫 봉 TR =
  오버나이트 갭이라 갭 3%에서 ATR 1.50배, 10%에서 2.75배 부풀었고
  클램프 없는 target=현재가+1.5×ATR 을 직접 왜곡했다
  → 진입 ATR == 청산 트레일링 ATR 보장
  ※ 부작용 승인됨: 09:30~10:10 은 ATR 부재로 폴백 -3.5% 가 실효 손절폭
- 양봉 판정이 형성 중 버킷을 보던 것 → 마지막 완성봉(거래량 게이트와 같은 봉)
- 1.5R 하한 보장(두 분기). risk=max(1.5×ATR,2%) 라 저변동 종목은
  R:R 0.21 까지 떨어졌다 — §7 "1.5R 분할" 규약 복귀, RSI_BB 와 동일

## DLQ 안전
- replay_dlq 나이 제한: 상태 2시간+같은날짜 / 기록 24시간 / close 는
  충돌판정만(7일). "닫아라"는 지시는 시간이 지나도 거짓이 되지 않는다 —
  격리하면 반대로 유령 active 행이 남는다(§8.4.1 129개와 같은 형태)
- 상태 충돌 확인(유령 생성 위험 시 expired/ 격리, 삭제 안 함)
- DB 장애 시 단락 — 없으면 500파일 × 접속타임아웃만큼 기동이 정지한다

## 진단·테스트
- 진입 reason 에 [진단 ATR=…, 갭=…], 탈락 후보 RSI·갭 집계 로그(하루 ~60줄)
- 잔고조회 실패 debug→warning (available_cash=0 → 모든 진입 조용히 차단)
- 벽시계 의존 테스트 재발 방지: autouse 가드(시각 미고정 시 AssertionError)
  + 호출부 전수 테스트
- 소스 문자열 매칭 단언을 동작 검증으로 전환(값·분기 판정 한정)

테스트 848 → 865 passed / 0 failed. 전략 임계값 변경 0건. alembic 불필요.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-30 16:50:54 +09:00
parent b5e0d6a9bb
commit 9de0715165
19 changed files with 4519 additions and 126 deletions

View File

@@ -16,6 +16,7 @@
추가: F2(ORB 전일 캔들 혼입 차단), F3(종료 drain 이 포지션 추적 계열 커밋),
F4(DLQ 재처리가 큐 경유).
"""
import asyncio
from datetime import date, datetime, time as dtime, timedelta
from unittest.mock import AsyncMock
@@ -382,12 +383,15 @@ async def test_replay_dlq_requeues_and_moves_files(tmp_path, monkeypatch):
import json as _json
good = dlq / "20260928_100000_upsert_active_trade_abcd1234.json"
# failed_at 은 _save_to_dlq 가 항상 기록한다. C2(나이 제한) 이후 상태 덮어쓰기
# 계열은 최근 것만 재큐잉되므로 '방금 실패'로 둔다(나이 정책 자체는 전용 테스트에서 검증).
good.write_text(_json.dumps({
"task": {"type": "upsert_active_trade",
"data": {"stock_code": "005930", "quantity": 10,
"entry_price": 70000},
"_retry_count": 3},
"error": "shutdown_drain",
"failed_at": datetime.now().isoformat(),
}), encoding="utf-8")
broken = dlq / "20260928_100001_upsert_portfolio_deadbeef.json"
broken.write_text("{not json", encoding="utf-8")
@@ -399,6 +403,11 @@ async def test_replay_dlq_requeues_and_moves_files(tmp_path, monkeypatch):
monkeypatch.setattr(db_worker, "enqueue_db_write", _fake_enqueue)
async def _no_conflict(task, entry):
return False, ""
monkeypatch.setattr(db_worker, "_dlq_state_conflict", _no_conflict)
count = await db_worker.replay_dlq()
assert count == 1
@@ -512,9 +521,69 @@ def test_scan_time_budget_below_guard_timeout():
from app.services.daytrading import scheduler as sched
assert es._SCAN_TIME_BUDGET < sched.STEP_TIMEOUT
src = __import__("inspect").getsource(es.scan_and_enter)
assert "_scan_offset" in src, "라운드로빈 오프셋 없음"
assert "미스캔" in src, "잘린 종목 수를 로그에 남겨야 한다"
@pytest.mark.asyncio
async def test_truncated_scan_rotates_start_offset_next_cycle(monkeypatch, caplog):
"""[M-5 동작 검증] 예산에 걸려 잘리면 다음 사이클은 다른 종목부터 스캔한다.
소스에 '_scan_offset' 이 있는지 보는 대신, 두 사이클을 실제로 돌려 **스캔된 종목
순서**가 회전하는지 본다. 잘린 종목 수가 로그에 남는지도 함께 확인한다.
"""
import logging
from app.services.daytrading import entry_service as es
from app.services.daytrading.position_manager import day_trading_state
scanned_order = []
async def slow_try_enter(code, name, ta, ac, is_lunch_break=False):
scanned_order.append(code)
await asyncio.sleep(0.05) # 예산을 넘기도록 느리게
return None
async def fake_prefetch(codes):
return len(codes)
monkeypatch.setattr(es, "_try_enter", slow_try_enter)
monkeypatch.setattr(es, "_prefetch_minute_charts", fake_prefetch)
monkeypatch.setattr(es, "_SCAN_TIME_BUDGET", 0.04)
monkeypatch.setattr(es.mdd_manager, "is_short_term_strategy_enabled",
AsyncMock(return_value=True))
monkeypatch.setattr(es.mdd_manager, "can_open_new_position",
AsyncMock(return_value=True))
monkeypatch.setattr(es, "is_daytrading_allowed_by_playbook", lambda *a, **k: True)
monkeypatch.setattr(es, "check_market_regime",
AsyncMock(return_value={"regime": "neutral"}))
monkeypatch.setattr(es.korea_investment_api, "get_balance", AsyncMock(
return_value={"output2": [{"tot_evlu_amt": "100000000",
"prvs_rcdl_excc_amt": "50000000"}]}))
monkeypatch.setattr(es.korea_investment_api, "get_buyable_cash",
AsyncMock(return_value=50_000_000.0))
# 벽시계 의존 제거 — 컷오프(15:00) 이후에 돌리면 조기 리턴해 빈 리스트가 된다
_freeze_entry_window(monkeypatch, es)
day_trading_state.positions.clear()
day_trading_state.closed_today.clear()
day_trading_state.daily_pnl = 0.0
day_trading_state.last_total_assets = 100_000_000.0
es._scan_offset = 0
es.set_watchlist([{"code": f"00000{i}", "name": f"종목{i}"} for i in range(4)])
with caplog.at_level(logging.WARNING):
await es.scan_and_enter()
first_cycle = list(scanned_order)
scanned_order.clear()
await es.scan_and_enter()
second_cycle = list(scanned_order)
assert first_cycle and second_cycle
assert first_cycle[0] != second_cycle[0], (
f"두 사이클이 같은 종목부터 시작했다 ({first_cycle[0]}) — 뒤쪽 종목이 영구 소외된다"
)
assert any("미스캔" in r.getMessage() for r in caplog.records), \
"잘린 종목 수가 로그에 남아야 한다"
es.set_watchlist([])
es._scan_offset = 0
# ── F7: 거래량 게이트 — 형성 중 봉이 아니라 완성봉끼리 비교 ─────────────────
@@ -672,15 +741,61 @@ async def test_watchlist_refresh_preserves_orb_holders(monkeypatch):
es.set_watchlist([])
def test_watchlist_uses_latest_analysis_date_not_today_only():
"""e0178aa(2026-09-22)의 `max(analysis_date <= today)` 완화가 살아 있는지."""
import inspect
@pytest.mark.asyncio
async def test_watchlist_uses_latest_analysis_date_not_today_only(monkeypatch):
"""[M-5 동작 검증] 당일 분석이 없어도 최근 분석일 결과로 watchlist 가 채워진다.
e0178aa(2026-09-22) 완화. 소스 문자열이 아니라 **결과**로 본다: 조회 결과에서
최신 분석일이 어제여도 종목이 채워져야 한다(아니면 09:00 ORB 수집이 0건이 된다).
"""
from app.services.daytrading import entry_service as es
src = inspect.getsource(es.populate_daytrading_watchlist)
assert "func.max(StockAnalysis.analysis_date)" in src
assert "StockAnalysis.analysis_date <= today" in src
assert "StockAnalysis.analysis_date == latest_date" in src
yesterday = date.today() - timedelta(days=1)
class _Row:
def __init__(self, code):
self.stock_code = code
self.stock_name = f"종목{code}"
self.score = 5
class _Res:
def __init__(self, rows=None, scalar=None):
self._rows = rows or []
self._scalar = scalar
def all(self):
return self._rows
def scalar_one_or_none(self):
return self._scalar
class _Session:
def __init__(self):
self._n = 0
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
async def execute(self, *a, **k):
self._n += 1
if self._n == 1:
return _Res(scalar=yesterday) # max(analysis_date <= today) = 어제
return _Res(rows=[_Row("005930"), _Row("000660")])
import app.core.database as dbmod
monkeypatch.setattr(dbmod, "AsyncSessionLocal", lambda: _Session())
from app.services.daytrading.position_manager import day_trading_state
day_trading_state.orb_cache.clear()
es.set_watchlist([])
count = await es.populate_daytrading_watchlist()
assert count == 2, "당일 분석이 없으면 watchlist 가 비어 ORB 수집이 0건이 된다"
assert set(es.get_watchlist()) == {"005930", "000660"}
es.set_watchlist([])
def test_scheduler_populates_watchlist_before_orb_collection():
@@ -835,7 +950,10 @@ async def test_daily_guard_redis_failure_is_not_silent(monkeypatch, caplog):
def test_daily_guard_save_is_wired_into_exit_path():
"""청산 경로가 변경 직후 저장을 호출하는지 (배선 누락 방지)."""
"""청산 경로가 변경 직후 저장을 호출하는지 (배선 누락 방지).
[M-5] 배선·순서 확인 전용. 저장/복원 동작은 별 테스트가 실제 호출로 검증한다.
"""
import inspect
from app.services.daytrading import exit_service as es
from app.services.daytrading import scheduler as sched
@@ -1232,7 +1350,13 @@ def test_candle_cache_ttl_covers_one_cycle():
# ── M4: swing 청산 경로도 일일 안전장치를 저장한다 ──────────────────────────
def test_swing_exit_persists_daily_guards():
"""스윙 손절이 등록한 closed_today 가 재기동 후에도 유지되어야 한다."""
"""스윙 손절이 등록한 closed_today 가 재기동 후에도 유지되어야 한다.
[M-5] 이것은 **배선 확인**이다(값·분기 판정이 아니다). 저장 동작 자체는
test_daily_guards_restored_on_same_day_restart / test_mdd_reduction_persists_daily_guards
가 실제 호출로 검증한다. 이 경로는 대형 함수 내부의 깊은 분기라 동작 검증에
필요한 목이 과도해지므로, 호출이 남아 있는지만 고정한다.
"""
import inspect
from app.services.swing import exit_service as ses
@@ -1274,8 +1398,10 @@ async def test_replay_dlq_restores_types_and_task_commits(tmp_path, monkeypatch)
"strategy_name": "day_trade",
},
}
degraded = _json.loads(_json.dumps({"task": original, "error": "shutdown_drain"},
default=str))
degraded = _json.loads(_json.dumps(
{"task": original, "error": "shutdown_drain",
"failed_at": _dt.now().isoformat()}, # C2: 나이 제한 통과용(방금 실패)
default=str))
assert isinstance(degraded["task"]["data"]["entry_price"], str)
(dlq / "20260928_103000_upsert_active_trade_aaaa1111.json").write_text(
_json.dumps(degraded), encoding="utf-8"
@@ -1288,6 +1414,11 @@ async def test_replay_dlq_restores_types_and_task_commits(tmp_path, monkeypatch)
monkeypatch.setattr(db_worker, "enqueue_db_write", _fake_enqueue)
async def _no_conflict(task, entry):
return False, ""
monkeypatch.setattr(db_worker, "_dlq_state_conflict", _no_conflict)
assert await db_worker.replay_dlq() == 1
data = enqueued[0]["data"]
assert isinstance(data["entry_price"], Decimal), "Decimal 이 문자열로 남았다"
@@ -1318,7 +1449,7 @@ async def test_replay_dlq_restores_date_for_bulk_rows(tmp_path, monkeypatch):
_json.dumps({"task": {
"type": "bulk_upsert_stock_analysis",
"data": [{"stock_code": "005930", "analysis_date": "2026-09-28"}],
}}), encoding="utf-8")
}, "failed_at": datetime.now().isoformat()}), encoding="utf-8")
enqueued = []
@@ -1400,5 +1531,440 @@ def test_scan_clears_cache_before_prefetch():
import inspect
from app.services.daytrading import entry_service as es
src = inspect.getsource(es.scan_and_enter)
# autouse 가드가 모듈 속성을 감싸므로 원본을 풀어서 본다
_fn = getattr(es.scan_and_enter, "__wrapped__", es.scan_and_enter)
src = inspect.getsource(_fn)
assert src.index("_candle_cache.clear()") < src.index("_prefetch_minute_charts(")
# ── C2(2026-09-30): replay_dlq 나이 제한 + 상태 충돌 격리 ───────────────────
def _write_dlq(dlq_dir, task_type, data, failed_at, name=None):
import json as _json
name = name or f"20260930_100000_{task_type}_{abs(hash(str(data))) % 10**8:08x}.json"
(dlq_dir / name).write_text(_json.dumps({
"task": {"type": task_type, "data": data},
"error": "shutdown_drain",
"failed_at": failed_at.isoformat() if hasattr(failed_at, "isoformat") else failed_at,
}), encoding="utf-8")
return dlq_dir / name
@pytest.fixture
def dlq_env(tmp_path, monkeypatch):
"""DLQ 디렉터리 3종을 tmp 로 돌리고 enqueue 를 캡처한다."""
from app.services import db_worker
dlq = tmp_path / "dlq"
dlq.mkdir()
monkeypatch.setattr(db_worker, "DLQ_DIR", dlq)
monkeypatch.setattr(db_worker, "DLQ_PROCESSED_DIR", dlq / "processed")
monkeypatch.setattr(db_worker, "DLQ_EXPIRED_DIR", dlq / "expired")
enqueued = []
async def _fake_enqueue(task):
enqueued.append(task)
monkeypatch.setattr(db_worker, "enqueue_db_write", _fake_enqueue)
# 상태 충돌 확인은 기본적으로 '충돌 없음' (개별 테스트가 필요 시 덮는다)
async def _no_conflict(task, entry):
return False, ""
monkeypatch.setattr(db_worker, "_dlq_state_conflict", _no_conflict)
return dlq, enqueued
@pytest.mark.asyncio
async def test_replay_dlq_quarantines_stale_state_task(dlq_env, caplog):
"""오래된 upsert_active_trade 는 재큐잉하지 않고 expired/ 로 격리된다.
9/25 drain 으로 빠진 payload 가 그 사이 전량 청산된 뒤 9/30 기동에 재큐잉되면,
핸들러가 active 행을 못 찾고 필수 키가 다 있으니 **새 active 행을 INSERT** 한다
→ §8.4.1 의 '9/23 유령 행 129개' 실패 모드 자동화.
"""
import logging
from datetime import datetime as _dt, timedelta as _td
from app.services import db_worker
dlq, enqueued = dlq_env
stale = _write_dlq(dlq, "upsert_active_trade",
{"stock_code": "005930", "quantity": 10, "entry_price": 70000},
_dt.now() - _td(days=5), name="20260925_100000_upsert_active_trade_a1.json")
with caplog.at_level(logging.WARNING):
assert await db_worker.replay_dlq() == 0
assert enqueued == [], "오래된 상태 태스크가 재큐잉됐다"
assert not stale.exists()
assert (dlq / "expired" / stale.name).exists(), "증거 보존을 위해 격리(삭제 금지)"
msgs = " ".join(r.getMessage() for r in caplog.records)
assert "격리" in msgs and "upsert_active_trade" in msgs and "나이" in msgs
@pytest.mark.asyncio
async def test_replay_dlq_requeues_recent_state_task(dlq_env):
"""최근(2시간 이내, 같은 날) 상태 태스크는 정상 재큐잉된다."""
from datetime import datetime as _dt, timedelta as _td
from app.services import db_worker
dlq, enqueued = dlq_env
recent = _write_dlq(dlq, "upsert_active_trade",
{"stock_code": "005930", "quantity": 10, "entry_price": 70000},
_dt.now() - _td(minutes=3))
assert await db_worker.replay_dlq() == 1
assert len(enqueued) == 1 and enqueued[0]["type"] == "upsert_active_trade"
assert (dlq / "processed" / recent.name).exists()
@pytest.mark.asyncio
async def test_replay_dlq_record_types_allow_longer_age(dlq_env):
"""기록 계열은 12시간 전도 허용, 48시간 전은 격리 (타입별 차등)."""
from datetime import datetime as _dt, timedelta as _td
from app.services import db_worker
dlq, enqueued = dlq_env
ok = _write_dlq(dlq, "insert_trade_history", {"stock_code": "005930"},
_dt.now() - _td(hours=12), name="a_insert_trade_history_1.json")
too_old = _write_dlq(dlq, "insert_trade_history", {"stock_code": "000660"},
_dt.now() - _td(hours=48), name="b_insert_trade_history_2.json")
assert await db_worker.replay_dlq() == 1
assert len(enqueued) == 1
assert enqueued[0]["data"]["stock_code"] == "005930"
assert (dlq / "processed" / ok.name).exists()
assert (dlq / "expired" / too_old.name).exists()
@pytest.mark.asyncio
async def test_replay_dlq_state_task_without_failed_at_is_quarantined(dlq_env):
"""failed_at 이 없으면 상태 계열은 보수적으로 격리, 기록 계열은 허용."""
from app.services import db_worker
dlq, enqueued = dlq_env
st = _write_dlq(dlq, "upsert_portfolio", {"stock_code": "005930", "quantity": 5},
"", name="a_upsert_portfolio_x.json")
rec = _write_dlq(dlq, "insert_trade_journal", {"stock_code": "005930"},
"", name="b_insert_trade_journal_y.json")
assert await db_worker.replay_dlq() == 1
assert len(enqueued) == 1 and enqueued[0]["type"] == "insert_trade_journal"
assert (dlq / "expired" / st.name).exists()
assert (dlq / "processed" / rec.name).exists()
@pytest.mark.asyncio
async def test_replay_dlq_quarantines_on_state_conflict(tmp_path, monkeypatch, caplog):
"""나이를 통과해도 현재 상태와 충돌하면 격리한다 (유령 행 차단)."""
import logging
from datetime import datetime as _dt, timedelta as _td
from app.services import db_worker
dlq = tmp_path / "dlq"
dlq.mkdir()
monkeypatch.setattr(db_worker, "DLQ_DIR", dlq)
monkeypatch.setattr(db_worker, "DLQ_PROCESSED_DIR", dlq / "processed")
monkeypatch.setattr(db_worker, "DLQ_EXPIRED_DIR", dlq / "expired")
enqueued = []
async def _fake_enqueue(task):
enqueued.append(task)
monkeypatch.setattr(db_worker, "enqueue_db_write", _fake_enqueue)
f = _write_dlq(dlq, "upsert_active_trade",
{"stock_code": "005930", "quantity": 10, "entry_price": 70000},
_dt.now() - _td(minutes=5))
# active 행 없음 + Portfolio 수량 0 → 유령 생성 위험
class _Sess:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
async def execute(self, *a, **k):
class _S:
def first(self_inner):
return None
class _R:
def scalars(self_inner):
return _S()
return _R()
import app.services.db_worker as dbw
monkeypatch.setattr(dbw, "AsyncSessionLocal", lambda: _Sess())
with caplog.at_level(logging.WARNING):
assert await db_worker.replay_dlq() == 0
assert enqueued == []
assert (dlq / "expired" / f.name).exists()
assert "충돌" in " ".join(r.getMessage() for r in caplog.records)
@pytest.mark.asyncio
async def test_dlq_state_conflict_degrades_to_age_only_on_db_failure(monkeypatch):
"""DB 판정 실패는 '충돌 아님'으로 돌려 기동을 막지 않는다."""
from app.services import db_worker
def _boom():
raise RuntimeError("db down")
monkeypatch.setattr(db_worker, "AsyncSessionLocal", _boom)
conflict, reason = await db_worker._dlq_state_conflict(
{"type": "upsert_active_trade", "data": {"stock_code": "005930"}},
{"failed_at": "2026-09-30T10:00:00"},
)
assert conflict is False and reason == ""
# ── M-2: DB 장애 시 충돌 판정 단락 (기동 지연 유계화) ───────────────────────
@pytest.mark.asyncio
async def test_conflict_check_short_circuits_after_first_db_failure(tmp_path, monkeypatch, caplog):
"""DB 가 안 뜬 상태에서 파일마다 접속을 재시도하지 않는다.
replay_dlq() 는 lifespan 에서 await 되므로 기동을 막는다. 단락이 없으면
(파일 수 × 접속 타임아웃)만큼 기동이 정지한다. 단락 후에도 **나이 제한은 계속
적용**되어야 한다.
"""
import logging
from datetime import datetime as _dt, timedelta as _td
from app.services import db_worker
dlq = tmp_path / "dlq"
dlq.mkdir()
monkeypatch.setattr(db_worker, "DLQ_DIR", dlq)
monkeypatch.setattr(db_worker, "DLQ_PROCESSED_DIR", dlq / "processed")
monkeypatch.setattr(db_worker, "DLQ_EXPIRED_DIR", dlq / "expired")
enqueued = []
async def _fake_enqueue(task):
enqueued.append(task)
monkeypatch.setattr(db_worker, "enqueue_db_write", _fake_enqueue)
attempts = []
def _boom_session():
attempts.append(1)
raise RuntimeError("connection refused")
monkeypatch.setattr(db_worker, "AsyncSessionLocal", _boom_session)
# 최근 상태 파일 5개 (모두 충돌 판정 대상)
for i in range(5):
_write_dlq(dlq, "upsert_active_trade",
{"stock_code": f"00593{i}", "quantity": 1, "entry_price": 1000},
_dt.now() - _td(minutes=1), name=f"f{i}_upsert_active_trade_{i}.json")
# 오래된 상태 파일 1개 — 단락 후에도 나이 제한으로 격리되어야 한다
old = _write_dlq(dlq, "upsert_active_trade",
{"stock_code": "000660", "quantity": 1, "entry_price": 1000},
_dt.now() - _td(days=3), name="z_upsert_active_trade_old.json")
with caplog.at_level(logging.WARNING):
requeued = await db_worker.replay_dlq()
assert len(attempts) == 1, (
f"DB 접속을 {len(attempts)}회 시도했다 — 첫 실패에서 단락해야 한다"
)
assert requeued == 5, "최근 파일은 나이 제한만으로 재큐잉된다"
assert (dlq / "expired" / old.name).exists(), "단락 후에도 나이 제한은 적용되어야 한다"
# 반복 로그 억제: 실패 warning 은 1회만
fails = [r for r in caplog.records if "상태 충돌 확인 실패" in r.getMessage()]
assert len(fails) == 1, f"충돌 판정 실패 로그가 {len(fails)}회 — 1회만 남겨야 한다"
@pytest.mark.asyncio
async def test_conflict_availability_resets_each_replay(tmp_path, monkeypatch):
"""이전 호출의 단락을 다음 기동이 물려받지 않는다."""
from app.services import db_worker
dlq = tmp_path / "dlq"
dlq.mkdir()
monkeypatch.setattr(db_worker, "DLQ_DIR", dlq)
db_worker._dlq_conflict_check_available = False
assert await db_worker.replay_dlq() == 0 # 파일 없음 → 조기 반환
# 파일이 하나라도 있으면 리셋된다
from datetime import datetime as _dt
_write_dlq(dlq, "insert_trade_journal", {"stock_code": "005930"}, _dt.now())
monkeypatch.setattr(db_worker, "DLQ_PROCESSED_DIR", dlq / "processed")
monkeypatch.setattr(db_worker, "DLQ_EXPIRED_DIR", dlq / "expired")
async def _fake_enqueue(task):
return None
monkeypatch.setattr(db_worker, "enqueue_db_write", _fake_enqueue)
await db_worker.replay_dlq()
assert db_worker._dlq_conflict_check_available is True
# ── M-4: close_active_trade 는 오래돼도 재큐잉, 새 포지션에는 걸린다 ────────
@pytest.mark.asyncio
async def test_stale_close_active_trade_is_requeued(dlq_env):
"""야간 재기동(2시간 초과)에도 close 는 재큐잉된다.
격리하면 이미 청산된 포지션의 ActiveTrade 가 active 로 영구 잔존해
§8.4.1 '유령 행' 과 같은 형태가 되고, 그 잔량이 유입 잔고 격리의 '시스템 보유'
하한으로 흘러들어 다른 안전장치를 무력화한다.
"""
from datetime import datetime as _dt, timedelta as _td
from app.services import db_worker
dlq, enqueued = dlq_env
f = _write_dlq(dlq, "close_active_trade",
{"stock_code": "005930", "strategy_name": "day_trade"},
_dt.now() - _td(days=3), name="a_close_active_trade_old.json")
assert await db_worker.replay_dlq() == 1
assert enqueued[0]["type"] == "close_active_trade"
assert (dlq / "processed" / f.name).exists()
@pytest.mark.asyncio
async def test_stale_close_is_quarantined_when_newer_position_exists(tmp_path, monkeypatch):
"""단, 현재 active 행이 failed_at 이후 생성이면(새 포지션) 격리한다."""
from datetime import datetime as _dt, timedelta as _td
from app.services import db_worker
dlq = tmp_path / "dlq"
dlq.mkdir()
monkeypatch.setattr(db_worker, "DLQ_DIR", dlq)
monkeypatch.setattr(db_worker, "DLQ_PROCESSED_DIR", dlq / "processed")
monkeypatch.setattr(db_worker, "DLQ_EXPIRED_DIR", dlq / "expired")
enqueued = []
async def _fake_enqueue(task):
enqueued.append(task)
monkeypatch.setattr(db_worker, "enqueue_db_write", _fake_enqueue)
failed_at = _dt.now() - _td(days=3)
f = _write_dlq(dlq, "close_active_trade",
{"stock_code": "005930", "strategy_name": "day_trade"},
failed_at, name="a_close_active_trade_conflict.json")
newer = _FakeActiveTrade(stock_code="005930", quantity=5)
newer.created_at = _dt.now() - _td(hours=1) # failed_at 이후 = 새 포지션
class _Sess:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
async def execute(self, *a, **k):
rec = newer
class _S:
def first(self_inner):
return rec
class _R:
def scalars(self_inner):
return _S()
return _R()
monkeypatch.setattr(db_worker, "AsyncSessionLocal", lambda: _Sess())
assert await db_worker.replay_dlq() == 0
assert enqueued == []
assert (dlq / "expired" / f.name).exists()
@pytest.mark.asyncio
async def test_close_age_limit_tightens_when_conflict_check_unavailable():
"""충돌 판정이 단락되면 close 허용 나이가 기록 계열 상한으로 내려간다."""
from app.services import db_worker
db_worker._dlq_conflict_check_available = True
assert db_worker._dlq_max_age_seconds("close_active_trade") == \
db_worker.DLQ_REPLAY_MAX_AGE_CLOSE_SECONDS
db_worker._dlq_conflict_check_available = False
assert db_worker._dlq_max_age_seconds("close_active_trade") == \
db_worker.DLQ_REPLAY_MAX_AGE_RECORD_SECONDS
# 상태 계열은 단락과 무관하게 2시간
assert db_worker._dlq_max_age_seconds("upsert_active_trade") == \
db_worker.DLQ_REPLAY_MAX_AGE_STATE_SECONDS
db_worker._dlq_conflict_check_available = True
# ── 재발 방지: 벽시계 의존 차단 (autouse) ───────────────────────────────────
#
# 이 파일에서 `scan_and_enter()` 를 시각 고정 없이 호출해 **두 번** 깨졌다
# (컷오프 15:00 이후 실행 시 조기 리턴 → 빈 결과). 개별 테스트의 주의력에 맡기면
# 새 테스트가 같은 함정에 다시 빠지므로, 호출 자체를 구조적으로 막는다.
#
# 가드는 **현재 시각을 보지 않고 무조건** 실패시킨다. 벽시계를 참고하면 낮에는
# 통과하고 밤에는 깨지는 '운 좋은 통과'가 생겨 재발 방지가 되지 않는다.
@pytest.fixture(autouse=True)
def _require_frozen_clock_for_time_gated_calls(monkeypatch):
"""`scan_and_enter()` 호출 시 `_freeze_entry_window` 적용을 강제한다.
청산 계열(`check_and_exit_positions`/`force_close_all`/`close_stale_positions`)도
내부에서 시각을 읽는다(EOD·시간손절). 이 파일에는 현재 호출부가 없지만, 추가할
때는 같은 방식으로 해당 모듈의 `datetime` 을 고정하고 이 가드를 확장할 것.
"""
from app.services.daytrading import entry_service as es
_real_scan = es.scan_and_enter
async def _guarded_scan(*args, **kwargs):
if es.datetime is datetime:
raise AssertionError(
"scan_and_enter() 를 시각 고정 없이 호출했다 — 진입 윈도우"
"(09:30~DAY_TRADE_ENTRY_CUTOFF) 밖에서는 조기 리턴해 테스트가 시간대에 따라"
" 깨진다. 테스트 안에서 `_freeze_entry_window(monkeypatch, es)` 를 먼저 호출하라."
)
return await _real_scan(*args, **kwargs)
# 원본을 노출해 소스 검사 테스트가 가드가 아닌 실제 함수를 보게 한다
_guarded_scan.__wrapped__ = _real_scan
monkeypatch.setattr(es, "scan_and_enter", _guarded_scan)
def test_wallclock_guard_rejects_unfrozen_scan(monkeypatch):
"""가드 자체가 동작하는지 — 시각 고정 없이 호출하면 실패해야 한다."""
from app.services.daytrading import entry_service as es
with pytest.raises(AssertionError, match="시각 고정"):
asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
es.scan_and_enter()
)
def test_every_scan_call_site_freezes_the_clock():
"""[전수 확인] 이 파일의 scan_and_enter 호출부가 모두 시각을 고정했는지.
가드(autouse)가 런타임에 막지만, 이 테스트는 '어느 테스트가 빠뜨렸는지'를
파일 단위로 드러내 리뷰에서 바로 보이게 한다.
"""
import pathlib
import re
src = pathlib.Path(__file__).read_text(encoding="utf-8")
# 테스트 함수 단위로 쪼개 scan_and_enter 호출과 freeze 호출을 함께 본다
blocks = re.split(r"\n(?=@pytest\.|def test_|async def test_)", src)
offenders = []
for b in blocks:
m = re.search(r"(?:async )?def (test_\w+)", b)
if not m:
continue
name = m.group(1)
if name in ("test_wallclock_guard_rejects_unfrozen_scan",
"test_every_scan_call_site_freezes_the_clock"):
continue
if "es.scan_and_enter()" in b and "_freeze_entry_window(" not in b:
offenders.append(name)
assert not offenders, (
f"시각 고정 없이 scan_and_enter() 를 호출하는 테스트: {offenders} — "
"_freeze_entry_window(monkeypatch, es) 를 추가하라"
)