| Overall Statistics |
|
Total Orders 29091 Average Win 0.22% Average Loss -0.13% Compounding Annual Return 28.734% Drawdown 9.900% Expectancy 0.069 Start Equity 1000000 End Equity 1281598.79 Net Profit 28.160% Sharpe Ratio 1.149 Sortino Ratio 1.815 Probabilistic Sharpe Ratio 71.019% Loss Rate 59% Win Rate 41% Profit-Loss Ratio 1.62 Alpha 0.156 Beta -0.077 Annual Standard Deviation 0.126 Annual Variance 0.016 Information Ratio 0.019 Tracking Error 0.169 Treynor Ratio -1.883 Total Fees $90426.38 Estimated Strategy Capacity $2500000.00 Lowest Capacity Asset TCK TJVJKNII45T1 Portfolio Turnover 310.11% |
# region imports from AlgorithmImports import * # endregion class Test(QCAlgorithm): def initialize(self): self.set_start_date(2024, 1, 1) self.set_end_date(2024, 12, 31) self.set_cash(1_000_000) self.settings.automatic_indicator_warm_up = True self._selected = [] # Set the parameters. self._universe_size = 1000 self._indicator_period = 14 # days self._stop_loss_atr_distance = 0.5 # 0.1 => 10% of ATR self._stop_loss_risk_size = 0.01 # 0.01 => Lose 1% of the portfolio if stop loss is hit self._max_positions = 15 self._opening_range_minutes = 5 self._leverage = 3 # Set timezone to Eastern Time self.SetTimeZone("America/New_York") #Add three attributes self.previous_portfolio_value = 1000000 self.drawdown_threshold = 0.05 # 1% self.can_trade = True self.set_security_initializer(CustomSecurityInitializer(self.brokerage_model, FuncSecuritySeeder(self.get_last_known_prices))) # Add SPY so there is at least 1 asset at minute resolution to step the algorithm along. self._spy = self.add_equity('SPY').symbol self.universe_settings.resolution = Resolution.DAILY self.universe_settings.schedule.on(self.date_rules.month_start(self._spy)) self._universe = self.add_universe( lambda fundamentals: [ f.symbol for f in sorted( [ f for f in fundamentals if f.price > 5 # price > 5 and f.symbol != self._spy ], key=lambda f: f.dollar_volume )[-self._universe_size:] ] ) # run 5 min after the market open self.schedule.on( self.date_rules.every_day(self._spy), self.time_rules.after_market_open(self._spy, self._opening_range_minutes), self._scan_for_entries ) # Scheduled entry scan after market open. # exit 1 min before market close self.schedule.on( self.date_rules.every_day(self._spy), self.time_rules.before_market_close(self._spy, 1), self._exit ) self.set_warm_up(timedelta(2*self._indicator_period)) # Set Benchmark self.SetBenchmark("SPY") # Variable to hold the last calculated benchmark value self.lastBenchmarkValue = None # Our inital benchmark value scaled to match our portfolio # dont know why the inital value of SP 500 is 1047215.721 at 2024/1/1, I scale back by just deducting a constant, still have some minor gap but should be negligible in the long term # need to adjust for the constant depends on the initial gap self.BenchmarkPerformance = self.Portfolio.TotalPortfolioValue - 47000 def on_securities_changed(self, changes): for security in changes.added_securities: security.atr = self.atr(security.symbol, self._indicator_period, resolution=Resolution.DAILY) security.volume_sma = SimpleMovingAverage(self._indicator_period) def _scan_for_entries(self): symbols = list(self._universe.selected) equities = [self.securities[symbol] for symbol in symbols] history = self.history(symbols, 5, Resolution.MINUTE) volume_sum = history.volume.unstack(0).sum() equities = [equity for equity in equities if equity.symbol in volume_sum] for equity in equities: volume = volume_sum.loc[equity.symbol] equity.relative_volume = volume / equity.volume_sma.current.value if equity.volume_sma.is_ready else None equity.volume_sma.update(self.time, volume) # self.log(str(equity.symbol)+" "+str(equity.atr.current.value)+" "+str(equity.volume_sma.current.value)) if self.is_warming_up: return # Filter 1: ATR > 0.5 equities = [equity for equity in equities if equity.atr.is_ready and equity.atr.current.value > 0.5] if not equities: return #filter for volume (removed) #equities = [equity for equity in equities if equity.volume_sma.is_ready and equity.volume_sma.current.value > 1_000_000] #if not equities: # return # Filter 2: Select assets with Relative Volume > 100% equities = [equity for equity in equities if equity.relative_volume and equity.relative_volume > 1] if not equities: return # Filter 3: Select the top 20 assets with the greatest Relative Volume. equities = sorted(equities, key=lambda equity: equity.relative_volume)[-self._max_positions:] history = history.loc[[equity.symbol for equity in equities]] open_by_symbol = history.open.unstack(0).iloc[0] close_by_symbol = history.close.unstack(0).iloc[-1] high_by_symbol = history.high.unstack(0).max() low_by_symbol = history.low.unstack(0).min() # Create orders for the target assets. # Calculate position sizes so that if you fill an order at the high (low) of the first 5-minute bar # and hit a stop loss based on 10% of the ATR, you only lose x% of portfolio value. orders = [] for symbol in close_by_symbol[close_by_symbol > open_by_symbol].index: equity = self.securities[symbol] orders.append({ 'equity': equity, 'entry_price': high_by_symbol.loc[equity.symbol], 'stop_price': high_by_symbol.loc[equity.symbol] - self._stop_loss_atr_distance * equity.atr.current.value # stop price }) for symbol in close_by_symbol[close_by_symbol < open_by_symbol].index: equity = self.securities[symbol] orders.append({ 'equity': equity, 'entry_price': low_by_symbol.loc[equity.symbol], 'stop_price': low_by_symbol.loc[equity.symbol] + self._stop_loss_atr_distance * equity.atr.current.value }) for order in orders: equity = order['equity'] self._selected.append(equity) self._create_empty_order_tickets(equity) self.add_security(equity.symbol, leverage=self._leverage) quantity = int((self._stop_loss_risk_size * self.portfolio.total_portfolio_value) / (order['entry_price'] - order['stop_price'])) quantity_limit = self.calculate_order_quantity(equity.symbol, 3/self._max_positions) quantity = min(abs(quantity), quantity_limit) * np.sign(quantity) if quantity: equity.stop_loss_price = order['stop_price'] equity.entry_ticket = self.stop_market_order(equity.symbol, quantity, order['entry_price'], tag='Entry') def on_data(self, data): current_time = self.Time # At market close (4:00 PM ET), save the portfolio value for drawdown calculation if current_time.hour == 16 and current_time.minute == 0: self.previous_portfolio_value = self.Portfolio.TotalPortfolioValue self.can_trade = True # Reset trading status for the next day # Calculate real-time drawdown during the trading day if self.previous_portfolio_value > 0: # Calculate drawdown as a percentage of the previous day's portfolio value drawdown = (self.previous_portfolio_value - self.Portfolio.TotalPortfolioValue) / self.previous_portfolio_value # If drawdown exceeds the threshold, stop trading and liquidate positions if drawdown < -self.drawdown_threshold: # Negative drawdown self._exit() # Liquidate all positions self.can_trade = False # Disable further trading self.Debug("Exiting all positions due to daily drawdown exceeding 5%.") return # Skip further processing # Get the current close price for the benchmark (SPY) self._update_stop_loss() benchmark = self.Securities["SPY"].Close # If we had a previous close, update our benchmark performance. if self.lastBenchmarkValue is not None and self.lastBenchmarkValue != 0: self.BenchmarkPerformance = self.BenchmarkPerformance * (benchmark / self.lastBenchmarkValue) # Save the current close price for the next update self.lastBenchmarkValue = benchmark self.Plot("Strategy vs Benchmark", "Portfolio Value", self.Portfolio.TotalPortfolioValue) self.Plot("Strategy vs Benchmark", "Benchmark", self.BenchmarkPerformance) def _update_stop_loss(self): # Only update stop loss for securities currently in the portfolio open_positions = [equity for equity in self._selected if equity.symbol in self.Portfolio and self.Portfolio[equity.symbol].Invested] if not open_positions: return # Update stop loss dynamically based on the last 60-minute high symbols = [equity.symbol for equity in open_positions] history = self.history(symbols, 60, Resolution.MINUTE) # Fetch 60-minute data high_by_symbol_60min = history.high.unstack(0).max() # High of the last 60 minutes low_by_symbol_60min = history.low.unstack(0).min() for equity in open_positions: # Only update the stop loss if the new stop loss is higher (for long positions) if self.Portfolio[equity.symbol].Quantity > 0: # Calculate the new stop loss based on the latest 60-minute high new_stop_loss = high_by_symbol_60min.loc[equity.symbol] - self._stop_loss_atr_distance * equity.atr.current.value if new_stop_loss > equity.stop_loss_price: # Cancel the existing stop loss order if equity.stop_loss_ticket is not None: equity.stop_loss_ticket.Cancel() # Submit a new stop loss order quantity = self.Portfolio[equity.symbol].Quantity equity.stop_loss_ticket = self.StopMarketOrder(equity.symbol, -quantity, new_stop_loss, tag='ATR Stop') # Update the stored stop loss price equity.stop_loss_price = new_stop_loss # For short positions, stop loss would be lower elif self.Portfolio[equity.symbol].Quantity < 0: # Calculate the new stop loss based on the latest 60-minute low new_stop_loss = low_by_symbol_60min.loc[equity.symbol] + self._stop_loss_atr_distance * equity.atr.current.value if new_stop_loss < equity.stop_loss_price: # Cancel the existing stop loss order if equity.stop_loss_ticket is not None: equity.stop_loss_ticket.Cancel() # Submit a new stop loss order quantity = self.Portfolio[equity.symbol].Quantity equity.stop_loss_ticket = self.StopMarketOrder(equity.symbol, -quantity, new_stop_loss, tag='ATR Stop') # Update the stored stop loss price equity.stop_loss_price = new_stop_loss def on_order_event(self, order_event: OrderEvent) -> None: if order_event.status != OrderStatus.FILLED: return security = self.securities[order_event.symbol] # When the entry order is hit, place the exit order: Stop loss based on ATR. if order_event.ticket == security.entry_ticket: security.stop_loss_ticket = self.stop_market_order(order_event.symbol, -security.entry_ticket.quantity, security.stop_loss_price, tag='ATR Stop') # When the stop loss order is hit, cancel the MOC order. elif order_event.ticket == security.stop_loss_ticket: self._create_empty_order_tickets(security) # Create some members on the Equity object to store each order ticket. def _create_empty_order_tickets(self, equity): equity.entry_ticket = None equity.stop_loss_ticket = None # Liquidate the portfolio, remove order tickets, remove the minute-resolution data subscriptions. def _exit(self): self.liquidate() for equity in self._selected: self._create_empty_order_tickets(equity) self.remove_security(equity.symbol) self._selected = [] class CustomSlippageModel: def get_slippage_approximation(self, asset: Security, order: Order): # We set a maximum 0.05% slippage that is linearly linked to the ratio of the order size versus the previous bar's volume. #return asset.price * 0.0005 return asset.price * 0.0005 * min(1, order.absolute_quantity / asset.volume) class CustomSecurityInitializer(BrokerageModelSecurityInitializer): def __init__(self, brokerage_model, security_seeder): super().__init__(brokerage_model, security_seeder) def initialize(self, security): super().initialize(security) # To set the slippage model to the custom one for all stocks entering the universe. security.set_slippage_model(CustomSlippageModel())
