[퀀트] [퀀트투자] 101 Formulaic Alphas Python 구현 코드 (3/3)

관

관리자 Lv.1

02-10 02:06 · 조회 1 · 추천 0

101 Formulaic Alphas Python 구현

pandas + numpy 기반 실전 구현 코드

1. 기본 연산자 라이브러리

먼저 논문에서 사용하는 모든 연산자를 Python 함수로 정의합니다.

import pandas as pd
import numpy as np

# === 횚단면(Cross-sectional) 연산자 ===

def rank(df):
    """횚단면 순위 (0~1 정규화)"""
    return df.rank(axis=1, pct=True)

def scale(df, k=1):
    """절대값 합이 k가 되도록 스케일링"""
    return df.mul(k).div(df.abs().sum(axis=1), axis=0)

def signedpower(df, a):
    """부호 보존 거듭제곱"""
    return df.apply(np.sign) * df.abs().pow(a)

# === 시계열(Time-series) 연산자 ===

def delay(df, d):
    """d일 전 값"""
    return df.shift(d)

def delta(df, d):
    """d일간 변화량"""
    return df.diff(d)

def ts_sum(df, d):
    """과거 d일 합계"""
    return df.rolling(d).sum()

def ts_mean(df, d):
    """과거 d일 평균"""
    return df.rolling(d).mean()

def ts_max(df, d):
    """과거 d일 최대값"""
    return df.rolling(d).max()

def ts_min(df, d):
    """과거 d일 최소값"""
    return df.rolling(d).min()

def ts_argmax(df, d):
    """과거 d일 중 최대값 위치"""
    return df.rolling(d).apply(np.argmax, raw=True) + 1

def ts_argmin(df, d):
    """과거 d일 중 최소값 위치"""
    return df.rolling(d).apply(np.argmin, raw=True) + 1

def ts_rank(df, d):
    """시계열 순위 (과거 d일 중 현재값의 순위)"""
    def _rank(x):
        return pd.Series(x).rank(pct=True).iloc[-1]
    return df.rolling(d).apply(_rank, raw=True)

def ts_stddev(df, d):
    """과거 d일 표준편차"""
    return df.rolling(d).std()

def ts_corr(x, y, d):
    """과거 d일 상관관계"""
    return x.rolling(d).corr(y)

def ts_cov(x, y, d):
    """과거 d일 공분산"""
    return x.rolling(d).cov(y)

def decay_linear(df, d):
    """선형 가중 이동평균"""
    weights = np.arange(1, d + 1, dtype=float)
    weights /= weights.sum()
    return df.rolling(d).apply(
        lambda x: np.dot(x, weights), raw=True
    )

2. 데이터 준비

# yfinance로 데이터 가져오기 (예시)
import yfinance as yf

# 복수 종목 데이터 다운로드
tickers = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'NVDA',
           'META', 'TSLA', 'JPM', 'V', 'JNJ']

data = yf.download(tickers, start='2024-01-01', 
                   end='2025-12-31', group_by='column')

# DataFrame 구성 (종목 x 날짜 피벗 테이블)
open_  = data['Open']
close  = data['Close']
high   = data['High']
low    = data['Low']
volume = data['Volume']

# 파생 변수 계산
returns = close.pct_change()
vwap = (close + high + low) / 3  # 근사값 (실제는 거래량 가중)
adv20 = ts_mean(volume, 20)

3. 핵심 알파 15개 구현

# ===== Alpha #1 =====
def alpha001(close, returns):
    inner = close.copy()
    inner[returns < 0] = ts_stddev(returns, 20)
    return rank(ts_argmax(signedpower(inner, 2), 5)) - 0.5

# ===== Alpha #2 =====
def alpha002(open_, close, volume):
    return -1 * ts_corr(
        rank(delta(np.log(volume), 2)),
        rank((close - open_) / open_),
        6
    )

# ===== Alpha #3 =====
def alpha003(open_, volume):
    return -1 * ts_corr(rank(open_), rank(volume), 10)

# ===== Alpha #5 =====
def alpha005(open_, close, vwap):
    return (
        rank(open_ - ts_mean(vwap, 10)) *
        (-1 * rank(close - vwap).abs())
    )

# ===== Alpha #9 =====
def alpha009(close):
    d = delta(close, 1)
    cond1 = ts_min(d, 5) > 0
    cond2 = ts_max(d, 5) < 0
    result = -1 * d
    result[cond1 | cond2] = d[cond1 | cond2]
    return result

# ===== Alpha #12 =====
def alpha012(close, volume):
    return np.sign(delta(volume, 1)) * (-1 * delta(close, 1))

# ===== Alpha #13 =====
def alpha013(close, volume):
    return -1 * rank(ts_cov(rank(close), rank(volume), 5))

# ===== Alpha #17 =====
def alpha017(close, volume, adv20):
    return (
        -1 * rank(ts_rank(close, 10)) *
        rank(delta(delta(close, 1), 1)) *
        rank(ts_rank(volume / adv20, 5))
    )

# ===== Alpha #22 =====
def alpha022(close, high, volume):
    return -1 * (
        delta(ts_corr(high, volume, 5), 5) *
        rank(ts_stddev(close, 20))
    )

