퀀트 팩터 투자 백테스팅: Python으로 모멘텀·밸류 전략 구현 가이드

# 기본 데이터 분석
pip install pandas numpy scipy matplotlib seaborn

# 주가 데이터 수집
pip install yfinance pykrx

# 백테스팅 프레임워크
pip install vectorbt backtrader

# 성과 분석
pip install quantstats pyfolio-reloaded

import pandas as pd
import numpy as np
from pykrx import stock
from datetime import datetime, timedelta

def get_kospi_universe(date: str) -> pd.DataFrame:
    """
    특정 날짜 기준 KOSPI 상장 종목의 기본 정보를 수집한다.
    date 형식: 'YYYYMMDD'
    """
    tickers = stock.get_market_ticker_list(date, market="KOSPI")

    records = []
    for ticker in tickers:
        try:
            name = stock.get_market_ticker_name(ticker)
            # 기본 재무 지표 (PER, PBR, 배당수익률)
            fundamental = stock.get_market_fundamental_by_ticker(
                date, market="KOSPI"
            )
            if ticker in fundamental.index:
                row = fundamental.loc[ticker]
                records.append({
                    'ticker': ticker,
                    'name': name,
                    'PER': row.get('PER', np.nan),
                    'PBR': row.get('PBR', np.nan),
                    'DIV': row.get('DIV', np.nan),
                })
        except Exception as e:
            continue

    return pd.DataFrame(records)


def get_price_data(
    tickers: list,
    start: str,
    end: str
) -> pd.DataFrame:
    """
    여러 종목의 종가 데이터를 수집하여 DataFrame으로 반환한다.
    tickers: 종목코드 리스트 (예: ['005930', '000660'])
    start, end: 'YYYYMMDD' 형식
    """
    price_dict = {}
    for ticker in tickers:
        try:
            df = stock.get_market_ohlcv_by_date(start, end, ticker)
            price_dict[ticker] = df['종가']
        except Exception:
            continue

    return pd.DataFrame(price_dict)


# 사용 예시
end_date = '20260301'
start_date = '20240301'  # 2년치 데이터

universe = get_kospi_universe(end_date)
print(f"KOSPI 유니버스: {len(universe)}종목")

# 상위 200종목 시가총액 기준 (유동성 확보)
cap_df = stock.get_market_cap_by_ticker(end_date, market="KOSPI")
top200 = cap_df.nlargest(200, '시가총액').index.tolist()

prices = get_price_data(top200, start_date, end_date)
print(f"수집 완료: {prices.shape}")

import yfinance as yf

def get_sp500_data(period: str = '5y') -> pd.DataFrame:
    """S&P 500 구성 종목의 종가 데이터를 수집한다."""
    # 위키피디아에서 S&P 500 목록 가져오기
    sp500_url = (
        'https://en.wikipedia.org/wiki/'
        'List_of_S%26P_500_companies'
    )
    table = pd.read_html(sp500_url)[0]
    tickers = table['Symbol'].str.replace('.', '-').tolist()

    # yfinance로 일괄 다운로드
    data = yf.download(
        tickers,
        period=period,
        auto_adjust=True,
        threads=True
    )
    return data['Close']


# S&P 500 종가 데이터
sp500_prices = get_sp500_data('5y')
print(f"S&P 500 데이터: {sp500_prices.shape}")

def calculate_momentum_factor(
    prices: pd.DataFrame,
    lookback: int = 252,
    skip_recent: int = 21
) -> pd.DataFrame:
    """
    12-1 모멘텀 팩터를 계산한다.

    Parameters
    ----------
    prices : pd.DataFrame
        종가 데이터 (columns: 종목, index: 날짜)
    lookback : int
        모멘텀 계산 기간 (거래일 기준, 기본 252일 = 약 12개월)
    skip_recent : int
        최근 제외 기간 (거래일 기준, 기본 21일 = 약 1개월)

    Returns
    -------
    pd.DataFrame
        각 종목의 모멘텀 스코어
    """
    # 전체 lookback 기간 수익률
    total_return = prices.pct_change(lookback)

    # 최근 skip_recent 기간 수익률
    recent_return = prices.pct_change(skip_recent)

    # 12-1 모멘텀: 전체에서 최근 제외
    # (1 + r_total) / (1 + r_recent) - 1
    momentum = (1 + total_return) / (1 + recent_return) - 1

    return momentum


