실전 전략 코드 템플릿

바로 사용 가능한 퀀트 투자 전략 Python 코드

1. 환경 설정

1.1 필수 패키지 설치

pip install pandas numpy matplotlib
pip install FinanceDataReader pykrx yfinance
pip install ta  # 기술적 지표 라이브러리

1.2 기본 임포트

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')

# 한글 폰트 설정 (차트용)
plt.rcParams['font.family'] = 'AppleGothic'  # Mac
# plt.rcParams['font.family'] = 'Malgun Gothic'  # Windows
plt.rcParams['axes.unicode_minus'] = False

# 데이터 수집
import FinanceDataReader as fdr

2. 데이터 수집 유틸리티

class DataCollector:
    """주식 데이터 수집 클래스"""

    @staticmethod
    def get_stock_data(ticker: str, start: str, end: str = None) -> pd.DataFrame:
        """
        개별 종목 데이터 수집

        Parameters:
        -----------
        ticker: 종목코드 (예: '005930' 삼성전자)
        start: 시작일 (예: '2020-01-01')
        end: 종료일 (기본: 오늘)
        """
        if end is None:
            end = datetime.now().strftime('%Y-%m-%d')

        df = fdr.DataReader(ticker, start, end)
        df = df.reset_index()
        df.columns = ['date', 'open', 'high', 'low', 'close', 'volume', 'change']
        return df

    @staticmethod
    def get_kospi_list() -> pd.DataFrame:
        """KOSPI 종목 리스트"""
        return fdr.StockListing('KOSPI')

    @staticmethod
    def get_kosdaq_list() -> pd.DataFrame:
        """KOSDAQ 종목 리스트"""
        return fdr.StockListing('KOSDAQ')

    @staticmethod
    def get_financial_data(ticker: str) -> dict:
        """
        재무 데이터 수집 (pykrx 사용)
        """
        try:
            from pykrx import stock
            # 최근 분기 재무 데이터
            today = datetime.now().strftime('%Y%m%d')
            fundamental = stock.get_market_fundamental(today, today, ticker)
            return fundamental.to_dict('records')[0] if len(fundamental) > 0 else {}
        except Exception as e:
            print(f"재무 데이터 수집 실패: {e}")
            return {}

    @staticmethod
    def get_multiple_stocks(tickers: list, start: str, end: str = None) -> dict:
        """
        여러 종목 데이터 한번에 수집

        Returns:
        --------
        dict: {ticker: DataFrame}
        """
        data = {}
        for ticker in tickers:
            try:
                df = DataCollector.get_stock_data(ticker, start, end)
                data[ticker] = df
                print(f"  ✓ {ticker} 수집 완료")
            except Exception as e:
                print(f"  ✗ {ticker} 수집 실패: {e}")
        return data


# 사용 예시
# collector = DataCollector()
# samsung = collector.get_stock_data('005930', '2023-01-01')

3. 전략 1: 이동평균 골든크로스

