借助 Python 实现海龟交易系统
前言
海龟交易系统本质上是一个趋势跟随的系统,但是最值得我们学习的,是资金管理尤其是分批建仓及动态止损的部分
一、趋势捕捉
唐奇安通道
该指标是有 Richard Donchian 发明的，是有 3 条不同颜色的曲线组成的，该指标用周期（一般都是 20 ）内的最高价和最低价来显示市场价格的波动性，当其通道窄时表示市场波动较小，反之通道宽则表示市场波动比较大。 如图所示：
该具体分析为：
当价格冲冲破上轨是就是可能的买的信号；反之，冲破下轨时就是可能的卖的信号。
该指标的计算方法为：
上线=Max （最高低， n ）
下线=Min （最低价， n ）
中线=（上线+下线）/2
海龟交易就是利用唐奇安通道的价格突破来捕捉趋势。
不过我们在向下突破 10 日唐奇安下沿卖出。
二、资金管理
2.1 、 N 值计算
N 值是仓位管理的核心，涉及加仓及止损。另外， N 值与技术指标平均真实波幅 ATR 很相似
首先介绍真实波幅： 真实波幅是以下三个值中的最大值
- 1 、当前交易日最高价和最低价的波幅
- 2 、前一交易日的收盘价与当前交易日最高价的波幅
- 3 、前一交易日的收盘价与当前交易日最低价的波幅
用公式写就是：
TrueRange=Max(High−Low,High−PreClose,PreClose−Low)接下来， N 值计算公式为：
N=PreN[−19 ：]+TrueRange20其中 preN 为前面 N 值， TrueRange 为当前的真实波幅,此公式的真是含义为计算之前 20 天（包括今天在内）的 N 的平均值。
另外，有些海龟交易系统用的是 ATR 来代替 N 值， ATR 为真实波幅的 20 日平均。
2.2 买卖单位及首次建仓
先给出公式：
Unit=1%∗AccountN首次建仓的时候，当捕捉到趋势，即价格突破唐奇安上轨时，买入 1 个 unit 。
其意义就是，让一个 N 值的波动与你总资金 1%的波动对应，如果买入 1unit 单位的资产，当天震幅使得总资产的变化不超过 1%。例如：
现在你有 10 万元资金， 1%波动就是 1000 元。假如标 X 的 N 值为 0.2 元， 1000 元÷0.2 元=5000 股。也就是说，你的第一笔仓位应该是在其突破上轨（假设为 5 元）时立刻买入 5000 股，耗资 25000 元。
2.3 加仓
若股价在上一次买入（或加仓）的基础上上涨了 0.5N ，则加仓一个 Unit 。
接上面的例子：假如 N 值仍为 0.2 。
价格来到 5 + 0.20.5 = 5.1 时，加仓 1 个 Unit ，买入 5000 股，耗资 25500 元，剩余资金 49500 元
价格来到 5.1 + 0.20.5 = 5.2 时再加仓 1 个 unit 。买入 5000 股，耗资 26000 元，剩余资金 23500 元
2.4 动态止损
当价格比最后一次买入价格下跌 2N 时，则卖出全部头寸止损。
接上面的例子，最后一次加仓价格为 5.2 。假如此时 N 值 0.2 元。 当价格下跌到 5.2 - 20.2 = 4.8 元时，清仓。
持仓成本为 （ 5+5.1+5.2 ）5000/15000 = 5.1 元。 此时亏损 （ 5.1-4.8 ）*15000 = 4500 元 对于 10 万来说 这波亏损 4.5%
2.5 止盈
当股价跌破 10 日唐奇安通道下沿，清空头寸结束本次交易。
三、代码实现
本代码用 ATR 代替 N 值进行计算,其他逻辑不变:
ATR=MA(TrueRange,20)我们以单只股票为标，建立海龟交易系统，当然，可以将总资产均分为 n 份，同时交易 n 个标。
计算 ATR 值用日线数据，监控价格突破采用分钟线。
0 初始化参数，在 initialize(account)写入
def initialize(account):
account.last_buy_prcie = 0 #上一次买入价
account.hold_flag = False # 是否持有头寸标志
account.limit_unit = 4 # 限制最多买入的单元数
account.unit = 0 # 现在买入1单元的股数
1 唐奇安通道计算及判断入场离场：
我们设计个函数，传入值为回测中 account.get_history()取得的某单个股票的历史数据、股票现价、T为计算唐奇安通道的数据长度,转化为dataframe格式
def IN_OR_OUT(data,price,T):
up = max(data['highPrice'].iloc[-T:])
down = min(data['lowPrice'].iloc[-int(T/2):]) # 这里是10日唐奇安下沿
if price>up:
return 1
elif price<down:
return -1
else:
return 0 2. ATR值计算：
def CalcATR(data):
TR_List = []
for i in range(1,21):
TR = max(data['highPrice'].iloc[i]-data['lowPrice'].iloc[i],abs(data['highPrice'].iloc[i]-data['closePrice'].iloc[i-1]),abs(data['closePrice'].iloc[i-1]-data['lowPrice'].iloc[i]))
TR_List.append(TR)
ATR = np.array(TR_List).mean()
return ATR计算unit,注意股数为100的整数倍
如果为数字货币，则不需要乘以100
def CalcUnit(perValue,ATR):
return int((perValue/ATR)/100)*1004.判断是否加仓或止损：
- 当价格相对上个买入价上涨 0.5ATR时，再买入一个unit
- 当价格相对上个买入价下跌 2ATR时，清仓
def Add_OR_Stop(price,lastprice,ATR):
if price >= lastprice + 0.5*ATR:
return 1
elif price <= lastprice - 2*ATR:
return -1
else:
return 05 判断上次卖出操作是否成功
判断上次卖出操作是否成功(可能出现当日买进，之后却判断需要卖出)
def SellComplete(hold_flag,security_position):
if len(security_position)>0 and hold_flag==False:
return True
else:
return False构建策略
分钟线回测时间略长啊~
先把上面写的函数集中下，方便微核充启后运行函数
################################################### 计算、判断函数 #####################################################################
def IN_OR_OUT(data,price,T):
up = max(data['highPrice'].iloc[-T:])
down = min(data['lowPrice'].iloc[-int(T/2):]) # 这里是10日唐奇安下沿
if price>up:
return 1
elif price<down:
return -1
else:
return 0
def CalcATR(data):
TR_List = []
for i in range(1,21):
TR = max(data['highPrice'].iloc[i]-data['lowPrice'].iloc[i],abs(data['highPrice'].iloc[i]-data['closePrice'].iloc[i-1]),abs(data['closePrice'].iloc[i-1]-data['lowPrice'].iloc[i]))
TR_List.append(TR)
ATR = np.array(TR_List).mean()
return ATR
def CalcUnit(perValue,ATR):
return int((perValue/ATR)/100)*100
def Add_OR_Stop(price,lastprice,ATR):
if price >= lastprice + 0.5*ATR:
return 1
elif price <= lastprice - 2*ATR:
return -1
else:
return 0
def SellComplete(hold_flag,security_position):
if len(security_position)>0 and hold_flag==False:
return True
else:
return False逻辑代码:
import numpy as np
import pandas as pd
from __future__ import division
from CAL.PyCAL import *
import matplotlib.pyplot as plt
start = '2012-01-01' # 回测起始时间
end = '2016-01-01' # 回测结束时间
benchmark = '000001.XSHE'
universe = ['000001.XSHE']
capital_base = 100000 # 起始资金
freq = 'm' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 1 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
#----------------------------------- 记录部分数据 -----------------------------
global record
record = {'break_up':{},'break_down':{},'stop_loss':{},'position':{},'ATR':{}} # 记录入场、离常、止损点、持仓比、ATR
#---------------------------------------------------------------------------------------
#****************************************** 策略主体 ********************************************
def initialize(account): # 初始化虚拟账户状态
account.last_buy_prcie = 0 #上一次买入价
account.hold_flag = False # 是否持有头寸标志
account.limit_unit = 4 # 限制最多买入的单元数
account.unit = 0 # 现在买入1单元的股数
account.add_time = 0 # 买入次数
def handle_data(account): # 每个交易日的买入卖出指令
T = 20
data = account.get_daily_history(T+1)
stk = universe[0]
data = data[stk]
data = pd.DataFrame(data)
prices = account.reference_price[stk]
today = Date.fromDateTime(account.current_date)
today = today.toISO()
# 0 如果停牌，直接跳过
if np.isnan(prices) or prices == 0: # 停牌或是还没有上市等原因不能交易
return
# 1 计算ATR
ATR = CalcATR(data)
record['ATR'].update({today:ATR})
# 2 判断上次卖出是否成功，若不成功，再次卖出
if SellComplete(account.hold_flag,account.security_position):
for stk in account.security_position:
order_to(stk,0)
# 3 判断加仓或止损
if account.hold_flag==True and len(account.security_position)>0: # 先判断是否持仓
temp = Add_OR_Stop(prices,account.last_buy_prcie,ATR)
if temp ==1and account.add_time<account.limit_unit: # 判断加仓
order_num = min(account.unit,account.cash) # 不够1unit时买入剩下全部
order_to(stk,account.unit)
account.last_buy_prcie = prices
account.add_time += 1
elif temp== -1: # 判断止损
order_to(stk,0)
initialize(account) # 重新初始化参数 very important here!
record['stop_loss'].update({today:prices})
# 4 判断入场离场
out = IN_OR_OUT(data,prices,T)
if out ==1 and account.hold_flag==False: #入场
value = account.reference_portfolio_value * 0.01
account.unit = CalcUnit(value,ATR)
order_to(stk,account.unit)
account.add_time = 1
account.hold_flag = True
account.last_buy_prcie = prices
record['break_up'].update({today:prices})
elif out==-1 and account.hold_flag ==True: #离场
order_to(stk,0)
initialize(account) # 重新初始化参数 very important here!
record['break_down'].update({today:prices})
# 5 计算持仓比
ratio = 1 - account.cash/account.reference_portfolio_value
record['position'].update({today:ratio}) # 虽然每分钟重算，但因为key是日期，最后覆盖为当日最终持仓比
return
四、海龟策略BTC现货版
看到平台上没有公开的python海龟策略，自己写个简单的抛个砖。
接近原版的海龟系统，没怎么优化，当做回测试验吧，也可以自己再优化下接实盘跑。
开仓：超过唐奇安上轨开仓
加仓：超过之前的价格的0.5ATR就加仓
止损止盈：跌破下轨或者跌破上次开仓价-2ATR就全部止盈
回测了1年数据，年化80%，最大回撤16%
现货资金利用率较低，改成合约版后收益会更高。
#!/root/anaconda3/bin/python3.6
# 我的托管者使用的是anaconda3解释器，所以，这里也必须指定绝对路径，否则在
# 导入包numpy 时会报错。
'''backtest
start: 2019-01-01 00:00:00
end: 2020-03-02 00:00:00
period: 1d
basePeriod: 1h
exchanges: [{"eid":"OKEX","currency":"BTC_USDT","stocks":0}]
args: [["DC_range",20],["atrlength",14]]
'''
import numpy as np
import pandas as pd
import datetime
# 默认参数（测试用，实际可改）
fresh_rete = 24 # 交易频率（小时）
trade_percent = 0.01 # 资产比率
DC_range = 20 # 唐奇安通道周期数（在这个周期计算最高点和最低点，作为判断入场和离场信号,离场取10日唐奇安下沿）
atrlength = 20 # atr周期数（在这个周期计算平均真实波幅average true-range）
data = {'ordertime':[],'id':[],'price':[]}
hisorder = pd.DataFrame(data)
def turtle():
#声明全局变量
global hisorder
acct = exchange.GetAccount()
records=exchange.GetRecords(fresh_rete*60*60)
ticker = exchange.GetTicker()
portfolio_value = acct.Balance+acct.FrozenBalance+(acct.Stocks+acct.FrozenStocks)*records[-1]['Close']
atr = TA.ATR(records, atrlength)[-1]
#计算得到unit大小
value = portfolio_value*trade_percent
unit = min(round(value/atr,4),round(acct.Balance/(ticker['Last']+100),4))
#unit = round(value/atr,2)
df = pd.DataFrame(records)
current_price = records[-1]['Close']
last_price = 0
if len(hisorder)!=0:
last_price = hisorder.iloc[-1]['price']
max_price = df[-DC_range:-2]['High'].max()
min_price = df[-int(DC_range/2):-2]['Low'].min()
opensign = len(hisorder)==0 and current_price > max_price
addsign = len(hisorder)!=0 and current_price > last_price + 0.5*atr
stopsign = len(hisorder)!=0 and current_price < min_price
closesign = len(hisorder)!=0 and current_price < (last_price - 2*atr)
# if _D(records[-1]['Time']/1000) == '2020-01-25 00:00:00':
# Log("records[-1]",records[-1])
if opensign | addsign:
if acct.Balance >= (ticker['Last']+10)*unit and unit >0:
id = exchange.Buy(ticker['Last']+10,unit)
orderinfo = exchange.GetOrder(id)
data = {'ordertime':_D(records[-1]['Time']/1000),'id':id,'price':records[-1]['Close']}
hisorder = hisorder.append(data,ignore_index=True)
Log('买入后，最新账户信息：', exchange.GetAccount())
Log("opensign",opensign,"addsign",addsign)
# else:
# Log('余额已不足，请充值......', exchange.GetAccount())
if stopsign | closesign:
exchange.Sell(-1, acct.Stocks+acct.FrozenStocks)
data = {'ordertime':[],'id':[],'price':[]}
hisorder = pd.DataFrame(data)
Log('卖出后，最新账户信息：', exchange.GetAccount())
Log("stopsign",stopsign,"closesign",closesign)
def main():
while True:
turtle()
Sleep(fresh_rete*60*60*1000) 注意： 我的托管者使用的是anaconda3解释器，所以，这里也必须指定绝对路径，否则在导入包numpy 时会报错。
在脚本第一行加上解释器。
#!/root/anaconda3/bin/python3.6模拟回测结果：
可以看到这个策略还是不错的，在几次大瀑布前都提前离场了，尤其在 2020-03-12 那天暴跌的情况下，提前空仓；在趋势上涨的情况下，可以连续加仓，吃掉趋势行情，不过，我们可以看出资金利用率比较低，还有一个就是在暴跌的情况下，没有抄底入场，而是等的时间太久了。
该策略地址：FMZ:海龟策略btc现货版
总结
本文主要介绍了海龟交易的细节，不过是面向一个投资目标的。当想投多只股票时，可以先设定几个坑位，平分资金，然后对每个坑位采用海龟交易策略。
海龟交易系统通常会用两个趋势捕捉系统，不同之处在于价格突破的上下线计算。系统 1 ：突破上线 20 日最高买，突破下线 10 日最低卖；系统 2 ：突破上线 55 日最高买，突破下线 20 日最低卖。 这部分可以通过修改参数实现。
原始的海龟交易采用唐奇安通道来捕捉趋势，虽然能捕捉到大趋势，但是在震荡的情况下表现不如人意，不过这也是所有趋势型策略的通病。
海龟交易策略的核心在于资金管理，可以看出策略的回撤比较小，并且还有优化的空间。资金管理不一定要与趋势型策略结合，是不是可以用到多因子策略上？动量反转？均值回归？这些就留给读者们自行尝试了~
转载文章：
优矿海龟交易系统的实现
FMZ:海龟策略btc现货版
为者常成，行者常至
自由转载-非商用-非衍生-保持署名（创意共享3.0许可证）
