파이썬 알고리즘 트레이딩 실전 가이드: 백테스팅 프레임워크·전략 개발·리스크 관리

import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta

def fetch_market_data(
    tickers: list,
    start_date: str = "2020-01-01",
    end_date: str = None,
    interval: str = "1d",
) -> dict:
    """시장 데이터 수집"""
    if end_date is None:
        end_date = datetime.now().strftime("%Y-%m-%d")

    data = {}
    for ticker in tickers:
        try:
            df = yf.download(
                ticker,
                start=start_date,
                end=end_date,
                interval=interval,
                progress=False,
            )
            if not df.empty:
                data[ticker] = df
                print(f"{ticker}: {len(df)} rows loaded ({df.index[0]} ~ {df.index[-1]})")
            else:
                print(f"{ticker}: No data available")
        except Exception as e:
            print(f"{ticker}: Error - {e}")

    return data

# 데이터 수집
tickers = ["AAPL", "MSFT", "GOOGL", "AMZN", "SPY"]
market_data = fetch_market_data(tickers, start_date="2020-01-01")

# 데이터 확인
aapl = market_data["AAPL"]
print(f"\nAAPL 데이터 요약:")
print(f"  기간: {aapl.index[0]} ~ {aapl.index[-1]}")
print(f"  데이터 포인트: {len(aapl)}")
print(f"  컬럼: {list(aapl.columns)}")

import requests
import pandas as pd

class AlphaVantageClient:
    """Alpha Vantage API 클라이언트"""

    BASE_URL = "https://www.alphavantage.co/query"

    def __init__(self, api_key: str):
        self.api_key = api_key

    def get_daily(self, symbol: str, outputsize: str = "full") -> pd.DataFrame:
        """일봉 데이터 조회"""
        params = {
            "function": "TIME_SERIES_DAILY_ADJUSTED",
            "symbol": symbol,
            "outputsize": outputsize,
            "apikey": self.api_key,
        }
        response = requests.get(self.BASE_URL, params=params)
        data = response.json()

        if "Time Series (Daily)" not in data:
            raise ValueError(f"API error: {data.get('Note', data.get('Error Message', 'Unknown'))}")

        df = pd.DataFrame.from_dict(data["Time Series (Daily)"], orient="index")
        df.columns = ["Open", "High", "Low", "Close", "Adj Close", "Volume", "Dividend", "Split"]
        df = df.astype(float)
        df.index = pd.to_datetime(df.index)
        df = df.sort_index()

        return df

    def get_intraday(self, symbol: str, interval: str = "5min") -> pd.DataFrame:
        """분봉 데이터 조회"""
        params = {
            "function": "TIME_SERIES_INTRADAY",
            "symbol": symbol,
            "interval": interval,
            "outputsize": "full",
            "apikey": self.api_key,
        }
        response = requests.get(self.BASE_URL, params=params)
        data = response.json()

        time_series_key = f"Time Series ({interval})"
        if time_series_key not in data:
            raise ValueError(f"API error: {data}")

        df = pd.DataFrame.from_dict(data[time_series_key], orient="index")
        df.columns = ["Open", "High", "Low", "Close", "Volume"]
        df = df.astype(float)
        df.index = pd.to_datetime(df.index)
        df = df.sort_index()

        return df

# 사용 예시
# client = AlphaVantageClient(api_key="YOUR_API_KEY")
# daily_data = client.get_daily("AAPL")

import pandas as pd
import numpy as np
from backtesting import Backtest, Strategy
from backtesting.lib import crossover

class MovingAverageCrossover(Strategy):
    """이동평균 교차 전략
    - 단기 이동평균이 장기 이동평균을 상향 돌파하면 매수
    - 단기 이동평균이 장기 이동평균을 하향 돌파하면 매도
    """
    fast_period = 10  # 단기 이동평균 기간
    slow_period = 30  # 장기 이동평균 기간

    def init(self):
        close = self.data.Close
        self.fast_ma = self.I(lambda x: pd.Series(x).rolling(self.fast_period).mean(), close)
        self.slow_ma = self.I(lambda x: pd.Series(x).rolling(self.slow_period).mean(), close)

    def next(self):
        # 골든 크로스: 매수
        if crossover(self.fast_ma, self.slow_ma):
            if not self.position:
                self.buy()

        # 데드 크로스: 매도
        elif crossover(self.slow_ma, self.fast_ma):
            if self.position:
                self.position.close()

