量化交易之海龜策略

代碼只能保證運行在ricequant上

海龜策略這里我就不著重介紹，感興趣可以去買書自己看。

下面是我搬運的，重點我加粗了。

一、海龜交易法

海龜交易法是著名的公開交易系統，1983年著名的商品投機家理查德. 丹尼斯在一個交易員培訓班上推廣而聞名于世，它涵蓋了交易系統的各個方面。其法則覆蓋了交易的各個方面，并且不給交易員留下一點主觀想象決策的余地。它具備一個完整的交易系統的所有成分。

著名的商品投機家理查德·丹尼斯想弄清楚偉大的交易員是天生造就的還是后天培養的。為此，在1983年他招募了13個人，教授給他們期貨交易的基本概念，以及他自己的交易方法和原則。 “學員們被稱為‘海龜’(丹尼斯先生說這項計劃開始時他剛剛從亞洲回來，他解釋了自己向別人說過的話，‘我們正在成長為交易員，就象在新加坡他們正在成長為海龜一樣’)?！薄?斯坦利.W.安格瑞斯特，《華爾街期刊》，1989年9月5日。

海龜成為交易史上最著名的實驗，因為在隨后的四年中海龜們取得了年均復利80%的收益。 丹尼斯證明用一套簡單的系統和法則，可以使僅有很少或根本沒有交易經驗的人成為優秀的交易員。 當時，海龜們認為應對理查德·丹尼斯負責，商定甚至在他們議定的10年保密協定于1993年終止后也不泄露這些法則。但是，有個別海龜在網站上出售海龜交易法則而謀取錢財。兩個原版海龜科蒂斯·費思和阿瑟·馬多克，為了阻止個別海龜對知識產權的偷竊和出售海龜交易法則而賺錢的行為，決定在網站上將海龜交易法則免費公之于眾。

海龜交易策略是一套非常完整的趨勢跟隨型的自動化交易策略。這個復雜的策略在入場條件、倉位控制、資金管理、止損止盈等各個環節，都進行了詳細的設計，這基本上可以作為復雜交易策略設計和開發的模板。

二、趨勢信號的捕捉

在趨勢信號的撲捉上，海龜交易法則使用了一個非常重要的技術指標—唐奇安通道（Donchian channel）。這個通道很類似布林通道（Bollinger Bands），只是在具體計算方式上有些不一樣。

唐奇安通道指標是Richard Donchian發明的，由3條不同顏色的曲線組成，該指標用周期（一般都是20，有的平臺系統設置時可以改變的，有的則設置的不可以）內的最高價和最低價來顯示市場價格的波動性，當其通道窄時表示市場波動較小，反之通道寬則表示市場波動比較大。

當價格沖破該通道的上軌道時，就是可能的買信號；反之，沖破下軌時就是可能的賣信號。唐奇安通道的各項指標的計算方法為：

上軌=Max（最高低，n）, n日最高價的最大值

下軌=Min（最低價，n）, n日最低價的最小值

中軌=（上軌+下軌）/2

在金融領域的多因子分析框架內，這個策略對突破后的價格走勢預測就是基于動量因子（momentum）的有效性假設。當然，這個因子的有效性也確實被嚴格地驗證過，并作為Fama-French三因素模型的補充，被廣泛地應用于金融市場。

當然，我們可以進行優化，使用更合理的趨勢突破型指標。

三、交易原理

那么，既然動量因子是已經公開并普遍使用的因子，那么海龜交易法則是憑什么能夠脫穎而出呢？答案很簡單，海龜交易法則定義了一整套非常嚴謹的倉位控制、止盈止損的規則。對于幣市的大漲大跌行情，海龜交易法則正是應付這種極端行情的利器。下面我們來一一剖析。

倉位的基本單位N

海龜法則的加倉原則是定義好一個小單位（Unit），使得該倉位的預期價值波動與總凈資產的1%對應。也就是說，如果買入了這1個小單位的資產，那當天該倉位的市值變動幅度不會超過總凈資產的1%。

那么，如何定義這個小單位？又如何預估這個小單位能帶來的價值波動呢？首先，在預估這個小單位帶來的價值波動（該價值波動被稱為N）上，海龜策略使用了對歷史的價格波動進行統計的方法。具體計算公式如下：

**TrueRange=Max(High?Low, High?PreClose, PreClose?Low)

