퀀트(Quant)는 수학·통계·프로그래밍을 이용해 시장에서 수익을 내는 방법입니다. 감(感)이나 뉴스가 아니라 데이터와 규칙으로 매매 결정을 내립니다.

import yfinance as yf
import pandas as pd
import numpy as np

# 1. 데이터 다운로드 (예: NASDAQ 100 구성 종목 일부)
tickers = ["AAPL", "MSFT", "NVDA", "AMZN", "META",
           "GOOGL", "TSLA", "AVGO", "AMD", "INTC"]

prices = yf.download(tickers, start="2022-01-01", end="2024-12-31")["Adj Close"]

# 2. 모멘텀 팩터: 최근 6개월(126거래일) 수익률
momentum = prices.pct_change(126)  # 각 종목의 6개월 수익률

# 3. 매월 리밸런싱: 상위 3개 종목 매수, 하위 3개 종목 공매도
monthly = prices.resample("ME").last()  # 월말 가격
mom_monthly = momentum.resample("ME").last()

portfolio_returns = []
for date in mom_monthly.index[1:]:
    scores = mom_monthly.loc[date].dropna()
    if len(scores) < 6:
        continue

    top3 = scores.nlargest(3).index     # 상위 3개 (롱)
    bot3 = scores.nsmallest(3).index    # 하위 3개 (숏)

    # 다음 달 수익률로 측정 (date→next_date 기간 수익률)
    try:
        next_date = mom_monthly.index[mom_monthly.index.get_loc(date) + 1]
        # ⚠️ 주의: .pct_change()는 Series에서 인접 원소 간 변화율 → 잘못된 결과
        # 올바른 방법: date→next_date 두 시점의 가격 비율로 계산
        long_ret  = (monthly.loc[next_date, top3] / monthly.loc[date, top3] - 1).mean()
        short_ret = (monthly.loc[next_date, bot3] / monthly.loc[date, bot3] - 1).mean()
        portfolio_returns.append(long_ret - short_ret)  # 롱-숏 수익률
    except:
        pass

# 4. 성과 분석
ret_series = pd.Series(portfolio_returns)
total_return = (1 + ret_series).prod() - 1
sharpe = ret_series.mean() / ret_series.std() * np.sqrt(12)  # 연율화 샤프

print(f"총 수익률: {total_return:.1%}")
print(f"샤프 비율: {sharpe:.2f}")
print(f"MDD: {(ret_series.cumsum().cummax() - ret_series.cumsum()).max():.1%}")

상관관계가 높은 두 종목의 가격 차이(스프레드)가 역사적 평균에서 벗어날 때 — 비싼 쪽을 공매도하고 싼 쪽을 매수 — 다시 평균으로 수렴할 때 이익을 냅니다. 이를 평균 회귀(Mean Reversion)라 합니다.

import yfinance as yf
import pandas as pd
import numpy as np
from statsmodels.tsa.stattools import coint

# 1. BTC, ETH 데이터 (USD 기준)
btc = yf.download("BTC-USD", start="2023-01-01")["Adj Close"]
eth = yf.download("ETH-USD", start="2023-01-01")["Adj Close"]

# 2. 공적분 검정
score, pvalue, _ = coint(btc, eth)
print(f"공적분 p-value: {pvalue:.4f}")  # < 0.05이면 페어 유효

# 3. 스프레드 계산 (로그 가격 차이)
log_btc = np.log(btc)
log_eth = np.log(eth)

# 헤지 비율 계산 (OLS 회귀)
from numpy.linalg import lstsq
hedge_ratio = lstsq(log_eth.values.reshape(-1,1), log_btc.values, rcond=None)[0][0]
spread = log_btc - hedge_ratio * log_eth

# 4. Z-score (60일 롤링 윈도우)
window = 60
z_score = (spread - spread.rolling(window).mean()) / spread.rolling(window).std()

# 5. 매매 신호
signals = pd.DataFrame(index=z_score.index)
signals['z_score'] = z_score
signals['signal'] = ''
signals.loc[z_score > 2,  'signal'] = 'SHORT_BTC_LONG_ETH'
signals.loc[z_score < -2, 'signal'] = 'LONG_BTC_SHORT_ETH'
signals.loc[abs(z_score) < 0.5, 'signal'] = 'CLOSE'

# 현재 신호 출력
current = signals.iloc[-1]
print(f"현재 Z-score: {current['z_score']:.2f}")
print(f"현재 신호: {current['signal'] or '관망'}")