class GoldenCrossStrategy:
    """
    골든크로스/데드크로스 전략

    매수: 단기 이평선 > 장기 이평선 (골든크로스)
    매도: 단기 이평선 < 장기 이평선 (데드크로스)
    """

    def __init__(self, short_window: int = 20, long_window: int = 60):
        self.short_window = short_window
        self.long_window = long_window
        self.name = f"GoldenCross_{short_window}_{long_window}"

    def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """매매 신호 생성"""
        df = df.copy()

        # 이동평균 계산
        df['ma_short'] = df['close'].rolling(self.short_window).mean()
        df['ma_long'] = df['close'].rolling(self.long_window).mean()

        # 신호 생성
        df['signal'] = 0
        df.loc[df['ma_short'] > df['ma_long'], 'signal'] = 1  # 매수 신호
        df.loc[df['ma_short'] < df['ma_long'], 'signal'] = -1  # 매도 신호

        # 포지션 변화 (실제 매매 시점)
        df['position'] = df['signal'].diff()

        return df

    def backtest(self, df: pd.DataFrame, initial_capital: float = 10000000) -> dict:
        """백테스트 실행"""
        df = self.generate_signals(df)

        cash = initial_capital
        holdings = 0
        portfolio_values = []
        trades = []

        for i, row in df.iterrows():
            price = row['close']
            position_change = row['position']

            # 매수
            if position_change == 2:  # -1 → 1
                if cash > 0:
                    shares = cash // price
                    cost = shares * price
                    cash -= cost
                    holdings += shares
                    trades.append({'date': row['date'], 'type': 'BUY', 'price': price, 'shares': shares})

            # 매도
            elif position_change == -2:  # 1 → -1
                if holdings > 0:
                    revenue = holdings * price
                    cash += revenue
                    trades.append({'date': row['date'], 'type': 'SELL', 'price': price, 'shares': holdings})
                    holdings = 0

            # 포트폴리오 가치 기록
            total_value = cash + (holdings * price)
            portfolio_values.append({
                'date': row['date'],
                'total_value': total_value,
                'cash': cash,
                'holdings_value': holdings * price
            })

        return {
            'portfolio': pd.DataFrame(portfolio_values),
            'trades': pd.DataFrame(trades),
            'final_value': portfolio_values[-1]['total_value'],
            'return': (portfolio_values[-1]['total_value'] / initial_capital - 1) * 100
        }

    def plot(self, df: pd.DataFrame, result: dict):
        """결과 시각화"""
        df = self.generate_signals(df)

        fig, axes = plt.subplots(2, 1, figsize=(14, 10))

        # 주가 + 이동평균
        ax1 = axes[0]
        ax1.plot(df['date'], df['close'], label='종가', alpha=0.7)
        ax1.plot(df['date'], df['ma_short'], label=f'MA{self.short_window}', linestyle='--')
        ax1.plot(df['date'], df['ma_long'], label=f'MA{self.long_window}', linestyle='--')

        # 매수/매도 포인트
        buys = df[df['position'] == 2]
        sells = df[df['position'] == -2]
        ax1.scatter(buys['date'], buys['close'], marker='^', color='red', s=100, label='매수')
        ax1.scatter(sells['date'], sells['close'], marker='v', color='blue', s=100, label='매도')

        ax1.set_title(f'{self.name} 전략')
        ax1.legend()
        ax1.grid(True, alpha=0.3)

        # 포트폴리오 가치
        ax2 = axes[1]
        portfolio = result['portfolio']
        ax2.plot(portfolio['date'], portfolio['total_value'], label='포트폴리오')
        ax2.axhline(y=10000000, color='gray', linestyle='--', label='초기자본')
        ax2.set_title(f'포트폴리오 가치 (수익률: {result["return"]:.2f}%)')
        ax2.legend()
        ax2.grid(True, alpha=0.3)

        plt.tight_layout()
        plt.savefig(f'{self.name}_result.png', dpi=150)
        plt.show()


# 사용 예시
"""
# 데이터 수집
df = DataCollector.get_stock_data('005930', '2022-01-01')

# 전략 실행
strategy = GoldenCrossStrategy(short_window=20, long_window=60)
result = strategy.backtest(df)

print(f"최종 수익률: {result['return']:.2f}%")
print(f"거래 횟수: {len(result['trades'])}")

# 시각화
strategy.plot(df, result)
"""

4. 전략 2: RSI 역추세 전략

