zephyrdark c3ec7f68a0 docs: add Phase 5 backtest engine design

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

2026-02-03 09:33:44 +09:00

15 KiB

Raw Blame History

Phase 5: Backtest Engine 설계

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: 퀀트 전략 백테스트 엔진 구현 - 과거 데이터 기반 전략 성과 시뮬레이션

핵심 기능:

비동기 백테스트 실행 (스레드 기반)
DB 저장으로 결과 조회/비교 가능
벤치마크(KOSPI) 대비 성과 비교
거래 비용(수수료, 슬리피지) 반영
리밸런싱 주기: 월별/분기별/반기별/연별

1. 데이터 모델

1.1 backtests (백테스트 작업)

CREATE TABLE backtests (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    strategy_type VARCHAR(50) NOT NULL,  -- multi_factor/quality/value_momentum
    strategy_params JSONB,                -- 전략 파라미터
    start_date DATE NOT NULL,
    end_date DATE NOT NULL,
    rebalance_period VARCHAR(20) NOT NULL, -- monthly/quarterly/semi_annual/annual
    initial_capital DECIMAL(20,2) NOT NULL,
    commission_rate DECIMAL(10,6) DEFAULT 0.00015,
    slippage_rate DECIMAL(10,6) DEFAULT 0.001,
    benchmark VARCHAR(20) DEFAULT 'KOSPI',
    status VARCHAR(20) DEFAULT 'pending', -- pending/running/completed/failed
    created_at TIMESTAMP DEFAULT NOW(),
    completed_at TIMESTAMP,
    error_message TEXT
);

1.2 backtest_results (결과 지표)

CREATE TABLE backtest_results (
    backtest_id INTEGER PRIMARY KEY REFERENCES backtests(id),
    total_return DECIMAL(10,4),      -- 총 수익률 (%)
    cagr DECIMAL(10,4),              -- 연평균 수익률 (%)
    mdd DECIMAL(10,4),               -- 최대 낙폭 (%)
    sharpe_ratio DECIMAL(10,4),
    volatility DECIMAL(10,4),        -- 연간 변동성 (%)
    benchmark_return DECIMAL(10,4),  -- 벤치마크 수익률 (%)
    excess_return DECIMAL(10,4)      -- 초과 수익률 (%)
);

1.3 backtest_equity_curve (자산 추이)

CREATE TABLE backtest_equity_curve (
    backtest_id INTEGER REFERENCES backtests(id),
    date DATE,
    portfolio_value DECIMAL(20,2),
    benchmark_value DECIMAL(20,2),
    drawdown DECIMAL(10,4),
    PRIMARY KEY (backtest_id, date)
);

1.4 backtest_holdings (리밸런싱 시점 보유 종목)

CREATE TABLE backtest_holdings (
    backtest_id INTEGER REFERENCES backtests(id),
    rebalance_date DATE,
    ticker VARCHAR(20),
    name VARCHAR(100),
    weight DECIMAL(10,4),
    shares INTEGER,
    price DECIMAL(12,2),
    PRIMARY KEY (backtest_id, rebalance_date, ticker)
);

1.5 backtest_transactions (거래 내역)

CREATE TABLE backtest_transactions (
    id SERIAL PRIMARY KEY,
    backtest_id INTEGER REFERENCES backtests(id),
    date DATE,
    ticker VARCHAR(20),
    action VARCHAR(10),  -- buy/sell
    shares INTEGER,
    price DECIMAL(12,2),
    commission DECIMAL(12,2)
);

2. API 엔드포인트

Method	Endpoint	설명
POST	/api/backtest	백테스트 생성 (작업 ID 반환)
GET	/api/backtest	내 백테스트 목록 조회
GET	/api/backtest/{id}	백테스트 상태/결과 조회
GET	/api/backtest/{id}/equity-curve	자산 추이 데이터
GET	/api/backtest/{id}/holdings	리밸런싱별 보유 종목
GET	/api/backtest/{id}/transactions	거래 내역
DELETE	/api/backtest/{id}	백테스트 삭제

요청 예시 (POST /api/backtest)

{
  "strategy_type": "multi_factor",
  "strategy_params": {
    "weights": {"value": 0.3, "quality": 0.3, "momentum": 0.4},
    "top_n": 30
  },
  "start_date": "2020-01-01",
  "end_date": "2024-12-31",
  "rebalance_period": "quarterly",
  "initial_capital": 100000000,
  "commission_rate": 0.00015,
  "slippage_rate": 0.001
}

응답 예시 (GET /api/backtest/{id})

{
  "id": 1,
  "status": "completed",
  "strategy_type": "multi_factor",
  "start_date": "2020-01-01",
  "end_date": "2024-12-31",
  "result": {
    "total_return": 125.4,
    "cagr": 17.8,
    "mdd": -32.5,
    "sharpe_ratio": 0.85,
    "volatility": 22.3,
    "benchmark_return": 45.2,
    "excess_return": 80.2
  }
}