폴리마켓은 실제 돈으로 미래 사건의 확률에 베팅하는 탈중앙화 예측시장입니다. 시장 참여자들의 집단 지성이 뉴스·여론조사보다 정확한 확률을 만들어냅니다. 퀀트 관점에서는: 확률의 비효율을 찾아 시스템적으로 베팅하는 것이 목표입니다.

import requests
import pandas as pd
from datetime import datetime

GAMMA_API = "https://gamma-api.polymarket.com"

def get_active_markets(limit=20, min_volume=10000):
    """활성 마켓 목록 가져오기 (볼륨 기준 필터링)"""
    res = requests.get(f"{GAMMA_API}/markets", params={
        "active": True,
        "limit": limit,
        "order": "volume24hr",
        "ascending": False
    })
    markets = res.json()

    result = []
    for m in markets:
        vol = float(m.get("volume24hr", 0) or 0)
        if vol < min_volume:
            continue

        # YES 토큰 현재 가격 파싱
        tokens = m.get("tokens", [])
        yes_price = next((t["price"] for t in tokens if t.get("outcome") == "Yes"), None)

        result.append({
            "question": m.get("question", ""),
            "yes_price": float(yes_price) if yes_price else None,
            "volume_24h": vol,
            "end_date": m.get("endDate", "")[:10],
            "slug": m.get("slug", "")
        })

    return pd.DataFrame(result)

# 실행 예시
df = get_active_markets(limit=50, min_volume=5000)
print(df.to_string(index=False))

# 특정 가격 구간 필터 (0.3~0.7: 불확실성 높음 → 비효율 발견 가능성↑)
uncertain = df[(df['yes_price'] > 0.3) & (df['yes_price'] < 0.7)]
print(f"\n불확실 마켓 {len(uncertain)}개:")
print(uncertain[['question', 'yes_price', 'volume_24h']].to_string())

폴리마켓에서 지속적으로 수익 내는 플레이어들의 공통점은 단순 뉴스 베팅이 아닙니다. 여러 시장의 "잘 맞추는 사람들의 중복 지점"을 찾아 시스템적으로 포지션을 잡습니다.

가격 데이터에서 피처(Feature)를 추출하고, 레이블(Label)을 정의한 뒤, 모델이 미래 수익/방향을 예측하게 합니다. 감이 아닌 패턴과 확률로 매매합니다.

import yfinance as yf
import pandas as pd
import numpy as np
from xgboost import XGBClassifier
from sklearn.metrics import accuracy_score, classification_report

# 1. 데이터 & 피처 공학
df = yf.download("BTC-USD", start="2020-01-01")["Adj Close"].rename("close").to_frame()

# 기술적 지표 피처
df['ret_1d']  = df['close'].pct_change(1)
df['ret_5d']  = df['close'].pct_change(5)
df['ret_20d'] = df['close'].pct_change(20)

# RSI (14일)
delta = df['close'].diff()
gain = delta.clip(lower=0).rolling(14).mean()
loss = (-delta.clip(upper=0)).rolling(14).mean()
df['rsi'] = 100 - 100 / (1 + gain / loss)

# 볼린저 밴드 %B
ma20 = df['close'].rolling(20).mean()
std20 = df['close'].rolling(20).std()
df['bb_pct'] = (df['close'] - (ma20 - 2*std20)) / (4*std20)

# 2. 레이블: 5일 후 수익률
df['label'] = (df['close'].shift(-5) / df['close'] - 1 > 0.01).astype(int)

df = df.dropna()
features = ['ret_1d', 'ret_5d', 'ret_20d', 'rsi', 'bb_pct']

# 3. 시간순 분리 (70% 훈련, 30% 테스트)
split = int(len(df) * 0.7)
X_train, y_train = df[features][:split], df['label'][:split]
X_test,  y_test  = df[features][split:], df['label'][split:]

# 4. 모델 학습
model = XGBClassifier(n_estimators=100, max_depth=4, learning_rate=0.05,
                      eval_metric='logloss', random_state=42)
model.fit(X_train, y_train)

# 5. 평가
preds = model.predict(X_test)
print(f"테스트 정확도: {accuracy_score(y_test, preds):.3f}")
print(classification_report(y_test, preds, target_names=['관망', '매수']))

# 피처 중요도
importance = pd.Series(model.feature_importances_, index=features).sort_values(ascending=False)
print(f"\n피처 중요도:\n{importance}")