# 데이터 준비
data = yf.download("AAPL", start="2020-01-01", end="2025-12-31", progress=False)
data.columns = data.columns.droplevel(1) if isinstance(data.columns, pd.MultiIndex) else data.columns

# 백테스트 실행
bt = Backtest(
    data,
    MovingAverageCrossover,
    cash=100000,
    commission=0.001,  # 0.1% 수수료
    exclusive_orders=True,
)

results = bt.run()
print("=== Moving Average Crossover Results ===")
print(f"총 수익률: {results['Return [%]']:.2f}%")
print(f"연간 수익률: {results['Return (Ann.) [%]']:.2f}%")
print(f"샤프 비율: {results['Sharpe Ratio']:.2f}")
print(f"최대 낙폭: {results['Max. Drawdown [%]']:.2f}%")
print(f"승률: {results['Win Rate [%]']:.2f}%")
print(f"총 거래 횟수: {results['# Trades']}")

# 파라미터 최적화
optimization_results = bt.optimize(
    fast_period=range(5, 25, 5),
    slow_period=range(20, 60, 10),
    maximize="Sharpe Ratio",
    constraint=lambda p: p.fast_period < p.slow_period,
)
print(f"\n최적 파라미터: fast={optimization_results._strategy.fast_period}, slow={optimization_results._strategy.slow_period}")

class RSIMeanReversion(Strategy):
    """RSI 평균회귀 전략
    - RSI가 과매도 구간(30 이하)에 진입하면 매수
    - RSI가 과매수 구간(70 이상)에 진입하면 매도
    """
    rsi_period = 14
    rsi_oversold = 30
    rsi_overbought = 70

    def init(self):
        close = pd.Series(self.data.Close)
        delta = close.diff()
        gain = delta.where(delta > 0, 0.0)
        loss = (-delta).where(delta < 0, 0.0)

        avg_gain = gain.rolling(window=self.rsi_period).mean()
        avg_loss = loss.rolling(window=self.rsi_period).mean()

        rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))

        self.rsi = self.I(lambda: rsi, name="RSI")

    def next(self):
        if self.rsi[-1] < self.rsi_oversold:
            if not self.position:
                self.buy()
        elif self.rsi[-1] > self.rsi_overbought:
            if self.position:
                self.position.close()

# 백테스트 실행
bt_rsi = Backtest(
    data,
    RSIMeanReversion,
    cash=100000,
    commission=0.001,
    exclusive_orders=True,
)

results_rsi = bt_rsi.run()
print("=== RSI Mean Reversion Results ===")
print(f"총 수익률: {results_rsi['Return [%]']:.2f}%")
print(f"샤프 비율: {results_rsi['Sharpe Ratio']:.2f}")
print(f"최대 낙폭: {results_rsi['Max. Drawdown [%]']:.2f}%")
print(f"승률: {results_rsi['Win Rate [%]']:.2f}%")

class BollingerBandsBreakout(Strategy):
    """볼린저 밴드 브레이크아웃 전략
    - 가격이 하단 밴드를 터치하면 매수 (평균 회귀 기대)
    - 가격이 상단 밴드를 터치하면 매도
    - 스톱로스: 진입가 대비 2% 하락 시 손절
    """
    bb_period = 20
    bb_std = 2.0
    stop_loss_pct = 0.02

    def init(self):
        close = pd.Series(self.data.Close)
        self.sma = self.I(lambda: close.rolling(self.bb_period).mean(), name="SMA")
        std = close.rolling(self.bb_period).std()
        self.upper = self.I(lambda: close.rolling(self.bb_period).mean() + self.bb_std * std, name="Upper")
        self.lower = self.I(lambda: close.rolling(self.bb_period).mean() - self.bb_std * std, name="Lower")

    def next(self):
        price = self.data.Close[-1]

        # 하단 밴드 터치: 매수
        if price <= self.lower[-1]:
            if not self.position:
                self.buy(sl=price * (1 - self.stop_loss_pct))

        # 상단 밴드 터치: 매도
        elif price >= self.upper[-1]:
            if self.position:
                self.position.close()

