🚀 정량적 자동 거래 엔진 (Quant Engine) 통합 시스템 구현 (개념 설계)

정량적 분석 방법(베타, 샤프 비율 등)과 제시된 API 엔드포인트를 통합하여 분석, 의사결정, 주문 실행까지 자동화된 거래 시스템을 실행 가능하도록 파이썬 기반의 아키텍처 및 핵심 코드를 구체적으로 제시합니다.

주의: 이 코드는 실제 금융 거래소와 연동되는 개념 설계 및 실행 가능한 파이썬 구조입니다. 실제 거래를 위해서는 브로커별 API 라이브러리를 사용한 BrokerConnector 및 **DataHandler**의 완벽한 구현이 필요하며, 반드시 Testnet 환경에서 검증해야 합니다.

---

1. ⚙️ 시스템 핵심 모듈 및 파일 구조

시스템은 기능을 분리하여 관리하기 쉽고 확장성 높은 모듈식 구조를 따릅니다.

파일/모듈	역할	주요 기능
main.py	메인 실행 루프	엔진 시작/종료, 주기적인 분석 및 실행 요청.
quant_analyzer.py	수학적 분석	베타, 샤프 비율, 상대 강도, 리스크/수익률 계산.
strategy_engine.py	의사 결정 및 관리	매매 신호 생성, 포지션 사이즈 계산, SL/TP 설정. (POST /api/position/recommend)
broker_connector.py	브로커 통신	API 연결, 잔고 조회, 주문 실행. (POST /api/order/execute)
data_handler.py	데이터/DB 관리	실시간/과거 데이터 수집 및 SQLite 영속성 관리.

---

2. 🔢 quant_analyzer.py (정량적 분석 모듈)

시장 대비 강도를 정량화하는 핵심 로직입니다.

Python

 

import numpy as np
import pandas as pd

class QuantAnalyzer:
    def __init__(self, market_returns: pd.Series):
        # 시장(지수) 수익률을 벤치마크로 설정
        self.Rm = market_returns
        self.Rf = 0.0001  # 일간 무위험 이자율 가정

    def calculate_beta(self, stock_returns: pd.Series) -> float:
        """베타 (β) 계산: 시장과의 민감도 측정"""
        cov = np.cov(stock_returns, self.Rm)[0, 1]
        var_m = np.var(self.Rm)
        if var_m == 0: return 1.0 # 시장 변동 없을 시
        return round(cov / var_m, 2)

    def calculate_sharpe_ratio(self, stock_returns: pd.Series) -> float:
        """샤프 비율 계산: 위험 대비 수익률 효율성 측정"""
        excess_return_mean = stock_returns.mean() - self.Rf
        std_dev = stock_returns.std()
        if std_dev == 0: return 0.0
        # (기대 수익률 - 무위험 이자율) / 표준편차
        return round(excess_return_mean / std_dev, 4)

    def select_strong_stocks(self, stock_data_dict: dict) -> pd.DataFrame:
        """
        수학적 지표를 기반으로 시장을 이길 종목 선별 및 순위화
        """
        results = []
        for symbol, data in stock_data_dict.items():
            returns = data['Close'].pct_change().dropna()
            if returns.empty: continue
            
            beta = self.calculate_beta(returns)
            sharpe = self.calculate_sharpe_ratio(returns)
            
            # **강도 점수 (Strength Score)** 계산: 샤프 비율을 중심으로 베타를 조정
            # Beta가 1보다 작으면 (방어적) 점수를 높게, 1보다 크면 (공격적) 변동성을 고려해 조정
            strength_score = sharpe * (2 - beta) # 1 < Beta < 2 가정
            
            results.append({
                'symbol': symbol,
                'beta': beta,
                'sharpe_ratio': sharpe,
                'strength_score': strength_score
            })
            
        # strength_score가 높은 종목이 '시장 대비 강함'으로 판단
        return pd.DataFrame(results).sort_values(by='strength_score', ascending=False)

---

3. 🛡️ strategy_engine.py (의사 결정 및 주문 실행)

주문 실행 및 리스크 관리 로직이 포함됩니다.

Python

 

# strategy_engine.py

