Experiment of Turtle trading strategy

Last post in 2020. See you, 2020. See you, in 2021.

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 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 turtle experiment here. You can quickly read through the below posts to get an idea of what turtle strategy is about:

• Analyzing alpha - turtle trading
• Medium reading - turtle trading

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 places. I’m gonna adopt the object-oriented programming concept 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.subportfolios in 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 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 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 average as 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 lose 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 2N 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 mean 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 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 us to 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:

1. The China stock market has the policy that restricts day-trading (selling the stocks that you purchased that day), leaving the risk of our positions uncontrolled.
2. One lot in China 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 price are higher as we need to relocate our capital to several stocks in the strategy.
3. 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 trend.

But there are other thoughts that we still can extend and develop upon the turtle strategy we built:

1. 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.
2. 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 strategy instead of the original momentum 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

Click here if you want to checkout the source code

# Inspired from: https://www.joinquant.com/post/1401
#                and https://www.joinquant.com/view/community/detail/284a9688a58e0112b3bad8c1283548bc
# 标题：Turtle Strategy that monitors multiple stocks
# 作者：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)

                # 每上涨0.5N，加仓一个单元
                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
                    )

                # 若当前价格低于前out_date天的收盘价的最小值, 则卖掉所有持仓
                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

                # 若当前价格低于 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          # dollars_per_share是标的股票每波动一个最小单位，1手股票的总价格变化量。
                                            # 在国内最小变化量是0.01元，所以就是0.01 × 100=1
        self.number_days = 20               # 计算N值的天数
        self.unit_limit = 4                 # 4 for the lnog direction
        self.risk_ratio = 0.1               # 为了尽量买到更多股票，我把风险比率设为了0.1（因为海龟原
                                            # 来是应用在美国的期货上的，10倍杠杆，A股没杠杆，所以把风
                                            # 险比率设为0.1，这样用于单个品种的买卖资金就和期货一样了），
                                            # 也就是说，按照平均波动幅度的每天最大亏损是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)

'''
================================================================================
总体回测前
================================================================================
'''
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,
    )

'''
================================================================================
每天开盘前
================================================================================
'''
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))

'''
================================================================================
每天交易时
================================================================================
'''
# 按分钟回测
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)

'''
================================================================================
每天收盘后
================================================================================
'''
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

'''
================================================================================
策略结束后
================================================================================
'''
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)
        ])