Lean: QuantConnect을 구동하는 오픈소스 알고리즘 트레이딩 엔진 — C# & Python 설정 2026

class MyAlgorithm(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2020, 1, 1)
        self.SetEndDate(2026, 1, 1)
        self.SetCash(100000)
        self.AddEquity("AAPL", Resolution.Daily)

// C# 알고리즘 구조
namespace QuantConnect.Algorithm.CSharp
{
    public class MyAlgorithm : QCAlgorithm
    {
        public override void Initialize()
        {
            SetStartDate(2020, 1, 1);
            SetEndDate(2026, 1, 1);
            SetCash(100000);
            AddEquity("SPY", Resolution.Daily);
        }

        public override void OnData(Slice data)
        {
            if (!Portfolio.Invested)
            {
                SetHoldings("SPY", 1.0);
            }
        }
    }
}

# Ubuntu/Debian
sudo apt-get update && sudo apt-get install -y dotnet-sdk-8.0 git

# macOS
brew install dotnet-sdk git

# Windows — https://dotnet.microsoft.com/download에서 다운로드

# 저장소 클론
git clone https://github.com/QuantConnect/Lean.git
cd Lean

# 엔진 빌드
dotnet build QuantConnect.Lean.sln

# 샘플 백테스트 실행
dotnet run --project Launcher --config Config.json

# Python.NET 설치 (Python 알고리즘에 필수)
pip install pythonnet

# Lean의 Python API wrapper 설치
pip install quantconnect-stubs

# 설치 확인
python -c "from Algorithm.Python import *; print('Lean Python ready')"

# 공식 이미지 가져오기
docker pull quantconnect/lean:latest

# 컨테이너에서 백테스트 실행
docker run -v "$(pwd)/Data:/Data" \
  -v "$(pwd)/Results:/Results" \
  quantconnect/lean:latest --backtest

from AlgorithmImports import *

class SmaCrossoverAlgorithm(QCAlgorithm):
    def Initialize(self):
        # 백테스트 기간
        self.SetStartDate(2020, 1, 1)
        self.SetEndDate(2026, 1, 1)
        self.SetCash(100000)
        
        # 주식 추가
        self.symbol = self.AddEquity("AAPL", Resolution.Daily).Symbol
        
        # SMA 지표 생성
        self.fast_sma = self.SMA(self.symbol, 20, Resolution.Daily)
        self.slow_sma = self.SMA(self.symbol, 50, Resolution.Daily)
        
        # 트레이딩 전에 지표 웜업
        self.SetWarmUp(50)
        
        # 크로스오버 감지를 위한 이전 상태 추적
        self.previous_fast = None
        self.previous_slow = None

    def OnData(self, data: Slice):
        if self.IsWarmingUp:
            return
        
        # 현재 SMA 값 가져오기
        fast_val = self.fast_sma.Current.Value
        slow_val = self.slow_sma.Current.Value
        
        # 첫 유효 데이터에서 크로스오버 확인
        if self.previous_fast is not None:
            # 골든 크로스: 빠른선이 느린선 위로 교차
            if self.previous_fast <= self.previous_slow and fast_val > slow_val:
                if not self.Portfolio[self.symbol].Invested:
                    self.SetHoldings(self.symbol, 1.0)
            
            # 데드 크로스: 빠른선이 느린선 아래로 교차
            elif self.previous_fast >= self.previous_slow and fast_val < slow_val:
                if self.Portfolio[self.symbol].Invested:
                    self.Liquidate(self.symbol)
        
        self.previous_fast = fast_val
        self.previous_slow = slow_val

# main.py로 저장한 후:
lean backtest "MyProject" --output results.json

from AlgorithmImports import *
import numpy as np

