基于策略设计模式的波动市场交易模型：
- 按照策略设计模式至少需要需要以下几个类：
- 环境类 包裹数据与策略池；
- 数据类；
- 策略类及其基类；
- 环境类设计：
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()
- 策略基类：
class Strategy(ABC):
@abstractclassmethod
def action_signal(self,data):
pass
- 基于策略基类的交易策略：
三个交易策略，就一直写自己的交易策略就好了：
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"
- 测试交易策略：
首先需要将策略注册入交易环境：
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)