# 백테스트 실행
bt_bb = Backtest(
    data,
    BollingerBandsBreakout,
    cash=100000,
    commission=0.001,
    exclusive_orders=True,
)

results_bb = bt_bb.run()
print("=== Bollinger Bands Breakout Results ===")
print(f"총 수익률: {results_bb['Return [%]']:.2f}%")
print(f"샤프 비율: {results_bb['Sharpe Ratio']:.2f}")
print(f"최대 낙폭: {results_bb['Max. Drawdown [%]']:.2f}%")
print(f"승률: {results_bb['Win Rate [%]']:.2f}%")

import numpy as np
import pandas as pd
from scipy import stats

class RiskMetrics:
    """리스크 관리 메트릭 계산기"""

    def __init__(self, returns: pd.Series, risk_free_rate: float = 0.04):
        """
        Args:
            returns: 일별 수익률 시리즈
            risk_free_rate: 무위험 수익률 (연율, 기본 4%)
        """
        self.returns = returns.dropna()
        self.risk_free_rate = risk_free_rate
        self.daily_rf = (1 + risk_free_rate) ** (1/252) - 1

    def sharpe_ratio(self) -> float:
        """샤프 비율 계산"""
        excess_returns = self.returns - self.daily_rf
        if excess_returns.std() == 0:
            return 0.0
        return np.sqrt(252) * excess_returns.mean() / excess_returns.std()

    def sortino_ratio(self) -> float:
        """소르티노 비율 계산 (하방 위험만 고려)"""
        excess_returns = self.returns - self.daily_rf
        downside_returns = excess_returns[excess_returns < 0]
        if len(downside_returns) == 0 or downside_returns.std() == 0:
            return 0.0
        downside_std = downside_returns.std()
        return np.sqrt(252) * excess_returns.mean() / downside_std

    def maximum_drawdown(self) -> float:
        """최대 낙폭 (MDD) 계산"""
        cumulative = (1 + self.returns).cumprod()
        peak = cumulative.expanding().max()
        drawdown = (cumulative - peak) / peak
        return drawdown.min()

    def value_at_risk(self, confidence: float = 0.95) -> float:
        """VaR (Value at Risk) 계산 - 히스토리컬 방법"""
        return np.percentile(self.returns, (1 - confidence) * 100)

    def conditional_var(self, confidence: float = 0.95) -> float:
        """CVaR (Conditional VaR) 계산"""
        var = self.value_at_risk(confidence)
        return self.returns[self.returns <= var].mean()

    def calmar_ratio(self) -> float:
        """칼마 비율 계산 (연간 수익률 / MDD)"""
        annual_return = (1 + self.returns.mean()) ** 252 - 1
        mdd = abs(self.maximum_drawdown())
        if mdd == 0:
            return 0.0
        return annual_return / mdd

    def summary(self) -> dict:
        """전체 리스크 메트릭 요약"""
        annual_return = (1 + self.returns.mean()) ** 252 - 1
        annual_volatility = self.returns.std() * np.sqrt(252)

        return {
            "연간 수익률": f"{annual_return:.2%}",
            "연간 변동성": f"{annual_volatility:.2%}",
            "샤프 비율": f"{self.sharpe_ratio():.2f}",
            "소르티노 비율": f"{self.sortino_ratio():.2f}",
            "최대 낙폭(MDD)": f"{self.maximum_drawdown():.2%}",
            "VaR(95%)": f"{self.value_at_risk():.2%}",
            "CVaR(95%)": f"{self.conditional_var():.2%}",
            "칼마 비율": f"{self.calmar_ratio():.2f}",
            "총 거래일": len(self.returns),
            "양의 수익일": f"{(self.returns > 0).sum()} ({(self.returns > 0).mean():.1%})",
        }