# ===== Alpha #23 =====
def alpha023(high):
    cond = ts_mean(high, 20) < high
    result = pd.DataFrame(0, index=high.index, columns=high.columns)
    result[cond] = -1 * delta(high, 2)[cond]
    return result

# ===== Alpha #33 =====
def alpha033(open_, close):
    return rank(-1 * (1 - open_ / close))

# ===== Alpha #34 =====
def alpha034(close, returns):
    return rank(
        (1 - rank(ts_stddev(returns, 2) / ts_stddev(returns, 5))) +
        (1 - rank(delta(close, 1)))
    )

# ===== Alpha #35 =====
def alpha035(close, high, low, volume, returns):
    return (
        ts_rank(volume, 32) *
        (1 - ts_rank(close + high - low, 16)) *
        (1 - ts_rank(returns, 32))
    )

# ===== Alpha #43 =====
def alpha043(close, volume, adv20):
    return (
        ts_rank(volume / adv20, 20) *
        ts_rank(-1 * delta(close, 7), 8)
    )

# ===== Alpha #101 =====
def alpha101(open_, close, high, low):
    return (close - open_) / (high - low + 0.001)

4. 실행 및 결과 확인

# 알파 계산
a001 = alpha001(close, returns)
a002 = alpha002(open_, close, volume)
a012 = alpha012(close, volume)
a017 = alpha017(close, volume, adv20)
a101 = alpha101(open_, close, high, low)

# 결과 확인
print("=== Alpha #12 (Latest) ===")
print(a012.tail(3))
print("\n=== Alpha #101 (Latest) ===")
print(a101.tail(3))

# 알파 간 상관관계 확인
alphas_df = pd.DataFrame({
    'A001': a001.mean(axis=1),
    'A002': a002.mean(axis=1),
    'A012': a012.mean(axis=1),
    'A017': a017.mean(axis=1),
    'A101': a101.mean(axis=1),
}).dropna()

print("\n=== 알파 간 상관관계 ===")
print(alphas_df.corr().round(3))

5. 백테스팅 프레임워크

def backtest_alpha(alpha_signal, returns, n_groups=5):
    """
    간단한 팩터 백테스트
    - alpha_signal: 알파 값 (DataFrame)
    - returns: 다음날 수익률
    - n_groups: 분위수 그룹 수
    """
    # 다음날 수익률
    fwd_returns = returns.shift(-1)
    
    results = {}
    for date in alpha_signal.dropna(how='all').index:
        signal = alpha_signal.loc[date].dropna()
        ret = fwd_returns.loc[date].reindex(signal.index).dropna()
        common = signal.index.intersection(ret.index)
        
        if len(common) < n_groups:
            continue
        
        # 분위수별 그룹화
        groups = pd.qcut(signal[common], n_groups, 
                        labels=False, duplicates='drop')
        
        for g in range(n_groups):
            mask = groups == g
            if mask.sum() > 0:
                avg_ret = ret[common][mask].mean()
                results.setdefault(g, []).append(avg_ret)
    
    # 결과 요약
    summary = pd.DataFrame({
        f'Q{g+1}': pd.Series(v).mean()
        for g, v in sorted(results.items())
    }, index=['Avg Daily Return'])
    
    # Long-Short 수익률
    summary['L/S'] = summary.iloc[0, -1] - summary.iloc[0, 0]
    return summary

# 예시: Alpha #12 백테스트
result = backtest_alpha(a012, returns)
print("\n=== Alpha #12 백테스트 결과 ===")
print(result.round(6))

6. 실전 활용 팁

주의사항

1. VWAP 계산: 실제 VWAP는 (sum(price * volume) / sum(volume))으로 tick 데이터가 필요합니다. (High+Low+Close)/3은 근사값입니다.

2. 데이터 품질: 주식 분할, 배당 조정된 가격(Adjusted) 사용을 권장합니다.

3. 생존자 편향: 상장 폐지 종목을 제외하면 백테스트 결과가 왕곡됩니다. Point-in-time 데이터를 사용하세요.

4. 거래 비용: 스프레드, 수수료, 슬리페이지를 반드시 반영해야 합니다. 단기 알파는 거래 비용에 민감합니다.

5. 데이터 양: 최소 200개 이상 종목, 2년 이상 데이터로 테스트해야 의미 있는 결과를 얻을 수 있습니다.

다음 단계 추천

1. 개별 알파 IC(Information Coefficient) 계산하여 예측력 평가

2. 여러 알파를 조합한 메가 알파 구성

3. 한국 주식 데이터(KRX)에 적용하여 국내 시장 테스트

4. 머신러닝으로 알파 가중치 최적화

참고 리소스:
- 논문: arxiv.org/abs/1601.00991
- GitHub: github.com/yli188/WorldQuant_alpha101_code
- pip install yfinance pandas numpy

◀ [퀀트투자] 101 Formulaic Alphas 주요 공식 상세 분석 (2/3) [2/10 장중] 코스피 5,300선 유지, 코스닥은 하락 전환 - 오늘 시장 정리 ▶

💬 0 로그인 후 댓글 작성

첫 댓글을 남겨보세요!

공유하기