Turtle-Style Stock Breakouts from Historical Highs and Lows
Summary
This event-driven stock strategy calculates the highest high and lowest low across a configurable number of recent daily bars for each symbol loaded from a CSV file. It subscribes to tick data and compares each new price with those stored levels. A move above the historical high triggers a long entry if there is no existing long position; the order price is set slightly above the current tick. Position quantity is derived from a per-symbol purchase amount, adjusted for price and rounded down to whole lots.
A move below the stored low prompts the strategy to close long positions for symbols in its configured list. Despite the Turtle name, the code does not update the breakout range as new bars arrive, and the lower-bound logic is an exit from longs rather than an explicit short entry. It gives implementation details but no test results, costs, or risk analysis. The code also depends on a specific broker SDK and CSV configuration, and its position-check and symbol handling should be reviewed before live use.
Key ideas
- The strategy uses a configurable lookback range of daily highs and lows as breakout thresholds.
- A tick above the stored high can trigger a long order when no long position is present.
- Position quantity is based on a configured amount and rounded to whole lots.
- A move below the stored low closes configured long positions rather than opening an explicit short.
- The document offers source code but no performance results, and the range is not refreshed after initialization.
Tags
Full text
# TurtleStrategy
# TurtleStrategy
## Source (Apache-2.0)
```python
# encoding: utf8
import logging
import logging.config
import pandas as pd
import numpy as np
from gmsdk import *
class TurtleStrategy(StrategyBase):
def __init__(self, *args, **kwargs):
super(TurtleStrategy, self).__init__(*args, **kwargs)
self.__get_param__()
self.__init_data__()
def __get_param__(self):
self.csv_file = self.config.get('para', 'csv_file')
self.period = self.config.getint('para', 'period')
self.hop = self.config.get('para', 'hop') or 0.1
def __init_data__(self):
'''
read stocks from csv file
:return:
'''
self.sec_ids = []
self.hist_data = dict()
self.positions = dict()
subscribe_symbols = []
stocks = pd.read_csv(self.csv_file, sep=',')
for r in stocks.iterrows():
exchange = r[1][0]
sec_id = r[1][1]
buy_amount = r[1][2]
t = "{0}.{1}".format(exchange, sec_id)
hl = self.get_highest_lowest_price(t)
#print (hl)
self.hist_data[t] = hl + (buy_amount,) ## high, low, buy_amount
self.sec_ids.append("{0}".format(sec_id))
subscribe_symbols.append(t + ".tick")
self.subscribe(",".join(subscribe_symbols))
def get_highest_lowest_price(self, symbol):
#print(symbol)
## get last N dailybars
hd = self.get_last_n_dailybars(symbol, self.period)
high_prices = [b.high for b in hd]
#high_prices.reverse()
low_prices = [b.low for b in hd]
#low_prices.reverse()
return np.max(high_prices), np.min(low_prices)
def on_tick(self, tick):
if not tick.sec_id in self.sec_ids:
pass
#self.logger.info( "received tick: %s %s %s %s" % (tick.exchange, tick.sec_id, round(tick.last_price,2), tick.last_volume))
else:
#self.logger.info( "received tick: %s %s, A: %s B: %s" % (tick.sec_id, round(tick.last_price,2), round(tick.asks[0][0],2), round(tick.bids[0][0],2)))
symbol = ".".join([tick.exchange, tick.sec_id])
data = self.hist_data.get(symbol)
self.hop = float(self.hop)
#print (data)
pa = self.get_positions()
#int pa
if data and tick.last_price > data[0] : ## check high
volume = int(data[2]/(tick.last_price+self.hop)/100)*100 ## get slots, then shares
if bool(self.get_position(tick.exchange,tick.sec_id,1)) == 0:
#print (to_dict(aaa[0]))
self.open_long(tick.exchange, tick.sec_id, tick.last_price + self.hop, volume)
#print (tick.exchange, tick.sec_id, tick.last_price + self.hop, volume)
#print (tick.sec_id)
elif data and tick.last_price < data[1]: ## check low
#p = self.get_position(tick.exchange, tick.sec_id, OrderSide_Bid)
ps = self.get_positions()
for p in ps:
sym = ".".join([p.exchange, p.sec_id])
self.positions[sym] = p
#print (p)
## if not in stock list, close long position
if p.sec_id in self.sec_ids:
self.close_long(p.exchange, p.sec_id, 0, p.volume)
if __name__ == '__main__':
ini_file = sys.argv[1] if len(sys.argv) > 1 else 'turtle.ini'
logging.config.fileConfig(ini_file)
st = TurtleStrategy(config_file=ini_file)
st.logger.info("Strategy turtle ready, waiting for data ...")
ret = st.run()
st.logger.info("Strategy turtle message %s" % st.get_strerror(ret))
```Shown in full with attribution under the source's licence. Licence: Apache-2.0
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.