알고리즘 트레이딩 시스템 구축: Python 기반 자동매매 아키텍처와 백테스트 전략

┌─────────────────────────────────────────────────────────┐
│                    Trading System                        │
│                                                          │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌─────────┐ │
│  │  Market   │  │ Strategy │  │   Risk   │  │  Order  │ │
│  │   Data    │──│  Engine  │──│ Manager  │──│Executor │ │
│  │  Handler  │  │          │  │          │  │         │ │
│  └──────────┘  └──────────┘  └──────────┘  └─────────┘ │
│       │              │              │             │      │
│       └──────────────┴──────────────┴─────────────┘      │
│                    Event Bus (Queue)                      │
│                                                          │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐              │
│  │ Position │  │  Logger  │  │ Portfolio│              │
│  │ Tracker  │  │ /Monitor │  │  Manager │              │
│  └──────────┘  └──────────┘  └──────────┘              │
└─────────────────────────────────────────────────────────┘

import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, Callable, Dict, List
from collections import defaultdict


class EventType(Enum):
    MARKET_DATA = "market_data"
    SIGNAL = "signal"
    ORDER = "order"
    FILL = "fill"
    POSITION_UPDATE = "position_update"
    RISK_BREACH = "risk_breach"
    SYSTEM_ERROR = "system_error"


@dataclass
class Event:
    event_type: EventType
    timestamp: datetime
    data: Dict[str, Any]
    source: str = ""
    event_id: str = field(default_factory=lambda: str(id(object())))


class EventBus:
    """
    비동기 이벤트 버스.
    각 이벤트 타입에 대해 여러 핸들러를 등록하고,
    이벤트 발생 시 등록된 모든 핸들러에 비동기적으로 전달한다.
    """

    def __init__(self):
        self._handlers: Dict[EventType, List[Callable]] = defaultdict(list)
        self._event_log: List[Event] = []
        self._max_log_size = 100_000

    def subscribe(self, event_type: EventType, handler: Callable) -> None:
        self._handlers[event_type].append(handler)

    def unsubscribe(self, event_type: EventType, handler: Callable) -> None:
        self._handlers[event_type].remove(handler)

    async def publish(self, event: Event) -> None:
        # 이벤트 로그 기록 (감사 추적용)
        self._event_log.append(event)
        if len(self._event_log) > self._max_log_size:
            self._event_log = self._event_log[-self._max_log_size:]

        handlers = self._handlers.get(event.event_type, [])
        if not handlers:
            return

        # 모든 핸들러에 병렬로 이벤트 전달
        tasks = []
        for handler in handlers:
            if asyncio.iscoroutinefunction(handler):
                tasks.append(handler(event))
            else:
                handler(event)

        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)

    def get_event_log(self, event_type: EventType = None, limit: int = 100) -> List[Event]:
        if event_type:
            filtered = [e for e in self._event_log if e.event_type == event_type]
            return filtered[-limit:]
        return self._event_log[-limit:]

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
import numpy as np
import pandas as pd


@dataclass
class Signal:
    symbol: str
    direction: str          # "BUY" or "SELL"
    strength: float         # 0.0 ~ 1.0, 시그널 강도
    strategy_name: str
    timestamp: datetime
    metadata: Dict[str, Any] = field(default_factory=dict)


class BaseStrategy(ABC):
    """모든 전략이 구현해야 하는 추상 인터페이스"""

    def __init__(self, name: str, params: Dict[str, Any]):
        self.name = name
        self.params = params
        self._data_buffer: Dict[str, pd.DataFrame] = {}

    @abstractmethod
    def on_bar(self, symbol: str, bar: pd.Series) -> Optional[Signal]:
        """새로운 OHLCV 바가 도착할 때 호출"""
        pass

    @abstractmethod
    def on_tick(self, symbol: str, price: float, volume: float) -> Optional[Signal]:
        """새로운 틱 데이터가 도착할 때 호출"""
        pass

    def update_data(self, symbol: str, df: pd.DataFrame) -> None:
        self._data_buffer[symbol] = df

    def get_required_history(self) -> int:
        """전략이 시그널 생성에 필요한 최소 과거 데이터 길이"""
        return self.params.get("lookback_period", 20)