class RSIStrategy:
    """
    RSI 과매도/과매수 역추세 전략

    매수: RSI < 30 (과매도)
    매도: RSI > 70 (과매수)
    """

    def __init__(self, period: int = 14, oversold: int = 30, overbought: int = 70):
        self.period = period
        self.oversold = oversold
        self.overbought = overbought
        self.name = f"RSI_{period}_{oversold}_{overbought}"

    def calculate_rsi(self, prices: pd.Series) -> pd.Series:
        """RSI 계산"""
        delta = prices.diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=self.period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=self.period).mean()

        rs = gain / loss
        rsi = 100 - (100 / (1 + rs))
        return rsi

    def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """매매 신호 생성"""
        df = df.copy()

        df['rsi'] = self.calculate_rsi(df['close'])

        df['signal'] = 0
        df.loc[df['rsi'] < self.oversold, 'signal'] = 1   # 과매도 → 매수
        df.loc[df['rsi'] > self.overbought, 'signal'] = -1  # 과매수 → 매도

        # 신호 변화 시점만 추출
        df['position'] = df['signal'].diff()

        return df

    def backtest(self, df: pd.DataFrame, initial_capital: float = 10000000) -> dict:
        """백테스트 실행"""
        df = self.generate_signals(df)

        cash = initial_capital
        holdings = 0
        portfolio_values = []
        trades = []

        for i, row in df.iterrows():
            price = row['close']
            signal = row['signal']
            position_change = row['position']

            # 매수 신호 발생
            if position_change > 0 and signal == 1:
                if cash > 0:
                    shares = cash // price
                    cost = shares * price
                    cash -= cost
                    holdings += shares
                    trades.append({'date': row['date'], 'type': 'BUY', 'price': price, 'shares': shares, 'rsi': row['rsi']})

            # 매도 신호 발생
            elif position_change < 0 and signal == -1:
                if holdings > 0:
                    revenue = holdings * price
                    cash += revenue
                    trades.append({'date': row['date'], 'type': 'SELL', 'price': price, 'shares': holdings, 'rsi': row['rsi']})
                    holdings = 0

            total_value = cash + (holdings * price)
            portfolio_values.append({
                'date': row['date'],
                'total_value': total_value,
                'rsi': row['rsi']
            })

        return {
            'portfolio': pd.DataFrame(portfolio_values),
            'trades': pd.DataFrame(trades),
            'final_value': portfolio_values[-1]['total_value'],
            'return': (portfolio_values[-1]['total_value'] / initial_capital - 1) * 100
        }

    def plot(self, df: pd.DataFrame, result: dict):
        """결과 시각화"""
        df = self.generate_signals(df)

        fig, axes = plt.subplots(3, 1, figsize=(14, 12))

        # 주가
        ax1 = axes[0]
        ax1.plot(df['date'], df['close'], label='종가')

        trades_df = result['trades']
        if len(trades_df) > 0:
            buys = trades_df[trades_df['type'] == 'BUY']
            sells = trades_df[trades_df['type'] == 'SELL']
            ax1.scatter(buys['date'], buys['price'], marker='^', color='red', s=100, label='매수')
            ax1.scatter(sells['date'], sells['price'], marker='v', color='blue', s=100, label='매도')

        ax1.set_title('주가 및 매매 시점')
        ax1.legend()
        ax1.grid(True, alpha=0.3)

        # RSI
        ax2 = axes[1]
        ax2.plot(df['date'], df['rsi'], label='RSI', color='purple')
        ax2.axhline(y=self.oversold, color='green', linestyle='--', label=f'과매도 ({self.oversold})')
        ax2.axhline(y=self.overbought, color='red', linestyle='--', label=f'과매수 ({self.overbought})')
        ax2.fill_between(df['date'], self.oversold, df['rsi'].clip(upper=self.oversold), alpha=0.3, color='green')
        ax2.fill_between(df['date'], self.overbought, df['rsi'].clip(lower=self.overbought), alpha=0.3, color='red')
        ax2.set_ylim(0, 100)
        ax2.set_title('RSI')
        ax2.legend()
        ax2.grid(True, alpha=0.3)

        # 포트폴리오
        ax3 = axes[2]
        portfolio = result['portfolio']
        ax3.plot(portfolio['date'], portfolio['total_value'], label='포트폴리오')
        ax3.axhline(y=10000000, color='gray', linestyle='--', label='초기자본')
        ax3.set_title(f'포트폴리오 가치 (수익률: {result["return"]:.2f}%)')
        ax3.legend()
        ax3.grid(True, alpha=0.3)

        plt.tight_layout()
        plt.savefig(f'{self.name}_result.png', dpi=150)
        plt.show()

5. 전략 3: 볼린저 밴드 전략

