shuim
9fab4018d4
하네스 규칙 검사 전역 위반 117→0, 회차별 검증(code-verifier opus 4도메인)으로 Critical/High 해소. 주문·리스크 - MDD 강제축소 반복 매도(종목별 완료 집합·재시도 상한·날짜 롤오버·잔고 오류 보류·epoch·장외 미주문) - KIS 거부 FAILED 기록, 결과 불명(전송 후) 재전송 금지·PENDING, 미전송 실패만 재시도(msg_cd 분류) - 국내 전량·손실 매도 시장가, 부분 익절 직전 현재가, 지정가만 미체결 취소(cancel_order_v2: TR 0013U·조직번호) + 0체결 롤백 - 누적 4%·현금·레버리지(신규분 포함)·사이클 내 매수 누적, 수동매수 공용 가드(buy_guards), 수동매도 DT 차감·락 - 가드 우회 엔드포인트 차단(/recommendations/auto-trade 위임, US /execute 매수 410·매도 가드) - 매수 슬롯=성공 매수, 평가 상한·회전, 업무거부/계좌 쿨다운, 프로세스 내 in-flight 주문 중복 차단 데이트레이딩 - 진입·청산 shield, 청산 진행 중 표식(소유 토큰), 2단계 청산 체크, 결과 불명 → 보유 감소 판정·비DT 차감 상한 - 전일 포지션 재시도(일시 실패 미산정), 분할매도 영속, ORB 구간 손절, /cycle 락, 사이클 내 현금 차감, 체결 수량 보정 스윙 - 손절 공백(판정 예외 시 하드 바닥), 단일 손절 소스·범위 검증, 매도 직전 KIS 잔고 상한(min(주문가능,보유)−DT) - 결과 불명 체결 대조·장부 반영, 대체가(나이·진입 이후·ATR 캐시), 분할 단계 기록, 국면 실패 fail-closed - 듀얼 플레이북·서킷브레이커 게이트, 평균회귀 우선·듀얼 독립, 시간손절 거래일 - 거래 확대(사용자 결정, 백테스트 미검증): RSI 46/35 정본 경로, 모멘텀 v2 ON, SIDEWAYS ADX AND, 트레일링 1차 익절 후, 20일 시간손절 ON 인프라·보안 - 계좌·손익 GET 인증, 세션 is_active·관리자 경로, 서명키 파생, realip, WS 토큰 마스킹 - KIS 토큰 즉시 실패(sleep 제거)·무효 시 캐시 삭제, 주문 5xx+JSON 본문 반환 - db_worker 종료 drain(멱등 재실행·커밋 불명 격리·예산·밴 커밋), stop_grace 40s, Redis 타임아웃·volatile-lru, /health DLQ·캐시 - KRX 휴장일, py3.11 비호환 f-string 수정 + Jenkins 이미지 내 compileall·헬스 실패 시 빌드 실패 하네스: rules_check 에 PEP 701 f-string 검사, 마커(allow-*), deploy 스모크 401 기준 CLAUDE.md: 실제 동작·플래그·백로그(§8.5-27~38) 갱신 backend 1451 passed / hooks 112 passed / tsc ok / py3.9 compile 0 errors Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
495 lines
22 KiB
Python
495 lines
22 KiB
Python
"""
|
|
전략 자동 최적화 서비스
|
|
|
|
[역할]
|
|
- 최근 거래 성과 약점 분석 → 파라미터 범위 결정
|
|
- BacktestEngine을 사용한 그리드 서치로 최적 파라미터 탐색
|
|
- 평가 지표: Sharpe * 0.4 + win_rate * 0.3 + (1 - |MDD|/100) * 0.3
|
|
- Sharpe 10% 이상 개선 시 strategy_config 자동 적용
|
|
(단 `settings.STRATEGY_AUTO_APPLY_ENABLED` 기본 False — 계산·로그만 하고 DB 쓰기는 차단)
|
|
- StrategyOptimizationLog 테이블에 이력 저장
|
|
|
|
[사용법]
|
|
from app.services.strategy_optimizer import run_optimization
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
result = await run_optimization(db, perf_data, trigger="manual")
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
from datetime import date, datetime, timedelta
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy import select, desc
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.database import AsyncSessionLocal
|
|
from app.models.models import StrategyOptimizationLog, StockDailyPrice
|
|
from app.services.strategy_config_service import strategy_cfg
|
|
from app.core.config import settings
|
|
from app.services.backtest_engine import BacktestEngine, BacktestConfig
|
|
from app.core.timeutil import now_kst
|
|
|
|
from app.core.log_throttle import warn_throttled # noqa: E402
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 결과 데이터클래스
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
@dataclass
|
|
class OptimizationResult:
|
|
best_params: Dict[str, Any]
|
|
best_score: float
|
|
best_sharpe: float
|
|
best_win_rate: float
|
|
best_mdd: float
|
|
current_sharpe: float
|
|
previous_params: Dict[str, Any]
|
|
applied: bool
|
|
apply_reason: str
|
|
total_combinations: int
|
|
# IS/OOS 검증 결과 (v2 추가)
|
|
oos_sharpe: float = 0.0
|
|
oos_win_rate: float = 0.0
|
|
oos_mdd: float = 0.0
|
|
is_oos_sharpe_ratio: float = 0.0 # OOS Sharpe / IS Sharpe (1.0에 가까울수록 건전)
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 파라미터 범위 결정 (성과 약점 기반)
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def _build_param_grid(perf_data: Dict[str, Any]) -> Dict[str, List[Any]]:
|
|
"""
|
|
성과 약점을 분석하여 그리드 서치할 파라미터 범위를 결정.
|
|
|
|
규칙:
|
|
- 승률 < 50% → RSI_OVERSOLD_LEVEL 강화 (더 낮은 값)
|
|
- avg_loss > 손절 기준 → ATR_STOP_MULTIPLIER, STOP_LOSS_PERCENT 조정
|
|
- MDD 과다(< -15%) → MAX_POSITION_SIZE, MAX_HOLDINGS_COUNT 조정
|
|
- 항상 RSI 기간 변화도 포함 (성과 개선 여지 높음)
|
|
"""
|
|
summary = perf_data.get("summary", {})
|
|
win_rate = summary.get("win_rate", 50.0)
|
|
avg_loss = summary.get("avg_loss_pct", -5.0)
|
|
|
|
grid: Dict[str, List[Any]] = {}
|
|
|
|
# RSI_OVERSOLD_LEVEL: 승률 기반
|
|
if win_rate < 40:
|
|
grid["RSI_OVERSOLD_LEVEL"] = [20.0, 25.0, 28.0]
|
|
elif win_rate < 50:
|
|
grid["RSI_OVERSOLD_LEVEL"] = [25.0, 28.0, 30.0]
|
|
else:
|
|
grid["RSI_OVERSOLD_LEVEL"] = [28.0, 30.0, 32.0]
|
|
|
|
# STOP_LOSS_PERCENT: 평균 손실 기반
|
|
if avg_loss < -8.0:
|
|
grid["STOP_LOSS_PERCENT"] = [-5.0, -6.0, -7.0]
|
|
elif avg_loss < -6.0:
|
|
grid["STOP_LOSS_PERCENT"] = [-5.0, -6.0]
|
|
else:
|
|
grid["STOP_LOSS_PERCENT"] = [-6.0, -7.0, -8.0]
|
|
|
|
# ATR_STOP_MULTIPLIER: 항상 포함
|
|
grid["ATR_STOP_MULTIPLIER"] = [2.0, 2.5, 3.0]
|
|
|
|
# BOLLINGER_STDDEV
|
|
grid["BOLLINGER_STDDEV"] = [1.8, 2.0, 2.2]
|
|
|
|
return grid
|
|
|
|
|
|
def _cartesian_product(grid: Dict[str, List[Any]]) -> List[Dict[str, Any]]:
|
|
"""딕셔너리 그리드의 카르테시안 곱을 반환."""
|
|
keys = list(grid.keys())
|
|
values = list(grid.values())
|
|
|
|
def _product(lists):
|
|
result = [[]]
|
|
for lst in lists:
|
|
result = [x + [y] for x in result for y in lst]
|
|
return result
|
|
|
|
combos = _product(values)
|
|
return [dict(zip(keys, combo)) for combo in combos]
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 현재 전략 파라미터 수집
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
async def _get_current_params() -> Dict[str, Any]:
|
|
"""현재 strategy_cfg 값들을 읽어 반환."""
|
|
await strategy_cfg._refresh()
|
|
return {
|
|
# [2026-09-27] 하드코딩 폴백 → config.py 정합 (우선순위: strategy_cfg(DB) > settings)
|
|
"RSI_OVERSOLD_LEVEL": strategy_cfg.get("RSI_OVERSOLD_LEVEL", settings.RSI_OVERSOLD_LEVEL), # 30.0 동일
|
|
"STOP_LOSS_PERCENT": strategy_cfg.get("STOP_LOSS_PERCENT", -abs(settings.STOP_LOSS_PERCENT)), # 옛 폴백 -6.0 → -7.0
|
|
"ATR_STOP_MULTIPLIER": strategy_cfg.get("ATR_STOP_MULTIPLIER", settings.ATR_STOP_MULTIPLIER), # 2.5 동일
|
|
"BOLLINGER_STDDEV": strategy_cfg.get("BOLLINGER_STDDEV", settings.BOLLINGER_STDDEV), # 2.0 동일
|
|
"MAX_POSITION_SIZE": strategy_cfg.get("MAX_POSITION_SIZE", settings.MAX_POSITION_SIZE), # 0.04 동일
|
|
"MAX_HOLDINGS_COUNT": strategy_cfg.get("MAX_HOLDINGS_COUNT", settings.MAX_HOLDINGS_COUNT), # 20 동일
|
|
}
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 백테스트로 파라미터 평가
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def _compute_score(sharpe: float, win_rate: float, mdd: float) -> float:
|
|
"""통합 평가 지표 계산: Sharpe*0.4 + win_rate*0.3 + (1-|mdd|/100)*0.3"""
|
|
mdd_penalty = max(0.0, 1.0 - abs(mdd) / 100.0)
|
|
return sharpe * 0.4 + (win_rate / 100.0) * 0.3 + mdd_penalty * 0.3
|
|
|
|
|
|
async def _evaluate_params(
|
|
params: Dict[str, Any],
|
|
stock_codes: List[str],
|
|
start_date: date,
|
|
end_date: date,
|
|
initial_capital: float,
|
|
) -> Tuple[float, float, float, float]:
|
|
"""
|
|
단일 파라미터 조합으로 백테스트 실행.
|
|
Returns: (score, sharpe, win_rate, mdd)
|
|
"""
|
|
try:
|
|
backtest_params = {
|
|
# [2026-09-27] 폴백 config.py 정합 (옛값: 30.0 / -6.0)
|
|
"MEAN_REVERSION_RSI_THRESHOLD": params.get("RSI_OVERSOLD_LEVEL", settings.RSI_OVERSOLD_LEVEL),
|
|
"MEAN_REVERSION_STOP_LOSS_PCT": params.get("STOP_LOSS_PERCENT", -abs(settings.STOP_LOSS_PERCENT)),
|
|
"MAX_TOTAL_POSITIONS": min(5, len(stock_codes)),
|
|
}
|
|
config = BacktestConfig(
|
|
stock_codes=stock_codes,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
initial_capital=initial_capital,
|
|
strategy_params=backtest_params,
|
|
max_positions=min(5, len(stock_codes)),
|
|
position_size=params.get("MAX_POSITION_SIZE", 0.2),
|
|
)
|
|
engine = BacktestEngine()
|
|
result = await engine.run_backtest(config)
|
|
|
|
sharpe = float(result.sharpe_ratio or 0.0)
|
|
win_rate = float(result.win_rate or 0.0)
|
|
mdd = float(result.mdd_pct or 0.0)
|
|
score = _compute_score(sharpe, win_rate, mdd)
|
|
return score, sharpe, win_rate, mdd
|
|
except Exception as e:
|
|
warn_throttled(logger, "strategy_optimizer:184", f"백테스트 실패 (파라미터: {params}): {e}")
|
|
return -1.0, 0.0, 0.0, 0.0
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 대표 종목 선택
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
async def _get_representative_stocks(period_days: int = 90) -> List[str]:
|
|
"""
|
|
최근 N일 거래가 있는 대표 종목 10개 반환.
|
|
백테스트 실행 시간을 줄이기 위해 종목 수 제한.
|
|
"""
|
|
from app.models.models import TradeHistory
|
|
from sqlalchemy import func, and_
|
|
from datetime import datetime, timedelta
|
|
|
|
since = now_kst() - timedelta(days=period_days)
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
stmt = (
|
|
select(TradeHistory.stock_code, func.count(TradeHistory.id).label("cnt"))
|
|
.where(TradeHistory.created_at >= since)
|
|
.group_by(TradeHistory.stock_code)
|
|
.order_by(desc("cnt"))
|
|
.limit(10)
|
|
)
|
|
result = await db.execute(stmt)
|
|
codes = [row[0] for row in result.fetchall()]
|
|
|
|
# 거래 이력 없으면 stock_daily_prices에서 최근 종목 fallback
|
|
if not codes:
|
|
async with AsyncSessionLocal() as db:
|
|
stmt = (
|
|
select(StockDailyPrice.stock_code)
|
|
.where(StockDailyPrice.trade_date >= (date.today() - timedelta(days=period_days)))
|
|
.group_by(StockDailyPrice.stock_code)
|
|
.limit(10)
|
|
)
|
|
result = await db.execute(stmt)
|
|
codes = [row[0] for row in result.fetchall()]
|
|
|
|
return codes or []
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 파라미터 적용
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
async def apply_params(params: Dict[str, Any], reason: str = "auto_optimizer") -> bool:
|
|
"""strategy_cfg에 파라미터를 일괄 적용.
|
|
|
|
[2026-09-29] `settings.STRATEGY_AUTO_APPLY_ENABLED` (기본 **False**) 가 꺼져 있으면
|
|
**아무것도 쓰지 않고 False 를 반환한다.** 이 함수가 DB `strategy_configs` 를 바꾸는
|
|
유일한 지점이고(§0 우선순위 2위), 호출 게이트가 "최근 7일 SELL ≥ 5건"이라 5거래
|
|
표본의 그리드 서치가 RSI 진입·손절·포지션 크기를 덮어쓸 수 있다 — CLAUDE.md
|
|
§8.1-5 / §8.4(파라미터당 100거래) / §8.5-1 과 정면 충돌한다.
|
|
|
|
가드는 **`try` 밖**에 둔다(§4). try 안에 두면 예외 시 가드가 조용히 통과한다.
|
|
호출부(`run_optimization`)에도 같은 가드가 있고 상세 로그를 남긴다 — 여기는
|
|
다른 호출부가 생겨도 뚫리지 않게 하는 최종 방어선이다.
|
|
"""
|
|
if not settings.STRATEGY_AUTO_APPLY_ENABLED:
|
|
logger.warning(
|
|
"🚫 전략 파라미터 자동 적용 차단 (STRATEGY_AUTO_APPLY_ENABLED=False) — "
|
|
"적용하려던 값: %s (사유: %s). 반영하려면 전략설정 UI 에서 수동으로 하거나 "
|
|
"CLAUDE.md §8.4 백테스트 요건 충족 후 플래그를 켤 것.",
|
|
params, reason,
|
|
)
|
|
return False
|
|
|
|
try:
|
|
for key, value in params.items():
|
|
await strategy_cfg.upsert(key, str(value), updated_by=reason)
|
|
logger.info(f"✅ 전략 파라미터 자동 적용 완료: {list(params.keys())} (사유: {reason})")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"❌ 파라미터 적용 실패: {e}", exc_info=True)
|
|
return False
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 최적화 이력 저장
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
async def save_optimization_log(db: AsyncSession, result: OptimizationResult, trigger: str) -> StrategyOptimizationLog:
|
|
"""최적화 결과를 DB에 저장."""
|
|
log = StrategyOptimizationLog(
|
|
run_date=date.today(),
|
|
trigger=trigger,
|
|
period_days=90,
|
|
best_params=result.best_params,
|
|
best_sharpe=result.best_sharpe,
|
|
best_win_rate=result.best_win_rate,
|
|
best_mdd=result.best_mdd,
|
|
current_sharpe=result.current_sharpe,
|
|
previous_params=result.previous_params,
|
|
applied=result.applied,
|
|
apply_reason=result.apply_reason,
|
|
total_combinations=result.total_combinations,
|
|
)
|
|
db.add(log)
|
|
await db.commit()
|
|
await db.refresh(log)
|
|
return log
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 메인 최적화 함수
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
async def run_optimization(
|
|
db: AsyncSession,
|
|
perf_data: Dict[str, Any],
|
|
trigger: str = "auto",
|
|
period_days: int = 90,
|
|
initial_capital: float = 10_000_000,
|
|
) -> OptimizationResult:
|
|
"""
|
|
성과 데이터 기반 전략 파라미터 최적화.
|
|
|
|
Args:
|
|
db: AsyncSession
|
|
perf_data: analyze_performance() 반환값
|
|
trigger: "auto" | "manual"
|
|
period_days: 백테스트 기간 (일)
|
|
initial_capital: 백테스트 초기 자본
|
|
|
|
Returns:
|
|
OptimizationResult
|
|
"""
|
|
logger.info(f"전략 최적화 시작 (trigger={trigger}, period={period_days}일)")
|
|
|
|
# 1. 현재 파라미터 저장 (롤백용)
|
|
prev_params = await _get_current_params()
|
|
|
|
# 2. 그리드 구성
|
|
param_grid = _build_param_grid(perf_data)
|
|
combinations = _cartesian_product(param_grid)
|
|
total_combos = len(combinations)
|
|
logger.info(f" 그리드 조합 수: {total_combos}")
|
|
|
|
# 3. 대표 종목 선택
|
|
stock_codes = await _get_representative_stocks(period_days)
|
|
if not stock_codes:
|
|
logger.warning("백테스트 대상 종목 없음. 최적화 건너뜀.")
|
|
return OptimizationResult(
|
|
best_params=prev_params,
|
|
best_score=0.0,
|
|
best_sharpe=0.0,
|
|
best_win_rate=0.0,
|
|
best_mdd=0.0,
|
|
current_sharpe=0.0,
|
|
previous_params=prev_params,
|
|
applied=False,
|
|
apply_reason="no_stock_data",
|
|
total_combinations=0,
|
|
)
|
|
|
|
# ── IS/OOS 기간 분할 (70% IS / 30% OOS, 날짜 기준) ──
|
|
end_date = date.today()
|
|
start_date = end_date - timedelta(days=period_days)
|
|
is_days = int(period_days * 0.7)
|
|
is_end_date = start_date + timedelta(days=is_days)
|
|
oos_start_date = is_end_date + timedelta(days=1)
|
|
oos_end_date = end_date
|
|
|
|
logger.info(
|
|
f" IS 기간: {start_date} ~ {is_end_date} ({is_days}일), "
|
|
f"OOS 기간: {oos_start_date} ~ {oos_end_date} ({period_days - is_days}일)"
|
|
)
|
|
|
|
# 4. 현재 파라미터 기준 Sharpe 계산 (IS 기간)
|
|
cur_score, cur_sharpe, cur_wr, cur_mdd = await _evaluate_params(
|
|
prev_params, stock_codes, start_date, is_end_date, initial_capital
|
|
)
|
|
logger.info(f" 현재 IS 성과 -> Sharpe: {cur_sharpe:.3f}, 승률: {cur_wr:.1f}%, MDD: {cur_mdd:.1f}%")
|
|
|
|
# 5. 그리드 서치 — IS 기간에서 최적 파라미터 탐색 (배치 5개씩 병렬)
|
|
best_score = cur_score
|
|
best_params = dict(prev_params)
|
|
best_sharpe = cur_sharpe
|
|
best_win_rate = cur_wr
|
|
best_mdd = cur_mdd
|
|
|
|
batch_size = 5
|
|
for i in range(0, len(combinations), batch_size):
|
|
batch = combinations[i:i + batch_size]
|
|
tasks = [
|
|
_evaluate_params(p, stock_codes, start_date, is_end_date, initial_capital)
|
|
for p in batch
|
|
]
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
for combo, res in zip(batch, results):
|
|
if isinstance(res, Exception):
|
|
continue
|
|
score, sharpe, wr, mdd = res
|
|
if score > best_score:
|
|
best_score = score
|
|
best_sharpe = sharpe
|
|
best_win_rate = wr
|
|
best_mdd = mdd
|
|
best_params = {**prev_params, **combo}
|
|
|
|
logger.info(f" IS 최적 성과 -> Sharpe: {best_sharpe:.3f}, 승률: {best_win_rate:.1f}%, MDD: {best_mdd:.1f}%")
|
|
|
|
# 6. OOS 기간에서 최적 파라미터 검증 (과적합 탐지)
|
|
oos_score, oos_sharpe, oos_wr, oos_mdd = await _evaluate_params(
|
|
best_params, stock_codes, oos_start_date, oos_end_date, initial_capital
|
|
)
|
|
is_oos_ratio = (oos_sharpe / best_sharpe) if best_sharpe > 0 else 0.0
|
|
logger.info(
|
|
f" OOS 검증 -> Sharpe: {oos_sharpe:.3f}, 승률: {oos_wr:.1f}%, MDD: {oos_mdd:.1f}%, "
|
|
f"IS/OOS Sharpe 비율: {is_oos_ratio:.2f}"
|
|
)
|
|
|
|
if is_oos_ratio < 0.5 and best_sharpe > 0:
|
|
logger.warning(
|
|
f" [과적합 경고] OOS Sharpe / IS Sharpe = {is_oos_ratio:.2f} < 0.5 "
|
|
"-- 파라미터가 IS 기간에 과적합되었을 가능성 높음"
|
|
)
|
|
|
|
# 7. 자동 적용 여부 결정
|
|
# 조건: IS Sharpe 10% 이상 개선 AND OOS Sharpe > 0 (OOS에서도 양의 성과)
|
|
improvement_threshold = 1.1
|
|
_is_improved = (
|
|
cur_sharpe > 0
|
|
and best_sharpe >= cur_sharpe * improvement_threshold
|
|
and best_params != prev_params
|
|
)
|
|
if cur_sharpe <= 0 and best_sharpe > 0:
|
|
_is_improved = True
|
|
|
|
# OOS 검증 게이트: OOS Sharpe > 0 이어야 실전 적용 허가
|
|
_oos_passed = oos_sharpe > 0
|
|
_should_apply = _is_improved and _oos_passed
|
|
|
|
applied = False
|
|
apply_reason = ""
|
|
|
|
if _should_apply:
|
|
improvement_pct = ((best_sharpe - cur_sharpe) / max(abs(cur_sharpe), 0.001)) * 100
|
|
apply_reason = (
|
|
f"IS Sharpe {improvement_pct:.1f}% 향상 ({cur_sharpe:.3f} -> {best_sharpe:.3f}), "
|
|
f"OOS Sharpe: {oos_sharpe:.3f}, IS/OOS 비율: {is_oos_ratio:.2f}"
|
|
)
|
|
# 최적화된 파라미터만 적용 (prev와 다른 키만)
|
|
changed = {k: v for k, v in best_params.items() if str(v) != str(prev_params.get(k, ""))}
|
|
# 가드는 try 밖 (§4) — 예외로 통과되면 실매매 파라미터가 바뀐다
|
|
if changed and not settings.STRATEGY_AUTO_APPLY_ENABLED:
|
|
_diff = ", ".join(
|
|
f"{k}: {prev_params.get(k, '(없음)')} → {v}" for k, v in sorted(changed.items())
|
|
)
|
|
logger.warning(
|
|
"🚫 [Optimizer] 자동 적용 차단 (STRATEGY_AUTO_APPLY_ENABLED=False). "
|
|
"적용될 뻔한 변경 %d건 — %s | IS Sharpe %.3f → %.3f, OOS Sharpe %.3f, "
|
|
"IS/OOS %.2f, 승률 %.1f%%, MDD %.1f%% | %s",
|
|
len(changed), _diff, cur_sharpe, best_sharpe, oos_sharpe, is_oos_ratio,
|
|
best_win_rate, best_mdd, apply_reason,
|
|
)
|
|
apply_reason = (
|
|
f"자동 적용 차단(STRATEGY_AUTO_APPLY_ENABLED=False) — 제안: {_diff} | 근거: {apply_reason}"
|
|
)
|
|
elif changed:
|
|
applied = await apply_params(changed, reason="auto_optimizer")
|
|
if applied:
|
|
logger.info(f"자동 적용 완료: {apply_reason}")
|
|
else:
|
|
if cur_sharpe <= 0 and best_sharpe <= 0:
|
|
apply_reason = "insufficient_data"
|
|
elif not _oos_passed and _is_improved:
|
|
apply_reason = f"OOS 검증 실패 (OOS Sharpe: {oos_sharpe:.3f} <= 0, 과적합 의심)"
|
|
elif best_sharpe < cur_sharpe * improvement_threshold:
|
|
pct = ((best_sharpe - cur_sharpe) / max(abs(cur_sharpe), 0.001)) * 100
|
|
apply_reason = f"IS 개선 미달 ({pct:+.1f}% < 10%)"
|
|
else:
|
|
apply_reason = "no_improvement"
|
|
logger.info(f"자동 적용 건너뜀: {apply_reason}")
|
|
|
|
return OptimizationResult(
|
|
best_params=best_params,
|
|
best_score=best_score,
|
|
best_sharpe=best_sharpe,
|
|
best_win_rate=best_win_rate,
|
|
best_mdd=best_mdd,
|
|
current_sharpe=cur_sharpe,
|
|
previous_params=prev_params,
|
|
applied=applied,
|
|
apply_reason=apply_reason,
|
|
total_combinations=total_combos,
|
|
oos_sharpe=oos_sharpe,
|
|
oos_win_rate=oos_wr,
|
|
oos_mdd=oos_mdd,
|
|
is_oos_sharpe_ratio=round(is_oos_ratio, 4),
|
|
)
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 마지막 최적화 이력 조회
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
async def get_last_optimization_date(db: AsyncSession) -> Optional[date]:
|
|
"""마지막 최적화 실행일 반환 (없으면 None)."""
|
|
stmt = select(StrategyOptimizationLog.run_date).order_by(
|
|
desc(StrategyOptimizationLog.created_at)
|
|
).limit(1)
|
|
result = await db.execute(stmt)
|
|
row = result.scalar_one_or_none()
|
|
return row
|