3. 서비스 구조

backend/app/services/backtest/
├── __init__.py
├── engine.py          # 메인 백테스트 엔진
├── portfolio.py       # 가상 포트폴리오 관리
├── metrics.py         # 성과 지표 계산
└── worker.py          # 백그라운드 작업 처리

3.1 BacktestEngine (engine.py)

class BacktestEngine:
    def __init__(self, db: Session):
        self.db = db

    def run(self, backtest_id: int) -> None:
        """백테스트 실행 메인 로직"""
        backtest = self._get_backtest(backtest_id)

        # 1. 리밸런싱 날짜 목록 생성
        rebalance_dates = self._generate_rebalance_dates(
            backtest.start_date,
            backtest.end_date,
            backtest.rebalance_period
        )

        # 2. 가상 포트폴리오 초기화
        portfolio = VirtualPortfolio(backtest.initial_capital)

        # 3. 벤치마크 데이터 로드
        benchmark_prices = self._load_benchmark_prices(
            backtest.benchmark,
            backtest.start_date,
            backtest.end_date
        )

        # 4. 전략 인스턴스 생성
        strategy = self._create_strategy(
            backtest.strategy_type,
            backtest.strategy_params
        )

        # 5. 시뮬레이션 실행
        equity_curve = []
        holdings_history = []
        transactions = []

        for date in self._get_trading_days(backtest.start_date, backtest.end_date):
            prices = self._get_prices_for_date(date)

            # 리밸런싱 날짜인 경우
            if date in rebalance_dates:
                # 전략 실행하여 종목 선정
                targets = strategy.run(date)

                # 포트폴리오 리밸런싱
                txns = portfolio.rebalance(
                    targets,
                    prices,
                    backtest.commission_rate,
                    backtest.slippage_rate
                )
                transactions.extend(txns)

                # 보유 종목 기록
                holdings_history.append({
                    'date': date,
                    'holdings': portfolio.get_holdings_with_weights(prices)
                })

            # 일별 가치 기록
            equity_curve.append({
                'date': date,
                'portfolio_value': portfolio.get_value(prices),
                'benchmark_value': benchmark_prices.get(date)
            })

        # 6. 성과 지표 계산
        metrics = MetricsCalculator.calculate_all(equity_curve)

        # 7. 결과 저장
        self._save_results(backtest_id, metrics, equity_curve, holdings_history, transactions)

3.2 VirtualPortfolio (portfolio.py)

class VirtualPortfolio:
    def __init__(self, initial_capital: Decimal):
        self.cash = initial_capital
        self.holdings: Dict[str, int] = {}  # ticker -> shares

    def rebalance(
        self,
        targets: List[StockFactor],
        prices: Dict[str, Decimal],
        commission_rate: Decimal,
        slippage_rate: Decimal
    ) -> List[Transaction]:
        """목표 비중으로 리밸런싱"""
        transactions = []
        total_value = self.get_value(prices)

        # 목표 비중 계산 (동일 비중)
        target_weight = Decimal("1") / len(targets)
        target_tickers = {t.ticker for t in targets}

        # 매도: 목표에 없는 종목
        for ticker in list(self.holdings.keys()):
            if ticker not in target_tickers:
                shares = self.holdings[ticker]
                price = prices[ticker] * (1 - slippage_rate)
                proceeds = shares * price
                commission = proceeds * commission_rate

                self.cash += proceeds - commission
                del self.holdings[ticker]

                transactions.append(Transaction(
                    ticker=ticker,
                    action='sell',
                    shares=shares,
                    price=price,
                    commission=commission
                ))

        # 매수: 목표 비중에 맞게
        for target in targets:
            ticker = target.ticker
            target_value = total_value * target_weight
            current_value = self.holdings.get(ticker, 0) * prices.get(ticker, 0)

            if target_value > current_value:
                buy_value = target_value - current_value
                price = prices[ticker] * (1 + slippage_rate)
                shares = int(buy_value / price)

                if shares > 0:
                    cost = shares * price
                    commission = cost * commission_rate

                    if self.cash >= cost + commission:
                        self.cash -= cost + commission
                        self.holdings[ticker] = self.holdings.get(ticker, 0) + shares

                        transactions.append(Transaction(
                            ticker=ticker,
                            action='buy',
                            shares=shares,
                            price=price,
                            commission=commission
                        ))

        return transactions

    def get_value(self, prices: Dict[str, Decimal]) -> Decimal:
        """총 포트폴리오 가치"""
        holdings_value = sum(
            shares * prices.get(ticker, Decimal("0"))
            for ticker, shares in self.holdings.items()
        )
        return self.cash + holdings_value

3.3 MetricsCalculator (metrics.py)

