| Overall Statistics |
|
Total Orders 102 Average Win 10.70% Average Loss -2.13% Compounding Annual Return 43.107% Drawdown 34.100% Expectancy 2.301 Start Equity 10000 End Equity 85434.44 Net Profit 754.344% Sharpe Ratio 1.21 Sortino Ratio 0.995 Probabilistic Sharpe Ratio 67.388% Loss Rate 45% Win Rate 55% Profit-Loss Ratio 5.01 Alpha 0.256 Beta 0.323 Annual Standard Deviation 0.241 Annual Variance 0.058 Information Ratio 0.708 Tracking Error 0.26 Treynor Ratio 0.902 Total Fees $901.53 Estimated Strategy Capacity $350000000.00 Lowest Capacity Asset MSTR RBGP9S2961YD Portfolio Turnover 1.90% |
from AlgorithmImports import * from sklearn.linear_model import LinearRegression from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score import numpy as np import pandas as pd # Custom fee model for 0.1% per trade class PercentageFeeModel(FeeModel): def GetOrderFee(self, parameters): security = parameters.Security order = parameters.Order fee = 0.001 * security.Price * abs(order.Quantity) currency = security.QuoteCurrency.Symbol return OrderFee(CashAmount(fee, currency)) class MLTradingAlgorithm(QCAlgorithm): def Initialize(self): # Algorithm Parameters self.SetStartDate(2019, 1, 1) # Start date self.SetEndDate(2024, 12, 31) # End date self.SetCash(10000) # Initial capital # Configurable ticker symbols and allocation percentage self.trading_ticker = self.GetParameter("trading_ticker", "MSTR") self.benchmark_ticker = self.GetParameter("benchmark_ticker", "SPY") self.allocation_percentage = self.GetParameter("allocation_percentage", 0.4) # Add trading equity with custom fee and slippage models trading_security = self.AddEquity(self.trading_ticker, Resolution.Daily) trading_security.SetFeeModel(PercentageFeeModel()) trading_security.SetSlippageModel(ConstantSlippageModel(0)) self.symbol = trading_security.Symbol # Add benchmark equity with custom fee and slippage models benchmark_security = self.AddEquity(self.benchmark_ticker, Resolution.Daily) benchmark_security.SetFeeModel(PercentageFeeModel()) benchmark_security.SetSlippageModel(ConstantSlippageModel(0)) self.benchmark_symbol = benchmark_security.Symbol # RollingWindow to store 200 days of TradeBar data for trading asset self.data = RollingWindow[TradeBar](200) # Warm-up period self.SetWarmUp(200) # Initialize Linear Regression model self.model = LinearRegression() self.training_count = 0 self.is_model_trained = False # Tracks if the model is trained # Schedule training every Monday at 10:00 AM self.Schedule.On(self.DateRules.Every(DayOfWeek.Monday), self.TimeRules.At(10, 0), self.TrainModel) # Initialize variables for benchmarking self.beat_benchmark_count = 0 self.trade_entry_price = None # To store the entry price of a trade self.benchmark_entry_price = None # To store the benchmark price at trade entry # Parameters for Grid Search optimization self.sma_short_window = int(self.GetParameter("sma_short_window", 10)) self.sma_long_window = int(self.GetParameter("sma_long_window", 60)) self.rsi_window = int(self.GetParameter("rsi_window",14)) self.macd_fast = int(self.GetParameter("macd_fast", 12)) self.macd_slow = int(self.GetParameter("macd_slow", 26)) self.macd_signal = int(self.GetParameter("macd_signal", 9)) self.hv_window = int(self.GetParameter("hv_window", 30)) def OnData(self, data): # Ensure data exists for trading symbol if not data.ContainsKey(self.symbol): return trade_bar = data[self.symbol] if trade_bar is None: return # Add TradeBar to Rolling Window self.data.Add(trade_bar) # Check if RollingWindow is ready if not self.data.IsReady or self.data.Count < 200: return # Ensure model is trained before making predictions if not self.is_model_trained: self.Debug("Model is not trained yet. Skipping prediction.") return # Extract features for prediction df = self.GetFeatureDataFrame() if df is None or len(df) < 1: return latest_features = df.iloc[-1, :-1].values.reshape(1, -1) # Make predictions using threshold try: pred_value = self.model.predict(latest_features)[0] prediction = 1 if pred_value > 0.5 else 0 except Exception as e: self.Debug(f"Error: Model prediction failed. {e}") return # Trading logic holdings = self.Portfolio[self.symbol].Quantity # Buy if prediction = 1 and not currently invested if prediction == 1 and holdings <= 0: self.SetHoldings(self.symbol, self.allocation_percentage) # Record the entry prices for the trade and benchmark self.trade_entry_price = trade_bar.Close if self.benchmark_symbol in data and data[self.benchmark_symbol] is not None: self.benchmark_entry_price = data[self.benchmark_symbol].Close else: self.benchmark_entry_price = None # Sell if prediction = 0 and currently invested elif prediction == 0 and holdings > 0: # Calculate trade return and benchmark return if self.trade_entry_price is not None and self.benchmark_entry_price is not None: trade_exit_price = trade_bar.Close trade_return = (trade_exit_price - self.trade_entry_price) / self.trade_entry_price if self.benchmark_symbol in data and data[self.benchmark_symbol] is not None: benchmark_exit_price = data[self.benchmark_symbol].Close benchmark_return = (benchmark_exit_price - self.benchmark_entry_price) / self.benchmark_entry_price # Compare trade return with benchmark return if trade_return > benchmark_return: self.beat_benchmark_count += 1 # Reset entry prices after the trade is closed self.trade_entry_price = None self.benchmark_entry_price = None # Execute the sell order self.Liquidate(self.symbol) def TrainModel(self): # Prepare training data df = self.GetFeatureDataFrame() if df is None or len(df) < 50: # Require enough data to train self.Debug("Insufficient data for training.") return # Split data chronologically (no shuffle) X = df.iloc[:, :-1] # Features y = df.iloc[:, -1] # Target (0 or 1) X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, shuffle=False, random_state=42 ) # Train Linear Regression model self.model.fit(X_train, y_train) self.is_model_trained = True # Evaluate model performance y_train_pred = self.model.predict(X_train) y_train_pred_binary = [1 if val > 0.5 else 0 for val in y_train_pred] train_accuracy = accuracy_score(y_train, y_train_pred_binary) y_test_pred = self.model.predict(X_test) y_test_pred_binary = [1 if val > 0.5 else 0 for val in y_test_pred] test_accuracy = accuracy_score(y_test, y_test_pred_binary) self.training_count += 1 self.Debug(f"Training #{self.training_count}: " f"Train Accuracy: {train_accuracy:.2%}, " f"Test Accuracy: {test_accuracy:.2%}") def GetFeatureDataFrame(self): # Wait until we have 200 data points in the rolling window if self.data.Count < 200: return None # Convert rolling window data to a DataFrame close_prices = [bar.Close for bar in self.data] df = pd.DataFrame(close_prices, columns=["Close"]) # Feature Engineering df["SMA_short"] = df["Close"].rolling(window=self.sma_short_window).mean() df["SMA_long"] = df["Close"].rolling(window=self.sma_long_window).mean() # RSI Calculation delta = df["Close"].diff() gain = (delta.where(delta > 0, 0)).rolling(self.rsi_window).mean() loss = (-delta.where(delta < 0, 0)).rolling(self.rsi_window).mean() rs = gain / loss df["RSI"] = 100 - (100 / (1 + rs)) # MACD Calculation df["MACD"] = df["Close"].ewm(span=self.macd_fast, adjust=False).mean() - df["Close"].ewm(span=self.macd_slow, adjust=False).mean() df["MACD_Signal"] = df["MACD"].ewm(span=self.macd_signal, adjust=False).mean() # Historical Volatility (HV) df["HV"] = df["Close"].pct_change().rolling(window=self.hv_window).std() * np.sqrt(252) # Define Target: 1 if next day's Close > today's Close, else 0 df["Target"] = (df["Close"].shift(-1) > df["Close"]).astype(int) # Remove rows with NaN values df.dropna(inplace=True) return df def OnEndOfAlgorithm(self): # Print the number of times the strategy beat the benchmark self.Log(f"Number of times strategy beat {self.benchmark_ticker}: {self.beat_benchmark_count}")