class MomentumStrategy(BaseStrategy):
    """
    듀얼 모멘텀 전략 구현.
    절대 모멘텀(시계열 모멘텀)과 상대 모멘텀(크로스 섹셔널)을
    결합하여 시그널을 생성한다.
    """

    def __init__(self, params: Dict[str, Any] = None):
        default_params = {
            "lookback_period": 60,       # 모멘텀 산출 기간 (거래일)
            "skip_period": 5,            # 최근 N일 제외 (단기 반전 효과 방지)
            "entry_threshold": 0.02,     # 진입 시그널 최소 수익률
            "exit_threshold": -0.01,     # 청산 시그널 수익률
            "vol_lookback": 20,          # 변동성 산출 기간
        }
        if params:
            default_params.update(params)
        super().__init__("DualMomentum", default_params)

    def _calculate_momentum(self, prices: pd.Series) -> float:
        lookback = self.params["lookback_period"]
        skip = self.params["skip_period"]

        if len(prices) < lookback + skip:
            return 0.0

        # 최근 skip일을 제외한 lookback 기간의 수익률
        end_price = prices.iloc[-(skip + 1)]
        start_price = prices.iloc[-(lookback + skip)]
        return (end_price / start_price) - 1.0

    def _calculate_volatility(self, prices: pd.Series) -> float:
        vol_lookback = self.params["vol_lookback"]
        if len(prices) < vol_lookback:
            return 0.0
        returns = prices.pct_change().dropna().tail(vol_lookback)
        return returns.std() * np.sqrt(252)

    def on_bar(self, symbol: str, bar: pd.Series) -> Optional[Signal]:
        if symbol not in self._data_buffer:
            return None

        prices = self._data_buffer[symbol]["close"]
        momentum = self._calculate_momentum(prices)
        volatility = self._calculate_volatility(prices)

        # 변동성 조정 모멘텀 스코어
        if volatility > 0:
            vol_adj_momentum = momentum / volatility
        else:
            vol_adj_momentum = 0.0

        entry_threshold = self.params["entry_threshold"]
        exit_threshold = self.params["exit_threshold"]

        if momentum > entry_threshold and vol_adj_momentum > 0.5:
            strength = min(vol_adj_momentum / 2.0, 1.0)
            return Signal(
                symbol=symbol,
                direction="BUY",
                strength=strength,
                strategy_name=self.name,
                timestamp=datetime.now(),
                metadata={
                    "momentum": momentum,
                    "volatility": volatility,
                    "vol_adj_momentum": vol_adj_momentum,
                },
            )
        elif momentum < exit_threshold:
            return Signal(
                symbol=symbol,
                direction="SELL",
                strength=abs(vol_adj_momentum),
                strategy_name=self.name,
                timestamp=datetime.now(),
                metadata={"momentum": momentum, "volatility": volatility},
            )

        return None

    def on_tick(self, symbol: str, price: float, volume: float) -> Optional[Signal]:
        # 모멘텀 전략은 바 단위로 동작하므로 틱 레벨에서는 시그널 없음
        return None