class StrategyEngine:
    MAX_RISK_PER_TRADE = 0.01  # 전체 계좌 잔고의 1%만 한 거래에 손실 허용

    def __init__(self, connector, analyzer):
        self.broker = connector
        self.analyzer = analyzer

    def get_position_recommendation(self, strong_rank: pd.DataFrame) -> dict:
        """
        POST /api/position/recommend 매핑: 포지션 사이즈 계산 및 추천
        """
        if strong_rank.empty: return None

        top_stock = strong_rank.iloc[0]
        symbol = top_stock['symbol']
        
        # 1. 잔고 및 가격 확인
        current_balance = self.broker.get_balance() # GET /api/broker/balance 호출
        current_price = self.broker.get_latest_price(symbol)
        
        if not current_balance or not current_price: return None

        # 2. SL/TP 가격 자동 설정 (엔지니어링 접근: 변동성 기반)
        # ATR(Average True Range)을 사용하지 않고 단순 %로 예시 (실제는 ATR 사용)
        risk_percent = 0.03 # 종목 진입 시 3% 하락 시 손절
        profit_percent = 0.06 # 종목 진입 시 6% 상승 시 익절
        
        stop_loss_price = round(current_price * (1 - risk_percent), 2)
        take_profit_price = round(current_price * (1 + profit_percent), 2)

        # 3. 포지션 사이즈 계산 (리스크 금액 기반)
        price_risk_unit = current_price - stop_loss_price # 주당 감수할 가격 위험
        max_loss_amount = current_balance * self.MAX_RISK_PER_TRADE # 총 허용 손실 금액
        
        if price_risk_unit <= 0: return None
        
        # 매수 수량 = 허용 손실 금액 / 주당 가격 위험
        quantity = int(max_loss_amount / price_risk_unit)
        
        if quantity > 0 and (quantity * current_price) < current_balance:
            return {
                'symbol': symbol,
                'action': 'BUY',
                'quantity': quantity,
                'entry_price': current_price,
                'stop_loss': stop_loss_price,
                'take_profit': take_profit_price
            }
        return None

    def execute_order(self, recommendation: dict):
        """
        POST /api/order/execute 매핑: 실제 주문 실행
        """
        if not recommendation:
            print("▶️ 매매 신호 없음.")
            return

        print(f"✅ 주문 실행 요청: {recommendation['symbol']} / 수량: {recommendation['quantity']}")
        
        # 브로커 커넥터 호출 (실제 API 통신)
        order_result = self.broker.send_order(
            symbol=recommendation['symbol'],
            side=recommendation['action'],
            quantity=recommendation['quantity'],
            price=recommendation['entry_price']
        )
        
        if order_result.get('status') == 'FILLED':
            # 주문 성공 시 SL/TP 자동 설정 요청
            self.broker.set_take_profit_stop_loss(
                order_id=order_result['id'],
                tp=recommendation['take_profit'],
                sl=recommendation['stop_loss']
            )
            self.broker.data_handler.save_trade_history(recommendation)

---

4. 🚀 main.py (자동화 실행 루프)

모든 모듈을 결합하고 주기적으로 자동 거래를 실행하는 메인 엔진입니다.

Python

 

# main.py

import time
from data_handler import DataHandler # 데이터 영속성 및 관리
from broker_connector import BrokerConnector # 브로커 통신 (가정된 구현)
from quant_analyzer import QuantAnalyzer
from strategy_engine import StrategyEngine

# --- ⚠️ 필수 설정 ---
BROKER_NAME = 'Binance_TESTNET'
API_KEY = 'YOUR_TESTNET_API_KEY'
SECRET_KEY = 'YOUR_TESTNET_SECRET_KEY'
# --------------------

def run_quant_engine():
    # 1. 시스템 초기화
    data_handler = DataHandler(db_path='quant_trades.sqlite')
    broker_connector = BrokerConnector(data_handler=data_handler)
    
    # 2. 브로커 연결 (인증)
    if not broker_connector.connect_broker(API_KEY, SECRET_KEY, BROKER_NAME):
        print("❌ 브로커 연결 실패. 시스템 종료.")
        return

    # 3. 메인 자동 거래 루프
    while True:
        try:
            print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] --- 분석 루틴 시작 ---")
            
            # A. 데이터 로드 및 준비
            market_data, stock_universe = data_handler.load_analysis_data(period='90D')
            
            if market_data.empty or stock_universe.empty:
                print("⚠️ 분석 데이터 부족. 다음 루프 대기.")
                time.sleep(600)
                continue

            # B. 정량적 분석 실행
            analyzer = QuantAnalyzer(market_data['Returns'])
            strong_rank = analyzer.select_strong_stocks(stock_universe)
            
            # C. 전략 결정 및 포지션 추천
            engine = StrategyEngine(broker_connector, analyzer)
            recommendation = engine.get_position_recommendation(strong_rank)
            
            # D. 주문 실행 (구입/판매)
            engine.execute_order(recommendation)
            
            # E. 전략 통계 업데이트 (GET /api/strategies/stats)
            data_handler.update_strategy_stats() 
            
            print(f"--- 분석 루틴 완료. {BROKER_NAME} 잔고: {broker_connector.get_balance()} ---")

        except Exception as e:
            print(f"🔥 치명적인 엔진 오류: {e}")
        
        # 1시간마다 루프 실행 (분석 주기는 전략에 따라 조정 가능)
        time.sleep(3600) 

if __name__ == '__main__':
    run_quant_engine()