N =(前19日的N值之和+當時的TrueRange)/20**

其中，High表示當日最高價，Low表示當日最低價，PreClose表示前一日收盤價。

我們可以從定義上看出，N值確實能很恰當地表達該資產在價格上的最近波動幅度。

這樣，一個Unit就應該是這樣計算出來的：

Unit = (1%*Total_net)/N， total_net就是總資產凈值

可以看出，一個Unit的資產的價格波動幅度 = 總凈資產的1%

什么時候建倉

建倉的動作來自于趨勢突破信號的產生。如果當前價格沖破上軌，就產生了一個買的建倉信號，如果當前價格跌破下軌，就產生了一個賣空的建倉信號（如果該市場支持賣空，某些數字資產市場是支持借幣賣空的?。?/p>

初始建倉的大小 = 1個Unit

什么時候加倉

如果開的底倉是多倉且資產的價格在上一次建倉（或者加倉）的基礎上又上漲了0.5N，就再加一個Unit的多倉；

如果開的底倉是空倉且資產的價格在上一次建倉（或者加倉）的基礎上又下跌了0.5N，就再加一個Unit的空倉。

我們看到，海龜策略其實是一個追漲殺跌的策略的。

怎么做動態止損

如果開的底倉是多倉且資產的價格在上一次建倉（或者加倉）的基礎上又下跌了2N，就賣出全部頭寸止損；

如果開的底倉是空倉且資產的價格在上一次建倉（或者加倉）的基礎上又上漲了2N，就buy cover全部的頭寸止損。

當然，用戶可以自定義動態止損方案，比如下跌了0.5N就開始部分平倉，而不用等到下跌了2N后才匆忙一次性清倉，畢竟沖擊成本擺在那里。

怎么做止盈，可以自定義動態止盈嗎

海龜法則里面，止盈信號是這樣產生的：

如果開的底倉是多倉且當前資產價格跌破了10日唐奇安通道的下軌，就清空所有頭寸結束策略；

如果開的底倉是空倉且當前資產價格升破了10日唐奇安通道的上軌，就清空所有頭寸結束策略。

當然，用戶可以自定義動態止盈方案，比如總凈資產/初始凈資產>1.5, 就止盈離場。

下面是我實現的代碼，較常規的海龜策略而言，我增加了每次開盤前有一個篩選股票的策略，我只會在市值最大的幾只股票進行操作。

效果還行，跑過了滬深300指數:)

# 可以自己import我們平臺支持的第三方python模塊，比如pandas、numpy等。
import numpy as np
import pandas as pd
import talib
import math
# 在這個方法中編寫任何的初始化邏輯。context對象將會在你的算法策略的任何方法之間做傳遞。
def init(context):
    # 在context中保存全局變量
    context.s = []
    context.vis = {}
    context.last_buy_price = {} 
    #上一次買入價格
    context.hold_flag = {}
    #False 
    #是否持有頭寸標志，即是否買入
    context.limit_unit = 4 
    #最多買入單元數
    context.unit = {}#0 
    #現在買入一單元股數
    context.add_time = {}#0 
    #買入次數 最多四次
    context.long_time = 55
    context.short_time = 20
    context.ten_time = 10
def CalcATR(high,low,close):
    TR_List = []
    for i in range(1,min(21,len(high),len(close))):
        TR = max(high[i]-low[i],abs(high[i]-close[i-1]),abs(close[i-1]-low[i]))
        TR_List.append(TR)
    ATR = np.array(TR_List).mean()
    return ATR
def CalcUnit(perValue,ATR):#計算unit
    return int((perValue/ATR)/100)*30
    # 乘了個系數，莽
def IN_OR_OUT(high,low,price): #判斷入場還是離場
    up = max(high)
    down = min(low) 
    if price>up:
        return 1
    elif price<down:
        return -1
    else:
        return 0 
def Add_OR_Stop(price,lastprice,ATR):
    if price >= lastprice + 0.5*ATR:
        return 1
    elif price <= lastprice - 0.5*ATR:
        return -1
    else:
        return 0    