# 사용 예시
# SPY의 일별 수익률 계산
spy = yf.download("SPY", start="2020-01-01", end="2025-12-31", progress=False)
daily_returns = spy["Close"].pct_change().dropna()

metrics = RiskMetrics(daily_returns.squeeze(), risk_free_rate=0.04)
summary = metrics.summary()

print("=== SPY Risk Metrics ===")
for key, value in summary.items():
    print(f"  {key}: {value}")

class PositionSizer:
    """포지션 사이징 알고리즘"""

    @staticmethod
    def kelly_criterion(win_rate: float, avg_win: float, avg_loss: float) -> float:
        """켈리 기준에 의한 최적 포지션 크기 계산

        Args:
            win_rate: 승률 (0~1)
            avg_win: 평균 이익률 (양수)
            avg_loss: 평균 손실률 (양수)

        Returns:
            최적 베팅 비율 (0~1)
        """
        if avg_loss == 0:
            return 0.0

        # Kelly Formula: f = (bp - q) / b
        # b = avg_win / avg_loss (odds ratio)
        # p = win_rate, q = 1 - win_rate
        b = avg_win / avg_loss
        p = win_rate
        q = 1 - p

        kelly = (b * p - q) / b

        # 음수면 베팅하지 않음
        return max(0.0, kelly)

    @staticmethod
    def half_kelly(win_rate: float, avg_win: float, avg_loss: float) -> float:
        """하프 켈리: 켈리 기준의 절반으로 보수적 접근"""
        full_kelly = PositionSizer.kelly_criterion(win_rate, avg_win, avg_loss)
        return full_kelly / 2

    @staticmethod
    def fixed_fractional(equity: float, risk_per_trade: float,
                          entry_price: float, stop_loss_price: float) -> int:
        """고정 비율(Fixed Fractional) 포지션 사이징

        Args:
            equity: 현재 자본금
            risk_per_trade: 거래당 위험 비율 (예: 0.02 = 2%)
            entry_price: 진입 가격
            stop_loss_price: 손절 가격

        Returns:
            매수할 주식 수
        """
        risk_amount = equity * risk_per_trade
        risk_per_share = abs(entry_price - stop_loss_price)

        if risk_per_share == 0:
            return 0

        shares = int(risk_amount / risk_per_share)
        return max(0, shares)

    @staticmethod
    def volatility_based(equity: float, target_volatility: float,
                          asset_volatility: float) -> float:
        """변동성 기반 포지션 사이징

        Args:
            equity: 현재 자본금
            target_volatility: 목표 포트폴리오 변동성 (연율)
            asset_volatility: 자산 변동성 (연율)

        Returns:
            포지션 비중 (0~1)
        """
        if asset_volatility == 0:
            return 0.0

        weight = target_volatility / asset_volatility
        return min(weight, 1.0)  # 최대 100%

# 사용 예시
sizer = PositionSizer()

# 켈리 기준 계산
win_rate = 0.55
avg_win = 0.03   # 평균 3% 이익
avg_loss = 0.02  # 평균 2% 손실

kelly = sizer.kelly_criterion(win_rate, avg_win, avg_loss)
half = sizer.half_kelly(win_rate, avg_win, avg_loss)
print(f"켈리 기준: {kelly:.2%}")
print(f"하프 켈리: {half:.2%}")

# 고정 비율 포지션 사이징
equity = 100000
entry = 150.0
stop_loss = 147.0
shares = sizer.fixed_fractional(equity, 0.02, entry, stop_loss)
print(f"매수 주식 수: {shares}주 (진입: {entry}, 손절: {stop_loss})")

import pandas as pd
import numpy as np
from backtesting import Backtest