class BollingerBandStrategy:
    """
    볼린저 밴드 전략

    매수: 가격이 하단 밴드 터치
    매도: 가격이 상단 밴드 터치
    """

    def __init__(self, window: int = 20, num_std: float = 2.0):
        self.window = window
        self.num_std = num_std
        self.name = f"BollingerBand_{window}_{num_std}"

    def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """매매 신호 생성"""
        df = df.copy()

        # 볼린저 밴드 계산
        df['ma'] = df['close'].rolling(self.window).mean()
        df['std'] = df['close'].rolling(self.window).std()
        df['upper'] = df['ma'] + (df['std'] * self.num_std)
        df['lower'] = df['ma'] - (df['std'] * self.num_std)

        # %B 계산 (0~1 사이 정규화)
        df['pct_b'] = (df['close'] - df['lower']) / (df['upper'] - df['lower'])

        # 신호 생성
        df['signal'] = 0
        df.loc[df['close'] <= df['lower'], 'signal'] = 1   # 하단 밴드 터치 → 매수
        df.loc[df['close'] >= df['upper'], 'signal'] = -1  # 상단 밴드 터치 → 매도

        df['position'] = df['signal'].diff()

        return df

    def backtest(self, df: pd.DataFrame, initial_capital: float = 10000000) -> dict:
        """백테스트 실행"""
        df = self.generate_signals(df)

        cash = initial_capital
        holdings = 0
        portfolio_values = []
        trades = []

        for i, row in df.iterrows():
            price = row['close']
            signal = row['signal']
            position_change = row['position']

            # 매수
            if position_change > 0 and signal == 1:
                if cash > 0:
                    shares = cash // price
                    cost = shares * price
                    cash -= cost
                    holdings += shares
                    trades.append({'date': row['date'], 'type': 'BUY', 'price': price, 'shares': shares})

            # 매도
            elif position_change < 0 and signal == -1:
                if holdings > 0:
                    revenue = holdings * price
                    cash += revenue
                    trades.append({'date': row['date'], 'type': 'SELL', 'price': price, 'shares': holdings})
                    holdings = 0

            total_value = cash + (holdings * price)
            portfolio_values.append({
                'date': row['date'],
                'total_value': total_value
            })

        return {
            'portfolio': pd.DataFrame(portfolio_values),
            'trades': pd.DataFrame(trades),
            'final_value': portfolio_values[-1]['total_value'],
            'return': (portfolio_values[-1]['total_value'] / initial_capital - 1) * 100
        }

    def plot(self, df: pd.DataFrame, result: dict):
        """결과 시각화"""
        df = self.generate_signals(df)

        fig, axes = plt.subplots(2, 1, figsize=(14, 10))

        # 주가 + 볼린저 밴드
        ax1 = axes[0]
        ax1.plot(df['date'], df['close'], label='종가', color='black')
        ax1.plot(df['date'], df['ma'], label='MA', color='blue', linestyle='--')
        ax1.fill_between(df['date'], df['lower'], df['upper'], alpha=0.2, color='blue', label='볼린저 밴드')

        trades_df = result['trades']
        if len(trades_df) > 0:
            buys = trades_df[trades_df['type'] == 'BUY']
            sells = trades_df[trades_df['type'] == 'SELL']
            ax1.scatter(buys['date'], buys['price'], marker='^', color='red', s=100, label='매수')
            ax1.scatter(sells['date'], sells['price'], marker='v', color='blue', s=100, label='매도')

        ax1.set_title(f'{self.name} 전략')
        ax1.legend()
        ax1.grid(True, alpha=0.3)

        # 포트폴리오
        ax2 = axes[1]
        portfolio = result['portfolio']
        ax2.plot(portfolio['date'], portfolio['total_value'], label='포트폴리오')
        ax2.axhline(y=10000000, color='gray', linestyle='--', label='초기자본')
        ax2.set_title(f'포트폴리오 가치 (수익률: {result["return"]:.2f}%)')
        ax2.legend()
        ax2.grid(True, alpha=0.3)

        plt.tight_layout()
        plt.savefig(f'{self.name}_result.png', dpi=150)
        plt.show()

6. 전략 4: 듀얼 모멘텀 전략

