Created with Highcharts 12.1.2EquityJan 2019May 2019Sep 2019Jan 2020May 2020Sep 2020Jan 2021May 2021Sep 2021Jan 2022May 2022Sep 20224k12k040-24000.24-1101000240M0100k048
Overall Statistics
Total Orders
22
Average Win
15.73%
Average Loss
-3.41%
Compounding Annual Return
9.425%
Drawdown
20.200%
Expectancy
1.041
Start Equity
7500
End Equity
10204.74
Net Profit
36.063%
Sharpe Ratio
0.507
Sortino Ratio
0.65
Probabilistic Sharpe Ratio
16.188%
Loss Rate
64%
Win Rate
36%
Profit-Loss Ratio
4.61
Alpha
0.073
Beta
-0.068
Annual Standard Deviation
0.119
Annual Variance
0.014
Information Ratio
-0.339
Tracking Error
0.37
Treynor Ratio
-0.892
Total Fees
$22.00
Estimated Strategy Capacity
$140000000.00
Lowest Capacity Asset
XLE RGRPZX100F39
Portfolio Turnover
0.91%
# region imports
from AlgorithmImports import *
import numpy as np
# endregion




class BootCampTask(QCAlgorithm):
    
    
   
    def initialize(self):

        self.set_start_date(2019, 1, 1)
        self.set_end_date(2022, 6, 1)
        self.set_cash(7500)

        # self.default_order_properties.time_in_force = TimeInForce.GOOD_TIL_CANCELED

        # Add underlying
        self.underlying = "XLE" #"TNA"  # "IWM" #"UUP", "XLK", "DBA", "USO", "GLD", "DIA", "SPY"

        # self.add_equity("SPY", Resolution.MINUTE, data_normalization_mode=DataNormalizationMode.RAW)
        
        self.add_equity(self.underlying, Resolution.DAILY, data_normalization_mode=DataNormalizationMode.ADJUSTED)
        self._symbol = self.underlying 

        self.settings.daily_precise_end_time = False
        
        self.set_benchmark(self.underlying) # set the benchmark

        # Build history for indicators
        
        self.history[TradeBar](self._symbol, 100, Resolution.DAILY)#, dataNormalizationMode=DataNormalizationMode.SCALED_RAW)

        # Warm up the close price and trade bar rolling windows with the previous 100-day trade bar data
        self.rollingwindow = RollingWindow[TradeBar](100)
        self.Consolidate(self._symbol, Resolution.DAILY, self.CustomBarHandler)

        

        # Risk parameters
        self.target_risk = 0.12 # percentage 
        self.instrument_look_back = 30 # number of days 
        self.instrument_risk_annualizer = 16 #math.sqrt(256/self.instrument_look_back)
        self.stop_price_instrument_ratio = 0.5 # percentage

        # Strategy parameters
        short_look_back = 16
        long_look_back = 64

        self.short_ma = self.sma(self._symbol, short_look_back) #, Resolution.DAILY)
        self.long_ma = self.sma(self._symbol, long_look_back) #, Resolution.DAILY)
        self.set_warm_up(long_look_back) # Warm up for the long ma

        self.strategy_direction = ""
        self.new_strategy_direction = ""
     
        # Order ticket for our stop order, Datetime when stop order was last hit
        self.stop_market_ticket = None



        
    def on_data(self, data):
        
        # Warming up the data and indicator 
        if not self.long_ma.is_ready: return
        if not self.rollingwindow.is_ready: return

        # Calculate the trade signal
        self.cal_trade_signal()
           

        # Get current close
        self.trade_bar_df = self.pandas_converter.get_data_frame[TradeBar](list(self.rollingwindow)[::-1]) 
        self.current_close = self.trade_bar_df.close[-1] 

        # self.debug(self.current_close)


        # Calculate annualized instrument risk
        self.cal_annualized_instrument_risk()

        self.plot("Risks", "Annualized Instrument Risk", self.annalized_instrument_risk)


        # Get stop adjustment
        stop_adjustment = self.annalized_instrument_risk*self.stop_price_instrument_ratio


        # Calculate the stop price
        self.starting_stop_price_long =  self.current_close - stop_adjustment
        self.starting_stop_price_short =  self.current_close + stop_adjustment 

        # Do not trade if the position was liquidated and no signal change happens to trigger an open trade.

        # if not self.portfolio.invested and self.new_strategy_direction == self.strategy_direction: return

        #if self.new_strategy_direction == self.strategy_direction: return
        

        if not self.portfolio.invested and self.new_strategy_direction != self.strategy_direction: 

            # Calculate the trade size
            self.cal_trade_size()

            # Open trade 
            self.open_trade()            


            # Open Trailing-Stop-Loss trade
            if self.new_strategy_direction == "long":
                self.stop_price = self.starting_stop_price_long
                self.open_tsl_trade(-self.size)
            else:
                self.stop_price = self.starting_stop_price_short
                self.open_tsl_trade(self.size)


            # Start tracking price for trailing stop loss order    
            self.highest_price = self.current_close 
            self.lowest_price = self.current_close 


        elif self.portfolio.invested and self.stop_market_ticket:
            # update trailing-stop-loss price
            self.update_tsl_order()

             

        #Plot the moving stop price on "Data Chart" with "Stop Price" series name
        self.plot_data()
        

            

    def on_order_event(self, order_event):
        
        if order_event.status != OrderStatus.FILLED:
            return
        
        self.Debug(order_event)
        


    def CustomBarHandler(self, bar):
        self.rollingwindow.add(bar)      


    def Instrument_Risk(self, df) -> float:

        self.daily_returns = df.close.pct_change()

        self.risk = self.current_close * self.daily_returns[-self.instrument_look_back-1:-1].std() # offset by -1 for not include the EoD price on the trading day

        return self.risk

    

    def cal_annualized_instrument_risk(self):
    
        self.instrument_risk = self.Instrument_Risk(self.trade_bar_df) 

        self.annalized_instrument_risk = self.instrument_risk * self.instrument_risk_annualizer # self.instrument_risk

        # self.Debug(f"Annualized risk by price points is {round(self.annalized_instrument_risk, 2)}")

    
    def cal_trade_size(self):
        
        self.target_exposure = self.target_risk*self.portfolio.total_portfolio_value

        self.Debug(f"Target exposure is {round(self.target_exposure, 2)}.")
        
        self.notional_exposure = self.target_exposure/self.annalized_instrument_risk

        self.size = math.floor(self.notional_exposure)

        # self.Debug(f"Trade size is {str(self.size)}.")



    def cal_trade_signal(self):
        if self.short_ma.current.value  > self.long_ma.current.value :
            self.new_strategy_direction = "long"
        else:
            self.new_strategy_direction = "short"

        if self.new_strategy_direction != self.strategy_direction:
            self.debug(f"Strategy direction changed to: {self.new_strategy_direction}.")
            # self.strategy_direction = self.new_strategy_direction
            



    def open_trade(self):
        self.strategy_direction = self.new_strategy_direction
                    
        if self.strategy_direction == "long":
            self.market_on_open_order(self.underlying, self.size)
        else:
            self.market_on_open_order(self.underlying, -1 * self.size)
        

        
    def open_tsl_trade(self, size):
        self.stop_market_ticket = self.stop_market_order(self.underlying, size, self.stop_price)
                


    def update_tsl_order(self):

        self.start_stop_price = self.stop_price

        update_fields = UpdateOrderFields()

        if self.strategy_direction == "long": 
            if self.current_close > self.highest_price:
                delta = self.current_close - self.highest_price 
                update_fields.stop_price = self.start_stop_price + delta 
                self.stop_market_ticket.update(update_fields)
                self.highest_price = self.current_close
                self.stop_price = update_fields.stop_price
            else:
                pass #self.stop_price = self.start_stop_price

            

        if self.strategy_direction == "short":
            if self.current_close < self.lowest_price:
                delta = self.lowest_price - self.current_close
                update_fields.stop_price = self.start_stop_price - delta
                self.stop_market_ticket.update(update_fields)
                self.lowest_price = self.current_close
                self.stop_price = update_fields.stop_price
            else:
                pass #self.stop_price = self.start_stop_price 
           



    def plot_data(self):
        self.plot("Data Chart", "Asset Price", self.current_close)

        self.plot("Data Chart", "Stop Price", self.stop_price)

        # self.plot("Data Chart", "Short MA", self.short_ma.current.value)
        # self.plot("Data Chart", "Long MA", self.long_ma.current.value)