針對VNPY策略遺傳算法優化怎樣寫了一個類GeneticOptimizeStrategy

這篇文章將為大家詳細講解有關針對VNPY策略遺傳算法優化怎樣寫了一個類GeneticOptimizeStrategy，文章內容質量較高，因此小編分享給大家做個參考，希望大家閱讀完這篇文章后對相關知識有一定的了解。

寫了一個類，GeneticOptimizeStrategy，

1. Parameterlist 字典； 簡化了調用參數和定義對應隨機范圍問題。只要在Parameterlist 中定義策略參數名稱和對應的隨機范圍就可以，其中兩個參數的元祖是兩個之間隨機數，調用random.uniform()，三個參數元祖是開始結束和中間步進,調用的是random.randrange(), 如果是數組就是在數組中間隨機選擇。

2. Symbollist 字典，維護回測品種和數據

3. poptoExcel方法，輸出一個Excel，包括參數和value；參數可以直接調用。同時把相同項合并。效果如下圖

   源代碼也可以去我的GitHub中

-------------------------------------------------------------------------------------------------------------

發現多線程時候有報錯

cPickle.PicklingError: Can't pickle <type 'instancemethod'>: attribute lookup builtin.instanceme
搜索一下，發現是python2.7多進程問題，pool.map沒法綁定包在類里面方法。

把evaluate的方法放在類外面做成靜態方法綁定，雖然解決了pickle，但是在多線程情況下，策略參數名字和值的對應經出出錯。
就把參數賦值的方式也改了，從
[value1,value2,value3...]變成[{key1:value1},{key2,value2},{key3,value3}...]，這樣?？梢詽M足交叉，突變要求。生成隨機DNA算法要改下。

    def parameter_generate(self):
        '''
        根據設置的起始值，終止值和步進，隨機生成待優化的策略參數
        '''
        parameter_list = []
        for key, value in self.parameterlist.items():
            if isinstance(value, tuple):
                if len(value) == 3:
                    parameter_list.append({key:random.randrange(value[0], value[1], value[2])})
                elif len(value) == 2:
                    parameter_list.append({key:random.uniform(value[0], value[1])})
            elif isinstance(value, list):
                parameter_list.append({key:random.choice(value)})
            else:
                parameter_list.append({key:value})
        return parameter_list

parameter_list是類似vnpy optimize的格式，不過有所增強。 strategy_avg返回變成了[{key1:value1},{key2,value2},{key3,value3}...]，包含字典的list格式也可以滿足交叉和突變方法。

在進化篩選方法object_func，使用下面遍歷把[{key1:value1},{key2,value2},{key3,value3}...]變回{key1,value1,key2,value2...}，這樣就可以進入回測

    setting = {}
    for item in range(len(strategy_avg)):
        setting.update(strategy_avg[item])

最后，因為進化篩選方法object_func放在類外面，但必須要把一些回測參數，比如品種，日期等的傳入，這里有兩種方式可以實現，一個是把傳入individual list改為Tuple list，變成[(individual, parameterPackage)..]這樣list，但是就要修改algorithms.eaMuPlusLambda，比較麻煩。
還有一個是增強individual, 加入這個回測參數集做為屬性。但是多線程也有些要注意，不得不把parameterPackage做為靜態屬性放在類里面，不然回提示parameterPackage為空。還沒有找到比較合適處理方法

creator.create("Individual", list, fitness=creator.FitnessMulti, parameterPackage=parameterPackage)

完整新代碼如下