class RiskParityAlgorithm(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2020, 1, 1)
        self.SetEndDate(2026, 1, 1)
        self.SetCash(100000)
        
        # 자산 유니버스 정의
        self.symbols = [
            self.AddEquity("SPY", Resolution.Daily).Symbol,   # S&P 500
            self.AddEquity("TLT", Resolution.Daily).Symbol,   # 20년 만기 국채
            self.AddEquity("GLD", Resolution.Daily).Symbol,   # 금
            self.AddEquity("VIXY", Resolution.Daily).Symbol,  # VIX
        ]
        
        # 변동성 계산 롤링 윈도우
        self.lookback = 60
        self.rebalance_interval = 30  # 일
        self.days_since_rebalance = 0

    def OnData(self, data: Slice):
        self.days_since_rebalance += 1
        
        if self.days_since_rebalance < self.rebalance_interval:
            return
        
        self.days_since_rebalance = 0
        
        # 역-변동성 가중치 계산
        volatilities = {}
        for symbol in self.symbols:
            history = self.History(symbol, self.lookback, Resolution.Daily)
            if len(history) < self.lookback:
                return
            returns = history["close"].pct_change().dropna()
            volatilities[symbol] = returns.std()
        
        # 역변동성 가중치
        inv_vol = {s: 1.0 / v for s, v in volatilities.items()}
        total = sum(inv_vol.values())
        weights = {s: v / total for s, v in inv_vol.items()}
        
        # 리밸런싱
        for symbol, weight in weights.items():
            self.SetHoldings(symbol, weight)
        
        self.Debug(f"Rebalanced: {weights}")

from AlgorithmImports import *

class OptionsStraddleAlgorithm(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2023, 1, 1)
        self.SetEndDate(2026, 1, 1)
        self.SetCash(50000)
        
        # 주식 및 옵션 체인 추가
        equity = self.AddEquity("SPY", Resolution.Minute)
        option = self.AddOption("SPY")
        option.SetFilter(-2, 2, timedelta(7), timedelta(30))
        self.symbol = option.Symbol
        
        self.Schedule.On(
            self.DateRules.WeekStart("SPY"),
            self.TimeRules.AfterMarketOpen("SPY", 30),
            self.TradeStraddle
        )

    def TradeStraddle(self):
        if self.Portfolio.Invested:
            return
        
        chain = self.CurrentSlice.OptionChains.get(self.symbol)
        if chain is None:
            return
        
        # ATM 옵션 찾기
        atm_strike = sorted(chain,
            key=lambda x: abs(x.Strike - chain.Underlying.Price))[0]
        
        atm_call = [x for x in chain if x.Strike == atm_strike.Strike and x.Right == OptionRight.Call][0]
        atm_put = [x for x in chain if x.Strike == atm_strike.Strike and x.Right == OptionRight.Put][0]
        
        # 스트래들 매수
        self.Buy(atm_call.Symbol, 1)
        self.Buy(atm_put.Symbol, 1)

from AlgorithmImports import *

class LiveSmaAlgorithm(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2026, 1, 1)
        self.SetCash(10000)
        
        # Interactive Brokers에서 실시간 데이터
        self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage)
        self.AddEquity("AAPL", Resolution.Minute)
        
        # 또는 QuantConnect로 모의 트레이딩
        # self.SetBrokerageModel(BrokerageName.QuantConnectBrokerage)

    def OnData(self, data):
        # 백테스트와 동일한 로직
        pass

{
  "environment": "live",
  "algorithm-type-name": "LiveSmaAlgorithm",
  "algorithm-language": "Python",
  "algorithm-location": "./MyAlgorithm.py",
  "job-user-id": "YOUR_USER_ID",
  "api-access-token": "YOUR_TOKEN",
  "ib-account": "DU123456",
  "ib-host": "127.0.0.1",
  "ib-port": 7497
}

from AlgorithmImports import *
import pickle
import numpy as np

