Turtle Trend Following with Donchian Breakouts and ATR Pyramiding
Summary
This strategy implements a Turtle style breakout system using Donchian channels. When flat, it calculates an entry channel and places stop orders for breakouts above or below its boundaries. It also calculates a shorter Donchian channel for exits. After an entry, the initial protective level is set two ATRs away from the recorded fill price, while exits use the nearer of that stop level and the opposing exit channel.
The code adds to a position in fixed size increments at half ATR intervals, up to four units, and uses tick updates to construct bars before processing each bar. Its evidence is the implementation itself; it reports no backtest or realized performance. The example also cancels open orders on each bar and only refreshes the ATR while flat, details that matter when assessing order behavior and whether stops and additions track changing volatility as intended.
Key ideas
- Donchian channel boundaries define breakout entry levels and a shorter channel supplies exit levels.
- ATR sets an initial stop distance and spaces successive additions to a position.
- The strategy scales into positions in fixed units, with a limit of four units in either direction.
- Tick data is aggregated into bars before the strategy updates its indicators and orders.
- The document provides strategy code but no backtest results or performance evidence.
Tags
Full text
# TurtleSignalStrategy
# TurtleSignalStrategy
## Source (MIT)
```python
from howtrader.app.cta_strategy import (
CtaTemplate,
StopOrder
)
from howtrader.trader.object import TickData, BarData, TradeData, OrderData, Direction
from howtrader.trader.utility import BarGenerator, ArrayManager
from decimal import Decimal
class TurtleSignalStrategy(CtaTemplate):
""""""
author = "用Python的交易员"
entry_window = 20
exit_window = 10
atr_window = 20
fixed_size = 1
entry_up = 0
entry_down = 0
exit_up = 0
exit_down = 0
atr_value = 0
long_entry = 0
short_entry = 0
long_stop = 0
short_stop = 0
parameters = ["entry_window", "exit_window", "atr_window", "fixed_size"]
variables = ["entry_up", "entry_down", "exit_up", "exit_down", "atr_value"]
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
""""""
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.bg = BarGenerator(self.on_bar)
self.am = ArrayManager()
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(20)
def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
self.bg.update_tick(tick)
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.cancel_all()
self.am.update_bar(bar)
if not self.am.inited:
return
# Only calculates new entry channel when no position holding
if not self.pos:
self.entry_up, self.entry_down = self.am.donchian(
self.entry_window
)
self.exit_up, self.exit_down = self.am.donchian(self.exit_window)
if not self.pos:
self.atr_value = self.am.atr(self.atr_window)
self.long_entry = 0
self.short_entry = 0
self.long_stop = 0
self.short_stop = 0
self.send_buy_orders(self.entry_up)
self.send_short_orders(self.entry_down)
elif self.pos > 0:
self.send_buy_orders(self.entry_up)
sell_price = max(self.long_stop, self.exit_down)
self.sell(Decimal(sell_price), Decimal(abs(self.pos)), True)
elif self.pos < 0:
self.send_short_orders(self.entry_down)
cover_price = min(self.short_stop, self.exit_up)
self.cover(Decimal(cover_price), Decimal(abs(self.pos)), True)
self.put_event()
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
if trade.direction == Direction.LONG:
self.long_entry = trade.price
self.long_stop = self.long_entry - 2 * self.atr_value
else:
self.short_entry = trade.price
self.short_stop = self.short_entry + 2 * self.atr_value
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
pass
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass
def send_buy_orders(self, price):
""""""
t = self.pos / self.fixed_size
if t < 1:
self.buy(Decimal(price), Decimal(self.fixed_size), True)
if t < 2:
price = price + self.atr_value * 0.5
self.buy(Decimal(price), Decimal(self.fixed_size), True)
if t < 3:
price = price + self.atr_value
self.buy(Decimal(price), Decimal(self.fixed_size), True)
if t < 4:
price = price + self.atr_value * 1.5
self.buy(Decimal(price), Decimal(self.fixed_size), True)
def send_short_orders(self, price):
""""""
t = self.pos / self.fixed_size
if t > -1:
self.short(Decimal(price), Decimal(self.fixed_size), True)
if t > -2:
price = price - self.atr_value * 0.5
self.short(Decimal(price), Decimal(self.fixed_size), True)
if t > -3:
price = price - self.atr_value
self.short(Decimal(price), Decimal(self.fixed_size), True)
if t > -4:
price = price - self.atr_value * 1.5
self.short(Decimal(price), Decimal(self.fixed_size), True)
```Shown in full with attribution under the source's licence. Licence: MIT
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.