Investment universe consists of 24 country ETFs (mostly iShares MSCI, along with iShares China Large-Cap ETF [FXI], and the U.S. is represented by SPDR S&P 500 ETF [SPY]).
ASSET CLASS: ETFs | REGION: Global | FREQUENCY:
Daily | MARKET: equities | KEYWORD: Trading Strategy
I. STRATEGY IN A NUTSHELL
Trade 24 country ETFs (e.g., SPY, FXI, MSCI funds). Identify double bottoms in long-term uptrends using 30-day rolling windows. Entry via limit orders at bottoms, with 3% trailing stop-loss and 6% profit target (R:R = 1:2). Max 4 positions, 2:1 leverage. Monthly scanning, daily trade execution.
II. ECONOMIC RATIONALE
Technical analysis captures market psychology alongside fundamentals. While imperfect, price patterns like double bottoms reflect investor behavior and information assimilation. TA provides practical insights beyond fundamental analysis alone.
III. SOURCE PAPER
Double Bottom Country Trading Strategy [Click to Open PDF]
”Cyril Dujava”, ”Filip Kalús”, ”Radovan Vojtko”, ”Quantpedia”
<Abstract>
The article from Quantpedia discusses the methodology of technical analysis, mainly focusing on the double bottom and double top trading strategies. Technical analysis is often viewed skeptically, with some likening it to ”astrology for men.” However, the article emphasizes that while both fundamentals and psychology drive the market, it would be unwise to dismiss technical analysis entirely. The history of technical analysis is traced back to figures like Sokyu Honma in the 1600s in Japan and the Dow Theory from the modern world. The article then delves into various technical analysis terms, such as supports, resistances, and moving averages. The main focus is on the double top and double bottom patterns used as reversal indicators in charts. These patterns can indicate strong resistance or support in price charts. The methodology for the study was based on feedback and questions received over the years, indicating that technical analysis remains a popular subject.
IV. BACKTEST PERFORMANCE
| Annualised Return | 7.79% |
| Volatility | 16.57% |
| Beta | 0.103 |
| Sharpe Ratio | 0.47 |
| Sortino Ratio | 0.181 |
| Maximum Drawdown | -30.38% |
| Win Rate | 53% |
V. FULL PYTHON CODE
from AlgorithmImports import *
from pandas.core.frame import dataframe
from datetime import datetime
# endregion
class DoubleBottomCountryTradingStrategy(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.country_etfs:List[str] = [
"EWA", # iShares MSCI Australia Index ETF
"EWO", # iShares MSCI Austria Investable Mkt Index ETF
"EWK", # iShares MSCI Belgium Investable Market Index ETF
"EWZ", # iShares MSCI Brazil Index ETF
"EWC", # iShares MSCI Canada Index ETF
"FXI", # iShares China Large-Cap ETF
"EWQ", # iShares MSCI France Index ETF
"EWG", # iShares MSCI Germany ETF
"EWH", # iShares MSCI Hong Kong Index ETF
"EWI", # iShares MSCI Italy Index ETF
"EWJ", # iShares MSCI Japan Index ETF
"EWM", # iShares MSCI Malaysia Index ETF
"EWW", # iShares MSCI Mexico Inv. Mt. Idx
"EWN", # iShares MSCI Netherlands Index ETF
"EWS", # iShares MSCI Singapore Index ETF
"EZA", # iShares MSCI South Africe Index ETF
"EWY", # iShares MSCI South Korea ETF
"EWP", # iShares MSCI Spain Index ETF
"EWD", # iShares MSCI Sweden Index ETF
"EWL", # iShares MSCI Switzerland Index ETF
"EWT", # iShares MSCI Taiwan Index ETF
"THD", # iShares MSCI Thailand Index ETF
"EWU", # iShares MSCI United Kingdom Index ETF
"SPY", # SPDR S&P 500 ETF
]
self.rolling_period:int = 30
# two local highs/lows max and min distance
self.min_distance:int = self.rolling_period
self.max_distance:int = 1000
# max vertical distance between two local highs/lows
self.max_percentage:float = .02
self.stop_loss_pct:float = .03
self.take_profit_pct:float = .06
self.trade_long = True
self.trade_short = False
self.max_positions_opened:int = 4
self.SetWarmup(self.max_distance, Resolution.Daily)
self.price_data:Dict[str, List[Tuple(datetime.date, float)]] = {}
# add Forex data
for etf in self.country_etfs:
data:Equity = self.AddEquity(etf, Resolution.Daily)
data.SetLeverage(10)
self.price_data[etf] = []
self.recent_month:int = -1
def OnData(self, data: Slice) -> None:
# trail long SL for each opened positions
active_open_buy_orders:List = list(self.Transactions.GetOrderTickets(lambda ticket: 'LIMIT_BUY' in ticket.Tag and ticket.Status == OrderStatus.Filled))
for ticket in active_open_buy_orders:
if self.Portfolio[ticket.Symbol].IsLong:
active_sl_orders:List = list(self.Transactions.GetOrderTickets(lambda sl_ticket: 'SL' in sl_ticket.Tag \
and sl_ticket.Symbol == ticket.Symbol \
and sl_ticket.Status not in [OrderStatus.Filled, OrderStatus.Invalid, OrderStatus.Canceled]))
if len(active_sl_orders) != 0:
active_sl_order = active_sl_orders[0]
# price moved into profit
if self.Securities[ticket.Symbol].Price > ticket.Get(OrderField.LimitPrice):
new_sl_price:float = round(self.Securities[ticket.Symbol].Price - (self.Securities[ticket.Symbol].Price * self.stop_loss_pct), 2)
if new_sl_price > active_sl_order.Get(OrderField.StopPrice):
active_sl_order.UpdateStopPrice(new_sl_price)
# trail short SL for each opened positions
active_open_buy_orders:List = list(self.Transactions.GetOrderTickets(lambda ticket: 'LIMIT_SELL' in ticket.Tag and ticket.Status == OrderStatus.Filled))
for ticket in active_open_buy_orders:
if self.Portfolio[ticket.Symbol].IsShort:
active_sl_orders:List = list(self.Transactions.GetOrderTickets(lambda sl_ticket: 'SL' in sl_ticket.Tag \
and sl_ticket.Symbol == ticket.Symbol \
and sl_ticket.Status not in [OrderStatus.Filled, OrderStatus.Invalid, OrderStatus.Canceled]))
if len(active_sl_orders) != 0:
active_sl_order = active_sl_orders[0]
# price moved into profit
if self.Securities[ticket.Symbol].Price < ticket.Get(OrderField.LimitPrice):
new_sl_price:float = round(self.Securities[ticket.Symbol].Price + (self.Securities[ticket.Symbol].Price * self.stop_loss_pct), 2)
if new_sl_price < active_sl_order.Get(OrderField.StopPrice):
active_sl_order.UpdateStopPrice(new_sl_price)
# look for new double top and double bottom once a month
rebalance_flag = False
if self.recent_month != self.Time.month:
rebalance_flag = True
self.recent_month = self.Time.month
# store daily prices
for etf in self.country_etfs:
if etf in data and data[etf]:
self.price_data[etf].append( (self.Time, data[etf].Value) )
if not rebalance_flag:
continue
if len(self.price_data[etf]) >= self.max_distance:
if not self.Portfolio[etf].Invested:
etf_hist:dataframe = pd.dataframe(self.price_data[etf], columns=['Date', etf])
etf_hist.set_index('Date', inplace=True)
current_datetime:datetime = etf_hist.index[-1]
etf_hist = self.find_support_and_resistance(etf, etf_hist, self.rolling_period, self.min_distance)
etf_hist = self.find_double_tops_and_bottoms(etf, etf_hist, self.min_distance, self.max_distance, self.max_percentage)
if self.trade_long:
# find not yet crossed bottom
last_bottom_datetime:dataframe = etf_hist[~etf_hist['Double_Bottom'].isnull()]
if not last_bottom_datetime.empty:
last_bottom_datetime:datetime = last_bottom_datetime.index[-1]
if all(x[1][etf] >= etf_hist.loc[last_bottom_datetime][etf] for x in etf_hist.loc[last_bottom_datetime:current_datetime].iterrows()):
limit_price:float = round(etf_hist.loc[last_bottom_datetime][etf], 2)
active_limit_orders:List = list(self.Transactions.GetOrderTickets(lambda order_ticket: order_ticket.Status not in [OrderStatus.Filled, OrderStatus.Invalid, OrderStatus.Canceled] and order_ticket.Symbol == etf and 'LIMIT_BUY' in order_ticket.Tag))
# submit orders
if len(active_limit_orders) != 0:
active_order = active_limit_orders[0]
# there is active order for different price
if active_order.Get(OrderField.LimitPrice) != limit_price:
new_quantity:int = self.Portfolio.TotalPortfolioValue // limit_price
active_order.UpdateLimitPrice(limit_price)
active_order.UpdateQuantity(new_quantity)
else:
# submit new limit order
if len([x for x in self.Portfolio if x.Value.Invested]) < self.max_positions_opened:
quantity:int = self.Portfolio.TotalPortfolioValue // limit_price
if abs(quantity) > 1:
self.LimitOrder(etf, quantity, limit_price, tag='LIMIT_BUY')
if self.trade_short:
# find not yet crossed top
last_top_datetime = etf_hist[~etf_hist['Double_Top'].isnull()]
if not last_top_datetime.empty:
last_top_datetime = last_top_datetime.index[-1]
if all(x[1][etf] <= etf_hist.loc[last_top_datetime][etf] for x in etf_hist.loc[last_top_datetime:current_datetime].iterrows()):
limit_price:float = round(etf_hist.loc[last_top_datetime][etf], 2)
active_limit_orders:List = list(self.Transactions.GetOrderTickets(lambda order_ticket: order_ticket.Status not in [OrderStatus.Filled, OrderStatus.Invalid, OrderStatus.Canceled] and order_ticket.Symbol == etf and 'LIMIT_SELL' in order_ticket.Tag))
# submit orders
if len(active_limit_orders) != 0:
active_order = active_limit_orders[0]
# there is active order for different price
if active_order.Get(OrderField.LimitPrice) != limit_price:
new_quantity:int = self.Portfolio.TotalPortfolioValue // limit_price
active_order.UpdateLimitPrice(limit_price)
active_order.UpdateQuantity(-new_quantity)
else:
# submit new limit order
if len([x for x in self.Portfolio if x.Value.Invested]) < self.max_positions_opened:
quantity:int = self.Portfolio.TotalPortfolioValue // limit_price
if abs(quantity) > 1:
self.LimitOrder(etf, -quantity, limit_price, tag='LIMIT_SELL')
def cancel_all_limit_orders(self) -> None:
# cancel all submitted limit orders for the symbol
active_limit_orders:List = list(self.Transactions.GetOrderTickets(lambda ticket: ('LIMIT_SELL' in ticket.Tag or 'LIMIT_BUY' in ticket.Tag) and ticket.Status not in [OrderStatus.Filled, OrderStatus.Invalid, OrderStatus.Canceled]))
for ticket in active_limit_orders:
response = ticket.Cancel()
def cancel_all_orders(self, symbol:Symbol) -> None:
# cancel all submitted orders for the symbol
active_limit_orders:List = list(self.Transactions.GetOrderTickets(lambda ticket: ticket.Symbol == symbol))
for ticket in active_limit_orders:
response = ticket.Cancel()
def OnOrderEvent(self, orderEvent: OrderEvent) -> None:
order = self.Transactions.GetOrderById(orderEvent.OrderId)
if orderEvent.Status == OrderStatus.Filled:
order_ticket = self.Transactions.GetOrderTicket(orderEvent.OrderId)
# if TP or SL was hit
if '_TP' in order_ticket.Tag or '_SL' in order_ticket.Tag:
# cancel all submitted orders for the symbol
self.cancel_all_orders(order_ticket.Symbol)
# set TP and SL if limit was hit
if orderEvent.Quantity > 0:
if 'LIMIT_BUY' in order_ticket.Tag:
if len([x for x in self.Portfolio if x.Value.Invested]) == self.max_positions_opened:
self.cancel_all_limit_orders()
tp_price:float = round(orderEvent.FillPrice + (orderEvent.FillPrice * self.take_profit_pct), 2)
stop_price:float = round(orderEvent.FillPrice - (orderEvent.FillPrice * self.stop_loss_pct), 2)
self.LimitOrder(orderEvent.Symbol, -orderEvent.Quantity, tp_price, tag='BUY_TP')
self.StopMarketOrder(orderEvent.Symbol, -orderEvent.Quantity, stop_price, tag='BUY_SL')
if orderEvent.Quantity < 0:
if 'LIMIT_SELL' in order_ticket.Tag:
if len([x for x in self.Portfolio if x.Value.Invested]) == self.max_positions_opened:
self.cancel_all_limit_orders()
tp_price:float = round(orderEvent.FillPrice - (orderEvent.FillPrice * self.take_profit_pct), 2)
stop_price:float = round(orderEvent.FillPrice + (orderEvent.FillPrice * self.stop_loss_pct), 2)
self.LimitOrder(orderEvent.Symbol, -orderEvent.Quantity, tp_price, tag='SELL_TP')
self.StopMarketOrder(orderEvent.Symbol, -orderEvent.Quantity, stop_price, tag='SELL_SL')
def find_support_and_resistance(self, symbol:str, stock_data:dataframe, rolling_period:int, lookback_period:int) -> dataframe:
for level_str, obj_function in [('Support', 'min'), ('Resistance', 'max')]:
level_prices = stock_data[symbol]
# TODO eval min and max
if obj_function == 'min':
pivot_level = level_prices.rolling(window=rolling_period*2+1, center=True).min().shift(-rolling_period)
else:
pivot_level = level_prices.rolling(window=rolling_period*2+1, center=True).max().shift(-rolling_period)
stock_data[level_str] = None
level_points = pivot_level.where(pivot_level == level_prices, None).dropna().index
valid_indices = []
for i, level_point in enumerate(level_points):
if i == 0:
valid_indices.append(level_point)
else:
timedelta = level_point - level_points[i - 1]
if timedelta.days > lookback_period:
valid_indices.append(level_point)
stock_data.loc[valid_indices, level_str] = pivot_level.loc[valid_indices]
return stock_data
def find_double_tops_and_bottoms(self, symbol:str, stock_data:dataframe, min_distance:int, max_distance:int, max_percentage:float) -> dataframe:
stock_data['Double_Top'] = None
stock_data['Double_Bottom'] = None
support_points = stock_data.dropna(subset=['Support']).index
resistance_points = stock_data.dropna(subset=['Resistance']).index
for i in range(len(resistance_points)):
for j in range(i + 1, len(resistance_points)):
timedelta = resistance_points[j] - resistance_points[i]
if timedelta.days < min_distance or timedelta.days > max_distance:
continue
price_pct_difference = abs(stock_data.loc[resistance_points[j], 'Resistance'] / stock_data.loc[resistance_points[i], 'Resistance'] - 1)
if price_pct_difference <= max_percentage:
lower_top_index = resistance_points[j] if stock_data.loc[resistance_points[j], 'Resistance'] < stock_data.loc[resistance_points[i], 'Resistance'] else resistance_points[i]
if all(stock_data.loc[resistance_points[x], symbol] <= stock_data.loc[lower_top_index, 'Resistance'] for x in range(i+1,j)):
stock_data.loc[lower_top_index, 'Double_Top'] = stock_data.loc[lower_top_index, 'Resistance']
for i in range(len(support_points)):
for j in range(i + 1, len(support_points)):
timedelta = support_points[j] - support_points[i]
if timedelta.days < min_distance or timedelta.days > max_distance:
continue
price_pct_difference = abs(stock_data.loc[support_points[j], 'Support'] / stock_data.loc[support_points[i], 'Support'] - 1)
if price_pct_difference <= max_percentage:
higher_bottom_index = support_points[j] if stock_data.loc[support_points[j], 'Support'] > stock_data.loc[support_points[i], 'Support'] else support_points[i]
if all(stock_data.loc[support_points[x], symbol] >= stock_data.loc[higher_bottom_index, 'Support'] for x in range(i+1,j)):
stock_data.loc[higher_bottom_index, 'Double_Bottom'] = stock_data.loc[higher_bottom_index, 'Support']
return stock_data