利率动量预测外汇汇率策略

1import data_tools
2from AlgorithmImports import *
3import numpy as np
4class InterestRatesMomentumPredictsFXRates(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100000)
8
9# Country symbol and currency future symbol.
10self.symbols = {
11                "USD" : "not_traded", # US Dollar index Futures, Continuous Contract #1
12                "EUR" : "CME_EC1", # Euro FX Futures, Continuous Contract #1
13                "GBP" : "CME_BP1", # British Pound Futures, Continuous Contract #1
14                "CHF" : "CME_SF1", # Swiss Franc Futures, Continuous Contract #1
15                "JPY" : "CME_JY1", # Japanese Yen Futures, Continuous Contract #1
16                }
17# Interest rate data.
18self.interest_rate = self.AddData(data_tools.InterestRate, 'InterestRate', Resolution.Daily).Symbol
19
20self.period = 15
21self.yield_difference = {}
22
23countries = [x[0] for x in self.symbols.items()]
24for i, country1 in enumerate(countries):
25    for j, country2 in enumerate(countries):
26        if i <= j: continue
27        self.yield_difference[country1 + country2] = RollingWindow[float](self.period)
28for country, currency_future in self.symbols.items():
29    # Currency futures data.
30    data = self.AddData(data_tools.QuantpediaFutures, currency_future, Resolution.Daily)
31    data.SetLeverage(10)
32    data.SetFeeModel(data_tools.CustomFeeModel())
33
34def OnData(self, data):
35if not self.Securities.ContainsKey(self.interest_rate): return
36
37interest_rate_date = self.Securities[self.interest_rate].GetLastData()
38if not interest_rate_date: return
39
40# make sure interest data is still comming in
41if (self.Time.date() - interest_rate_date.Time.date()).days >= 5:
42    self.Liquidate()
43    return
44
45countries = [x[0] for x in self.symbols.items()]
46signal = {}
47for i, country1 in enumerate(countries):
48    sub_signal = {}
49    for j, country2 in enumerate(countries):
50        if i <= j: continue
51    
52        index = country1 + country2
53        yield1 = interest_rate_date[country1]
54        yield2 = interest_rate_date[country2]
55        diff = yield1 - yield2
56        
57        self.yield_difference[index].Add(diff)
58        
59        if self.yield_difference[index].IsReady:
60            avg_diff = np.mean([x for x in self.yield_difference[index]])
61            sub_signal[index] = (diff - avg_diff) / abs(avg_diff)
62    
63    if len(sub_signal) > 0:
64        abs_percentile = np.percentile([abs(x[1]) for x in sub_signal.items()], 50)
65        
66        for signal_index, sig in sub_signal.items():
67            iter_country1 = signal_index[:3]
68            iter_country2 = signal_index[-3:]
69            if abs(sig) > abs_percentile:
70                if iter_country1 != 'USD':
71                    iter_future1 = self.symbols[iter_country1]
72                    if iter_future1 not in signal:
73                        signal[iter_future1] = 0
74                    signal[iter_future1] += np.sign(sig)
75                    
76                if iter_country2 != 'USD':
77                    iter_future2 = self.symbols[iter_country2]
78                    if iter_future2 not in signal:
79                        signal[iter_future2] = 0
80                    signal[iter_future2] -= np.sign(sig)
81
82if len(signal) != 0:
83    futures_invested = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
84    for currency_future in futures_invested:
85        if currency_future not in signal:
86            self.Liquidate(currency_future)
87    
88    foo = 3
89    
90    for currency_future, country_signal in signal.items():
91        self.SetHoldings(currency_future, country_signal)