class MeanReversionStrategy(BaseStrategy):
    """
    볼린저 밴드 기반 평균 회귀 전략.
    가격이 하단 밴드를 이탈하면 매수, 상단 밴드를 이탈하면 매도.
    RSI를 보조 확인 지표로 사용한다.
    """

    def __init__(self, params: Dict[str, Any] = None):
        default_params = {
            "bb_period": 20,
            "bb_std": 2.0,
            "rsi_period": 14,
            "rsi_oversold": 30,
            "rsi_overbought": 70,
            "min_holding_bars": 5,   # 최소 보유 기간
        }
        if params:
            default_params.update(params)
        super().__init__("MeanReversion", default_params)

    def _bollinger_bands(self, prices: pd.Series):
        period = self.params["bb_period"]
        std_mult = self.params["bb_std"]

        sma = prices.rolling(window=period).mean()
        std = prices.rolling(window=period).std()

        upper = sma + (std * std_mult)
        lower = sma - (std * std_mult)

        return upper, sma, lower

    def _rsi(self, prices: pd.Series) -> pd.Series:
        period = self.params["rsi_period"]
        delta = prices.diff()

        gain = delta.where(delta > 0, 0.0)
        loss = -delta.where(delta < 0, 0.0)

        avg_gain = gain.rolling(window=period, min_periods=period).mean()
        avg_loss = loss.rolling(window=period, min_periods=period).mean()

        rs = avg_gain / avg_loss
        rsi = 100.0 - (100.0 / (1.0 + rs))
        return rsi

    def on_bar(self, symbol: str, bar: pd.Series) -> Optional[Signal]:
        if symbol not in self._data_buffer:
            return None

        prices = self._data_buffer[symbol]["close"]
        bb_period = self.params["bb_period"]

        if len(prices) < bb_period + self.params["rsi_period"]:
            return None

        upper, sma, lower = self._bollinger_bands(prices)
        rsi = self._rsi(prices)

        current_price = prices.iloc[-1]
        current_rsi = rsi.iloc[-1]
        current_lower = lower.iloc[-1]
        current_upper = upper.iloc[-1]
        current_sma = sma.iloc[-1]

        # %B 지표: 현재가가 볼린저 밴드 내 어디에 위치하는지
        bb_width = current_upper - current_lower
        if bb_width > 0:
            pct_b = (current_price - current_lower) / bb_width
        else:
            return None

        # 매수 조건: 하단 밴드 이탈 + RSI 과매도
        if current_price <= current_lower and current_rsi < self.params["rsi_oversold"]:
            strength = min((self.params["rsi_oversold"] - current_rsi) / 30.0, 1.0)
            return Signal(
                symbol=symbol,
                direction="BUY",
                strength=strength,
                strategy_name=self.name,
                timestamp=datetime.now(),
                metadata={"pct_b": pct_b, "rsi": current_rsi, "bb_lower": current_lower},
            )

        # 매도 조건: 상단 밴드 이탈 + RSI 과매수
        if current_price >= current_upper and current_rsi > self.params["rsi_overbought"]:
            strength = min((current_rsi - self.params["rsi_overbought"]) / 30.0, 1.0)
            return Signal(
                symbol=symbol,
                direction="SELL",
                strength=strength,
                strategy_name=self.name,
                timestamp=datetime.now(),
                metadata={"pct_b": pct_b, "rsi": current_rsi, "bb_upper": current_upper},
            )

        return None

    def on_tick(self, symbol: str, price: float, volume: float) -> Optional[Signal]:
        return None

import vectorbt as vbt
import pandas as pd
import numpy as np