def create_momentum_portfolio(
    prices: pd.DataFrame,
    rebalance_freq: str = 'M',
    n_quantiles: int = 5,
    long_quantile: int = 5,
    short_quantile: int = 1
) -> pd.DataFrame:
    """
    모멘텀 기반 롱숏 포트폴리오를 구성한다.

    Parameters
    ----------
    prices : pd.DataFrame
        종가 데이터
    rebalance_freq : str
        리밸런싱 빈도 ('M': 월간, 'Q': 분기)
    n_quantiles : int
        분위수 개수 (기본 5 = 퀸타일)
    long_quantile : int
        매수할 분위 (기본 5 = 상위 20%)
    short_quantile : int
        매도할 분위 (기본 1 = 하위 20%)

    Returns
    -------
    pd.DataFrame
        일별 포트폴리오 수익률
    """
    momentum = calculate_momentum_factor(prices)

    # 리밸런싱 날짜 추출
    rebal_dates = prices.resample(rebalance_freq).last().index

    daily_returns = prices.pct_change()
    portfolio_returns = pd.Series(index=prices.index, dtype=float)
    portfolio_returns.iloc[:] = 0.0

    for i in range(len(rebal_dates) - 1):
        date = rebal_dates[i]
        next_date = rebal_dates[i + 1]

        # 해당 날짜의 모멘텀 스코어
        mom_scores = momentum.loc[:date].iloc[-1].dropna()

        if len(mom_scores) < n_quantiles * 2:
            continue

        # 분위수 할당
        quantiles = pd.qcut(
            mom_scores, n_quantiles, labels=False
        ) + 1

        # 롱 포트폴리오 (상위 모멘텀)
        long_stocks = quantiles[quantiles == long_quantile].index
        # 숏 포트폴리오 (하위 모멘텀)
        short_stocks = quantiles[quantiles == short_quantile].index

        # 동일가중 롱숏 수익률
        period_mask = (
            (daily_returns.index > date) &
            (daily_returns.index <= next_date)
        )
        long_ret = daily_returns.loc[
            period_mask, long_stocks
        ].mean(axis=1)
        short_ret = daily_returns.loc[
            period_mask, short_stocks
        ].mean(axis=1)

        portfolio_returns.loc[period_mask] = long_ret - short_ret

    return portfolio_returns.dropna()


# 실행
momentum_returns = create_momentum_portfolio(prices)
cumulative = (1 + momentum_returns).cumprod()
print(f"누적 수익률: {cumulative.iloc[-1]:.2%}")
print(f"연환산 수익률: {momentum_returns.mean() * 252:.2%}")
print(f"연환산 변동성: {momentum_returns.std() * np.sqrt(252):.2%}")
print(f"샤프 비율: {momentum_returns.mean() / momentum_returns.std() * np.sqrt(252):.2f}")

def calculate_value_factor(
    fundamentals: pd.DataFrame,
    metric: str = 'PBR'
) -> pd.Series:
    """
    밸류 팩터 스코어를 계산한다.
    PBR이 낮을수록 저평가 = 밸류 스코어가 높음.

    Parameters
    ----------
    fundamentals : pd.DataFrame
        종목별 재무 지표 (PBR, PER 등)
    metric : str
        사용할 밸류 지표 (기본 'PBR')

    Returns
    -------
    pd.Series
        밸류 스코어 (높을수록 저평가)
    """
    values = fundamentals[metric].copy()

    # 음수 또는 0 제거 (적자 기업, 자본잠식 등)
    values = values[values > 0]

    # 역수를 취해서 낮은 PBR = 높은 스코어
    value_score = 1.0 / values

    # 극단값 처리: 상하위 1% Winsorize
    lower = value_score.quantile(0.01)
    upper = value_score.quantile(0.99)
    value_score = value_score.clip(lower, upper)

    # Z-score 정규화
    value_score = (
        (value_score - value_score.mean()) / value_score.std()
    )

    return value_score


def backtest_value_strategy(
    prices: pd.DataFrame,
    fundamentals_by_date: dict,
    rebalance_freq: str = 'Q',
    top_pct: float = 0.2
) -> pd.Series:
    """
    밸류 전략 백테스트를 실행한다.

    Parameters
    ----------
    prices : pd.DataFrame
        종가 데이터
    fundamentals_by_date : dict
        날짜별 재무 데이터 딕셔너리
    rebalance_freq : str
        리밸런싱 빈도
    top_pct : float
        상위 비율 (기본 0.2 = 상위 20%)

    Returns
    -------
    pd.Series
        일별 포트폴리오 수익률
    """
    rebal_dates = prices.resample(rebalance_freq).last().index
    daily_returns = prices.pct_change()
    portfolio_returns = pd.Series(
        0.0, index=prices.index
    )

    for i in range(len(rebal_dates) - 1):
        date = rebal_dates[i]
        next_date = rebal_dates[i + 1]

        # 해당 분기의 재무 데이터
        date_key = date.strftime('%Y%m%d')
        if date_key not in fundamentals_by_date:
            continue

        fund = fundamentals_by_date[date_key]
        value_scores = calculate_value_factor(fund, 'PBR')

        # 상위 20% 저평가 종목 선정
        n_stocks = max(1, int(len(value_scores) * top_pct))
        selected = value_scores.nlargest(n_stocks).index

        # 동일가중 포트폴리오 수익률 계산
        valid_stocks = [
            s for s in selected if s in daily_returns.columns
        ]
        period_mask = (
            (daily_returns.index > date) &
            (daily_returns.index <= next_date)
        )
        if valid_stocks:
            port_ret = daily_returns.loc[
                period_mask, valid_stocks
            ].mean(axis=1)
            portfolio_returns.loc[period_mask] = port_ret

    return portfolio_returns.dropna()