class WalkForwardOptimizer:
    """워크포워드 최적화"""

    def __init__(self, data: pd.DataFrame, strategy_class,
                 train_period: int = 252, test_period: int = 63):
        """
        Args:
            data: OHLCV 데이터
            strategy_class: 전략 클래스
            train_period: 학습 기간 (거래일, 기본 1년)
            test_period: 테스트 기간 (거래일, 기본 3개월)
        """
        self.data = data
        self.strategy_class = strategy_class
        self.train_period = train_period
        self.test_period = test_period

    def run(self, optimization_params: dict, maximize: str = "Sharpe Ratio") -> list:
        """워크포워드 분석 실행"""
        results = []
        total_days = len(self.data)
        start_idx = 0

        fold = 1
        while start_idx + self.train_period + self.test_period <= total_days:
            train_end = start_idx + self.train_period
            test_end = train_end + self.test_period

            train_data = self.data.iloc[start_idx:train_end]
            test_data = self.data.iloc[train_end:test_end]

            # 학습 기간에서 파라미터 최적화
            bt_train = Backtest(
                train_data, self.strategy_class,
                cash=100000, commission=0.001,
            )
            opt_result = bt_train.optimize(
                **optimization_params,
                maximize=maximize,
            )

            # 최적화된 파라미터 추출
            best_params = {}
            for param_name in optimization_params:
                best_params[param_name] = getattr(opt_result._strategy, param_name)

            # 테스트 기간에서 검증
            bt_test = Backtest(
                test_data, self.strategy_class,
                cash=100000, commission=0.001,
            )
            # 최적화된 파라미터로 테스트 실행
            test_result = bt_test.run(**best_params)

            fold_result = {
                "fold": fold,
                "train_start": train_data.index[0],
                "train_end": train_data.index[-1],
                "test_start": test_data.index[0],
                "test_end": test_data.index[-1],
                "best_params": best_params,
                "train_return": opt_result["Return [%]"],
                "test_return": test_result["Return [%]"],
                "test_sharpe": test_result["Sharpe Ratio"],
                "test_mdd": test_result["Max. Drawdown [%]"],
            }
            results.append(fold_result)

            print(f"Fold {fold}: Train Return={fold_result['train_return']:.2f}%, "
                  f"Test Return={fold_result['test_return']:.2f}%, "
                  f"Params={best_params}")

            start_idx += self.test_period
            fold += 1

        return results

    def summary(self, results: list) -> dict:
        """워크포워드 결과 요약"""
        test_returns = [r["test_return"] for r in results]
        test_sharpes = [r["test_sharpe"] for r in results]

        return {
            "총 Fold 수": len(results),
            "평균 테스트 수익률": f"{np.mean(test_returns):.2f}%",
            "테스트 수익률 표준편차": f"{np.std(test_returns):.2f}%",
            "양의 수익률 Fold 비율": f"{sum(1 for r in test_returns if r > 0) / len(test_returns):.1%}",
            "평균 테스트 샤프": f"{np.mean(test_sharpes):.2f}",
        }

# 사용 예시
# wfo = WalkForwardOptimizer(data, MovingAverageCrossover)
# results = wfo.run(
#     optimization_params={
#         "fast_period": range(5, 25, 5),
#         "slow_period": range(20, 60, 10),
#     },
# )
# print(wfo.summary(results))

class OverfitDetector:
    """과적합 감지기"""

    @staticmethod
    def check_overfit(train_sharpe: float, test_sharpe: float,
                       threshold: float = 0.5) -> dict:
        """과적합 여부 판단

        Args:
            train_sharpe: 학습 기간 샤프 비율
            test_sharpe: 테스트 기간 샤프 비율
            threshold: 허용 성능 감소 비율

        Returns:
            과적합 진단 결과
        """
        if train_sharpe <= 0:
            return {"is_overfit": True, "reason": "학습 기간 성과 자체가 음수"}

        degradation = 1 - (test_sharpe / train_sharpe)
        is_overfit = degradation > threshold

        return {
            "is_overfit": is_overfit,
            "train_sharpe": train_sharpe,
            "test_sharpe": test_sharpe,
            "performance_degradation": f"{degradation:.1%}",
            "recommendation": (
                "과적합 의심: 파라미터 수를 줄이거나 학습 기간을 늘리세요"
                if is_overfit
                else "적절한 범위 내의 성능 차이"
            ),
        }

    @staticmethod
    def parameter_sensitivity(results_grid: dict) -> dict:
        """파라미터 민감도 분석
        최적 파라미터 주변에서 성능이 급격히 떨어지면 과적합 가능성 높음
        """
        sharpe_values = list(results_grid.values())
        mean_sharpe = np.mean(sharpe_values)
        std_sharpe = np.std(sharpe_values)
        max_sharpe = max(sharpe_values)

        # 최적 성능이 평균보다 2 표준편차 이상 높으면 과적합 의심
        is_sensitive = (max_sharpe - mean_sharpe) > 2 * std_sharpe

        return {
            "is_sensitive": is_sensitive,
            "max_sharpe": max_sharpe,
            "mean_sharpe": mean_sharpe,
            "std_sharpe": std_sharpe,
            "recommendation": (
                "파라미터 민감도 높음: 과적합 위험"
                if is_sensitive
                else "파라미터에 대해 안정적인 성과"
            ),
        }

