基于自适应移动平均线的市场时机策略

1import numpy as np
2from AlgorithmImports import *
3class AdaptiveMovingAveragesMarketTiming(QCAlgorithm):
4def Initialize(self):
5self.SetStartDate(2010, 1, 1)
6self.SetCash(100000)
7
8data = self.AddEquity('SPY', Resolution.Daily)
9data.SetLeverage(5)
10self.symbol = data.Symbol
11self.period = 4 * 12 * 21
12self.SetWarmUp(self.period, Resolution.Daily)
13self.data = RollingWindow[float](self.period)
14
15self.ma = {}
16self.ma_signal = {}
17
18sma_periods = [x for x in range(1, 101, 4)]
19lma_periods = [x for x in range(5, 991, 20)]
20
21ma_combinations = [[i, j] for i in sma_periods for j in lma_periods if i<j]
22for sma, lma in ma_combinations:
23    self.ma[sma,lma] = [self.SMA(self.symbol, sma, Resolution.Daily), self.SMA(self.symbol, lma, Resolution.Daily)]
24    self.ma_signal[sma,lma] = RollingWindow[float](self.period)
25
26def OnData(self, data):
27# Update SPY price every day.
28symbol_obj = self.Symbol(self.symbol)
29if symbol_obj in data and data[symbol_obj]:
30    self.data.Add(data[symbol_obj].Value)
31
32# Store vector value for current day and optimize from previous values.
33avg_return = {}
34for sma_lma in self.ma:
35    sma = self.ma[sma_lma][0]
36    lma = self.ma[sma_lma][1]
37    if sma.IsReady and lma.IsReady:
38        sma_value = sma.Current.Value
39        lma_value = lma.Current.Value
40        
41        # MA SIGNAL = if short term MA is OVER long term MA == 1 else -1
42        self.ma_signal[sma_lma].Add(1 if sma_value > lma_value else -1)
43        
44        # 4 years of SPY data is ready.
45        if self.data.IsReady:
46            values = np.array([x for x in self.data])
47            daily_changes = values[:-1] / values[1:] - 1
48        
49            # Find optimal sma_lma pair.
50            if self.ma_signal[sma_lma].IsReady:
51                # Multiply both vectors to get daily performance for sma_lma pair.
52                # Ignore last value from ma_signal since it's today's value. It will be used to trade decision for next day.
53                ma_signal_vector = [x for x in self.ma_signal[sma_lma]][1:]
54                return_vector = ma_signal_vector * daily_changes
55                
56                # Store avg daily performance for sma_lma pair.
57                avg_return[sma_lma] = np.average([x for x in return_vector])
58        
59    else:
60        # MA is not ready yet.
61        self.ma_signal[sma_lma].Add(0)
62if self.IsWarmingUp: return
63if len(avg_return) == 0: return
64
65# Optimalization
66optimal_sma_lma = max(avg_return, key=avg_return.get)
67
68# Trading
69last_signal = self.ma_signal[optimal_sma_lma][0]
70if last_signal == 1:
71    self.SetHoldings(self.symbol, 1)
72elif last_signal == -1:
73    self.SetHoldings(self.symbol, -1)