# encoding: UTF-8
"""
展示如何執行參數優化。
"""
from __future__ import division
from __future__ import print_function
from vnpy.trader.app.ctaStrategy.ctaBacktesting import BacktestingEngine, MINUTE_DB_NAME, OptimizationSetting
from vnpy.trader.app.ctaStrategy.strategy.strategyBollChannel import BollChannelStrategy
import random
import numpy as np
from deap import creator, base, tools, algorithms
import multiprocessing
import time, datetime
import pandas as pd
def object_func(strategy_avgTuple):
    """
    本函數為優化目標函數，根據隨機生成的策略參數，運行回測后自動返回2個結果指標：收益回撤比和夏普比率
    """
    strategy_avg = strategy_avgTuple
    paraSet = strategy_avgTuple.parameterPackage
    symbol = paraSet["symbol"]
    strategy = paraSet["strategy"]
    # 創建回測引擎對象
    engine = BacktestingEngine()
    # 設置回測使用的數據
    engine.setBacktestingMode(engine.BAR_MODE)  # 設置引擎的回測模式為K線
    engine.setDatabase("VnTrader_1Min_Db", symbol["vtSymbol"])  # 設置使用的歷史數據庫
    engine.setStartDate(symbol["StartDate"])  # 設置回測用的數據起始日期
    engine.setEndDate(symbol["EndDate"])  # 設置回測用的數據起始日期
    # 配置回測引擎參數
    engine.setSlippage(symbol["Slippage"])  # 1跳
    engine.setRate(symbol["Rate"])  # 傭金大小
    engine.setSize(symbol["Size"])  # 合約大小
    engine.setPriceTick(symbol["Slippage"])  # 最小價格變動
    engine.setCapital(symbol["Capital"])
    setting = {}
    for item in range(len(strategy_avg)):
        setting.update(strategy_avg[item])
    engine.clearBacktestingResult()
    # 加載策略
    engine.initStrategy(strategy, setting)
    # 運行回測，返回指定的結果指標
    engine.runBacktesting()  # 運行回測
    # 逐日回測
    # engine.calculateDailyResult()
    backresult = engine.calculateBacktestingResult()
    try:
        capital = round(backresult['capital'], 3)  # 收益回撤比
        profitLossRatio = round(backresult['profitLossRatio'], 3)  # 夏普比率                 #夏普比率
        sharpeRatio = round(backresult['sharpeRatio'], 3)
    except Exception, e:
        print("Error: %s, %s" %(str(Exception),str(e)))
        sharpeRatio = 0
        profitLossRatio = 0  # 收益回撤比
        averageWinning = 0  # 夏普比率                 #夏普比率
        capital = 0
    return capital, sharpeRatio, profitLossRatio