class DualMomentumStrategy:
    """
    듀얼 모멘텀 전략 (Gary Antonacci)

    1. 절대 모멘텀: 자산 수익률 > 무위험 수익률
    2. 상대 모멘텀: 여러 자산 중 수익률 높은 것 선택
    """

    def __init__(self, lookback: int = 12, risk_free_rate: float = 0.03):
        """
        Parameters:
        -----------
        lookback: 모멘텀 계산 기간 (월)
        risk_free_rate: 연간 무위험 수익률
        """
        self.lookback = lookback
        self.risk_free_rate = risk_free_rate
        self.monthly_rf = (1 + risk_free_rate) ** (1/12) - 1
        self.name = f"DualMomentum_{lookback}M"

    def calculate_momentum(self, prices: pd.Series) -> float:
        """모멘텀(수익률) 계산"""
        if len(prices) < self.lookback:
            return np.nan
        return (prices.iloc[-1] / prices.iloc[-self.lookback] - 1)

    def run_monthly(self, data: dict, start_date: str, end_date: str,
                    initial_capital: float = 10000000) -> dict:
        """
        월간 리밸런싱 백테스트

        Parameters:
        -----------
        data: {ticker: DataFrame} 형태의 데이터
        """
        # 월간 데이터로 리샘플링
        monthly_data = {}
        for ticker, df in data.items():
            df = df.set_index('date')
            monthly = df['close'].resample('M').last()
            monthly_data[ticker] = monthly

        # 공통 기간 추출
        combined = pd.DataFrame(monthly_data)
        combined = combined.dropna()

        cash = initial_capital
        holdings = {}  # {ticker: shares}
        current_position = None
        portfolio_values = []
        trades = []

        for i in range(self.lookback, len(combined)):
            date = combined.index[i]
            current_prices = combined.iloc[i]

            # 각 자산의 모멘텀 계산
            momentums = {}
            for ticker in combined.columns:
                prices = combined[ticker].iloc[i-self.lookback:i+1]
                mom = self.calculate_momentum(prices)
                momentums[ticker] = mom

            # 상대 모멘텀: 가장 높은 수익률 자산
            best_ticker = max(momentums, key=momentums.get)
            best_momentum = momentums[best_ticker]

            # 절대 모멘텀: 무위험 수익률보다 높은가?
            threshold = (1 + self.monthly_rf) ** self.lookback - 1

            if best_momentum > threshold:
                # 투자
                target_position = best_ticker
            else:
                # 현금 보유
                target_position = 'CASH'

            # 포지션 변경
            if target_position != current_position:
                # 기존 포지션 청산
                if current_position and current_position != 'CASH':
                    price = current_prices[current_position]
                    revenue = holdings.get(current_position, 0) * price
                    cash += revenue
                    trades.append({
                        'date': date,
                        'type': 'SELL',
                        'ticker': current_position,
                        'price': price,
                        'shares': holdings.get(current_position, 0)
                    })
                    holdings[current_position] = 0

                # 새 포지션 진입
                if target_position != 'CASH':
                    price = current_prices[target_position]
                    shares = cash // price
                    cost = shares * price
                    cash -= cost
                    holdings[target_position] = shares
                    trades.append({
                        'date': date,
                        'type': 'BUY',
                        'ticker': target_position,
                        'price': price,
                        'shares': shares
                    })

                current_position = target_position

            # 포트폴리오 가치 계산
            holdings_value = sum(
                holdings.get(t, 0) * current_prices.get(t, 0)
                for t in combined.columns
            )
            total_value = cash + holdings_value

            portfolio_values.append({
                'date': date,
                'total_value': total_value,
                'position': current_position,
                'momentums': momentums.copy()
            })

        return {
            'portfolio': pd.DataFrame(portfolio_values),
            'trades': pd.DataFrame(trades),
            'final_value': portfolio_values[-1]['total_value'],
            'return': (portfolio_values[-1]['total_value'] / initial_capital - 1) * 100
        }


# 사용 예시
"""
# 데이터 수집 (KOSPI, KOSDAQ, 채권 대용으로 KODEX 국고채)
tickers = {
    'KOSPI': 'KS11',      # KOSPI 지수
    'KOSDAQ': 'KQ11',     # KOSDAQ 지수
    'BOND': '148070',     # KODEX 국고채 (또는 다른 채권 ETF)
}

data = {}
for name, ticker in tickers.items():
    data[name] = DataCollector.get_stock_data(ticker, '2015-01-01')

# 전략 실행
strategy = DualMomentumStrategy(lookback=12)
result = strategy.run_monthly(data, '2016-01-01', '2024-12-31')
print(f"수익률: {result['return']:.2f}%")
"""

7. 전략 5: 가치+모멘텀 팩터 전략