class MLPredictionAlgorithm(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2023, 1, 1)
        self.SetEndDate(2026, 1, 1)
        self.SetCash(50000)
        
        self.symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
        
        # 사전 훈련된 모델 로드
        model_path = "./models/spy_predictor.pkl"
        with open(model_path, 'rb') as f:
            self.model = pickle.load(f)
        
        # 특성 가격 이력
        self.price_history = RollingWindow[float](20)

    def OnData(self, data: Slice):
        if not data.ContainsKey(self.symbol):
            return
        
        price = data[self.symbol].Close
        self.price_history.Add(float(price))
        
        if not self.price_history.IsReady:
            return
        
        # 가격 이력에서 특성 생성
        features = np.array(list(self.price_history)).reshape(1, -1)
        prediction = self.model.predict(features)[0]
        
        # 1 = 상승 예측, 0 = 하락 예측
        if prediction == 1 and not self.Portfolio[self.symbol].Invested:
            self.SetHoldings(self.symbol, 1.0)
        elif prediction == 0 and self.Portfolio[self.symbol].Invested:
            self.Liquidate(self.symbol)

from AlgorithmImports import *

class CustomAlphaModel(AlphaModel):
    def __init__(self):
        self.name = "CustomAlpha"
        self.securities = []
    
    def Update(self, algorithm: QCAlgorithm, data: Slice) -> List[Insight]:
        insights = []
        
        for security in self.securities:
            symbol = security.Symbol
            history = algorithm.History(symbol, 30, Resolution.Daily)
            
            if len(history) < 30:
                continue
            
            # 평균 회귀 신호
            sma = history["close"].mean()
            price = algorithm.Securities[symbol].Price
            
            if price < sma * 0.95:  # SMA보다 5% 아래 = 매수 신호
                insights.append(Insight.Price(
                    symbol, timedelta(5), InsightDirection.Up
                ))
            elif price > sma * 1.05:  # SMA보다 5% 위 = 매도 신호
                insights.append(Insight.Price(
                    symbol, timedelta(5), InsightDirection.Down
                ))
        
        return insights
    
    def OnSecuritiesChanged(self, algorithm, changes):
        self.securities.extend(changes.AddedSecurities)
        for removed in changes.RemovedSecurities:
            self.securities.remove(removed)

from AlgorithmImports import *

class MaxDrawdownRiskManagement(RiskManagementModel):
    def __init__(self, max_drawdown=0.10):
        self.max_drawdown = max_drawdown
        self.peak_value = 0
    
    def ManageRisk(self, algorithm: QCAlgorithm, targets: List[PortfolioTarget]):
        current_value = algorithm.Portfolio.TotalPortfolioValue
        
        if current_value > self.peak_value:
            self.peak_value = current_value
        
        drawdown = (self.peak_value - current_value) / self.peak_value
        
        if drawdown > self.max_drawdown:
            algorithm.Error(f"Max drawdown hit: {drawdown:.2%}. Liquidating.")
            algorithm.Liquidate()
            return []
        
        return targets

from AlgorithmImports import *

class FundamentalUniverseAlgorithm(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2022, 1, 1)
        self.SetEndDate(2026, 1, 1)
        self.SetCash(100000)
        
        # 시가총액 상위 50개 주식 선택
        self.AddUniverse(
            self.CoarseSelectionFilter,
            self.FineSelectionFilter
        )
        self.UniverseSettings.Resolution = Resolution.Daily
    
    def CoarseSelectionFilter(self, coarse):
        # 유동성 높은 주식 필터링
        sorted_by_dollar_volume = sorted(
            coarse, 
            key=lambda x: x.DollarVolume, 
            reverse=True
        )
        return [x.Symbol for x in sorted_by_dollar_volume[:100]]
    
    def FineSelectionFilter(self, fine):
        # 펀더멘털 기준 선택
        sorted_by_market_cap = sorted(
            fine,
            key=lambda x: x.MarketCap,
            reverse=True
        )
        return [x.Symbol for x in sorted_by_market_cap[:50]]

    def OnData(self, data):
        # 월별 리밸런싱
        pass