构建自己的交易策略模型

基于策略设计模式的波动市场交易模型：

1. 按照策略设计模式至少需要需要以下几个类：

• 环境类 包裹数据与策略池；
• 数据类；
• 策略类及其基类；

2. 环境类设计：

class Model:
    def __init__(self):
        self.current_money = 300000
        self.current_price = 0
        self.total_shares = 0
        self.cost_for_each_trading = 1000
        self.cast_current_value_ratio = self.current_money/self.current_value
        self.current_value_all = []
        self._data = np.random.normal(0,1,3000).cumsum()[2000:]
        self.strategies = {}
        self.strategy = None
        self.action_signals = {"buy":self.buy,"sell":self.sell,"look":self.look}

    @property
    def data(self):
        if self._data.min()<0:
            self._data += -self._data.min()+1
        for d in self._data:
            yield d
    
    @property
    def current_value(self):
        return self.current_money + self.total_shares*self.current_price

    def register(self,strategy,strategy_name,**kwargs):
        self.strategies[strategy_name] = strategy(**kwargs)
    
    def run(self,strategy):
        self.strategy = self.strategies[strategy]
        for price in self.data:
            self.current_price = price
            self.action_signals[self.strategy.action_signal(price)](price)

    def buy(self,price):
        if self.current_money>0 and self.current_money/self.current_value>0.4:
            self.shares = self.cost_for_each_trading/self.current_price
            self.current_money -= self.cost_for_each_trading
            self.total_shares += self.shares
        self.current_value_all.append(self.current_value)

    def sell(self,price):
        if self.total_shares>0:
            self.shares = self.cost_for_each_trading/self.current_price
            self.current_money += self.cost_for_each_trading
            self.total_shares -= self.shares
        self.current_value_all.append(self.current_value)
        
    def look(self,price):
        self.current_value_all.append(self.current_value)
        
    def evalult(self):
        try:
            return (self._data[-1]-self._data[0])/self._data[0],(self.current_value-300000)/300000
        except ZeroDivisionError:
            return 0
    
    def plot(self):
        plt.subplot(211).plot(list(self.data))
        plt.subplot(212).plot(self.current_value_all)
        plt.show()

3. 策略基类：

class Strategy(ABC):
    @abstractclassmethod
    def action_signal(self,data):
        pass

4. 基于策略基类的交易策略：
   三个交易策略，就一直写自己的交易策略就好了：

class Awlays_buying(Strategy):
###一直买入的策略
    def action_signal(self,price):
        return "buy"

class Enforce_Awlays_buying(Strategy):
###下跌就买入的策略
    def __init__(self):
        self.enforce_last_price    = None
        self.enforce_current_price = None
        
    def action_signal(self,price):
        self.enforce_current_price = price
        if self.enforce_last_price:
            if self.enforce_current_price < self.enforce_last_price:
                self.enforce_last_price = self.enforce_current_price
                return "buy"
            else:
                self.enforce_last_price = self.enforce_current_price
                return "look"
        self.enforce_last_price = self.enforce_current_price
        return "look"

class Moving_average(Strategy):
###移动均线金叉死叉策略
    def __init__(self,short_range_count=5,long_range_count=20):
        self.short_range = deque([None]*short_range_count)
        self.long_range  = deque([None]*long_range_count)
        self.current_short_range_average = 0
        self.current_long_range_average  = 0
        self.last_short_range_average    = 0
        self.last_long_range_average     = 0

    def action_signal(self,price):
        self.short_range.appendleft(price)
        self.short_range.pop()
        self.long_range.appendleft(price)
        self.long_range.pop()
        if not None in self.short_range and not None in self.long_range:
            self.current_short_range_average = np.average(self.short_range)
            self.current_long_range_average  = np.average(self.long_range)
            if self.last_short_range_average and self.last_long_range_average:
                gold_x = (self.last_short_range_average<self.last_long_range_average) and (self.current_short_range_average>self.current_long_range_average)
                dead_x = (self.last_short_range_average>self.last_long_range_average) and (self.current_short_range_average<self.current_long_range_average)
                upper_towards = self.last_short_range_average<self.current_short_range_average and self.last_long_range_average<self.current_long_range_average
                down_towards = self.last_short_range_average>self.current_short_range_average and self.last_long_range_average>self.current_long_range_average
                
                if gold_x and upper_towards:
                    return "buy"
                if  dead_x and down_towards:
                    return "sell"
                else:
                    return "look"
        self.last_short_range_average = self.current_short_range_average
        self.last_long_range_average = self.current_long_range_average
        return "look"

5. 测试交易策略：

首先需要将策略注册入交易环境：

model = Model()
model.register(Awlays_buying,"Awlays_buying")
model.register(Moving_average,"Moving_average",short_range_count=20,long_range_count=40)
model.register(Enforce_Awlays_buying,"Enforce_Awlays_buying")

逐个测试每个策略：

model.run("Awlays_buying")
model.plot()
print(model.evalult())
print(model.current_money)

model.run("Enforce_Awlays_buying")
model.plot()
print(model.evalult())
print(model.current_money)

model.run("Moving_average")
model.plot()
print(model.evalult())
print(model.current_money)