def run_momentum_backtest(
    symbols: list[str],
    start_date: str = "2020-01-01",
    end_date: str = "2025-12-31",
    momentum_window: int = 60,
    top_n: int = 5,
):
    """
    VectorBT를 활용한 크로스 섹셔널 모멘텀 백테스트.
    N개 종목 중 모멘텀 상위 top_n 종목을 매월 리밸런싱.
    """
    # 데이터 다운로드 (Yahoo Finance)
    prices = vbt.YFData.download(
        symbols,
        start=start_date,
        end=end_date,
    ).get("Close")

    # 모멘텀 스코어 계산: 60일 수익률 (최근 5일 제외)
    momentum = prices.shift(5).pct_change(momentum_window)

    # 월말 리밸런싱 시그널 생성
    month_end = momentum.resample("M").last()

    # 각 월말에 모멘텀 상위 top_n 종목 선택
    entries = pd.DataFrame(False, index=prices.index, columns=prices.columns)
    exits = pd.DataFrame(False, index=prices.index, columns=prices.columns)

    for date in month_end.index:
        if date not in prices.index:
            continue
        scores = month_end.loc[date].dropna()
        if len(scores) < top_n:
            continue
        top_symbols = scores.nlargest(top_n).index
        bottom_symbols = scores.index.difference(top_symbols)

        # 가장 가까운 거래일 찾기
        nearest_date = prices.index[prices.index.get_indexer([date], method="ffill")[0]]
        entries.loc[nearest_date, top_symbols] = True
        exits.loc[nearest_date, bottom_symbols] = True

    # 포트폴리오 시뮬레이션
    portfolio = vbt.Portfolio.from_signals(
        prices,
        entries=entries,
        exits=exits,
        size=1.0 / top_n,          # 균등 배분
        size_type="targetpercent",
        fees=0.001,                # 거래 수수료 0.1%
        slippage=0.001,            # 슬리피지 0.1%
        freq="1D",
        init_cash=100_000_000,     # 초기 자본 1억 원
    )

    # 성과 지표 출력
    print("=== 백테스트 결과 ===")
    print(f"총 수익률: {portfolio.total_return():.2%}")
    print(f"연환산 수익률: {portfolio.annualized_return():.2%}")
    print(f"샤프 비율: {portfolio.sharpe_ratio():.4f}")
    print(f"소르티노 비율: {portfolio.sortino_ratio():.4f}")
    print(f"최대 낙폭(MDD): {portfolio.max_drawdown():.2%}")
    print(f"칼마 비율: {portfolio.calmar_ratio():.4f}")
    print(f"승률: {portfolio.trades.win_rate():.2%}")

    return portfolio

from dataclasses import dataclass
from enum import Enum
from typing import Dict, Optional, Tuple
import logging

logger = logging.getLogger(__name__)


class RiskAction(Enum):
    APPROVE = "approve"
    REDUCE_SIZE = "reduce_size"
    REJECT = "reject"
    EMERGENCY_CLOSE = "emergency_close"


@dataclass
class RiskCheckResult:
    action: RiskAction
    original_size: float
    adjusted_size: float
    reasons: list[str]