# 6. 현재 신호
latest = df[features].iloc[[-1]]
prob = model.predict_proba(latest)[0][1]
signal = "🟢 매수" if prob > 0.6 else ("🔴 매도" if prob < 0.4 else "⚪ 관망")
print(f"\n현재 상승 확률: {prob:.1%} → {signal}")

승률 60%짜리 전략도 잘못된 베팅 사이즈면 파산합니다. 반대로 승률 55%도 켈리 기준으로 포지션을 잡으면 장기 복리 수익을 최대화할 수 있습니다. 과베팅이 파산의 주원인입니다.

import numpy as np

def kelly_polymarket(my_prob, market_price, capital, half_kelly=True):
    """
    폴리마켓 켈리 기준 포지션 사이저
    my_prob   : 내가 추정한 YES 확률 (0~1)
    market_price: 현재 YES 토큰 가격 (= 시장 확률)
    capital   : 총 자본 (USDC)
    """
    if my_prob <= market_price:
        return {"action": "NO_EDGE", "bet_size": 0}

    # 켈리 공식: f* = (p - m) / (1 - m)
    f_star = (my_prob - market_price) / (1 - market_price)

    # Half-Kelly 적용 (추정 오차 대비)
    f_actual = f_star / 2 if half_kelly else f_star

    # 최대 포지션 제한 (자본의 20% 이하)
    f_capped = min(f_actual, 0.20)

    bet_usdc = capital * f_capped
    expected_profit = bet_usdc * (my_prob * (1 - market_price) - (1 - my_prob) * market_price) / market_price

    return {
        "action": "BUY_YES",
        "full_kelly_pct": round(f_star * 100, 1),
        "bet_pct": round(f_capped * 100, 1),
        "bet_size": round(bet_usdc, 2),
        "expected_profit": round(expected_profit, 2)
    }

# 예시 실행
result = kelly_polymarket(
    my_prob=0.75,          # 내 추정: 75%
    market_price=0.60,     # 시장 가격: 60%
    capital=1000,          # 자본: $1000
    half_kelly=True
)
print(f"액션:         {result['action']}")
print(f"풀켈리:       자본의 {result['full_kelly_pct']}%")
print(f"실제 베팅:    자본의 {result['bet_pct']}% = ${result['bet_size']}")
print(f"기대 수익:    ${result['expected_profit']}")

# 출력:
# 액션:         BUY_YES
# 풀켈리:       자본의 37.5%
# 실제 베팅:    자본의 18.8% = $187.5
# 기대 수익:    $37.5

동일한 전략도 시장 레짐(환경)에 따라 수익과 손실이 갈립니다. 모멘텀 전략은 추세장에서 강하고 횡보장에서 약합니다. 레짐을 감지해 전략을 자동으로 전환하는 것이 2025년 실전 퀀트의 핵심입니다.

import yfinance as yf
import pandas as pd
import numpy as np

# 1. 데이터: S&P500(SPY), VIX, 금(GLD)
spy = yf.download("SPY", start="2022-01-01")["Adj Close"]
vix = yf.download("^VIX", start="2022-01-01")["Adj Close"]

df = pd.DataFrame({'spy': spy, 'vix': vix}).dropna()
df['sma200'] = df['spy'].rolling(200).mean()
df['ret_vol'] = df['spy'].pct_change().rolling(20).std() * np.sqrt(252)

# 2. 레짐 감지 규칙 기반 (Rule-Based)
def detect_regime(row):
    if pd.isna(row['sma200']): return 'UNKNOWN'
    if row['vix'] > 30 or row['ret_vol'] > 0.30:
        return 'VOLATILE'   # 변동 레짐
    elif row['spy'] > row['sma200'] and row['vix'] < 20:
        return 'BULL'       # 강세 레짐
    else:
        return 'BEAR'       # 약세 레짐

df['regime'] = df.apply(detect_regime, axis=1)

# 3. 레짐별 전략 선택
STRATEGY = {
    'BULL':     '모멘텀 전략 ON — 상위 모멘텀 종목 롱',
    'BEAR':     '헤지 전략 ON  — 현금 비중 높이기, GLD 롱',
    'VOLATILE': '포지션 축소  — 켈리 사이즈 50% 감소',
}

current_regime = df['regime'].iloc[-1]
current_vix    = df['vix'].iloc[-1]
print(f"현재 레짐: {current_regime} (VIX: {current_vix:.1f})")
print(f"전략:     {STRATEGY.get(current_regime, '대기')}")

