알고리즘 페이퍼 트레이딩을 위한 오버나이트 전략 토너먼트 시스템 구축 방법

Source: Dev.to

위에 제공된 소스 링크 아래에 번역하고 싶은 텍스트를 붙여 주시면, 해당 내용을 한국어로 번역해 드리겠습니다.

트레이딩 전략 테스트의 문제점

트레이더에게 여러 전략을 테스트하는 일은 매우 힘든 일입니다. 수동 백테스팅은 느리고 일관성이 없으며, 잠자는 동안 10개의 전략을 동시에 실행할 수 없습니다. 대부분의 사람들은 분석이라고 포장된 직감에 의존하게 됩니다.

이 문제를 해결하기 위해 TradeSight를 만들었습니다: 밤새 전략 토너먼트를 진행해 전략들을 병렬로 실행하고, 실제 성과 지표로 순위를 매겨 아침에 리더보드를 제공하는 시스템입니다.

핵심 아이디어는 간단합니다 — 실제 시장 데이터(Alpaca를 통한 페이퍼 트레이딩)에서 전략들을 서로 겨루게 하고, 샤프 비율 + 승률로 순위를 매겨 가장 좋은 전략만 살아남게 합니다. 진화적 압력을 트레이딩 알고리즘에 적용하는 셈이죠.

매일 밤

1. 모든 등록된 전략이 티커 감시 목록에 대한 OHLCV 데이터를 가져옵니다.
2. 각 전략은 자체 지표 로직을 사용해 매수/매도 신호를 계산합니다.
3. 페이퍼 주문이 Alpaca에 제출됩니다.
4. 아침 마감 시 결과가 점수화되고 대시보드에 순위가 매겨집니다.

이 간단함을 유지하는 네 가지 구성 요소

• Flask 대시보드 – 토너먼트를 설정하고, 실시간 포지션과 리더보드를 확인합니다.
• 전략 실행기 – 전략을 병렬로 실행합니다(스레딩 사용, 비동기 아님 — 이 경우 더 간단합니다).
• Alpaca 통합 – 주문 제출 및 포지션 추적을 위한 페이퍼 트레이딩 API.
• Cron 자동화 – 시장 마감 시 토너먼트를 시작하고, 개장 시 점수를 매깁니다.

---

프로젝트 구조

TradeSight/
├── app/
│   ├── dashboard.py      # Flask routes
│   ├── runner.py         # Strategy execution engine
│   └── alpaca_client.py  # Alpaca API wrapper
├── strategies/
│   ├── base.py           # Strategy base class
│   ├── macd_strategy.py
│   └── rsi_strategy.py
└── cron/
    └── overnight_run.sh

---

전략 베이스 클래스 예시

# strategies/base.py
from abc import ABC, abstractmethod

class Strategy(ABC):
    name: str

    def __init__(self, alpaca):
        self.alpaca = alpaca

    @abstractmethod
    def compute_signals(self, ohlcv_data):
        """Return array of 1 (buy), -1 (sell), 0 (hold)."""
        pass

    @abstractmethod
    def submit_orders(self, signals, symbol, qty=10):
        """Submit paper orders based on signals."""
        pass

MACD 교차 구현

# strategies/macd_strategy.py
import numpy as np
from strategies.base import Strategy

class MACDStrategy(Strategy):
    name = "MACD Crossover"

    def compute_signals(self, ohlcv_data):
        close = ohlcv_data['close'].values

        # EMA calculations
        ema12 = self._ema(close, 12)
        ema26 = self._ema(close, 26)
        macd_line = ema12 - ema26
        signal_line = self._ema(macd_line, 9)

        # Crossover detection
        signals = np.zeros(len(close))
        for i in range(1, len(macd_line)):
            if macd_line[i] > signal_line[i] and macd_line[i-1] = signal_line[i-1]:
                signals[i] = -1  # bearish crossover
        return signals

    def submit_orders(self, signals, symbol, qty=10):
        latest = signals[-1]
        if latest == 1:
            self.alpaca.submit_order(symbol=symbol, qty=qty, side='buy', type='market')
        elif latest == -1:
            self.alpaca.submit_order(symbol=symbol, qty=qty, side='sell', type='market')

---

Built‑in Strategies

TradeSight ships with six ready‑to‑use strategies:

1. MACD Crossover
2. RSI Overbought/Oversold
3. Bollinger Band Squeeze
4. EMA Ribbon
5. Volume Spike + Price Confirmation
6. Mean Reversion (Z‑score based)

---

토너먼트 실행

import pandas as pd
import numpy as np
from alpaca_trade_api import TimeFrame

def run_tournament(symbols, strategies, alpaca):
    results = []

    for strategy in strategies:
        strategy_results = []
        for symbol in symbols:
            # Fetch OHLCV from Alpaca
            bars = alpaca.get_bars(symbol, TimeFrame.Minute, limit=200)
            ohlcv = pd.DataFrame([b._raw for b in bars])

            # Get signals
            signals = strategy.compute_signals(ohlcv)

            # Submit paper orders
            strategy.submit_orders(signals, symbol)

            # Track for scoring
            strategy_results.append({
                'symbol': symbol,
                'signal': signals[-1],
                'entry_price': ohlcv['close'].iloc[-1]
            })

        results.append({'strategy': strategy.name, 'trades': strategy_results})

    return results

---

스코어링 전략

def score_strategy(trades):
    winning_trades = [t for t in trades if t['pnl'] > 0]
    win_rate = len(winning_trades) / len(trades) if trades else 0

    returns = [t['pnl_pct'] for t in trades]
    sharpe = (np.mean(returns) / np.std(returns)) * np.sqrt(252) if returns else 0

    # Combined score: weights Sharpe more than win rate
    score = (sharpe * 0.6) + (win_rate * 0.4)
    return {'sharpe': sharpe, 'win_rate': win_rate, 'score': score}

Flask 대시보드는 실시간 리더보드, 현재 오픈 포지션, 그리고 과거 토너먼트 결과를 표시하여 전략이 시간에 따라 개선되는지(또는 사라지는지) 확인할 수 있게 해줍니다.

사용자 정의 전략 추가

1. strategies/ 폴더에 파일을 넣습니다.
2. Strategy 클래스를 상속합니다.
3. compute_signals와 submit_orders 메서드를 구현합니다.

class MyCustomStrategy(Strategy):
    name = "My Custom Strategy"

    def compute_signals(self, ohlcv_data):
        # Your logic here
        pass

    def submit_orders(self, signals, symbol, qty=10):
        # Your execution logic here
        pass

토너먼트 실행기는 시작 시 모든 전략 클래스를 자동으로 탐색합니다.

---

Overnight 토너먼트 운영에서 얻은 인사이트

몇 주간 밤새 토너먼트를 운영하면서 수동 백테스팅 몇 달보다 전략 행동에 대해 더 많이 배웠다.

• 실행 타이밍이 중요 – 종이상에서는 괜찮아 보이는 전략도 주문 지연과 슬리피지 때문에 실시간 데이터에서는 붕괴한다. 페이퍼 트레이딩이 완벽한 것은 아니지만, 완벽한 사후예측을 가진 백테스팅보다 현실에 훨씬 가깝다.
• 일관성이 최고 성과를 능가 – 토너먼트의 60 %를 승리하는 전략이 한 번은 화려하게 승리하고 나머지는 실패하는 전략보다 더 가치 있다.

Source Code

전체 소스는 **github.com/rmbell09-lang/tradesight**에서 확인할 수 있습니다. 별표와 PR을 환영합니다 — 특히 새로운 전략 구현을 기다립니다.