Validate and Experiment on Legendary Turtle Trading Strategy
The Turtle Traders experiment was conducted in the early 1980s by Richard Dennis and William Eckhardt to see whether anyone could be taught how to make money in trading. Therefore, the trading strategy that has been taught to these apprentices is named after this experiment as “ Turtle Strategy “.
In this post, we’re going to quickly go through the turtle strategy itself, and I’m going to experiment with a few things against the backtesting platform that I’m checking out recently: JoinQuant.
This post will be split into the below sections:
- Quick intro
- Objectives
- Backtesting
- Summary
- Reference
- Code Reference
Here is the Chinese version of this post in case anyone is interested: 经典海龟策略-股票多标的-短线趋势长线追击双系统.
Quick intro
I’m going to piggyback the existing posts so I don’t have to repeat the lengthy history of the turtle experiment here. You can quickly read through the below posts to get an idea of what turtle strategy is about:
In short, the turtle strategy is a momentum strategy that uses the Donchian channel as the indicator to buy when we spotted the bullish trend and to sell when the trend goes bearish. Turtle strategy also builds two similar systems to capture both strong and weak movements, along with an internal mechanism to manage the risk of loss of capital to 2% in one trading day.
Objectives
Implementing the turtle strategy is not a new thing to the quant industry. Therefore, I would like to instead experiment with a few additional things that might benefit the process of the quant research:
- Most of the quant strategies that you can see in the quant-trading backtest platforms scattered their function definitions and variables all over the place. I’m gonna adopt the
object-oriented programmingconcept in this implementation to increase the reusability and readability of the code. - By instantiating the instance of
class TurtleSystemManager(), we persist the data in this instance across the entire backtesting time span. - To decrease the complexity of the positions, capital, and available cash of these two systems in one algorithm trading script, we use
context.subportfoliosin JoinQuant to simplify the logic and codes.
Backtesting
The setup in my version of implementation are:
- Run every 15 minutes to calculate and inspect the trading signals
- The stock pool updates each month with the latest fundamental data
- The trading target in the original turtle strategy is the stock futures, but we’re looking at the stocks in China stock market. So we need to make several adjustments accordingly:
- risk_ratio: The leverage_ratio in the original turtle strategy is 10 while trading stock futures. Here we make it 0.1 so that our greatest loss per day is limited to 0.1%, and also we’re able to buy more stocks.
- capacity_per_system: The stock pool that we update each month contains more than 20 stocks. As turtle strategy is about buying positions at several prices along with the price soar (vise versa in the sell/short side), we place a limit of a maximum of 8 stocks per system to make sure we place our capital in the stocks we preferred other than spreading the capital in dozens of stocks.
- We use
40-day moving average > 200-day moving averageas the secondary trading signal to confirm the bullish trend. - Apply Donchian middle line to replace the Donchian low watermark as the new exit signal.
- As the China stock market policy prevents us from selling the positions that we buy on the same day, we apply another rule in our strategy that we need to hold the stock at least till the next available bar (which is the next day). See detail in 海龟策略应用在中国A股（股票）里的缺陷讨论（按分钟回测）.
And here are the results of the backtesting of the turtle strategy:
Backtest 1
Setup: Run the strategy once per day, and no specific sorting rules in the stock pool.
Result:
Comment: Ummm… The performance is fairly poor but expected. We trade only once per day on the signal without monitoring the price movement for the rest of the day. It is pretty much like you buy a stock when you see the right price on the TV, and then you go play with your cats and dogs until the next day. If any trader can make money like this, anyone on earth can be a trader and no one loses money in the market.
Backtest 2
Setup:
Run every 15 minutes, no specific sorting rules in the stock pool, and use daily close price to build the 40-day and 200-day moving average.
Result:
Comment:
The risk control is stronger when you monitor the stock movements every 15 minutes, and you’ll be able to control the daily loss within the range of2N as instructed in the turtle experiment. As for the upward trend, the 15-minute interval also helps the program more accurately capture the entry signal to gain profit. However, the log messages also reveal that there are several times that you won't be able to close your positions on the day. This represents that your position is exposed outside the risk control management system, causing a tremendous loss when the stock price dropping.
Backtest 3
Setup:
Run every 15 minutes, no specific sorting rules in the stock pool, and use 15-minute close price to build the 40-day and 200-day moving average.
Result:
Comment: Urgh….. I thought if I increase the granularity of the secondary indicator (40MA > 200MA) from 1 day to 15 minutes, the more accurate entry signals and more profit would be generated. By looking at the diagrams below, more signals do generate as the available cash is lower in the 15-minute scenario.
However, from the log messages you’ll notice that due to the limitation of the China stock market mentioned above, that there are risks that you buy the stock in the morning but things turned sour in the afternoon. Then all you can do is to look at the price go south and sweating all over your face. So we can kind of concluding that using daily close price would be a means of smoothing the data to avoid the zig-zag on price movement.
Backtest 4
Setup:
Run every 15 minutes, sort and rank the stock pool monthly, and use daily close price to build the 40-day and 200-day moving average.
Result:
Comment: According to what has been described in the book of turtle strategy and the characteristics of the volatility of future itself, I sorted and ranked the stocks in Shanghai Shenzhen 300 index in order to identify the stocks that have higher profitability than the others. The rules and the factors that I used are as follow:
- goods_sale_and_service_to_revenue: Of course the sales revenue should represent a higher percentage in the total revenue income, indicating the major business is doing great.
- peg_ratio: A ratio to replace the traditional pe_ratio as Peter Lynch suggested. We’re looking at under-estimated stocks(omitting the stocks that peg_ratio is negative).
- debt_to_equity_ratio: The less debt the better.
- FCF (Free Cash Flow): The more free cash flow the better.
- turnover_ratio: Turtle strategy suggests we target the markets that have a higher turnover ratio as in essence, turtle strategy is still a momentum strategy that looks for stocks that have high liquidity.
- pb_ratio: Also, we need to find stocks that are underestimated.
I was hoping that my petty trick would work and make the portfolio return better off. Sadly, it didn’t.
Summary
Even though the backtesting results above are not promising nor profitable for us to proceed to live trading, the objectives of this experiment have been achieved. We can use an object-oriented programming way of coding style inside these backtesting platforms. By doing so, the code can be reused when we implement our own automatic trading script later, reducing the time to rewrite everything the second time.
Other than that, we have also identified some limitations of running turtle strategy in the China stock market:
- The China stock market has a policy that restricts day-trading (selling the stocks that you purchased that day), leaving the risk of our positions uncontrolled.
- One lot in China's stock market represents 100 shares. Any number of shares that are under 100 are not able to be purchased. This would raise the bar of purchasing stocks whose prices are higher as we need to relocate our capital to several stocks in the strategy.
- Monitoring and trading several stocks in one turtle strategy is not preferable since it increases the complexity of managing positions during both bullish and bearish trends.
But there are other thoughts that we still can extend and develop upon the turtle strategy we built:
- We can see that the Donchian channel in turtle strategy is a lag indicator, meaning the momentum might have already passed when our indicators tell us to buy or sell. Therefore we might be able to switch to another indicator such as MACD or RSI so that we can spot the buy/sell opportunities in a quicker fashion.
- We can also reverse the signals to buy when we see sell signals and to sell when we see buy signals. This means that we’re going to use the turtle strategy as a
mean reversion strategyinstead of the originalmomentum strategy.
To summarize, this strategy is not going toward the live trading stage in the short run. The code can only be used as a template or reference for anyone here to implement their version of the turtle system by overwriting the detail in each class function.
Enjoy! Cheers!
References
- 【量化课堂】海龟策略
- 海龟策略精讲
- The Original Turtle Trading Rules
- Turtle trading rules trend-following investing based on 20 amp 55-day highs
- 唐奇安通道
- 增强版唐奇安通道策略
Code Reference
# Inspired from: https://www.joinquant.com/post/1401
# and https://www.joinquant.com/view/community/detail/284a9688a58e0112b3bad8c1283548bc
# Title：Turtle Strategy that monitors multiple stocks
# Author：Michael Hsia
import pandas as pd
import talib
from prettytable import *
from datetime import timedelta
enable_profile()
class ChinaMarketHelper():
def __init__(self):
pass
@classmethod
def normalize_position(self, position):
return int(position / 100) * 100
class TurtleSystemManager():
def __init__(self, short_in_date=20, short_out_date=10, long_in_date=55, long_out_date=20):
self.system = [
TurtleSystem(short_in_date, short_out_date, True),
TurtleSystem(long_in_date, long_out_date, False)
]
# Capacity in portfolio
self.capacity_per_system = 8
def update_turtle_params(self, context):
for pindex in range(len(context.subportfolios)):
total_value = context.subportfolios[pindex].total_value
for symbol in context.subportfolios[pindex].long_positions.keys():
self.system[pindex].update_turtle_params_by_symbol(symbol, total_value)
def start_running(self, context, symbol, current_price):
for pindex in range(len(context.subportfolios)):
total_value = context.subportfolios[pindex].total_value
donchian_high_price = max(attribute_history(symbol, self.system[pindex].in_date, '1d', ('high', 'close'))['close'])
donchian_low_price = min(attribute_history(symbol, self.system[pindex].out_date, '1d', ('low', 'close'))['close'])
donchian_mid_price = (donchian_high_price + donchian_low_price) / 2
if symbol not in context.subportfolios[pindex].long_positions.keys():
# Limit the number of assets in our portfolio to make sure we're able to invest enough money to complete one turtle strategy cycle
if len(context.subportfolios[pindex].long_positions) >= self.capacity_per_system:
continue
ma_40 = mean(attribute_history(symbol, 40, '1d', ('close'))['close'])
ma_200 = mean(attribute_history(symbol, 200, '1d', ('close'))['close'])
self.system[pindex].update_turtle_params_by_symbol(symbol, total_value)
if (current_price > donchian_high_price) and (ma_40 > ma_200):
if pindex == 1:
# Reset previous_state_winning status
self.system[0].previous_state_winning.discard(symbol)
self.system[pindex].market_in(
symbol,
pindex
)
else:
avg_cost = context.subportfolios[pindex].long_positions[symbol].acc_avg_cost
previous_purchased_price = self.system[pindex].get_previous_purchased_price_by_symbol(symbol)
system_N = self.system[pindex].get_N_from_turtle_params_by_symbol(symbol)
# The reason to add this block has been described in this article
# https://www.joinquant.com/view/community/detail/30694
# Clean up the remaining position in the next possible bar
current_position = context.subportfolios[pindex].long_positions[symbol].total_amount + context.subportfolios[pindex].long_positions[symbol].locked_amount
if (current_position != 0) and (self.system[pindex].system_positions[symbol]['unit'] == 0):
order_target(symbol, 0, style=MarketOrderStyle(), pindex=pindex)
# Long one unit for every 0.5N increased
if current_price >= (previous_purchased_price + 0.5 * system_N):
if pindex == 1:
# Reset previous_state_winning status if system 2 market in or market add
self.system[0].previous_state_winning.discard(symbol)
self.system[pindex].market_add(
symbol,
pindex
)
if current_price < donchian_mid_price:
ret = self.system[pindex].market_out(
symbol,
pindex
)
if ret == True:
if current_price >= avg_cost:
self.system[pindex].previous_state_winning.add(symbol)
else:
self.system[pindex].previous_state_winning.discard(symbol)
# Don't need to run stop loss as this position has already been removed
continue
# Stop loss if current price lower than 2N
if (current_price - previous_purchased_price) < (-2 * system_N):
ret = self.system[pindex].market_stop_loss(
symbol,
pindex
)
if ret == True:
self.system[pindex].previous_state_winning.discard(symbol)
class TurtleSystem():
TURTLE_N = 'N'
TURTLE_DOLLAR_VOLATILITY = 'DOLLAR_VOLATILITY'
TURTLE_UNIT = 'UNIT'
def __init__(self, in_date, out_date, ignore_state=False):
self.in_date = in_date
self.out_date = out_date
self.ignore_state = ignore_state
self.system_positions = {}
self.turtle_params = {}
self.previous_state_winning = set()
self.ignore_state = True
self.dollars_per_share = 1
self.number_days = 20
self.unit_limit = 4
self.risk_ratio = 0.1
def market_in(self, symbol, pindex):
# System 1 breakout entry signals would be ignored if the last breakout would have resulted in a winning trade
# All breakouts for System 2 would be taken whether the previous breakout had been a winner or not.
if (not self.ignore_state) and (symbol in self.previous_state_winning):
# 假如是系统1，就要看之前的trade是win的话，本次就不进入市场了
self.remove_turtle_params_by_symbol(symbol)
return
log.info("[System {}] - [{}] - 入仓".format(pindex, symbol))
self.add_position_by_symbol(
symbol,
pindex
)
def market_add(self, symbol, pindex):
log.info("[System {}] - [{}] - 加仓".format(pindex, symbol))
self.add_position_by_symbol(symbol, pindex)
def market_out(self, symbol, pindex):
log.info("[System {}] - [{}] - 减仓".format(pindex, symbol))
return self.reduce_position_by_symbol(
symbol,
pindex
)
def market_stop_loss(self, symbol, pindex):
log.debug('[System {}] - [{}] 开始止损'.format(pindex, symbol))
return self.reduce_position_by_symbol(
symbol,
pindex
)
def add_position_by_symbol(self, symbol, pindex):
if self.system_positions.get(symbol, None) == None:
self.system_positions[symbol] = {}
self.system_positions[symbol]['unit'] = 0
self.system_positions[symbol]['previous_purchased_price'] = 0
if self.system_positions[symbol]['unit'] >= self.unit_limit:
# Reaching unit limit
return
position = self.get_unit_from_turtle_params_by_symbol(symbol)
position = ChinaMarketHelper.normalize_position(position * self.risk_ratio)
if position < 100:
return
res = order(symbol, position, style=MarketOrderStyle(), pindex=pindex)
if res is not None:
if res.status not in [OrderStatus.canceled, OrderStatus.rejected]:
self.system_positions[symbol]['unit'] += 1
self.system_positions[symbol]['previous_purchased_price'] = res.price
else:
log.warning('[System {}] - [{}] order failed: {}.'.format(pindex, symbol, res))
def reduce_position_by_symbol(self, symbol, pindex):
if self.system_positions.get(symbol, None) == None:
log.warning('Reduce position failed. It does not exist')
return None
res = order_target(symbol, 0, style=MarketOrderStyle(), pindex=pindex)
if res is not None:
if res.status not in [OrderStatus.canceled, OrderStatus.rejected]:
self.system_positions[symbol]['unit'] = 0
# https://www.joinquant.com/view/community/detail/30694
# self.remove_position_by_symbol_if_empty(symbol)
return True
else:
log.warning('[System {}] - [{}] order failed: {}.'.format(pindex, symbol, res))
return False
def remove_position_by_symbol_if_empty(self, symbol):
if self.system_positions.get(symbol, None) == None:
return True
if self.system_positions[symbol]['unit'] == 0:
self.system_positions.pop(symbol)
self.remove_turtle_params_by_symbol(symbol)
return True
log.warning('Unit of [{}] is not empty'.format(symbol))
return False
def get_previous_purchased_price_by_symbol(self, symbol):
if self.system_positions.get(symbol, None) is None:
raise Exception(' [{}] - System position param does not exist'.format(symbol))
if self.system_positions[symbol].get('previous_purchased_price', None) is None:
raise Exception('[{}] - N does not exist'.format(symbol))
return self.system_positions[symbol]['previous_purchased_price']
def update_turtle_params_by_symbol(self, symbol, account_value):
if not self.is_turtle_params_existed_by_symbol(symbol):
self.turtle_params[symbol] = {}
df = attribute_history(
count=self.number_days + 1,
unit='1d',
fields=['low', 'close', 'high'],
security=symbol,
df=True
)
df['pdc'] = df['close'].shift(1)
df['h_l'] = (df['high'] - df['low']).abs()
df['h_pdc'] = (df['high'] - df['pdc']).abs()
df['pdc_l'] = (df['pdc'] - df['low']).abs()
N = df[['h_l', 'h_pdc', 'pdc_l']].max(axis=1)[1:].mean()
high = df['high']
low = df['low']
pdc = df['close']
del df
self.turtle_params[symbol][self.TURTLE_N] = talib.ATR(high, low, pdc, timeperiod=self.number_days)[-1]
self.turtle_params[symbol][self.TURTLE_DOLLAR_VOLATILITY] = self.dollars_per_share * self.turtle_params[symbol][self.TURTLE_N]
self.turtle_params[symbol][self.TURTLE_UNIT] = (account_value * 0.01) / self.turtle_params[symbol][self.TURTLE_DOLLAR_VOLATILITY]
return True
def get_N_from_turtle_params_by_symbol(self, symbol):
if not self.is_turtle_params_existed_by_symbol(symbol):
raise Exception('N of [{}] does not exist'.format(symbol))
if self.turtle_params[symbol].get(self.TURTLE_N, None) is None:
raise Exception('N of [{}] does not exist'.format(symbol))
return self.turtle_params[symbol][self.TURTLE_N]
def get_unit_from_turtle_params_by_symbol(self, symbol):
if not self.is_turtle_params_existed_by_symbol(symbol):
raise Exception('N of [{}] does not exist'.format(symbol))
return self.turtle_params[symbol][self.TURTLE_UNIT]
def is_turtle_params_existed_by_symbol(self, symbol):
if self.turtle_params.get(symbol, None) is None:
return False
else:
return True
def remove_turtle_params_by_symbol(self, symbol):
if self.is_turtle_params_existed_by_symbol(symbol):
self.turtle_params.pop(symbol)
'''
=================================
Initialization
=================================
'''
def initialize(context):
g.security = None
set_benchmark('000300.XSHG')
set_option('use_real_price',True)
set_option("avoid_future_data", True)
log.set_level('order','error')
# System 1 cash
ratio_system1 = 0.8
# Set up two separate systems:
# subportfolios[0] is system 1 in the turtle strategy
# and subportfolios[1] is the system 2
set_subportfolios(
[SubPortfolioConfig(cash=context.portfolio.starting_cash * ratio_system1, type='stock'),
SubPortfolioConfig(cash=context.portfolio.starting_cash * (1 - ratio_system1), type='stock')]
)
SHORT_IN_DATE = 20
SHORT_OUT_DATE = 10
LONG_IN_DATE = 55
LONG_OUT_DATE = 20
g.turtle = TurtleSystemManager(
SHORT_IN_DATE,
SHORT_OUT_DATE,
LONG_IN_DATE,
LONG_OUT_DATE,
)
'''
=================================
Everyday before market open
=================================
'''
def before_trading_start(context):
set_slip_fee(context)
set_tradable_stocks(context)
g.turtle.update_turtle_params(context)
def set_slip_fee(context):
set_slippage(FixedSlippage(0))
set_order_cost(OrderCost(open_tax=0, close_tax=0.001, open_commission=0.0003, close_commission=0.0003, close_today_commission=0, min_commission=5), type='stock')
def set_tradable_stocks(context):
# Run every month
if context.current_dt.day == 1 or g.security is None:
# Test one scenario
test = False
if test:
# 上证50
g.security = get_index_stocks('000016.XSHG')
else:
# Get yesterday's datetime string
date = (context.current_dt - timedelta(days=1)).strftime('%Y-%m-%d')
sh300 = get_index_stocks(normalize_code('000300.sh'))
sh300_df = get_fundamentals(
query(
valuation.code,
valuation.pe_ratio,
valuation.turnover_ratio,
balance.total_owner_equities,
balance.total_sheet_owner_equities,
cash_flow.cash_equivalent_increase,
cash_flow.cash_equivalents_at_beginning,
cash_flow.cash_and_equivalents_at_end,
indicator.goods_sale_and_service_to_revenue,
indicator.inc_net_profit_year_on_year,
valuation.pb_ratio,
).filter(
valuation.code.in_(sh300),
valuation.pe_ratio > 0,
indicator.inc_net_profit_year_on_year > 0
),
date=date
)
sh300_df = sh300_df.dropna()
sh300_df['peg_ratio'] = sh300_df.pe_ratio / sh300_df.inc_net_profit_year_on_year
sh300_df['debt_to_equity_ratio'] = ((sh300_df.total_sheet_owner_equities - sh300_df.total_owner_equities)/sh300_df.total_owner_equities)
sh300_df['FCF'] = (sh300_df.cash_and_equivalents_at_end - sh300_df.cash_equivalents_at_beginning)
cols = ['goods_sale_and_service_to_revenue', 'peg_ratio', 'debt_to_equity_ratio', 'FCF', 'turnover_ratio', 'pb_ratio']
sh300_df.index = sh300_df.code
sh300_df.index.name = 'Symbol'
sh300_df = sh300_df[cols]
sh300_df['goods_sale_and_service_to_revenue'] = sh300_df['goods_sale_and_service_to_revenue'].rank(ascending=False)
sh300_df['peg_ratio'] = sh300_df['peg_ratio'].rank(ascending=True)
sh300_df['debt_to_equity_ratio'] = sh300_df['debt_to_equity_ratio'].rank(ascending=True)
sh300_df['FCF'] = sh300_df['FCF'].rank(ascending=False)
sh300_df['turnover_ratio'] = sh300_df['turnover_ratio'].rank(ascending=False)
sh300_df['pb_ratio'] = sh300_df['pb_ratio'].rank(ascending=True)
zscore = sh300_df.sum(axis=1).sort_values(ascending=True)
g.security = zscore.index.tolist()
for pindex in range(len(context.subportfolios)):
g.security += context.subportfolios[pindex].long_positions.keys()
# Remove duplicated positions from the list
g.security = list(set(g.security))
'''
=================================
Perform every 15 minutes
=================================
'''
def handle_data(context, data):
# 15 minutes timer
timer = 15
if context.current_dt.minute % timer != 0:
return
elif (context.current_dt.hour == 9) and (context.current_dt.minute == 30):
return
for sec in g.security:
current_data = get_current_data()
if current_data[sec].paused or current_data[sec].is_st:
continue
current_price = data[sec].price
g.turtle.start_running(context, sec, current_price)
'''
=================================
After everyday market close
=================================
'''
def after_trading_end(context):
for pindex in range(len(context.subportfolios)):
if pindex == 0:
record(cash1=context.subportfolios[pindex].available_cash)
record(value1=context.subportfolios[pindex].total_value)
else:
record(cash2=context.subportfolios[pindex].available_cash)
record(value2=context.subportfolios[pindex].total_value)
return
'''
=================================
After the backtest is competed
=================================
'''
def on_strategy_end(context):
x = PrettyTable(['System', 'TotalValue', 'Avail Cash', 'Number of positions'])
for pindex in range(len(context.subportfolios)):
x.add_row([
f'System {pindex}',
context.subportfolios[pindex].total_value,
context.subportfolios[pindex].available_cash,
len(context.subportfolios[pindex].long_positions)
])Disclaimer: Nothing herein is financial advice or even a recommendation to trade real money. Many platforms exist for simulated trading (paper trading) which can be used for building and developing the strategies discussed. Please use common sense and consult a professional before trading or investing your hard-earned money.