Files
auto-stock/backend/tests/test_daytrading_atr_session_boundary_20260930.py
shuim 9de0715165 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>
2026-09-30 16:50:54 +09:00

571 lines
26 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""C1 회귀 테스트 (2026-09-30): 워밍업 시계열이 진입 ATR 을 오염시키던 결함.
증상: 진입 판정 경로는 지표 워밍업을 위해 전일 봉이 섞인 5분봉을 쓴다
(`get_minute_chart_full_day_v2(include_prev_day=True)`). `true_ranges` 가
`TR_i = max(H-L, |H-C_{i-1}|, |L-C_{i-1}|)` 이라 **당일 첫 봉의 TR = 오버나이트 갭**
이고, `atr_sma` 는 마지막 14개 TR 의 단순평균이므로 당일 5분봉이 14개 쌓이기
전(≈10:10)에는 갭 TR 이 윈도에 남았다. 그 ATR 이 손절·목표가를 직접 만든다.
수정: `day_trading_strategy._atr` 이 내부에서 `_last_session_candles()` 로 세션을 끊는다.
임계값은 하나도 바꾸지 않았다 — ATR 에 무엇을 먹이는지만 고쳤다.
고정하는 불변식 3건:
1. 당일 첫 봉이 갭으로 벌어져도 진입 ATR 이 그 갭을 먹지 않는다
2. 진입 ATR 과 청산 트레일링 ATR 이 같은 기준이다 (같은 날 데이터 → 같은 값)
3. RSI(14)/BB(20) 판정은 여전히 워밍업 혜택을 받는다 (09:30 에 판정 도달)
"""
from datetime import time as dtime
import pytest
from app.core.config import settings
from app.services.day_trading_strategy import (
MinuteCandle, _atr, _bollinger, _rsi,
check_day_trade_exit, check_vwap_bounce_signal, check_rsi_bb_signal,
)
PREV = "20260929"
TODAY = "20260930"
def _bar(date: str, hhmm: str, close: float, *, spread: float = 10.0,
volume: int = 1000) -> MinuteCandle:
"""종가 기준 ±spread 의 5분봉 1개."""
return MinuteCandle(
date=date, time=hhmm + "00",
open=close, high=close + spread, low=close - spread, close=close,
volume=volume, cum_volume=0, cum_amount=0.0,
)
def _series(date: str, n: int, base: float, *, step: float = 0.0,
spread: float = 10.0, start_min: int = 0) -> list[MinuteCandle]:
out = []
for i in range(n):
total = 9 * 60 + start_min + i * 5
out.append(_bar(
date, f"{total // 60:02d}{total % 60:02d}",
base + i * step, spread=spread,
))
return out
def _completed_bull_bar(bar: MinuteCandle) -> MinuteCandle:
"""마지막 **완성** 5분봉을 '양봉 + 거래량 급증'으로 만든다.
[2026-09-30] 양봉 판정이 형성 중 버킷(`[-1]`) → 마지막 완성봉(`[-2]`)으로
바뀌었으므로, 신호를 발화시키려면 완성봉 쪽이 양봉이어야 한다.
"""
return MinuteCandle(
date=bar.date, time=bar.time,
open=bar.close - 20, high=bar.high, low=bar.low, close=bar.close,
volume=5000, cum_volume=0, cum_amount=0.0,
)
# ── 불변식 1: 갭 TR 이 진입 ATR 에 들어오지 않는다 ────────────────────────────
def test_entry_atr_ignores_overnight_gap():
"""전일 종가가 어디에 있든 당일 ATR 값은 같다 (갭 TR 미포함)."""
today = _series(TODAY, 20, 10_000.0, spread=10.0)
# 전일 종가가 당일 시가와 비슷한 경우 / 30% 아래인 경우
small_gap = _series(PREV, 20, 9_900.0, spread=10.0, start_min=300) + today
huge_gap = _series(PREV, 20, 7_000.0, spread=10.0, start_min=300) + today
atr_small = _atr(small_gap, 14)
atr_huge = _atr(huge_gap, 14)
atr_today = _atr(today, 14)
assert atr_today is not None
assert atr_small == atr_today, "전일 봉이 진입 ATR 에 섞였다"
assert atr_huge == atr_today, "오버나이트 갭이 진입 ATR 을 부풀렸다"
def test_entry_atr_is_none_before_todays_warmup():
"""당일 봉이 period+1 개 미만이면 None — 워밍업 도입 이전과 같은 동작.
호출부는 이때 고정 폴백(DAY_TRADE_REVERSION_STOP_LOSS_PCT)을 쓴다.
전일 봉으로 ATR 을 '만들어내는' 것이 바로 C1 결함이었다.
"""
today_6 = _series(TODAY, 6, 10_000.0) # 09:30 시점
mixed = _series(PREV, 20, 9_000.0, start_min=300) + today_6
assert _atr(today_6, 14) is None
assert _atr(mixed, 14) is None, "전일 봉으로 ATR 을 합성하고 있다"
def test_vwap_signal_stop_and_target_ignore_prev_day():
"""신호의 손절·목표가가 전일 봉 유무에 영향받지 않는다.
ATR 이 `target = 현재가 + 1.5R` 과 `raw_stop` 에 직접 들어가므로,
같은 당일 데이터에서 두 입력이 같은 값을 내야 한다.
"""
# 하락 후 마지막 봉 양봉 + 거래량 급증 → VWAP 반등 신호 조건
today = _series(TODAY, 25, 10_600.0, step=-20.0, spread=8.0)
for c in today:
c.volume = 1000
today[-2] = _completed_bull_bar(today[-2])
last_close = today[-1].close
today[-1] = MinuteCandle( # 형성 중 봉: 양봉
date=TODAY, time=today[-1].time, open=last_close - 30, high=last_close + 10,
low=last_close - 40, close=last_close, volume=1200,
cum_volume=0, cum_amount=0.0,
)
price = today[-1].close
vwap = price # VWAP 밴드 중앙
# 워밍업은 **갭 없이 이어지는** 전일 하락 봉 — RSI 게이트가 양쪽에서 통과해야
# 손절·목표가 비교가 성립한다(갭을 주면 RSI 가 튀어 신호 자체가 사라진다).
mixed = _series(PREV, 20, 11_000.0, step=-20.0, spread=8.0, start_min=300) + today
sig_today = check_vwap_bounce_signal("005930", "삼성", today, vwap, price)
sig_mixed = check_vwap_bounce_signal("005930", "삼성", mixed, vwap, price)
assert sig_today is not None and sig_mixed is not None, "테스트 데이터가 신호를 못 만든다"
assert sig_mixed.stop_loss == sig_today.stop_loss, "전일 봉이 진입 손절을 움직였다"
assert sig_mixed.target_price == sig_today.target_price, "전일 봉이 목표가를 움직였다"
def test_rsi_bb_signal_stop_and_target_ignore_prev_day():
"""RSI_BB 경로도 동일 (risk → target 전파 차단 확인)."""
today = _series(TODAY, 25, 10_600.0, step=-20.0, spread=8.0)
today[-2] = _completed_bull_bar(today[-2])
last_close = today[-1].close
today[-1] = MinuteCandle(
date=TODAY, time=today[-1].time, open=last_close - 30, high=last_close + 10,
low=last_close - 40, close=last_close, volume=1200,
cum_volume=0, cum_amount=0.0,
)
price = today[-1].close * 0.97 # BB 하단 근처로 내려 조건 충족
mixed = _series(PREV, 20, 11_000.0, step=-20.0, spread=8.0, start_min=300) + today
sig_today = check_rsi_bb_signal("005930", "삼성", today, price)
sig_mixed = check_rsi_bb_signal("005930", "삼성", mixed, price)
assert sig_today is not None and sig_mixed is not None, "테스트 데이터가 신호를 못 만든다"
assert sig_mixed.stop_loss == sig_today.stop_loss
assert sig_mixed.target_price == sig_today.target_price
def test_fallback_stop_used_when_todays_atr_unavailable():
"""당일 ATR 이 없으면 고정 폴백 손절폭이 쓰인다 (ATR 합성 금지).
stop = min(현재가×(1−REVERSION_STOP_LOSS_PCT), 현재가×(1−MIN_STOP_LOSS_PCT))
— §7 의 '장중은 느슨한 쪽' 규약. 임계값은 settings 를 참조한다(§4.1).
"""
# 당일 5분봉 21개면 BB(20)·RSI(14)는 되지만 ATR(14)에는 15개가 필요하므로
# 당일 봉을 14개로 맞추고 워밍업으로 RSI/BB 만 채운다.
today = _series(TODAY, 14, 10_300.0, step=-20.0, spread=8.0)
today[-2] = _completed_bull_bar(today[-2])
last_close = today[-1].close
today[-1] = MinuteCandle(
date=TODAY, time=today[-1].time, open=last_close - 30, high=last_close + 10,
low=last_close - 40, close=last_close, volume=1200,
cum_volume=0, cum_amount=0.0,
)
mixed = _series(PREV, 20, 10_700.0, step=-20.0, spread=8.0, start_min=300) + today
price = today[-1].close
assert _atr(mixed, 14) is None, "당일 봉 14개로 ATR 이 나오면 전제가 깨진다"
sig = check_vwap_bounce_signal("005930", "삼성", mixed, price, price)
assert sig is not None, "워밍업 덕에 RSI 판정은 되어야 한다"
expected = min(
price * (1 - settings.DAY_TRADE_REVERSION_STOP_LOSS_PCT),
price * (1 - settings.DAY_TRADE_MIN_STOP_LOSS_PCT),
)
assert sig.stop_loss == pytest.approx(round(expected, 0))
# ── 불변식 2: 진입 ATR == 청산 트레일링 ATR ───────────────────────────────────
def test_entry_and_exit_atr_share_one_basis():
"""진입(워밍업 포함 시계열)과 청산(당일 전용 시계열)의 ATR 이 같다.
청산 경로(`daytrading/exit_service`)는 `get_minute_chart_full_day`(당일 전용)를
쓰고 진입 경로는 `_v2(include_prev_day=True)` 를 쓴다. 두 경로가 다른 ATR 을
쓰면 진입 손절과 트레일링 스탑의 기준이 엇갈린다.
"""
today = _series(TODAY, 20, 10_000.0, spread=25.0)
entry_series = _series(PREV, 20, 7_500.0, start_min=300) + today # 진입 경로 입력
exit_series = today # 청산 경로 입력
assert _atr(entry_series, 14) == _atr(exit_series, 14) is not None
def test_trailing_stop_identical_for_both_series():
"""같은 포지션·같은 당일 데이터에서 트레일링 청산 판정이 동일하다."""
today = _series(TODAY, 20, 10_000.0, spread=25.0)
mixed = _series(PREV, 20, 7_500.0, start_min=300) + today
atr = _atr(today, 14)
assert atr is not None
entry_price = 9_000.0
highest = 10_500.0
trail = highest - settings.DAY_TRADE_TRAILING_ATR_MULT * atr
current = trail - 1 # 트레일 바로 아래 → SELL_ALL 기대
kwargs = dict(
entry_price=entry_price, current_price=current,
stop_loss=entry_price * 0.9, target_price=entry_price * 1.5,
highest_since_entry=highest, current_time=dtime(11, 0), entry_time=None,
)
ex_today = check_day_trade_exit(candles_5min=today, **kwargs)
ex_mixed = check_day_trade_exit(candles_5min=mixed, **kwargs)
assert ex_today is not None and ex_today.action == "SELL_ALL"
assert ex_mixed is not None
assert ex_today.action == ex_mixed.action
assert ex_today.reason == ex_mixed.reason, "트레일링 기준값이 시계열에 따라 달라졌다"
# ── 불변식 3: RSI/BB 워밍업 혜택 유지 (C3 목적 보존) ─────────────────────────
def test_rsi_and_bb_still_benefit_from_warmup():
"""09:30(당일 6봉)에도 워밍업으로 RSI(14)/BB(20) 판정이 가능해야 한다."""
today_6 = _series(TODAY, 6, 10_000.0, step=-15.0)
mixed = _series(PREV, 20, 10_200.0, step=-5.0, start_min=300) + today_6
# 당일 봉만으로는 둘 다 불가
assert _rsi(today_6, 14) is None
assert _bollinger(today_6, 20) is None
# 워밍업이 붙으면 판정 도달 — C1 수정이 이걸 깨지 않았는지 확인
assert _rsi(mixed, 14) is not None, "워밍업이 무력화됐다 (C3 목적 상실)"
assert _bollinger(mixed, 20) is not None, "워밍업이 무력화됐다 (C3 목적 상실)"
def test_atr_session_filter_is_wallclock_independent():
"""`_atr` 은 벽시계를 쓰지 않는다 (§2) — **동작으로** 검증한다.
날짜가 1999년이든 2099년이든 '마지막 봉의 영업일자'만으로 세션이 갈려야 한다.
`datetime.now()` 에 의존하면 과거 날짜 입력에서 당일 봉이 0건이 되어 값이 달라진다
(휴장일·재기동·백테스트에서 기준이 흔들리는 것과 같은 증상).
"""
def _pair(prev_date, today_date):
prev = [
MinuteCandle(date=prev_date, time=f"{13 + i // 12:02d}{(i * 5) % 60:02d}00",
open=9_000.0, high=9_020.0, low=8_980.0, close=9_000.0,
volume=10, cum_volume=0, cum_amount=0.0)
for i in range(20)
]
today = [
MinuteCandle(date=today_date, time=f"{9 + i // 12:02d}{(i * 5) % 60:02d}00",
open=10_000.0, high=10_025.0, low=9_975.0, close=10_000.0,
volume=10, cum_volume=0, cum_amount=0.0)
for i in range(20)
]
return prev + today
past = _atr(_pair("19990104", "19990105"), 14)
future = _atr(_pair("20990104", "20990105"), 14)
recent = _atr(_pair(PREV, TODAY), 14)
assert past == future == recent is not None, "세션 판별이 벽시계에 의존한다"
# 같은 하루 변동폭(±25)에서 나온 값이므로 갭(1,000원)이 섞이지 않았음도 함께 확인
assert past == pytest.approx(50.0)
# 날짜 필드가 비어 있는 입력(목/레거시)은 기존 동작을 유지한다
undated = [
MinuteCandle(date="", time=f"{9 + i // 12:02d}{(i * 5) % 60:02d}00",
open=100.0, high=101.0, low=99.0, close=100.0,
volume=10, cum_volume=0, cum_amount=0.0)
for i in range(20)
]
assert _atr(undated, 14) is not None
# 날짜 필드가 비어 있는 입력(목/레거시)은 기존 동작을 유지한다
undated = [
MinuteCandle(date="", time=f"{9 + i // 12:02d}{(i * 5) % 60:02d}00",
open=100.0, high=101.0, low=99.0, close=100.0,
volume=10, cum_volume=0, cum_amount=0.0)
for i in range(20)
]
assert _atr(undated, 14) is not None
# ── 양봉 판정: 형성 중 버킷 → 마지막 완성봉 (2026-09-30) ──────────────────────
def _oversold_today(n: int = 25) -> list[MinuteCandle]:
"""RSI/BB 게이트를 통과할 하락 시계열 (마지막 완성봉은 호출부가 정한다)."""
return _series(TODAY, n, 10_600.0, step=-20.0, spread=8.0)
def _with_forming(bars: list[MinuteCandle], *, bull: bool) -> list[MinuteCandle]:
"""형성 중 버킷([-1])을 양봉/음봉으로 바꾼다 (판정에 영향을 주면 안 된다)."""
out = list(bars)
c = out[-1]
out[-1] = MinuteCandle(
date=c.date, time=c.time,
open=c.close - 30 if bull else c.close + 30,
high=c.close + 10, low=c.close - 40, close=c.close,
volume=1200, cum_volume=0, cum_amount=0.0,
)
return out
def test_bull_check_decided_by_completed_bar_not_forming_bucket():
"""형성 중 버킷이 양봉이든 음봉이든 판정은 **완성봉**으로 결정된다.
30초 주기 스캔에서 같은 5분 버킷을 약 10회 보므로, 형성 중 봉으로 판정하면
같은 시장 상태에 사이클마다 다른 답이 나온다(재현성 없는 게이트).
"""
base = _oversold_today()
base[-2] = _completed_bull_bar(base[-2]) # 완성봉 = 양봉
price = base[-1].close
sig_bull_forming = check_vwap_bounce_signal(
"005930", "삼성", _with_forming(base, bull=True), price, price)
sig_bear_forming = check_vwap_bounce_signal(
"005930", "삼성", _with_forming(base, bull=False), price, price)
assert sig_bull_forming is not None
assert sig_bear_forming is not None, "형성 중 버킷이 음봉이라고 게이트가 닫혔다"
assert sig_bull_forming.stop_loss == sig_bear_forming.stop_loss
assert sig_bull_forming.target_price == sig_bear_forming.target_price
def test_bear_completed_bar_blocks_even_if_forming_bucket_is_bullish():
"""완성봉이 음봉이면 형성 중 버킷이 양봉이어도 차단된다 (역방향 확인)."""
base = _oversold_today()
c = base[-2]
base[-2] = MinuteCandle( # 완성봉 = 음봉 + 거래량 급증
date=c.date, time=c.time, open=c.close + 20, high=c.high, low=c.low,
close=c.close, volume=5000, cum_volume=0, cum_amount=0.0,
)
price = base[-1].close
assert check_vwap_bounce_signal(
"005930", "삼성", _with_forming(base, bull=True), price, price) is None
assert check_rsi_bb_signal(
"005930", "삼성", _with_forming(base, bull=True), price * 0.97) is None
def test_bull_check_uses_same_bar_as_volume_gate():
"""양봉 판정 대상 봉 == 거래량 게이트가 비교한 봉 (`session[-2]`)."""
from app.services.day_trading_strategy import (
_last_completed_candle, volume_surge_ratio,
)
base = _oversold_today()
base[-2] = _completed_bull_bar(base[-2])
mixed = _series(PREV, 20, 11_000.0, step=-20.0, spread=8.0, start_min=300) + base
assert _last_completed_candle(mixed) is mixed[-2]
assert volume_surge_ratio(mixed) == pytest.approx(
mixed[-2].volume / mixed[-3].volume
)
def test_bull_check_never_crosses_session_boundary():
"""당일 완성봉이 없으면 미통과 — 전일 봉으로 게이트를 열지 않는다.
개장 직후(당일 봉 1개)에는 워밍업 시계열의 마지막 완성봉이 **전일 종가 동시호가
봉**이다. 그걸 보면 게이트가 전일 결과로 열린다(C1 과 같은 함정).
"""
from app.services.day_trading_strategy import _last_completed_candle
prev_bull = _series(PREV, 20, 11_000.0, step=-20.0, spread=8.0, start_min=300)
c = prev_bull[-1]
prev_bull[-1] = MinuteCandle( # 전일 마지막 봉을 강한 양봉으로
date=c.date, time=c.time, open=c.close - 100, high=c.close + 10,
low=c.close - 120, close=c.close, volume=9999, cum_volume=0, cum_amount=0.0,
)
today_one = _series(TODAY, 1, 9_000.0)
assert _last_completed_candle(prev_bull + today_one) is None, (
"전일 봉이 '마지막 완성봉'으로 새어 들어왔다"
)
# 데이터 부족은 통과 쪽으로 열지 않는다
assert check_vwap_bounce_signal(
"005930", "삼성", prev_bull + today_one, 9_000.0, 9_000.0) is None
def test_last_completed_candle_is_wallclock_independent():
"""§2 순수성 — 과거/미래 날짜에서도 같은 위치의 봉을 고른다 (동작 검증)."""
from app.services.day_trading_strategy import _last_completed_candle
for d_prev, d_today in (("19990104", "19990105"), ("20990104", "20990105"),
(PREV, TODAY)):
prev = _series(d_prev, 20, 11_000.0, step=-20.0, spread=8.0, start_min=300)
today = _series(d_today, 5, 10_000.0, step=-20.0, spread=8.0)
got = _last_completed_candle(prev + today)
assert got is today[-2], f"{d_today}: 마지막 완성봉 선택이 벽시계에 의존한다"
# ── 진단 정보 (갭·ATR) — 게이트가 아니라 관찰 수단 (2026-09-30) ───────────────
def test_overnight_gap_pct_from_data_only():
"""갭률은 데이터(전일 마지막 종가 vs 당일 첫 시가)로만 계산된다."""
from app.services.day_trading_strategy import _overnight_gap_pct
prev = _series(PREV, 20, 10_000.0, spread=10.0, start_min=300)
today = _series(TODAY, 20, 9_700.0, spread=10.0) # 시가 = 종가 = 9,700
gap = _overnight_gap_pct(prev + today)
assert gap == pytest.approx((9_700 - 10_000) / 10_000 * 100) # -3.00%
# 전일 봉이 없으면 계산 불가 → None (억지로 만들지 않는다)
assert _overnight_gap_pct(today) is None
# 날짜 필드가 빈 목/레거시 입력도 None
undated = [
MinuteCandle(date="", time="0900%02d" % 0, open=100.0, high=101.0,
low=99.0, close=100.0, volume=1, cum_volume=0, cum_amount=0.0)
]
assert _overnight_gap_pct(undated) is None
def test_reason_carries_gap_and_atr_diagnostics():
"""진입 reason 에 갭·ATR 이 **진단 정보**로 남는다 (게이트 아님).
RSI 갭 섭취는 유지 결정이 났고(차트 표준·평균회귀 의도와 방향 일치) 대신 관찰
대상이 됐다. 원장(`trade_history.entry_reason`)만으로 "갭 하락 종목에 쏠렸는가"를
집계할 수 있어야 한다.
"""
base = _oversold_today()
base[-2] = _completed_bull_bar(base[-2])
price = base[-1].close
mixed = _series(PREV, 20, 11_000.0, step=-20.0, spread=8.0, start_min=300) + base
sig = check_vwap_bounce_signal("005930", "삼성", mixed, price, price)
assert sig is not None
assert "[진단 " in sig.reason, "진단 정보가 게이트 조건과 구분되지 않는다"
assert "갭=" in sig.reason and "%" in sig.reason
assert "ATR=" in sig.reason
sig_rb = check_rsi_bb_signal("005930", "삼성", mixed, price * 0.97)
assert sig_rb is not None and "갭=" in sig_rb.reason and "ATR=" in sig_rb.reason
def test_reason_marks_fallback_stop_when_atr_absent():
"""ATR 이 없으면 reason 이 '폴백손절'임을 밝힌다.
09:30~10:10 은 당일 5분봉 15개 미만이라 ATR 이 없고 손절이
`DAY_TRADE_REVERSION_STOP_LOSS_PCT`(폴백)로 결정된다 — 원장에서 구분되어야 한다.
"""
today = _series(TODAY, 14, 10_300.0, step=-20.0, spread=8.0)
today[-2] = _completed_bull_bar(today[-2])
mixed = _series(PREV, 20, 10_700.0, step=-20.0, spread=8.0, start_min=300) + today
price = today[-1].close
from app.services.day_trading_strategy import _atr as _atr_fn
assert _atr_fn(mixed, 14) is None
sig = check_vwap_bounce_signal("005930", "삼성", mixed, price, price)
assert sig is not None
assert "ATR=없음→폴백손절" in sig.reason
assert sig.stop_loss == pytest.approx(round(min(
price * (1 - settings.DAY_TRADE_REVERSION_STOP_LOSS_PCT),
price * (1 - settings.DAY_TRADE_MIN_STOP_LOSS_PCT),
), 0))
def test_diagnostics_do_not_change_any_decision():
"""진단 정보 추가가 판정·손절·목표를 바꾸지 않는다.
`reason` 외의 모든 필드가 전일 봉 유무와 무관하게 동일해야 한다
(진단 문자열만 달라진다).
"""
base = _oversold_today()
base[-2] = _completed_bull_bar(base[-2])
price = base[-1].close
mixed = _series(PREV, 20, 11_000.0, step=-20.0, spread=8.0, start_min=300) + base
a = check_vwap_bounce_signal("005930", "삼성", base, price, price)
b = check_vwap_bounce_signal("005930", "삼성", mixed, price, price)
assert a is not None and b is not None
for field in ("strategy", "entry_price", "stop_loss", "target_price", "confidence"):
assert getattr(a, field) == getattr(b, field), f"{field} 가 달라졌다"
# 당일 전용 입력에는 갭이 없다(계산 불가) → 진단에서 생략된다
assert "갭=" not in a.reason and "갭=" in b.reason
def test_orb_reason_has_atr_but_no_gap():
"""ORB 경로는 당일 전용 시계열이라 갭을 계산할 수 없다 — ATR 만 남는다."""
from app.services.day_trading_strategy import OpeningRange, check_orb_signal
# 당일 1분봉 60개: 09:00~09:59, 마지막 완성 5분봉 양봉 + 거래량 급증
bars = []
for i in range(60):
total = 9 * 60 + i
close = 10_000 + (i if i < 50 else 60 + i)
bars.append(MinuteCandle(
date=TODAY, time=f"{total // 60:02d}{total % 60:02d}00",
open=close - 5, high=close + 5, low=close - 10, close=close,
volume=5000 if i >= 50 else 100, cum_volume=0, cum_amount=0.0,
))
orb = OpeningRange(high=10_020.0, low=9_950.0, total_volume=3000,
avg_vol_per_min=100.0, open_price=10_000.0)
price = bars[-1].close
sig = check_orb_signal("005930", "삼성", bars, orb, price)
if sig is not None: # 조건 미충족이면 이 단정은 의미가 없다
assert "ATR=" in sig.reason
assert "갭=" not in sig.reason
# ── M-1: 1.5R 하한 보장 (2026-09-30) ─────────────────────────────────────────
def _rr(sig) -> float:
"""보상/위험 배수."""
return (sig.target_price - sig.entry_price) / (sig.entry_price - sig.stop_loss)
def _signal_pair(n_today: int):
"""(VWAP 신호, RSI_BB 신호) — 당일 봉 수로 ATR 유무를 가른다.
n_today=25 → 당일 ATR(14) 산출 가능 / n_today=14 → ATR 없음(폴백 손절폭).
"""
today = _series(TODAY, n_today, 10_600.0, step=-20.0, spread=8.0)
today[-2] = _completed_bull_bar(today[-2])
mixed = _series(PREV, 20, 11_000.0, step=-20.0, spread=8.0, start_min=300) + today
price = today[-1].close
return (
check_vwap_bounce_signal("005930", "삼성", mixed, price, price),
check_rsi_bb_signal("005930", "삼성", mixed, price * 0.97),
)
@pytest.mark.parametrize("n_today,label", [(25, "ATR 있음"), (14, "ATR 폴백")])
def test_target_guarantees_target_multiplier_r(n_today, label):
"""목표가는 항상 위험의 `DAY_TRADE_TARGET_MULTIPLIER` 배 이상이다.
`target_price` 는 `daytrading/exit_service` 의 분할매도 트리거로 실제 쓰인다(§7).
위험보다 작은 목표는 R:R 역전이다 — ATR 유무 양쪽에서 막는다.
"""
sig_vwap, sig_rb = _signal_pair(n_today)
assert sig_vwap is not None and sig_rb is not None, "테스트 데이터가 신호를 못 만든다"
for name, sig in (("VWAP", sig_vwap), ("RSI_BB", sig_rb)):
risk = sig.entry_price - sig.stop_loss
assert risk > 0
assert sig.target_price - sig.entry_price >= risk * settings.DAY_TRADE_TARGET_MULTIPLIER - 1e-6, (
f"{label}/{name}: {_rr(sig):.2f}R < "
f"{settings.DAY_TRADE_TARGET_MULTIPLIER}R "
f"(진입 {sig.entry_price:.0f} 손절 {sig.stop_loss:.0f} 목표 {sig.target_price:.0f})"
)
def test_low_volatility_vwap_entry_is_not_r_inverted():
"""저변동 종목(1.5×ATR < 최소손절폭)에서도 R:R 이 역전되지 않는다.
수정 전 실측: 진입 10,120 / ATR 29 → 손절 -2.00% / 목표 +0.43% = **0.21R**.
손절이 `min()` 으로 -2% 쪽으로 느슨해지는데 목표는 `1.5×ATR` 이라 위험보다 작았다.
"""
sig_vwap, _ = _signal_pair(25)
assert sig_vwap is not None
from app.services.day_trading_strategy import _atr as _atr_fn
today = _series(TODAY, 25, 10_600.0, step=-20.0, spread=8.0)
today[-2] = _completed_bull_bar(today[-2])
atr = _atr_fn(today, 14)
assert atr is not None
# 전제: 이 데이터는 저변동(1.5×ATR < 최소손절폭)이라 옛 공식이면 R:R < 1 이었다
assert settings.DAY_TRADE_TARGET_MULTIPLIER * atr < (
sig_vwap.entry_price * settings.DAY_TRADE_MIN_STOP_LOSS_PCT
)
assert _rr(sig_vwap) >= settings.DAY_TRADE_TARGET_MULTIPLIER - 1e-6
def test_target_uses_the_larger_of_two_candidates():
"""기존 목표식이 더 크면 그 값이 유지된다 (1.5R 하한은 '하한'일 뿐).
RSI_BB 는 `max(bb_middle, 1.5R)` 이므로 BB 중앙선이 멀면 그쪽이 이긴다.
"""
_, sig_rb = _signal_pair(25)
assert sig_rb is not None
assert _rr(sig_rb) > settings.DAY_TRADE_TARGET_MULTIPLIER, (
"BB 중앙선이 1.5R 보다 먼 데이터인데 1.5R 로 깎였다"
)