# before_trading此函數會在每天策略交易開始前被調用，當天只會被調用一次
# 選股策略，選三個滿足特征的幾個，然后來進行排序
def before_trading(context): # 獲得stock list
    num_stocks = 5
    fundamental_df = get_fundamentals(
        query(
            fundamentals.eod_derivative_indicator.pb_ratio,
            fundamentals.eod_derivative_indicator.pe_ratio,
            fundamentals.eod_derivative_indicator.peg_ratio,
        )
        .filter(
            fundamentals.eod_derivative_indicator.pe_ratio<60
        )
        .filter(
            fundamentals.eod_derivative_indicator.pb_ratio<5
        )
        .filter(
            fundamentals.eod_derivative_indicator.peg_ratio<0.9
        )
        .order_by(
            fundamentals.eod_derivative_indicator.market_cap.desc()
        ).limit(
            num_stocks
        )
    )
    '''
    fundamental_df = get_fundamentals(
        query(
            fundamentals.eod_derivative_indicator.market_cap
        ).filter(
            fundamentals.eod_derivative_indicator.market_cap<5e10
        ).order_by(
            fundamentals.eod_derivative_indicator.market_cap.asc()   
        ).limit(
            num_stocks
        )
    )
    '''
    drop_list = []
    for stock in context.s:
        if context.add_time.get(stock,0) == 0:
            drop_list.append(stock)
    for stock in drop_list:
        init_stock(context,stock)
    if len(context.s) < 20:
        for stock in fundamental_df:
            if stock not in context.vis:
                context.s.append(stock)
                context.vis[stock] = 1
    logger.info(str(context.s))
# 你選擇的證券的數據更新將會觸發此段邏輯，例如日或分鐘歷史數據切片或者是實時數據切片更新
def handle_bar(context, bar_dict):
    # 開始編寫你的主要的算法邏輯
    stock_list = []
    for stock in context.s:
        stock_list.append(stock)
    for stock in stock_list:
        do_trade(context,stock)
def init_stock(context,stock):
    context.s.remove(stock)
    if stock in context.last_buy_price:
        del context.last_buy_price[stock]
    if stock in context.hold_flag:
        del context.hold_flag[stock]
    if stock in context.unit:
        del context.unit[stock] 
    if stock in context.add_time:
        del context.add_time[stock]
    if stock in context.vis:
        del context.vis[stock]
def do_trade(context,stock):
    high_price = history_bars(stock,context.short_time+1, '1d', 'high')
    low_price = history_bars(stock,context.short_time+1, '1d', 'low')
    close_price = history_bars(stock,context.short_time+2, '1d', 'close')
    close_price_atr = close_price[:-1]
    high_price_55 = history_bars(stock,context.long_time+1, '1d', 'high')
    low_price_55 = history_bars(stock,context.long_time+1, '1d', 'low')
    high_price_10 = history_bars(stock,context.ten_time+1, '1d', 'high')
    low_price_10 = history_bars(stock,context.ten_time+1, '1d', 'low')
    price_minute = history_bars(stock,context.short_time,'1m','close')
    price = price_minute[-1] #股票最新價格
    # 計算ATR
    ATR = CalcATR(high_price,low_price,close_price_atr)
    out = IN_OR_OUT(high_price_55,low_price_55,price) #判斷入場還是離場
    if out==1 and context.hold_flag.get(stock,False)==False: #入場
        value = context.portfolio.cash*0.01  # 我的資金的1%
        context.unit[stock] = CalcUnit(value,ATR)
        order_shares(stock, context.unit[stock])
        context.add_time[stock]=1
        context.hold_flag[stock]=True
        context.last_buy_price[stock]=price
    elif out==-1 and context.hold_flag.get(stock,False) ==True: #離場
        order_percent(stock,-1)
        init_stock(context,stock) #重新初始化
    #判斷加倉還是止損
    if context.hold_flag.get(stock,False)==True :
        temp = Add_OR_Stop(price,context.last_buy_price.get(stock,0),ATR)
        if temp==1 and context.add_time.get(stock,4) < context.limit_unit:#加倉
            order_shares(stock,context.unit[stock])
            context.last_buy_price[stock]=price
            context.add_time[stock]+=1
        elif temp==-1:
            if context.add_time.get(stock,0) == 1:
                order_percent(stock,-1)
                init_stock(context,stock)
            else:
                order_shares(stock,context.unit.get(stock,0))
                context.add_time[stock]-=1
# after_trading函數會在每天交易結束后被調用，當天只會被調用一次
def after_trading(context):
    pass