class GeneticOptimizeStrategy(object):
    Strategy = BollChannelStrategy
    Symbollist ={
                    "vtSymbol": 'rb0000',
                    "StartDate": "20140601",
                    "EndDate": "20141101",
                    "Slippage": 1,
                    "Size": 10,
                    "Rate": 2 / 10000.0,
                    "Capital": 10000
                    }
    Parameterlist = {
                    'bollWindow': (10,50,1),       #布林帶窗口
                    'bollDev': (2,10,1),        #布林帶通道閾值
                    'slMultiplier':(3,6),
                    'barMins':[2,3,5,10,15,20],
    }
    parameterPackage = {
    "symbol":Symbollist,
    "strategy":Strategy
    }
    # ------------------------------------------------------------------------
    def __init__(self, Strategy, Symbollist, Parameterlist):
        self.strategy = Strategy
        self.symbol = Symbollist
        self.parameterlist = Parameterlist
        self.parameterPackage = {
            "strategy":self.strategy,
            "symbol":self.symbol
        }
    creator.create("FitnessMulti", base.Fitness, weights=(1.0, 1.0, 1.0))  # 1.0 求最大值；-1.0 求最小值
    creator.create("Individual", list, fitness=creator.FitnessMulti, parameterPackage=parameterPackage)
    # ------------------------------------------------------------------------
    def parameter_generate(self):
        '''
        根據設置的起始值，終止值和步進，隨機生成待優化的策略參數
        '''
        parameter_list = []
        for key, value in self.parameterlist.items():
            if isinstance(value, tuple):
                if len(value) == 3:
                    parameter_list.append({key:random.randrange(value[0], value[1], value[2])})
                elif len(value) == 2:
                    parameter_list.append({key:random.uniform(value[0], value[1])})
            elif isinstance(value, list):
                parameter_list.append({key:random.choice(value)})
            else:
                parameter_list.append({key:value})
        return parameter_list
    def mutArrayGroup(self, individual, parameterlist, indpb):
        size = len(individual)
        paralist = parameterlist()
        for i in xrange(size):
            if random.random() < indpb:
                individual[i] = paralist[i]
        return individual,
    def optimize(self):
        # 設置優化方向：最大化收益回撤比，最大化夏普比率
        toolbox = base.Toolbox()  # Toolbox是deap庫內置的工具箱，里面包含遺傳算法中所用到的各種函數
        pool = multiprocessing.Pool(processes=(multiprocessing.cpu_count()-1))
        toolbox.register("map", pool.map)
        # 初始化
        toolbox.register("individual", tools.initIterate, creator.Individual,
                         self.parameter_generate)  # 注冊個體：隨機生成的策略參數parameter_generate()
        toolbox.register("population", tools.initRepeat, list,
                         toolbox.individual)  # 注冊種群：個體形成種群
        toolbox.register("mate", tools.cxTwoPoint)  # 注冊交叉：兩點交叉
        toolbox.register("mutate", self.mutArrayGroup, parameterlist=self.parameter_generate,
                         indpb=0.6)  # 注冊變異：隨機生成一定區間內的整數
        toolbox.register("evaluate", object_func)  # 注冊評估：優化目標函數object_func()
        toolbox.register("select", tools.selNSGA2)  # 注冊選擇:NSGA-II(帶精英策略的非支配排序的遺傳算法)
        # 遺傳算法參數設置
        MU = 8  # 設置每一代選擇的個體數
        LAMBDA = 5  # 設置每一代產生的子女數
        pop = toolbox.population(20)  # 設置族群里面的個體數量
        CXPB, MUTPB, NGEN = 0.5, 0.3, 10 # 分別為種群內部個體的交叉概率、變異概率、產生種群代數
        hof = tools.ParetoFront()  # 解的集合：帕累托前沿(非占優最優集)
        # 解的集合的描述統計信息
        # 集合內平均值，標準差，最小值，最大值可以體現集合的收斂程度
        # 收斂程度低可以增加算法的迭代次數
        stats = tools.Statistics(lambda ind: ind.fitness.values)
        np.set_printoptions(suppress=True)  # 對numpy默認輸出的科學計數法轉換
        stats.register("mean", np.mean, axis=0)  # 統計目標優化函數結果的平均值
        stats.register("std", np.std, axis=0)  # 統計目標優化函數結果的標準差
        stats.register("min", np.min, axis=0)  # 統計目標優化函數結果的最小值
        stats.register("max", np.max, axis=0)  # 統計目標優化函數結果的最大值
        # 運行算法
        algorithms.eaMuPlusLambda(pop, toolbox, MU, LAMBDA, CXPB, MUTPB, NGEN, stats,
                                  halloffame=hof, verbose=True)  # esMuPlusLambda是一種基于(μ+λ)選擇策略的多目標優化分段遺傳算法
        return pop
    def poptoExcel(self, pop, number = 1000, path = "C:/data/"):
        #按照輸入統計數據隊列和路徑，輸出excel，這里不提供新增模式，如果想，可以改
        #dft.to_csv(path,index=False,header=True, mode = 'a')
        path = path +  self.strategy.className + "_" + self.symbol[ "vtSymbol"] + str(datetime.date.today())+ ".xls"
        summayKey = ["StrategyParameter","TestValues"]
        best_ind = tools.selBest(pop, number)
        dft = pd.DataFrame(columns=summayKey)
        for i in range(0,len(best_ind)-1):
            if i == 0:
                # new = pd.DataFrame([{"StrategyParameter":self.complieString(best_ind[i])},{"TestValues":best_ind[i].fitness.values}], index=["0"])
                dft = dft.append([{"StrategyParameter":self.complieString(best_ind[i]),"TestValues":best_ind[i].fitness.values}], ignore_index=True)
            elif str(best_ind[i-1]) == (str(best_ind[i])):
                pass
            else:
                #new = pd.DataFrame({"StrategyParameter":self.complieString(best_ind[i]),"TestValues":best_ind[i].fitness.values}, index=["0"])
                dft = dft.append([{"StrategyParameter":self.complieString(best_ind[i]),"TestValues":best_ind[i].fitness.values}], ignore_index=True)
        dft.to_excel(path,index=False,header=True)
        print("回測統計結果輸出到" + path)
    def complieString(self,individual):
        setting = {}
        for item in range(len(individual)):
            setting.update(individual[item])
        return str(setting)
if __name__ == "__main__":
    Strategy = BollChannelStrategy
    Symbollist ={
                    "vtSymbol": 'rb0000',
                    "StartDate": "20140601",
                    "EndDate": "20141101",
                    "Slippage": 1,
                    "Size": 10,
                    "Rate": 2 / 10000.0,
                    "Capital": 10000
                    }
    Parameterlist = {
                    'bollWindow': (10,50,1),       #布林帶窗口
                    'bollDev': (2,10,1),        #布林帶通道閾值
                    'slMultiplier':(3,6),
                    'barMins':[2,3,5,10,15,20],
    }
    parameterPackage = {
    "symbol":Symbollist,
    "parameterlist":Parameterlist,
    "strategy":Strategy
    }
    GE = GeneticOptimizeStrategy(Strategy,Symbollist,Parameterlist)
    GE.poptoExcel(GE.optimize())
    print("-- End of (successful) evolution --")

-------------------------------------------------------------------------------------------------------------

關于針對VNPY策略遺傳算法優化怎樣寫了一個類GeneticOptimizeStrategy就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。