SW Architectural Desgin - FinAI6/arbitrage-trading GitHub Wiki

인터페이스 기반 차익거래 봇 구조

파일 구조

arbitrage_bot/
├── config.py              # 모든 설정
├── exchanges/
│   ├── __init__.py
│   ├── base_exchange.py   # 거래소 공통 인터페이스
│   ├── binance.py         # 바이낸스 구현
│   └── bybit.py           # 바이비트 구현
├── strategies/
│   ├── __init__.py
│   ├── base_strategy.py   # 전략 공통 인터페이스
│   ├── market_strategy.py # 시장가 전략
│   ├── limit_strategy.py  # 지정가 전략
│   └── hybrid_strategy.py # 혼합 전략
├── main.py               # 메인 실행
└── requirements.txt

1. 거래소 인터페이스 설계

exchanges/base_exchange.py

from abc import ABC, abstractmethod

class BaseExchange(ABC):
    """모든 거래소가 구현해야 하는 공통 인터페이스"""
    
    def __init__(self, api_key: str, secret: str):
        self.api_key = api_key
        self.secret = secret
        self.exchange = None
    
    @abstractmethod
    def connect(self) -> bool:
        """거래소 연결"""
        pass
    
    @abstractmethod
    def get_symbols(self) -> set:
        """거래 가능한 심볼 목록"""
        pass
    
    @abstractmethod
    def get_prices(self) -> dict:
        """모든 심볼의 현재가"""
        pass
    
    @abstractmethod
    def get_volumes(self) -> dict:
        """24시간 거래량"""
        pass
    
    @abstractmethod
    def convert_symbol(self, raw_symbol: str) -> str:
        """심볼 형식 변환"""
        pass
    
    @abstractmethod
    def create_market_order(self, symbol: str, side: str, quantity: float) -> dict:
        """시장가 주문"""
        pass
    
    @abstractmethod
    def create_limit_order(self, symbol: str, side: str, quantity: float, price: float) -> dict:
        """지정가 주문"""
        pass
    
    @abstractmethod
    def get_order_status(self, order_id: str, symbol: str) -> dict:
        """주문 상태 조회"""
        pass
    
    @abstractmethod
    def get_balance(self, asset: str = 'USDT') -> float:
        """잔고 조회"""
        pass

exchanges/binance.py

import ccxt
from .base_exchange import BaseExchange

class BinanceExchange(BaseExchange):
    """바이낸스 거래소 구현"""
    
    def connect(self) -> bool:
        try:
            self.exchange = ccxt.binance({
                'apiKey': self.api_key,
                'secret': self.secret,
                'options': {'defaultType': 'future'},
                'enableRateLimit': True
            })
            return True
        except Exception as e:
            print(f"바이낸스 연결 실패: {e}")
            return False
    
    def get_symbols(self) -> set:
        # 바이낸스 특화 로직
        pass
    
    def get_prices(self) -> dict:
        # 바이낸스 API 호출
        pass
    
    # ... 나머지 메서드 구현

exchanges/bybit.py

import ccxt
from .base_exchange import BaseExchange

class BybitExchange(BaseExchange):
    """바이비트 거래소 구현"""
    
    def connect(self) -> bool:
        try:
            self.exchange = ccxt.bybit({
                'apiKey': self.api_key,
                'secret': self.secret,
                'enableRateLimit': True,
                'options': {
                    'defaultType': 'swap',
                    'defaultSubType': 'linear',
                }
            })
            return True
        except Exception as e:
            print(f"바이비트 연결 실패: {e}")
            return False
    
    def get_symbols(self) -> set:
        # 바이비트 특화 로직 (category=linear 등)
        pass
    
    # ... 나머지 메서드 구현

2. 전략 인터페이스 설계

strategies/base_strategy.py

from abc import ABC, abstractmethod
from typing import Dict, Any, Optional

