| Overall Statistics |
|
Total Orders 114 Average Win 19.75% Average Loss -7.19% Compounding Annual Return 83.046% Drawdown 70.900% Expectancy 1.432 Start Equity 10000 End Equity 370374.35 Net Profit 3603.743% Sharpe Ratio 1.364 Sortino Ratio 1.199 Probabilistic Sharpe Ratio 57.647% Loss Rate 35% Win Rate 65% Profit-Loss Ratio 2.75 Alpha 0.613 Beta 0.871 Annual Standard Deviation 0.516 Annual Variance 0.267 Information Ratio 1.207 Tracking Error 0.496 Treynor Ratio 0.808 Total Fees $5234.78 Estimated Strategy Capacity $260000000.00 Lowest Capacity Asset MSTR RBGP9S2961YD Portfolio Turnover 5.24% |
from AlgorithmImports import * from sklearn.linear_model import LogisticRegression from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score from sklearn.preprocessing import StandardScaler 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", 1) # 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 Logistic Regression model and scaler self.model = LogisticRegression(random_state=42) self.scaler = StandardScaler() # Retained for feature scaling 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 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) # Transform features using the fitted scaler latest_features_scaled = self.scaler.transform(latest_features) # Make predictions using probability threshold try: prob_class = self.model.predict_proba(latest_features_scaled)[0][1] # Probability of class 1 prediction = 1 if prob_class > 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 # Fixed typo ) # Fit the scaler on training data and transform both train and test sets X_train_scaled = self.scaler.fit_transform(X_train) X_test_scaled = self.scaler.transform(X_test) # Train Logistic Regression model on scaled data self.model.fit(X_train_scaled, y_train) self.is_model_trained = True # Evaluate model performance y_train_prob = self.model.predict_proba(X_train_scaled)[:, 1] y_train_pred_binary = [1 if val > 0.5 else 0 for val in y_train_prob] train_accuracy = accuracy_score(y_train, y_train_pred_binary) y_test_prob = self.model.predict_proba(X_test_scaled)[:, 1] y_test_pred_binary = [1 if val > 0.5 else 0 for val in y_test_prob] 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_10"] = df["Close"].rolling(window=10).mean() df["SMA_50"] = df["Close"].rolling(window=50).mean() # RSI Calculation delta = df["Close"].diff() gain = (delta.where(delta > 0, 0)).rolling(14).mean() loss = (-delta.where(delta < 0, 0)).rolling(14).mean() rs = gain / loss df["RSI"] = 100 - (100 / (1 + rs)) # MACD Calculation df["MACD"] = df["Close"].ewm(span=12, adjust=False).mean() - df["Close"].ewm(span=26, adjust=False).mean() df["MACD_Signal"] = df["MACD"].ewm(span=9, adjust=False).mean() # Historical Volatility (HV_30) df["HV_30"] = df["Close"].pct_change().rolling(window=30).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}")