class RiskManager:
    """
    다층 리스크 관리 시스템.
    Layer 1: 개별 주문 리스크 (포지션 사이즈, 슬리피지 예측)
    Layer 2: 포트폴리오 리스크 (섹터 집중도, 상관관계)
    Layer 3: 계좌 리스크 (일일 손실 한도, 드로다운 한도)
    Layer 4: 시스템 리스크 (킬 스위치, 서킷 브레이커)
    """

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.daily_pnl = 0.0
        self.weekly_pnl = 0.0
        self.peak_equity = config.get("initial_capital", 100_000_000)
        self.current_equity = self.peak_equity
        self.positions: Dict[str, float] = {}
        self.killed = False  # 킬 스위치 상태

        # 리스크 한도 설정
        self.max_position_pct = config.get("max_position_pct", 0.05)      # 종목당 최대 5%
        self.max_sector_pct = config.get("max_sector_pct", 0.25)          # 섹터당 최대 25%
        self.max_daily_loss_pct = config.get("max_daily_loss_pct", 0.03)  # 일일 최대 손실 3%
        self.max_weekly_loss_pct = config.get("max_weekly_loss_pct", 0.05)  # 주간 최대 손실 5%
        self.max_drawdown_pct = config.get("max_drawdown_pct", 0.15)      # 최대 낙폭 15%
        self.max_correlation = config.get("max_correlation", 0.8)          # 종목간 최대 상관관계
        self.risk_per_trade_pct = config.get("risk_per_trade_pct", 0.01)  # 거래당 리스크 1%

    def check_order(self, signal: Signal, current_price: float) -> RiskCheckResult:
        """주문 전 다층 리스크 검증 수행"""
        reasons = []

        if self.killed:
            return RiskCheckResult(
                action=RiskAction.REJECT,
                original_size=0,
                adjusted_size=0,
                reasons=["KILL SWITCH ACTIVATED - 모든 신규 주문 거부"],
            )

        # --- Layer 1: 개별 주문 리스크 ---
        position_size = self._calculate_position_size(signal, current_price)

        # Kelly Criterion 기반 최적 사이즈 조정
        kelly_size = self._kelly_criterion_size(signal)
        position_size = min(position_size, kelly_size)

        # --- Layer 2: 포트폴리오 리스크 ---
        position_value = position_size * current_price
        position_pct = position_value / self.current_equity

        if position_pct > self.max_position_pct:
            old_size = position_size
            position_size = (self.max_position_pct * self.current_equity) / current_price
            reasons.append(
                f"포지션 한도 초과: {position_pct:.1%} -> {self.max_position_pct:.1%}"
            )

        # --- Layer 3: 계좌 리스크 ---
        daily_loss_pct = abs(self.daily_pnl) / self.current_equity

        if daily_loss_pct >= self.max_daily_loss_pct:
            logger.warning(f"일일 손실 한도 도달: {daily_loss_pct:.2%}")
            return RiskCheckResult(
                action=RiskAction.REJECT,
                original_size=position_size,
                adjusted_size=0,
                reasons=[f"일일 손실 한도 {self.max_daily_loss_pct:.1%} 도달"],
            )

        # 드로다운 체크
        current_drawdown = (self.peak_equity - self.current_equity) / self.peak_equity
        if current_drawdown >= self.max_drawdown_pct:
            logger.critical(f"최대 낙폭 도달: {current_drawdown:.2%}")
            self.killed = True
            return RiskCheckResult(
                action=RiskAction.EMERGENCY_CLOSE,
                original_size=position_size,
                adjusted_size=0,
                reasons=[
                    f"최대 낙폭 {self.max_drawdown_pct:.1%} 도달 - 킬 스위치 발동"
                ],
            )

        action = RiskAction.REDUCE_SIZE if reasons else RiskAction.APPROVE
        return RiskCheckResult(
            action=action,
            original_size=position_size,
            adjusted_size=position_size,
            reasons=reasons if reasons else ["모든 리스크 검증 통과"],
        )

    def _calculate_position_size(self, signal: Signal, price: float) -> float:
        """
        ATR 기반 포지션 사이징.
        거래당 리스크 금액 = 계좌 자본 * risk_per_trade_pct
        포지션 사이즈 = 리스크 금액 / (ATR * 2)
        """
        risk_amount = self.current_equity * self.risk_per_trade_pct
        atr = signal.metadata.get("atr", price * 0.02)  # ATR 없으면 가격의 2% 사용
        stop_distance = atr * 2  # 2 ATR 스톱

        if stop_distance <= 0:
            return 0

        size = risk_amount / stop_distance
        return int(size)

    def _kelly_criterion_size(self, signal: Signal) -> float:
        """
        켈리 기준(Kelly Criterion)에 기반한 최적 포지션 사이즈.
        f* = (bp - q) / b
        여기서 b=보상/리스크 비율, p=승률, q=패률
        실전에서는 풀 켈리의 25~50%만 사용 (Half-Kelly 또는 Quarter-Kelly)
        """
        win_rate = signal.metadata.get("win_rate", 0.5)
        avg_win = signal.metadata.get("avg_win", 0.02)
        avg_loss = signal.metadata.get("avg_loss", 0.01)

        if avg_loss == 0:
            return float("inf")

        b = avg_win / avg_loss  # Payoff ratio
        p = win_rate
        q = 1 - p

        kelly_fraction = (b * p - q) / b

        # Half-Kelly 적용 (보수적 접근)
        kelly_fraction = max(kelly_fraction * 0.5, 0)

        return self.current_equity * kelly_fraction

    def update_pnl(self, pnl: float) -> None:
        self.daily_pnl += pnl
        self.weekly_pnl += pnl
        self.current_equity += pnl
        self.peak_equity = max(self.peak_equity, self.current_equity)

    def reset_daily(self) -> None:
        logger.info(f"일일 P&L 리셋: {self.daily_pnl:,.0f}")
        self.daily_pnl = 0.0

    def reset_weekly(self) -> None:
        logger.info(f"주간 P&L 리셋: {self.weekly_pnl:,.0f}")
        self.weekly_pnl = 0.0

    def activate_kill_switch(self, reason: str) -> None:
        logger.critical(f"킬 스위치 발동: {reason}")
        self.killed = True

    def deactivate_kill_switch(self) -> None:
        logger.warning("킬 스위치 해제 - 수동 확인 필요")
        self.killed = False