class ValueMomentumStrategy:
    """
    가치 + 모멘텀 복합 팩터 전략

    1. 가치: PER, PBR 기준 저평가 종목
    2. 모멘텀: 최근 수익률 상위 종목
    3. 두 팩터 복합 점수로 종목 선정
    """

    def __init__(self, n_stocks: int = 20, value_weight: float = 0.5):
        self.n_stocks = n_stocks
        self.value_weight = value_weight
        self.momentum_weight = 1 - value_weight
        self.name = f"ValueMomentum_Top{n_stocks}"

    def calculate_scores(self, stock_data: pd.DataFrame) -> pd.DataFrame:
        """
        팩터 점수 계산

        Parameters:
        -----------
        stock_data: 종목별 데이터 (columns: ticker, per, pbr, return_12m, ...)
        """
        df = stock_data.copy()

        # 가치 점수 (낮을수록 좋음 → 역순위)
        df['per_rank'] = df['per'].rank(ascending=True, pct=True)
        df['pbr_rank'] = df['pbr'].rank(ascending=True, pct=True)
        df['value_score'] = (df['per_rank'] + df['pbr_rank']) / 2

        # 모멘텀 점수 (높을수록 좋음)
        df['momentum_score'] = df['return_12m'].rank(ascending=False, pct=True)

        # 복합 점수
        df['total_score'] = (
            df['value_score'] * self.value_weight +
            df['momentum_score'] * self.momentum_weight
        )

        return df

    def select_stocks(self, stock_data: pd.DataFrame) -> list:
        """상위 N개 종목 선정"""
        df = self.calculate_scores(stock_data)

        # 필터링 (기본 조건)
        df = df[df['per'] > 0]  # 적자 기업 제외
        df = df[df['pbr'] > 0]  # 자본잠식 제외

        # 상위 N개 선정
        top_stocks = df.nlargest(self.n_stocks, 'total_score')
        return top_stocks['ticker'].tolist()


# 종목 스크리닝 예시 코드
def screen_stocks():
    """월간 종목 스크리닝"""
    from pykrx import stock
    import time

    today = datetime.now().strftime('%Y%m%d')

    # KOSPI 종목 리스트
    tickers = stock.get_market_ticker_list(today, market="KOSPI")

    results = []
    for ticker in tickers[:100]:  # 테스트용 100개
        try:
            # 기본 정보
            fundamental = stock.get_market_fundamental(today, today, ticker)
            if fundamental.empty:
                continue

            per = fundamental['PER'].iloc[0]
            pbr = fundamental['PBR'].iloc[0]

            # 12개월 수익률
            ohlcv = stock.get_market_ohlcv(
                (datetime.now() - timedelta(days=365)).strftime('%Y%m%d'),
                today,
                ticker
            )
            if len(ohlcv) < 200:
                continue

            return_12m = (ohlcv['종가'].iloc[-1] / ohlcv['종가'].iloc[0] - 1) * 100

            results.append({
                'ticker': ticker,
                'per': per,
                'pbr': pbr,
                'return_12m': return_12m
            })

            time.sleep(0.1)  # API 부하 방지

        except Exception as e:
            continue

    return pd.DataFrame(results)

8. 성과 분석 유틸리티