def combine_factors(
    momentum_scores: pd.Series,
    value_scores: pd.Series,
    mom_weight: float = 0.5,
    val_weight: float = 0.5
) -> pd.Series:
    """
    모멘텀과 밸류 팩터를 결합한 복합 스코어를 계산한다.

    Parameters
    ----------
    momentum_scores : pd.Series
        모멘텀 Z-score
    value_scores : pd.Series
        밸류 Z-score
    mom_weight : float
        모멘텀 가중치
    val_weight : float
        밸류 가중치

    Returns
    -------
    pd.Series
        복합 팩터 스코어
    """
    # 공통 종목만 추출
    common = momentum_scores.index.intersection(
        value_scores.index
    )
    mom = momentum_scores.loc[common]
    val = value_scores.loc[common]

    # Z-score 정규화
    mom_z = (mom - mom.mean()) / mom.std()
    val_z = (val - val.mean()) / val.std()

    # 가중 합산
    combined = mom_weight * mom_z + val_weight * val_z

    return combined.sort_values(ascending=False)


def backtest_multifactor(
    prices: pd.DataFrame,
    fundamentals_by_date: dict,
    rebalance_freq: str = 'M',
    n_stocks: int = 30,
    mom_weight: float = 0.5,
    val_weight: float = 0.5
) -> dict:
    """
    멀티팩터 전략의 백테스트를 실행하고 성과 지표를 반환한다.
    """
    rebal_dates = prices.resample(rebalance_freq).last().index
    daily_returns = prices.pct_change()
    portfolio_returns = []

    for i in range(len(rebal_dates) - 1):
        date = rebal_dates[i]
        next_date = rebal_dates[i + 1]
        date_key = date.strftime('%Y%m%d')

        # 모멘텀 스코어 계산
        mom = calculate_momentum_factor(prices)
        mom_scores = mom.loc[:date].iloc[-1].dropna()

        # 밸류 스코어 계산
        if date_key in fundamentals_by_date:
            val_scores = calculate_value_factor(
                fundamentals_by_date[date_key]
            )
        else:
            continue

        # 복합 스코어
        combined = combine_factors(
            mom_scores, val_scores,
            mom_weight, val_weight
        )

        # 상위 n_stocks 종목 선정
        selected = combined.head(n_stocks).index.tolist()
        valid = [
            s for s in selected
            if s in daily_returns.columns
        ]

        period_mask = (
            (daily_returns.index > date) &
            (daily_returns.index <= next_date)
        )
        if valid:
            ret = daily_returns.loc[
                period_mask, valid
            ].mean(axis=1)
            portfolio_returns.append(ret)

    if not portfolio_returns:
        return {}

    result = pd.concat(portfolio_returns)

    # 성과 지표 계산
    annual_return = result.mean() * 252
    annual_vol = result.std() * np.sqrt(252)
    sharpe = annual_return / annual_vol if annual_vol > 0 else 0
    cumulative = (1 + result).cumprod()
    max_dd = (
        cumulative / cumulative.cummax() - 1
    ).min()

    return {
        'annual_return': annual_return,
        'annual_volatility': annual_vol,
        'sharpe_ratio': sharpe,
        'max_drawdown': max_dd,
        'cumulative_return': cumulative.iloc[-1] - 1,
        'daily_returns': result
    }

import quantstats as qs
import matplotlib.pyplot as plt