import asyncio
from dataclasses import dataclass
from typing import List, Optional
import logging

logger = logging.getLogger(__name__)


@dataclass
class ChildOrder:
    parent_order_id: str
    symbol: str
    side: str
    quantity: int
    scheduled_time: datetime
    executed: bool = False
    fill_price: Optional[float] = None
    fill_quantity: int = 0


class TWAPExecutor:
    """
    TWAP(Time-Weighted Average Price) 실행 알고리즘.
    총 주문 수량을 지정된 시간 동안 균등하게 분할하여 실행한다.
    """

    def __init__(self, broker_client, config: Dict[str, Any] = None):
        self.broker = broker_client
        self.config = config or {}
        self.max_participation_rate = self.config.get("max_participation_rate", 0.1)
        self.randomize_interval = self.config.get("randomize_interval", True)

    async def execute(
        self,
        symbol: str,
        side: str,
        total_quantity: int,
        duration_minutes: int = 60,
        num_slices: int = 12,
    ) -> List[ChildOrder]:
        """
        TWAP 실행.
        total_quantity를 num_slices개의 자식 주문으로 분할하여
        duration_minutes 동안 균등 간격으로 실행한다.
        """
        slice_quantity = total_quantity // num_slices
        remainder = total_quantity % num_slices
        interval_seconds = (duration_minutes * 60) / num_slices

        child_orders = []
        order_id = f"TWAP-{symbol}-{datetime.now().strftime('%Y%m%d%H%M%S')}"

        for i in range(num_slices):
            qty = slice_quantity + (1 if i < remainder else 0)

            # 예측 가능한 패턴 방지를 위한 간격 랜덤화
            if self.randomize_interval and i > 0:
                jitter = np.random.uniform(-0.3, 0.3) * interval_seconds
                wait_time = interval_seconds + jitter
            else:
                wait_time = interval_seconds

            if i > 0:
                await asyncio.sleep(wait_time)

            # 현재 시장 상태 확인
            current_spread = await self.broker.get_spread(symbol)
            current_volume = await self.broker.get_current_volume(symbol)

            # 참여율 체크: 현재 거래량 대비 주문 비율 제한
            if current_volume > 0:
                participation = qty / current_volume
                if participation > self.max_participation_rate:
                    qty = int(current_volume * self.max_participation_rate)
                    logger.warning(
                        f"참여율 제한 적용: {participation:.2%} -> "
                        f"{self.max_participation_rate:.2%}, 수량: {qty}"
                    )

            # 스프레드가 비정상적으로 넓은 경우 대기
            avg_spread = await self.broker.get_avg_spread(symbol, period="1h")
            if current_spread > avg_spread * 3:
                logger.warning(
                    f"스프레드 이상 감지: {current_spread} vs avg {avg_spread}. "
                    f"주문 지연 30초"
                )
                await asyncio.sleep(30)

            # 지정가 주문으로 실행 (시장가 대비 슬리피지 감소)
            child = ChildOrder(
                parent_order_id=order_id,
                symbol=symbol,
                side=side,
                quantity=qty,
                scheduled_time=datetime.now(),
            )

            try:
                result = await self.broker.place_limit_order(
                    symbol=symbol,
                    side=side,
                    quantity=qty,
                    price=await self._calculate_limit_price(symbol, side),
                )
                child.executed = True
                child.fill_price = result.get("fill_price")
                child.fill_quantity = result.get("fill_quantity", qty)
            except Exception as e:
                logger.error(f"자식 주문 실행 실패 [{i+1}/{num_slices}]: {e}")
                child.executed = False

            child_orders.append(child)

        # 실행 결과 요약
        filled = [o for o in child_orders if o.executed]
        total_filled = sum(o.fill_quantity for o in filled)
        avg_price = (
            sum(o.fill_price * o.fill_quantity for o in filled) / total_filled
            if total_filled > 0
            else 0
        )

        logger.info(
            f"TWAP 실행 완료: {symbol} {side} "
            f"{total_filled}/{total_quantity} 체결, "
            f"평균가 {avg_price:,.0f}"
        )

        return child_orders

    async def _calculate_limit_price(self, symbol: str, side: str) -> float:
        """매수 시 매도호가, 매도 시 매수호가에 소폭 유리하게 지정"""
        orderbook = await self.broker.get_orderbook(symbol)
        if side == "BUY":
            # 매도 1호가에서 약간 아래로
            return orderbook["asks"][0]["price"] * 0.9995
        else:
            # 매수 1호가에서 약간 위로
            return orderbook["bids"][0]["price"] * 1.0005