class MetricsCalculator:
    @staticmethod
    def calculate_all(equity_curve: List[dict]) -> BacktestMetrics:
        values = [e['portfolio_value'] for e in equity_curve]
        benchmark_values = [e['benchmark_value'] for e in equity_curve]

        returns = MetricsCalculator._calculate_returns(values)

        return BacktestMetrics(
            total_return=MetricsCalculator.calculate_total_return(values),
            cagr=MetricsCalculator.calculate_cagr(values, len(values) / 252),
            mdd=MetricsCalculator.calculate_mdd(values),
            sharpe_ratio=MetricsCalculator.calculate_sharpe(returns),
            volatility=MetricsCalculator.calculate_volatility(returns),
            benchmark_return=MetricsCalculator.calculate_total_return(benchmark_values),
            excess_return=... # total_return - benchmark_return
        )

    @staticmethod
    def calculate_total_return(values: List[Decimal]) -> Decimal:
        return (values[-1] - values[0]) / values[0] * 100

    @staticmethod
    def calculate_cagr(values: List[Decimal], years: float) -> Decimal:
        total_return = values[-1] / values[0]
        return (total_return ** (1 / years) - 1) * 100

    @staticmethod
    def calculate_mdd(values: List[Decimal]) -> Decimal:
        peak = values[0]
        max_dd = Decimal("0")
        for value in values:
            if value > peak:
                peak = value
            dd = (peak - value) / peak * 100
            if dd > max_dd:
                max_dd = dd
        return -max_dd

    @staticmethod
    def calculate_sharpe(returns: List[Decimal], risk_free: Decimal = Decimal("0.03")) -> Decimal:
        avg_return = sum(returns) / len(returns) * 252
        std_return = ... # 표준편차 * sqrt(252)
        return (avg_return - risk_free) / std_return

3.4 Worker (worker.py)

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=2)

def submit_backtest(backtest_id: int, db_url: str):
    """백테스트 작업 제출"""
    executor.submit(_run_backtest_job, backtest_id, db_url)

def _run_backtest_job(backtest_id: int, db_url: str):
    """백그라운드에서 실행"""
    engine = create_engine(db_url)
    Session = sessionmaker(bind=engine)
    db = Session()

    try:
        # status → running
        backtest = db.query(Backtest).get(backtest_id)
        backtest.status = 'running'
        db.commit()

        # 백테스트 실행
        engine = BacktestEngine(db)
        engine.run(backtest_id)

        # status → completed
        backtest.status = 'completed'
        backtest.completed_at = datetime.utcnow()
        db.commit()

    except Exception as e:
        backtest.status = 'failed'
        backtest.error_message = str(e)
        db.commit()
    finally:
        db.close()

4. 프론트엔드 페이지

4.1 /backtest (목록 + 생성)

새 백테스트 폼:

전략 선택 드롭다운
전략 파라미터 (전략별 동적 폼)
기간 선택 (DatePicker)
리밸런싱 주기 라디오 버튼
초기 자본금 입력
거래 비용 설정 (접이식 고급 옵션)
실행 버튼

백테스트 목록 테이블: | 전략 | 기간 | 수익률 | CAGR | MDD | 상태 | 생성일 |

4.2 /backtest/[id] (결과 상세)

성과 요약 카드:

┌─────────────────────────────────────────────────────┐
│  총 수익률     CAGR      MDD       샤프     변동성   │
│  +125.4%     17.8%    -32.5%     0.85     22.3%   │
├─────────────────────────────────────────────────────┤
│  벤치마크 수익률: +45.2%    초과 수익률: +80.2%      │
└─────────────────────────────────────────────────────┘

자산 추이 차트:

포트폴리오 라인 (파란색)
벤치마크 라인 (회색)
낙폭 영역 (빨간색, 아래 서브차트)

탭: 리밸런싱 이력 / 거래 내역

5. 구현 태스크

Task 1: Backtest Pydantic 스키마

backend/app/schemas/backtest.py 생성
backend/app/schemas/__init__.py 업데이트

Task 2: Backtest SQLAlchemy 모델

backend/app/models/backtest.py 생성
backend/app/models/__init__.py 업데이트
Alembic 마이그레이션 생성/실행

Task 3: MetricsCalculator 서비스

backend/app/services/backtest/metrics.py 생성

Task 4: VirtualPortfolio 서비스

backend/app/services/backtest/portfolio.py 생성

Task 5: BacktestEngine 서비스

backend/app/services/backtest/engine.py 생성
backend/app/services/backtest/__init__.py 생성

Task 6: Worker 서비스

backend/app/services/backtest/worker.py 생성

Task 7: Backtest API 엔드포인트

backend/app/api/backtest.py 생성
backend/app/api/__init__.py 업데이트
backend/app/main.py 업데이트

Task 8: Frontend 백테스트 목록/생성 페이지

frontend/src/app/backtest/page.tsx 생성

Task 9: Frontend 백테스트 결과 페이지

frontend/src/app/backtest/[id]/page.tsx 생성

Task 10: 통합 검증

Frontend 빌드 확인
Git 커밋 히스토리 확인

문서 버전: 1.0 작성일: 2026-02-03

15 KiB Raw Blame History