def generate_performance_report(
    returns: pd.Series,
    benchmark_returns: pd.Series = None,
    strategy_name: str = "Multifactor Strategy"
):
    """
    전략의 성과를 분석하고 시각화한다.

    Parameters
    ----------
    returns : pd.Series
        전략 일별 수익률
    benchmark_returns : pd.Series
        벤치마크 일별 수익률
    strategy_name : str
        전략 이름
    """
    # 기본 통계
    print(f"=== {strategy_name} 성과 분석 ===")
    print(f"연환산 수익률: {qs.stats.cagr(returns):.2%}")
    print(f"연환산 변동성: {qs.stats.volatility(returns):.2%}")
    print(f"샤프 비율: {qs.stats.sharpe(returns):.2f}")
    print(f"소르티노 비율: {qs.stats.sortino(returns):.2f}")
    print(f"최대 낙폭 (MDD): {qs.stats.max_drawdown(returns):.2%}")
    print(f"칼마 비율: {qs.stats.calmar(returns):.2f}")
    print(f"승률: {qs.stats.win_rate(returns):.2%}")

    # HTML 리포트 생성
    qs.reports.html(
        returns,
        benchmark=benchmark_returns,
        title=strategy_name,
        output=f'{strategy_name}_report.html'
    )

    # 핵심 차트
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))

    # 1. 누적 수익률
    cumulative = (1 + returns).cumprod()
    axes[0, 0].plot(cumulative.index, cumulative.values)
    axes[0, 0].set_title('Cumulative Returns')
    axes[0, 0].set_ylabel('Growth of $1')

    # 2. 드로다운
    drawdown = cumulative / cumulative.cummax() - 1
    axes[0, 1].fill_between(
        drawdown.index, drawdown.values, 0,
        alpha=0.5, color='red'
    )
    axes[0, 1].set_title('Drawdown')

    # 3. 월별 수익률 히트맵 데이터
    monthly = returns.resample('M').apply(
        lambda x: (1 + x).prod() - 1
    )
    axes[1, 0].bar(monthly.index, monthly.values)
    axes[1, 0].set_title('Monthly Returns')

    # 4. 롤링 샤프 비율 (12개월)
    rolling_sharpe = (
        returns.rolling(252).mean()
        / returns.rolling(252).std()
        * np.sqrt(252)
    )
    axes[1, 1].plot(rolling_sharpe.index, rolling_sharpe.values)
    axes[1, 1].axhline(y=0, color='r', linestyle='--')
    axes[1, 1].set_title('Rolling 12M Sharpe Ratio')

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

def apply_transaction_costs(
    returns: pd.Series,
    turnover: pd.Series,
    commission: float = 0.00015,   # 편도 수수료 0.015%
    tax: float = 0.0018,           # 증권거래세 0.18% (매도만)
    slippage: float = 0.001        # 슬리피지 0.1%
) -> pd.Series:
    """
    거래 비용을 반영한 순수익률을 계산한다.

    Parameters
    ----------
    returns : pd.Series
        거래비용 미반영 수익률
    turnover : pd.Series
        일별 회전율 (0~1, 포트폴리오 대비 교체 비율)
    commission : float
        편도 수수료율
    tax : float
        매도 시 세금 (한국: 증권거래세)
    slippage : float
        슬리피지 (시장 충격)

    Returns
    -------
    pd.Series
        비용 반영 순수익률
    """
    # 매수 비용: 수수료 + 슬리피지
    buy_cost = (commission + slippage) * turnover

    # 매도 비용: 수수료 + 세금 + 슬리피지
    sell_cost = (commission + tax + slippage) * turnover

    total_cost = buy_cost + sell_cost

    return returns - total_cost

import vectorbt as vbt

# 모멘텀 시그널 기반 간단한 백테스트 예시
def vectorbt_momentum_backtest(
    prices: pd.DataFrame,
    lookback_windows: list = [63, 126, 189, 252],
    skip_windows: list = [5, 10, 21]
):
    """
    VectorBT를 이용한 모멘텀 전략 파라미터 그리드 서치.
    다양한 lookback/skip 조합을 동시에 테스트한다.
    """
    results = {}

    for lookback in lookback_windows:
        for skip in skip_windows:
            # 모멘텀 계산
            total_ret = prices.pct_change(lookback)
            recent_ret = prices.pct_change(skip)
            momentum = (1 + total_ret) / (1 + recent_ret) - 1

            # 상위 20% 종목만 매수 시그널
            threshold = momentum.quantile(0.8, axis=1)
            entries = momentum.gt(threshold, axis=0)

            # 하위 20%로 내려가면 매도 시그널
            exit_threshold = momentum.quantile(0.5, axis=1)
            exits = momentum.lt(exit_threshold, axis=0)

            # VectorBT 포트폴리오
            pf = vbt.Portfolio.from_signals(
                close=prices,
                entries=entries,
                exits=exits,
                freq='1D',
                init_cash=100_000_000,  # 1억 원
                fees=0.00015,
                slippage=0.001
            )

            results[(lookback, skip)] = {
                'total_return': pf.total_return(),
                'sharpe': pf.sharpe_ratio(),
                'max_dd': pf.max_drawdown(),
                'calmar': pf.calmar_ratio()
            }

    # 결과를 DataFrame으로 정리
    results_df = pd.DataFrame(results).T
    results_df.index.names = ['lookback', 'skip']

    print("=== 파라미터 그리드 서치 결과 ===")
    print(results_df.sort_values(
        'sharpe', ascending=False
    ).head(10))

    return results_df