class SlippageModel:
    """
    현실적인 슬리피지 모델.
    시장 충격(Market Impact), 스프레드, 지연(Latency)을
    종합적으로 모델링한다.
    """

    def __init__(self, config: Dict[str, Any] = None):
        self.config = config or {}
        # 시장 충격 계수 (Almgren-Chriss 모델 기반)
        self.impact_coefficient = self.config.get("impact_coefficient", 0.1)
        self.permanent_impact_ratio = self.config.get("permanent_impact_ratio", 0.3)

    def estimate_slippage(
        self,
        order_size: int,
        price: float,
        daily_volume: int,
        spread: float,
        volatility: float,
    ) -> Dict[str, float]:
        """
        예상 슬리피지를 다중 요인으로 추정한다.

        Args:
            order_size: 주문 수량
            price: 현재가
            daily_volume: 일 평균 거래량
            spread: 매수-매도 스프레드
            volatility: 일일 변동성
        """
        participation_rate = order_size / daily_volume if daily_volume > 0 else 1.0

        # 1. 스프레드 비용: 항상 반 스프레드만큼 지불
        spread_cost = spread / 2.0

        # 2. 일시적 시장 충격 (Temporary Impact)
        # Almgren-Chriss: temporary_impact = eta * sigma * (Q/V)^0.6
        temp_impact = (
            self.impact_coefficient
            * volatility
            * (participation_rate ** 0.6)
            * price
        )

        # 3. 영구적 시장 충격 (Permanent Impact)
        perm_impact = temp_impact * self.permanent_impact_ratio

        # 4. 타이밍 리스크 (실행 지연에 따른 가격 변동)
        timing_risk = volatility * price * 0.01  # 대략적 추정

        total_slippage = spread_cost + temp_impact + perm_impact + timing_risk
        slippage_bps = (total_slippage / price) * 10000  # basis points

        return {
            "total_slippage": total_slippage,
            "slippage_bps": slippage_bps,
            "spread_cost": spread_cost,
            "temporary_impact": temp_impact,
            "permanent_impact": perm_impact,
            "timing_risk": timing_risk,
            "participation_rate": participation_rate,
        }

    def should_reduce_order(self, slippage_result: Dict) -> Tuple[bool, str]:
        """슬리피지가 임계값을 초과하면 주문 축소 권고"""
        if slippage_result["slippage_bps"] > 50:  # 50bp 이상
            return True, "슬리피지 50bp 초과 - 주문 분할 또는 축소 필요"
        if slippage_result["participation_rate"] > 0.1:  # 10% 이상
            return True, "참여율 10% 초과 - 시장 충격 위험"
        return False, "정상 범위"