def check_survivorship_bias(tickers: list, start_date: str) -> dict:
    """생존 편향 검사
    현재 존재하는 종목만으로 백테스트하면 생존 편향이 발생

    권장: 상장폐지/합병 종목도 포함된 데이터셋 사용
    """
    warnings = []

    # 현재 시점의 인덱스 구성 종목으로만 테스트하는 경우 경고
    if all(yf.Ticker(t).info.get("marketCap", 0) > 0 for t in tickers[:5]):
        warnings.append(
            "현재 상장 중인 종목만 포함됨. "
            "과거에 상장폐지되거나 합병된 종목이 누락되어 "
            "수익률이 과대평가될 수 있음"
        )

    return {
        "ticker_count": len(tickers),
        "warnings": warnings,
        "recommendation": "Point-in-time 데이터셋 사용 권장 (예: CRSP, Sharadar)",
    }

def validate_no_lookahead(strategy_code: str) -> list:
    """미래 참조 편향 검사 (코드 정적 분석)"""
    warnings = []

    # 미래 데이터를 참조하는 패턴 검사
    dangerous_patterns = [
        ("shift(-", "미래 데이터를 참조하는 shift(-N) 감지"),
        (".iloc[-1]", "마지막 행 참조 - 문맥에 따라 미래 참조 가능"),
        ("resample", "리샘플링 시 미래 데이터 포함 가능"),
    ]

    for pattern, description in dangerous_patterns:
        if pattern in strategy_code:
            warnings.append(f"경고: {description} - '{pattern}' 발견")

    if not warnings:
        warnings.append("명시적인 미래 참조 패턴은 감지되지 않음")

    return warnings

class RealisticBacktestConfig:
    """실전에 가까운 백테스트 설정"""

    @staticmethod
    def get_config(asset_type: str = "us_equity") -> dict:
        """자산 유형별 현실적인 거래 비용 설정"""
        configs = {
            "us_equity": {
                "commission": 0.001,     # 0.1% 수수료
                "slippage": 0.0005,      # 0.05% 슬리피지
                "spread": 0.0001,        # 0.01% 스프레드 (대형주)
                "market_impact": 0.0002, # 0.02% 시장 충격
            },
            "kr_equity": {
                "commission": 0.00015,   # 0.015% (증권사 수수료)
                "tax": 0.0018,           # 0.18% (증권거래세, 2026년 기준)
                "slippage": 0.001,       # 0.1% 슬리피지
                "spread": 0.0005,        # 0.05% 스프레드
            },
            "crypto": {
                "commission": 0.001,     # 0.1% (메이커 수수료)
                "slippage": 0.002,       # 0.2% 슬리피지
                "spread": 0.001,         # 0.1% 스프레드
            },
        }
        return configs.get(asset_type, configs["us_equity"])

    @staticmethod
    def total_cost_per_trade(config: dict) -> float:
        """거래당 총 비용 계산"""
        return sum(config.values())

# 비용 확인
for asset_type in ["us_equity", "kr_equity", "crypto"]:
    config = RealisticBacktestConfig.get_config(asset_type)
    total = RealisticBacktestConfig.total_cost_per_trade(config)
    print(f"{asset_type}: 거래당 총 비용 약 {total:.3%}")