Following the discussion from the previous post [Technical Indicator - Relative Strength Index (RSI)], we will implement a simple trading strategy using the RSI indicator.
Trading Logic:
Suppose we apply RSI(14) for the daily closing price of USDJPY over the year of 2020.
- Based on a sliding window approach to collect the previous 14 closing price
- Calculate the latest RSI value
- Order open conditions:
- if we have NO outstanding position,
- if RSI value < 30, we open a buy order
- if RSI value > 70, we open a sell order
- Order close conditions:
- if we have outstanding position,
- if we previously submit a buy order and RSI value reverses back to or above 50, we close the buy order
- if we previously submit a sell order and RSI value reverses back to or below 50, we close the sell order
- Repeat the process until the backtest period end
Now, let's write down our trading algorithm step-by-step.
Step 1. Calculate RSI value
First of all, we define 'rsi_period' and 'instrument' at initialization.
1 2 3 4 5 6 7 8 9 10 | from AlgoAPI import AlgoAPIUtil, AlgoAPI_Backtest from datetime import datetime, timedelta from talib import RSI import numpy as np class AlgoEvent: def __init__(self): self.timer = datetime(1970,1,1) self.rsi_period = 14 self.instrument = "USDJPY" |
We use the API function getHistoricalBar to collect historical observations. Then, we apply a python library 'talib.RSI' to calculate the latest RSI value.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | def on_marketdatafeed(self, md, ab): # execute stratey every 24 hours if md.timestamp >= self.timer+timedelta(hours=24): # get last 14 closing price res = self.evt.getHistoricalBar({"instrument":self.instrument}, self.rsi_period+1, "D") arr = [res[t]['c'] for t in res] # calculate the current RSI value RSI_cur = RSI(np.array(arr), self.rsi_period)[-1] # print out RSI value to console self.evt.consoleLog("RSI_cur = ", RSI_cur) # update timer self.timer = md.timestamp |
Step 2. Order Entry Conditions
We further define 'rsi_overbought', 'rsi_oversold' and 'position' at initialization.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | from AlgoAPI import AlgoAPIUtil, AlgoAPI_Backtest from datetime import datetime, timedelta from talib import RSI import numpy as np class AlgoEvent: def __init__(self): self.timer = datetime(1970,1,1) self.rsi_period = 14 self.rsi_overbought = 70 self.rsi_oversold = 30 self.position = 0 self.instrument = "USDJPY" |
We also create a function 'open_order' to handle order submissions.
1 2 3 4 5 6 7 8 | def open_order(self, buysell): order = AlgoAPIUtil.OrderObject() order.instrument = self.instrument order.openclose = 'open' order.buysell = buysell #1=buy, -1=sell order.ordertype = 0 #0=market, 1=limit order.volume = 0.01 self.evt.sendOrder(order) |
We also update the system's function 'on_marketdatafeed' for order open logic.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 | def on_marketdatafeed(self, md, ab): # execute stratey every 24 hours if md.timestamp >= self.timer+timedelta(hours=24): # get last 14 closing price res = self.evt.getHistoricalBar({"instrument":self.instrument}, self.rsi_period+1, "D") arr = [res[t]['c'] for t in res] # calculate the current RSI value RSI_cur = RSI(np.array(arr), self.rsi_period)[-1] # print out RSI value to console self.evt.consoleLog("RSI_cur = ", RSI_cur) # open an order if we have no outstanding position if self.position==0: # open a sell order if it is overbought if RSI_cur>self.rsi_overbought: self.open_order(-1) # open a buy order if it is oversold elif RSI_cur<self.rsi_oversold: self.open_order(1) # update timer self.timer = md.timestamp |
Step 3. Order Close Condition
We firstly edit the system's 'on_orderfeed' function. When there is order status update from the system, we record the system tradeID and update our outstanding position.
1 2 3 4 5 6 7 8 | def on_orderfeed(self, of): # when system confirm an order, update last_tradeID and position if of.status=="success": self.position += of.fill_volume*of.buysell if self.position==0: self.last_tradeID = "" else: self.last_tradeID = of.tradeID |
Secondly, we create a 'close_order' function.
1 2 3 4 5 | def close_order(self): order = AlgoAPIUtil.OrderObject() order.openclose = 'close' order.tradeID = self.last_tradeID self.evt.sendOrder(order) |
Finally, we update 'on_marketdatafeed' function for order closing logic.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 | def on_marketdatafeed(self, md, ab): # execute stratey every 24 hours if md.timestamp >= self.timer+timedelta(hours=24): # get last 14 closing price res = self.evt.getHistoricalBar({"instrument":self.instrument}, self.rsi_period+1, "D") arr = [res[t]['c'] for t in res] # calculate the current RSI value RSI_cur = RSI(np.array(arr), self.rsi_period)[-1] # print out RSI value to console self.evt.consoleLog("RSI_cur = ", RSI_cur) # open an order if we have no outstanding position if self.position==0: # open a sell order if it is overbought if RSI_cur>self.rsi_overbought: self.open_order(-1) # open a buy order if it is oversold elif RSI_cur<self.rsi_oversold: self.open_order(1) # check condition to close an order else: # close a position if we have previously open a buy order and RSI now reverse above 50 if self.position>0 and RSI_cur>50: self.close_order() # close a position if we have previously open a sell order and RSI now reverse below 50 elif self.position<0 and RSI_cur<50: self.close_order() # update timer self.timer = md.timestamp |
Full Source Code
Combining all above, the full script is presented below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | from AlgoAPI import AlgoAPIUtil, AlgoAPI_Backtest from datetime import datetime, timedelta from talib import RSI import numpy as np class AlgoEvent: def __init__(self): self.timer = datetime(1970,1,1) self.rsi_period = 14 self.rsi_overbought = 70 self.rsi_oversold = 30 self.position = 0 self.last_tradeID = "" self.instrument = "USDJPY" def start(self, mEvt): self.evt = AlgoAPI_Backtest.AlgoEvtHandler(self, mEvt) self.evt.start() def on_marketdatafeed(self, md, ab): # execute stratey every 24 hours if md.timestamp >= self.timer+timedelta(hours=24): # get last 14 closing price res = self.evt.getHistoricalBar({"instrument":self.instrument}, self.rsi_period+1, "D") arr = [res[t]['c'] for t in res] # calculate the current RSI value RSI_cur = RSI(np.array(arr), self.rsi_period)[-1] # print out RSI value to console self.evt.consoleLog("RSI_cur = ", RSI_cur) # open an order if we have no outstanding position if self.position==0: # open a sell order if it is overbought if RSI_cur>self.rsi_overbought: self.open_order(-1) # open a buy order if it is oversold elif RSI_cur<self.rsi_oversold: self.open_order(1) # check condition to close an order else: # close a position if we have previously open a buy order and RSI now reverse above 50 if self.position>0 and RSI_cur>50: self.close_order() # close a position if we have previously open a sell order and RSI now reverse below 50 elif self.position<0 and RSI_cur<50: self.close_order() # update timer self.timer = md.timestamp def open_order(self, buysell): order = AlgoAPIUtil.OrderObject() order.instrument = self.instrument order.openclose = 'open' order.buysell = buysell #1=buy, -1=sell order.ordertype = 0 #0=market, 1=limit order.volume = 0.01 self.evt.sendOrder(order) def close_order(self): order = AlgoAPIUtil.OrderObject() order.openclose = 'close' order.tradeID = self.last_tradeID self.evt.sendOrder(order) def on_bulkdatafeed(self, isSync, bd, ab): pass def on_newsdatafeed(self, nd): pass def on_weatherdatafeed(self, wd): pass def on_econsdatafeed(self, ed): pass def on_orderfeed(self, of): # when system confirm an order, update last_tradeID and position if of.status=="success": self.position += of.fill_volume*of.buysell if self.position==0: self.last_tradeID = "" else: self.last_tradeID = of.tradeID def on_dailyPLfeed(self, pl): pass def on_openPositionfeed(self, op, oo, uo): pass |
Results
Now, we are prepared to backtest this strategy.
Backtest Settings:
- Instrument: USDJPY
- Period: 2020.01 - 2020.12
- Initial Capital: US$1,000
- Data Interval: 1-day bar
- Allow Shortsell: True
Backtest Result
Final Thoughts
Although the backtest result above looks promising, some questions are raised below to brainstorm how to further make this strategy practical.
- From the logics above, we assume short selling is possible. What if it is not feasible in stock market?
- We are using a 14-day RSI.
- Any better choice than 14?
- Is it applicable to other timeframes, eg. 1-min, 1-hour?
- Will multiple RSIs produce a more reliable result?
- The RSI value of 30/70 are used as reference points for order entry.
- Any other better combinations?
- Can this entry points be dynamically changing?
- Can the 30/70 rule generally applicable to other financial instruments and asset classes?
- In our order exit logic, we close a trade when RSI reverses back to 50.
- Any better choice than 50?
- We should also think about stop loss conditions.
- Suppose we enter a buy order when RSI is low. If that instrument's price keep droping, its calculated RSI will maintain at a low level and won't reverse back to 50. Thus, according to current trading logics, we will keep losing money and never cut loss.
- Will it produce a more accurate signal when combining RSI with other indicators?
Like and follow me
If you find my articles inspiring, like this post and follow me here to receive my latest updates.
Enter my promote code "AjpQDMOSmzG2" for any purchase on ALGOGENE, you will automatically get 5% discount.