# 레짐별 일수 분포
print(f"\n레짐 분포:\n{df['regime'].value_counts()}")

전통 퀀트가 쓰는 주가·재무 데이터는 이미 모두가 씁니다. 알파가 사라졌습니다. 2025년 실전 엣지는 남들이 아직 안 쓰는 비정형 데이터에서 나옵니다.

import praw          # Reddit API
import requests
from transformers import pipeline

# 1. Reddit 감성 분석기 (Hugging Face FinBERT)
sentiment_pipe = pipeline("text-classification",
                           model="ProsusAI/finbert",
                           top_k=None)

def get_reddit_sentiment(keyword, limit=25):
    """Reddit 최신 게시물 감성 점수 (-1~+1)"""
    reddit = praw.Reddit(client_id="YOUR_ID",
                         client_secret="YOUR_SECRET",
                         user_agent="quant_bot")
    posts = list(reddit.subreddit("politics+news").search(
                 keyword, sort="new", limit=limit))

    scores = []
    for p in posts:
        result = sentiment_pipe(p.title[:512])[0]
        pos = next((r['score'] for r in result if r['label']=='positive'), 0)
        neg = next((r['score'] for r in result if r['label']=='negative'), 0)
        scores.append(pos - neg)

    return sum(scores) / len(scores) if scores else 0

# 2. 폴리마켓 시장과 연결
def alt_data_signal(keyword, market_yes_price):
    sentiment = get_reddit_sentiment(keyword)
    # 감성 → 확률 변환 (보수적: 기존 시장가 ±15% 이내)
    my_prob = market_yes_price + sentiment * 0.15
    my_prob = max(0.05, min(0.95, my_prob))

    edge = my_prob - market_yes_price
    if edge > 0.05:
        kelly_bet = (my_prob - market_yes_price) / (1 - market_yes_price) / 2  # Half-Kelly
        return {"signal": "BUY_YES", "sentiment": sentiment, "edge": edge, "kelly": kelly_bet}
    elif edge < -0.05:
        return {"signal": "BUY_NO", "sentiment": sentiment, "edge": edge}
    return {"signal": "HOLD", "sentiment": sentiment}

동일 사건이 서로 다른 예측 시장에서 다른 확률로 거래됩니다. 이 가격 차이가 위험 없는 차익(Arbitrage) 또는 기대값 양수 베팅의 기회입니다.

import requests
import pandas as pd
from fuzzywuzzy import process  # 유사 질문 매칭

# 1. Polymarket 데이터
def get_polymarket():
    r = requests.get("https://gamma-api.polymarket.com/markets",
                     params={"active": True, "limit": 100, "order": "volume24hr", "ascending": False})
    markets = []
    for m in r.json():
        tokens = m.get("tokens", [])
        yes_p = next((float(t["price"]) for t in tokens if t.get("outcome")=="Yes"), None)
        if yes_p: markets.append({"q": m["question"], "poly_p": yes_p})
    return pd.DataFrame(markets)

# 2. Metaculus 데이터
def get_metaculus():
    r = requests.get("https://www.metaculus.com/api2/questions/",
                     params={"type": "forecast", "status": "open", "limit": 100,
                             "order_by": "-activity"})
    markets = []
    for q in r.json().get("results", []):
        cp = q.get("community_prediction", {}).get("full", {}).get("q2")
        if cp: markets.append({"q": q["title"], "meta_p": cp})
    return pd.DataFrame(markets)

# 3. 유사 질문 매칭 → 괴리 계산
poly_df = get_polymarket()
meta_df = get_metaculus()

opportunities = []
for _, pm in poly_df.iterrows():
    match, score = process.extractOne(pm['q'], meta_df['q'].tolist())
    if score >= 70:  # 유사도 70% 이상
        meta_p = meta_df.loc[meta_df['q']==match, 'meta_p'].values[0]
        gap = meta_p - pm['poly_p']
        if abs(gap) > 0.08:  # 8% 이상 괴리
            opportunities.append({
                'question': pm['q'][:60],
                'polymarket': pm['poly_p'],
                'metaculus': meta_p,
                'gap': gap,
                'signal': 'BUY_YES' if gap > 0 else 'BUY_NO'
            })

opp_df = pd.DataFrame(opportunities).sort_values('gap', key=abs, ascending=False)
print("=== 멀티마켓 차익 기회 ===")
print(opp_df.to_string(index=False))