Stock-trading-programming/app/strategy/volatility_breakout.py

"""
strategy/volatility_breakout.py
변동성 돌파 전략 + AI 필터
기획서 v2.1 기준
"""
import json
import logging
import os
from datetime import datetime
from app.config import (
    STRATEGY_K, TP1_PCT, TP2_PCT, TP1_RATIO,
    ENTRY_START, ENTRY_END,
    AI_CONTEXT_PATH, AI_MIN_SCORE,
    AI_BOOST_MULTI, MIN_TRADE_AMOUNT,
    KOSPI_MIN_CHG, TICKER_REENTRY_COOLDOWN_MIN
)

logger = logging.getLogger(__name__)

# AI fallback 기본값
DEFAULT_CONTEXT = {
    "trade_allowed"          : True,
    "market_sentiment"       : "중립",
    "sentiment_score"        : 50,
    "risk_level"             : "보통",
    "hot_sectors"            : [],
    "avoid_sectors"          : [],
    "boosted_tickers"        : [],
    "blacklist_tickers"      : [],
    "position_size_multiplier": 0.8,
    "reason"                 : "AI 판단 실패 - 기본값 적용",
}


class VolatilityBreakout:
    """
    변동성 돌파 전략
    목표가 = 당일 시가 + (전일 고가 - 전일 저가) × K
    현재가 >= 목표가 → 진입 신호
    """

    def __init__(self):
        self.context     = DEFAULT_CONTEXT.copy()
        self.prev_data   = {}   # ticker → {high, low, amount} 전일 데이터
        self.today_open  = {}   # ticker → 당일 시가
        self.targets     = {}   # ticker → 목표가
        self._entry_times: dict = {}  # ticker → 마지막 진입 datetime (쿨다운 추적)

    # ── AI 컨텍스트 로드 ──

    def load_ai_context(self) -> dict:
        """daily_context.json 로드, 실패 시 fallback"""
        try:
            if not os.path.exists(AI_CONTEXT_PATH):
                logger.warning("daily_context.json 없음 → fallback")
                return DEFAULT_CONTEXT.copy()

            with open(AI_CONTEXT_PATH, encoding="utf-8") as f:
                ctx = json.load(f)

            # 날짜 체크 (오늘 날짜 파일인지 확인)
            today = datetime.now().strftime("%Y-%m-%d")
            if ctx.get("date") != today:
                logger.warning(f"AI 컨텍스트 날짜 불일치: {ctx.get('date')} → fallback")
                return DEFAULT_CONTEXT.copy()

            # 최소 감성 점수 체크
            if ctx.get("sentiment_score", 50) < AI_MIN_SCORE:
                ctx["trade_allowed"] = False
                logger.info(f"감성점수 {ctx['sentiment_score']} < {AI_MIN_SCORE} → 거래 중단")

            self.context = ctx
            logger.info(
                f"AI 컨텍스트 로드: {ctx['market_sentiment']}({ctx['sentiment_score']}점) "
                f"/ {ctx['reason']}"
            )
            return ctx

        except Exception as e:
            logger.error(f"AI 컨텍스트 로드 실패: {e} → fallback")
            return DEFAULT_CONTEXT.copy()

    # ── 목표가 계산 ──

    def set_prev_data(self, ticker: str, high: float,
                      low: float, amount: float):
        """전일 고가/저가/거래대금 저장"""
        self.prev_data[ticker] = {
            "high"  : high,
            "low"   : low,
            "amount": amount,
        }
    
    def has_prev_data(self, ticker: str) -> bool:
        """전일 데이터 캐시 여부 확인"""
        return ticker in self.prev_data

    def set_today_open(self, ticker: str, open_price: float):
        """당일 시가로 목표가 계산"""
        prev = self.prev_data.get(ticker)
        if not prev:
            logger.info(f"목표가 제외({ticker}): 전일 데이터 없음")
            return
        if prev["amount"] < MIN_TRADE_AMOUNT:
            logger.info(
                f"목표가 제외({ticker}): 전일 거래대금 {prev['amount']/1e8:.0f}억"
                f" < 기준 {MIN_TRADE_AMOUNT/1e8:.0f}억"
            )
            return

        prev_range = prev["high"] - prev["low"]
        if prev_range <= 0:
            return

        target = open_price + prev_range * STRATEGY_K
        self.today_open[ticker] = open_price
        self.targets[ticker]    = target

    def get_target(self, ticker: str) -> float:
        return self.targets.get(ticker, 0.0)

    # ── 진입 신호 판단 ──

    def check_entry(self, ticker: str, name: str,
                    current_price: float, sector: str = "") -> dict:
        """
        진입 신호 체크
        반환: {"signal": bool, "reason": str, "boosted": bool, "multiplier": float}
        """
        result = {"signal": False, "reason": "", "boosted": False, "multiplier": 1.0}

        # 시간 체크
        now_dt = datetime.now()
        now = now_dt.strftime("%H:%M")
        if not (ENTRY_START <= now <= ENTRY_END):
            result["reason"] = f"진입 시간 외 ({now})"
            return result

        # 동일 종목 재진입 쿨다운 체크
        last_entry = self._entry_times.get(ticker)
        if last_entry is not None:
            elapsed = (now_dt - last_entry).total_seconds() / 60
            if elapsed < TICKER_REENTRY_COOLDOWN_MIN:
                result["reason"] = f"재진입 쿨다운 ({elapsed:.0f}분 / {TICKER_REENTRY_COOLDOWN_MIN}분)"
                return result

        # 목표가 확인
        target = self.targets.get(ticker, 0)
        if target <= 0:
            result["reason"] = "목표가 없음"
            return result

        # 기술적 조건: 현재가 >= 목표가
        if current_price < target:
            result["reason"] = f"목표가 미달 ({current_price:,} < {target:,.0f})"
            return result

        # ── AI 필터 ──
        ctx = self.context

        # trade_allowed 체크
        if not ctx.get("trade_allowed", True):
            result["reason"] = f"AI 거래 중단: {ctx.get('reason', '')}"
            return result

        # 블랙리스트 체크
        if ticker in ctx.get("blacklist_tickers", []):
            result["reason"] = "AI 블랙리스트"
            return result

        # 섹터 회피 체크
        avoid = ctx.get("avoid_sectors", [])
        if sector and any(s in sector for s in avoid):
            result["reason"] = f"AI 섹터 회피 ({sector})"
            return result

        # 부스트 체크
        boosted    = ticker in ctx.get("boosted_tickers", [])
        multiplier = ctx.get("position_size_multiplier", 1.0)
        if boosted:
            multiplier = min(multiplier * AI_BOOST_MULTI, 1.5)

        self._entry_times[ticker] = now_dt  # 진입 시간 기록
        result.update({
            "signal"    : True,
            "reason"    : f"목표가 돌파 ({current_price:,} >= {target:,.0f})",
            "boosted"   : boosted,
            "multiplier": multiplier,
            "target"    : target,
        })
        return result

    # ── 청산 신호 판단 ──

    def check_exit(self, ticker: str, entry_price: float,
                   current_price: float, qty: int,
                   tp1_done: bool, sl_pct: float) -> dict:
        """
        청산 신호 체크
        우선순위: 손절 > 1차 익절 > 2차 익절
        반환: {"signal": bool, "reason": str, "qty": int}
        """
        result = {"signal": False, "reason": "", "qty": 0}

        sl_price  = entry_price * (1 - sl_pct)
        tp1_price = entry_price * (1 + TP1_PCT)
        tp2_price = entry_price * (1 + TP2_PCT)

        # 손절 (최우선)
        if current_price <= sl_price:
            result.update({
                "signal": True,
                "reason": "SL",
                "qty"   : qty,
                "price" : sl_price,
            })
            return result

        # 2차 익절 — 잔여 전량 청산
        if current_price >= tp2_price:
            result.update({
                "signal": True,
                "reason": "TP2",
                "qty"   : qty,
                "price" : tp2_price,
            })
            return result

        # 1차 익절 (아직 안 했으면)
        if not tp1_done and current_price >= tp1_price:
            result.update({
                "signal": True,
                "reason": "TP1",
                "qty"   : max(1, int(qty * TP1_RATIO)),
                "price" : tp1_price,
            })
            return result

        return result