class PerformanceAnalyzer:
    """백테스트 성과 분석"""

    @staticmethod
    def calculate_metrics(portfolio: pd.DataFrame, risk_free_rate: float = 0.03) -> dict:
        """종합 성과 지표 계산"""
        returns = portfolio['total_value'].pct_change().dropna()

        # 기본 수익률
        total_return = (portfolio['total_value'].iloc[-1] / portfolio['total_value'].iloc[0]) - 1
        n_days = len(portfolio)
        n_years = n_days / 252
        cagr = (1 + total_return) ** (1 / n_years) - 1

        # 리스크 지표
        volatility = returns.std() * np.sqrt(252)
        downside_returns = returns[returns < 0]
        downside_vol = downside_returns.std() * np.sqrt(252) if len(downside_returns) > 0 else 0

        # 위험조정 수익률
        sharpe = (cagr - risk_free_rate) / volatility if volatility > 0 else 0
        sortino = (cagr - risk_free_rate) / downside_vol if downside_vol > 0 else 0

        # MDD
        cummax = portfolio['total_value'].cummax()
        drawdown = (portfolio['total_value'] - cummax) / cummax
        mdd = drawdown.min()
        calmar = cagr / abs(mdd) if mdd != 0 else 0

        # 승률 (일간 기준)
        winning_days = (returns > 0).sum()
        total_days = len(returns)
        win_rate = winning_days / total_days if total_days > 0 else 0

        return {
            '총 수익률': f'{total_return:.2%}',
            'CAGR (연평균)': f'{cagr:.2%}',
            '변동성': f'{volatility:.2%}',
            '샤프 비율': f'{sharpe:.2f}',
            '소르티노 비율': f'{sortino:.2f}',
            '최대 낙폭 (MDD)': f'{mdd:.2%}',
            '칼마 비율': f'{calmar:.2f}',
            '승률 (일간)': f'{win_rate:.2%}',
        }

    @staticmethod
    def compare_strategies(results: dict) -> pd.DataFrame:
        """여러 전략 성과 비교"""
        comparison = []
        for name, result in results.items():
            metrics = PerformanceAnalyzer.calculate_metrics(result['portfolio'])
            metrics['전략'] = name
            comparison.append(metrics)
        return pd.DataFrame(comparison).set_index('전략')

    @staticmethod
    def plot_comparison(results: dict):
        """전략 비교 시각화"""
        fig, ax = plt.subplots(figsize=(14, 6))

        for name, result in results.items():
            portfolio = result['portfolio']
            normalized = portfolio['total_value'] / portfolio['total_value'].iloc[0] * 100
            ax.plot(portfolio['date'], normalized, label=name)

        ax.axhline(y=100, color='gray', linestyle='--', alpha=0.5)
        ax.set_title('전략별 수익률 비교 (초기값 = 100)')
        ax.set_ylabel('누적 수익률 (%)')
        ax.legend()
        ax.grid(True, alpha=0.3)

        plt.tight_layout()
        plt.savefig('strategy_comparison.png', dpi=150)
        plt.show()


# 사용 예시
"""
# 여러 전략 실행
results = {
    '골든크로스': golden_cross_result,
    'RSI': rsi_result,
    '볼린저밴드': bollinger_result,
}

# 비교 분석
comparison = PerformanceAnalyzer.compare_strategies(results)
print(comparison)

# 시각화
PerformanceAnalyzer.plot_comparison(results)
"""

9. 실행 예시 (통합)

def main():
    """메인 실행 함수"""

    # 1. 데이터 수집
    print("📊 데이터 수집 중...")
    df = DataCollector.get_stock_data('005930', '2022-01-01')  # 삼성전자
    print(f"  데이터 수: {len(df)}개")

    # 2. 전략 실행
    strategies = {
        '골든크로스': GoldenCrossStrategy(20, 60),
        'RSI': RSIStrategy(14, 30, 70),
        '볼린저밴드': BollingerBandStrategy(20, 2.0),
    }

    results = {}
    for name, strategy in strategies.items():
        print(f"\n🔄 {name} 전략 실행 중...")
        result = strategy.backtest(df)
        results[name] = result
        print(f"  수익률: {result['return']:.2f}%")
        print(f"  거래 횟수: {len(result['trades'])}")

    # 3. 비교 분석
    print("\n📈 성과 비교")
    comparison = PerformanceAnalyzer.compare_strategies(results)
    print(comparison)

    # 4. 시각화
    PerformanceAnalyzer.plot_comparison(results)


if __name__ == "__main__":
    main()

10. 주의사항

⚠️ 백테스트 ≠ 실전

1. 거래 비용: 수수료, 세금, 슬리피지 반드시 반영
2. 유동성: 실제 체결 가능 여부 확인
3. 과최적화: Out-of-sample 테스트 필수
4. 시장 변화: 과거 성과가 미래를 보장하지 않음

✅ 실전 적용 전 체크리스트

• [ ] 최소 5년 이상 백테스트
• [ ] 여러 시장 환경에서 검증 (상승장, 하락장, 횡보장)
• [ ] 거래 비용 반영 후에도 수익 발생
• [ ] MDD가 감당 가능한 수준
• [ ] 페이퍼 트레이딩으로 검증

#퀀트전략 #Python #백테스팅 #알고리즘트레이딩 #코드템플릿