| Overall Statistics |
|
Total Orders 33211 Average Win 0.44% Average Loss -0.31% Compounding Annual Return -40.447% Drawdown 48.100% Expectancy -0.058 Start Equity 1000000 End Equity 617346.59 Net Profit -38.265% Sharpe Ratio -1.271 Sortino Ratio -2.133 Probabilistic Sharpe Ratio 0.639% Loss Rate 61% Win Rate 39% Profit-Loss Ratio 1.41 Alpha -0.301 Beta -0.197 Annual Standard Deviation 0.261 Annual Variance 0.068 Information Ratio -1.704 Tracking Error 0.288 Treynor Ratio 1.686 Total Fees $273706.54 Estimated Strategy Capacity $4700000.00 Lowest Capacity Asset MBLY Y2XYU2EUA8O5 Portfolio Turnover 271.20% |
# 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 = 20 self._opening_range_minutes = 5 self._leverage = 4 # 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 #and f.volume > 1_000_000 this will affect the performance a lot, dk why ], 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): # register automatic indicators when add a new security for security in changes.added_securities: security.atr = self.atr(security.symbol, self._indicator_period, resolution=Resolution.DAILY) security.volume_sma = self.sma(security.symbol, self._indicator_period, resolution=Resolution.DAILY) for security in changes.removed_securities: # Stop updating the indicators of assets that leave the universe to release computation resources. self.deregister_indicator(security.atr) self.deregister_indicator(security.volume_sma) 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) if self.is_warming_up: return # Filter 1: Select assets with abnormally high volume for the day so far. (Relative Volume > 100%) equities = [equity for equity in equities if equity.relative_volume and equity.relative_volume > 1] if not equities: return # Filter 2: 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, 4/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): self._update_stop_loss() # Get the current close price for the benchmark (SPY) 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 30-minute high symbols = [equity.symbol for equity in open_positions] history = self.history(symbols, 60, Resolution.MINUTE) # Fetch 30-minute data high_by_symbol_30min = history.high.unstack(0).max() # High of the last 30 minutes low_by_symbol_30min = history.high.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 30-minute high new_stop_loss = high_by_symbol_30min.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 30-minute low new_stop_loss = low_by_symbol_30min.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 = []