class CircuitBreaker:
    """
    서킷 브레이커: 비정상적 시장 상황을 감지하고
    자동으로 트레이딩을 중단한다.
    """

    def __init__(self, config: Dict[str, Any] = None):
        self.config = config or {}
        self.price_history: Dict[str, list] = {}
        self.triggered = False
        self.trigger_time: Optional[datetime] = None

        # 임계값 설정
        self.price_change_threshold = self.config.get("price_change_threshold", 0.05)
        self.time_window_seconds = self.config.get("time_window_seconds", 300)
        self.cooldown_minutes = self.config.get("cooldown_minutes", 30)
        self.spread_multiplier_threshold = self.config.get("spread_multiplier_threshold", 5.0)

    def check(self, symbol: str, current_price: float, current_spread: float,
              avg_spread: float, timestamp: datetime) -> bool:
        """
        서킷 브레이커 조건 체크.
        True 반환 시 트레이딩 중단 필요.
        """
        if self.triggered:
            elapsed = (timestamp - self.trigger_time).total_seconds() / 60
            if elapsed < self.cooldown_minutes:
                return True
            else:
                logger.info(f"서킷 브레이커 쿨다운 완료 ({self.cooldown_minutes}분)")
                self.triggered = False

        # 가격 이력 업데이트
        if symbol not in self.price_history:
            self.price_history[symbol] = []

        self.price_history[symbol].append((timestamp, current_price))

        # 시간 윈도우 밖의 데이터 제거
        cutoff = timestamp.timestamp() - self.time_window_seconds
        self.price_history[symbol] = [
            (t, p) for t, p in self.price_history[symbol]
            if t.timestamp() > cutoff
        ]

        if len(self.price_history[symbol]) < 2:
            return False

        # 조건 1: 급격한 가격 변동
        oldest_price = self.price_history[symbol][0][1]
        price_change = abs(current_price - oldest_price) / oldest_price

        if price_change >= self.price_change_threshold:
            self._trigger(f"가격 급변 감지: {price_change:.2%} "
                         f"({self.time_window_seconds}초 이내)")
            return True

        # 조건 2: 비정상적 스프레드 확대
        if avg_spread > 0 and current_spread > avg_spread * self.spread_multiplier_threshold:
            self._trigger(f"스프레드 비정상 확대: {current_spread/avg_spread:.1f}배")
            return True

        return False

    def _trigger(self, reason: str) -> None:
        self.triggered = True
        self.trigger_time = datetime.now()
        logger.critical(f"서킷 브레이커 발동: {reason}")

# 나쁜 예: Python 루프
def slow_moving_average(prices, window):
    result = []
    for i in range(len(prices)):
        if i < window:
            result.append(None)
        else:
            result.append(sum(prices[i-window:i]) / window)
    return result

# 좋은 예: Pandas 벡터 연산
def fast_moving_average(prices: pd.Series, window: int) -> pd.Series:
    return prices.rolling(window=window).mean()

from numba import njit

@njit
def fast_rsi(prices: np.ndarray, period: int = 14) -> np.ndarray:
    """Numba로 최적화된 RSI 계산"""
    n = len(prices)
    rsi = np.empty(n)
    rsi[:period] = np.nan

    gains = np.zeros(n)
    losses = np.zeros(n)

    for i in range(1, n):
        change = prices[i] - prices[i-1]
        if change > 0:
            gains[i] = change
        else:
            losses[i] = -change

    avg_gain = np.mean(gains[1:period+1])
    avg_loss = np.mean(losses[1:period+1])

    if avg_loss == 0:
        rsi[period] = 100.0
    else:
        rs = avg_gain / avg_loss
        rsi[period] = 100.0 - (100.0 / (1.0 + rs))

    for i in range(period + 1, n):
        avg_gain = (avg_gain * (period - 1) + gains[i]) / period
        avg_loss = (avg_loss * (period - 1) + losses[i]) / period

        if avg_loss == 0:
            rsi[i] = 100.0
        else:
            rs = avg_gain / avg_loss
            rsi[i] = 100.0 - (100.0 / (1.0 + rs))

    return rsi

# CSV 대비 Parquet의 이점
# - 컬럼 기반 저장으로 필요한 컬럼만 읽기 가능
# - 자동 압축으로 저장 공간 50~80% 절감
# - 데이터 타입 보존 (CSV의 타입 추론 문제 없음)

import pyarrow.parquet as pq

# 저장
df.to_parquet("market_data.parquet", engine="pyarrow", compression="snappy")

# 특정 컬럼만 읽기 (메모리 효율적)
df = pd.read_parquet("market_data.parquet", columns=["close", "volume"])