class BaseStrategy(ABC):
    """모든 전략이 구현해야 하는 공통 인터페이스"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.spread_threshold = config['spread_threshold']
        self.exit_percent = config['exit_percent']
        self.spread_hold_count = config['spread_hold_count']
    
    @abstractmethod
    def should_enter(self, symbol: str, spread_data: Dict, history: Dict) -> bool:
        """진입 조건 확인"""
        pass
    
    @abstractmethod
    def should_exit(self, symbol: str, current_spread: float, position: Dict) -> bool:
        """청산 조건 확인"""
        pass
    
    @abstractmethod
    def create_entry_orders(self, long_exchange, short_exchange, symbol: str, 
                           quantity: float, prices: Dict) -> Dict:
        """진입 주문 생성"""
        pass
    
    @abstractmethod
    def create_exit_orders(self, long_exchange, short_exchange, position: Dict) -> Dict:
        """청산 주문 생성"""
        pass
    
    def get_strategy_name(self) -> str:
        """전략 이름 반환"""
        return self.__class__.__name__.replace('Strategy', '').lower()

strategies/market_strategy.py

from .base_strategy import BaseStrategy

class MarketStrategy(BaseStrategy):
    """시장가 전략"""
    
    def should_enter(self, symbol: str, spread_data: Dict, history: Dict) -> bool:
        # 기존 should_enter_position 로직
        return abs(spread_data['spread_pct']) >= self.spread_threshold
    
    def should_exit(self, symbol: str, current_spread: float, position: Dict) -> bool:
        # 기존 청산 조건 로직
        entry_spread = position['entry_spread']
        return abs(current_spread) < entry_spread - self.exit_percent
    
    def create_entry_orders(self, long_exchange, short_exchange, symbol: str, 
                           quantity: float, prices: Dict) -> Dict:
        # 시장가 주문 생성
        long_order = long_exchange.create_market_order(symbol, 'buy', quantity)
        short_order = short_exchange.create_market_order(symbol, 'sell', quantity)
        
        return {
            'long_order': long_order,
            'short_order': short_order,
            'order_type': 'market'
        }
    
    def create_exit_orders(self, long_exchange, short_exchange, position: Dict) -> Dict:
        # 시장가 청산
        pass

strategies/limit_strategy.py

from .base_strategy import BaseStrategy

class LimitStrategy(BaseStrategy):
    """지정가 전략"""
    
    def create_entry_orders(self, long_exchange, short_exchange, symbol: str, 
                           quantity: float, prices: Dict) -> Dict:
        # 지정가 주문 생성
        buy_price = prices['lower_price'] * 1.001
        sell_price = prices['higher_price'] * 0.999
        
        long_order = long_exchange.create_limit_order(symbol, 'buy', quantity, buy_price)
        short_order = short_exchange.create_limit_order(symbol, 'sell', quantity, sell_price)
        
        return {
            'long_order': long_order,
            'short_order': short_order,
            'order_type': 'limit',
            'needs_monitoring': True  # 체결 모니터링 필요
        }

3. Factory 패턴으로 객체 생성

exchanges/init.py

from .binance import BinanceExchange
from .bybit import BybitExchange

def create_exchange(exchange_name: str, api_key: str, secret: str):
    """거래소 팩토리"""
    exchanges = {
        'binance': BinanceExchange,
        'bybit': BybitExchange,
        # 새로운 거래소 추가 시 여기만 수정
    }
    
    if exchange_name not in exchanges:
        raise ValueError(f"지원하지 않는 거래소: {exchange_name}")
    
    return exchanges[exchange_name](api_key, secret)

strategies/init.py

from .market_strategy import MarketStrategy
from .limit_strategy import LimitStrategy
from .hybrid_strategy import HybridStrategy

def create_strategy(strategy_name: str, config: dict):
    """전략 팩토리"""
    strategies = {
        'market': MarketStrategy,
        'limit': LimitStrategy,
        'hybrid': HybridStrategy,
        # 새로운 전략 추가 시 여기만 수정
    }
    
    if strategy_name not in strategies:
        raise ValueError(f"지원하지 않는 전략: {strategy_name}")
    
    return strategies[strategy_name](config)

4. main.py에서 사용

import os
from config import STRATEGIES, COMMON_CONFIG
from exchanges import create_exchange
from strategies import create_strategy

def main():
    # 전략 선택
    strategy_name = input("전략 선택 (market/limit/hybrid): ")
    strategy_config = STRATEGIES[strategy_name]
    
    # 거래소 생성
    binance = create_exchange('binance', 
                             os.getenv('BINANCE_API_KEY'), 
                             os.getenv('BINANCE_SECRET'))
    bybit = create_exchange('bybit', 
                           os.getenv('BYBIT_API_KEY'), 
                           os.getenv('BYBIT_SECRET'))
    
    # 전략 생성
    strategy = create_strategy(strategy_name, strategy_config)
    
    # 거래소 연결
    if not binance.connect() or not bybit.connect():
        print("거래소 연결 실패")
        return
    
    # 메인 루프
    while True:
        # 스프레드 데이터 수집
        spread_data = collect_spread_data(binance, bybit)
        
        # 전략 실행
        for item in spread_data:
            if strategy.should_enter(item['symbol'], item, {}):
                execute_entry(strategy, binance, bybit, item)
            
            # 기존 포지션 청산 확인
            check_exit_conditions(strategy, binance, bybit)
        
        time.sleep(5)

if __name__ == "__main__":
    main()

새로운 거래소/전략 추가 방법

새로운 거래소 추가 (예: Upbit)

exchanges/upbit.py 파일 생성
BaseExchange 상속받아 구현
exchanges/__init__.py의 팩토리에 추가

# exchanges/upbit.py
class UpbitExchange(BaseExchange):
    def connect(self):
        # Upbit 특화 연결 로직
        pass
    
    # 나머지 메서드 구현...

# exchanges/__init__.py에 추가
exchanges = {
    'binance': BinanceExchange,
    'bybit': BybitExchange,
    'upbit': UpbitExchange,  # 이 줄만 추가!
}

새로운 전략 추가 (예: Scalping)

strategies/scalping_strategy.py 파일 생성
BaseStrategy 상속받아 구현
strategies/__init__.py의 팩토리에 추가

장점

✅ 확장성: 새로운 거래소/전략 추가가 쉬움
✅ 일관성: 모든 거래소/전략이 동일한 인터페이스 사용
✅ 유지보수: 공통 로직은 base 클래스에서 관리
✅ 테스트: 각 구현체별로 독립적 테스트 가능
✅ 단순함: 복잡한 의존성 없이 깔끔한 구조

이렇게 하면 새로운 거래소나 전략을 추가할 때 기존 코드에 거의 영향을 주지 않고 확장할 수 있습니다!