## 재기동 내구성 - scheduler: 복원 직후 reset_if_new_day()가 복원분을 전부 폐기하던 버그 (begin_restore()로 날짜 경계를 선확정. 복원 성공 로그는 뜨는데 실제로는 지워졌다) - 일일 안전장치(daily_pnl/loss_streak_today/closed_today) Redis 영속화 날짜를 키와 페이로드 양쪽에 실어 이중 게이트, 복원은 보수적 병합 (손익 min/연패 max/청산목록 합집합 — 복원이 안전장치를 약화시킬 수 없다) - db_worker 종료 drain이 insert_trade_history만 커밋하고 포지션 추적 행은 DLQ로 버리던 비대칭 제거 + DLQ 재처리 경로 신설(쓰기 전용이었다) - 장중 기동 시 ORB 소급 수집(backfill), 전일 캔들 혼입 차단 - swing/exit_service "보유 종목 없음" debug→info, 10회 연속 warning ## DT 진입 복구 - 워밍업 회귀 복구: 1b9257c(2026-09-28)가 조회 계층에 당일 필터를 넣어 5분봉 RSI(14)/BB(20) 워밍업이 죽고 VWAP/RSI_BB가 10:15/10:40 이전 발화 불가가 됐다. 필터를 삭제하지 않고 용도별로 분리 (get_minute_chart_full_day_v2 신설, 기존 메서드·3개 호출부 동작 무변경) - 거래량 게이트 비교 대상 교정: 형성 중 봉 → 완성봉 (5분봉·ORB 1분봉 양쪽) 임계값(1.3/1.5)은 불변, 고친 것은 비교 대상뿐 - 세션 경계 차단: _last_session_candles()로 거래량 비교가 전일 봉을 잡아 개장 직후 게이트가 자동 통과하는 것을 방지 - 스캔 선조회(동시성 4)·종목 타임아웃·시간예산·라운드로빈 + 잘린 종목 수 로그 - watchlist 30분 갱신 시 ORB 보유 종목 우선 유지 ## 유입 잔고 격리 - sync_portfolio()가 시스템 매수 근거 없는 KIS 잔고를 자기 스윙 포지션으로 편입해 자동 손절로 처분하던 문제 → strategy_type='external' 격리 (승격만 허용, 강등 없음. 기존 행은 수동 SQL 필요) - 격리분 손절선 이탈은 집행 없이 warning + summary 노출 - 수동 청산 경로에 record_trade_result 누락 보강, 전략 간 당일 쿨다운 ## 4% 상한 - market_data 표시 경로 클램프 누락 + confidence 배율 후 재클램프 - 회귀 방지 테스트(AST로 가드가 try 밖에 있는지 검증) ## 프론트 - 총자산 보정 가드(3%·종목수·가격결손·국내분 한정), 원인 미규명 명시 - cumulative_realized_pl이 /portfolio/combined에 없어 항상 0이던 문제 테스트 579 → 677 passed / 0 failed (소켓 차단 하 통과, 실호출 없음) 전략 임계값 변경 0건 ⚠️ 워밍업 복구의 효과 크기는 미확정 — KIS 분봉 API에 날짜 파라미터가 없어 미래 ref_time에 전일 데이터를 주는지가 미검증 전제다. 배포 후 "[분봉] 워밍업 5분봉 N봉 (당일 a + 전일 b)" 로그의 b로 확정할 것. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
110 lines
4.6 KiB
Python
110 lines
4.6 KiB
Python
"""종목당 4% 절대 상한(MAX_POSITION_SIZE) 회귀 테스트 — CLAUDE.md §5.1 / §4.
|
|
|
|
배경(2026-09-27): 패턴메모리 조정 뒤의 상한 재클램프가 `try` 블록 **안**에 있어
|
|
`get_pattern_adjusted_invest_amount()` 가 예외를 던지면 클램프를 건너뛰고
|
|
적응형 포지션(6~7%)이 그대로 주문 금액이 됐다. §4 는 "리스크 가드·한도 클램프를
|
|
`try` 안에 두지 말 것" 으로 이것을 규칙화했다.
|
|
|
|
여기서는 두 층을 검증한다.
|
|
1) 구조: `_execute_buy` 의 절대 상한 클램프가 패턴메모리 `try` **밖**에 있다
|
|
(예외 발생 케이스를 실제 주문 경로 없이 확정적으로 잡는다).
|
|
2) 수치: 적응형 Normal(6%)·Aggressive(7%) 산출액이 클램프 후 총자산 대비
|
|
MAX_POSITION_SIZE 를 넘지 않는다.
|
|
|
|
기대값은 전부 `settings.*` 참조다(§4.1 — config 값 하드코딩 금지).
|
|
"""
|
|
import ast
|
|
import inspect
|
|
|
|
import pytest
|
|
|
|
from app.core.config import settings
|
|
from app.services import trade_executor
|
|
from app.services.sizing_core import clamp_by_ratios
|
|
|
|
|
|
# ── 1) 구조 검증 ────────────────────────────────────────────────────────────
|
|
|
|
def _execute_buy_ast() -> ast.AST:
|
|
src = inspect.getsource(trade_executor._execute_buy)
|
|
return ast.parse(ast.unparse(ast.parse(_dedent(src))))
|
|
|
|
|
|
def _dedent(src: str) -> str:
|
|
import textwrap
|
|
return textwrap.dedent(src)
|
|
|
|
|
|
def _contains_call(node: ast.AST, name: str) -> bool:
|
|
return any(
|
|
isinstance(n, ast.Call)
|
|
and (
|
|
(isinstance(n.func, ast.Name) and n.func.id == name)
|
|
or (isinstance(n.func, ast.Attribute) and n.func.attr == name)
|
|
)
|
|
for n in ast.walk(node)
|
|
)
|
|
|
|
|
|
def _mentions_max_position_size(node: ast.AST) -> bool:
|
|
return any(
|
|
isinstance(n, ast.Attribute) and n.attr == "MAX_POSITION_SIZE"
|
|
for n in ast.walk(node)
|
|
)
|
|
|
|
|
|
def test_pattern_memory_clamp_is_outside_try():
|
|
"""패턴메모리 호출은 try 안, 절대 상한 클램프는 그 try 뒤의 문장이어야 한다."""
|
|
tree = _execute_buy_ast()
|
|
fn = tree.body[0]
|
|
assert isinstance(fn, (ast.AsyncFunctionDef, ast.FunctionDef))
|
|
|
|
body = fn.body
|
|
pm_try_idx = None
|
|
for idx, stmt in enumerate(body):
|
|
if isinstance(stmt, ast.Try) and _contains_call(stmt, "get_pattern_adjusted_invest_amount"):
|
|
pm_try_idx = idx
|
|
break
|
|
assert pm_try_idx is not None, "패턴메모리 조정 try 블록을 찾지 못했다"
|
|
|
|
# try 블록 안에는 절대 상한 클램프가 없어야 한다
|
|
assert not (
|
|
_contains_call(body[pm_try_idx], "clamp_by_ratios")
|
|
and _mentions_max_position_size(body[pm_try_idx])
|
|
), "절대 상한 클램프가 패턴메모리 try 안에 있다 (예외 시 가드가 통과된다)"
|
|
|
|
# try 이후 문장 중에 MAX_POSITION_SIZE 클램프가 있어야 한다
|
|
after = body[pm_try_idx + 1:]
|
|
assert any(
|
|
_contains_call(s, "clamp_by_ratios") and _mentions_max_position_size(s)
|
|
for s in after
|
|
), "패턴메모리 try 뒤에 MAX_POSITION_SIZE 재클램프가 없다"
|
|
|
|
|
|
def test_adaptive_position_cap_is_clamped_at_source():
|
|
"""적응형 position_size 를 소비할 때 min(..., MAX_POSITION_SIZE) 가 적용된다."""
|
|
src = inspect.getsource(trade_executor._execute_buy)
|
|
assert "min(_adp.position_size, settings.MAX_POSITION_SIZE)" in src
|
|
|
|
|
|
# ── 2) 수치 검증 ────────────────────────────────────────────────────────────
|
|
|
|
TOTAL_ASSETS = 10_000_000.0
|
|
AVAILABLE_CASH = 10_000_000.0
|
|
|
|
|
|
@pytest.mark.parametrize("mode_position_size", [0.06, 0.07]) # Normal / Aggressive
|
|
def test_adaptive_modes_never_exceed_hard_cap(mode_position_size):
|
|
"""적응형 6~7% 예산도 최종 클램프 후에는 MAX_POSITION_SIZE 이하."""
|
|
adaptive_cap = min(mode_position_size, settings.MAX_POSITION_SIZE)
|
|
raw = TOTAL_ASSETS * adaptive_cap
|
|
final = clamp_by_ratios(raw, TOTAL_ASSETS, [settings.MAX_POSITION_SIZE], AVAILABLE_CASH)
|
|
assert final / TOTAL_ASSETS <= settings.MAX_POSITION_SIZE + 1e-9
|
|
|
|
|
|
def test_pattern_memory_uplift_is_clamped():
|
|
"""패턴메모리가 +30% 상향해도 최종 클램프가 상한을 지킨다."""
|
|
raw = TOTAL_ASSETS * settings.MAX_POSITION_SIZE * 1.3
|
|
final = clamp_by_ratios(raw, TOTAL_ASSETS, [settings.MAX_POSITION_SIZE], AVAILABLE_CASH)
|
|
assert final / TOTAL_ASSETS <= settings.MAX_POSITION_SIZE + 1e-9
|