在FOMC目标利率变化之前的美元

1from AlgorithmImports import *
2from pandas.tseries.offsets import BDay
3from dateutil.relativedelta import relativedelta
4#endregion
5class TheDollarAheadOfFOMCTargetRateChanges(QCAlgorithm):
6def initialize(self) -> None:
7 self.set_start_date(2000, 1, 1)
8 self.set_cash(100000)
9 self.spread:float|None = None
10 self.last_target_rate:float|None = None
11 self.last_target_rate_update:datetime.date|None = None
12 self.max_missing_days_target_rate:int = 5
13 self.days_counter:int = 0
14 self.holding_period: int = 3
15 self.spread_threshold: float = 0.125
16 self.leverage:int = 5
17 data = self.add_data(QuantpediaFutures, 'ICE_DX1', Resolution.DAILY)
18 data.set_fee_model(CustomFeeModel())
19 data.set_leverage(self.leverage)
20 
21 self.symbol:Symbol = data.symbol
22 self.fed_fund_futures:Symbol = self.add_data(NasdaqFedFuture, 'CHRIS/CME_FF3', Resolution.DAILY).symbol
23 self.fed_fund_rate:Symbol = self.add_data(FedRate, 'FEDFUNDS', Resolution.DAILY).symbol
24 
25 csv_string_file:str = self.download('data.quantpedia.com/backtesting_data/economic/fed_days.csv')
26 dates:list[str] = csv_string_file.split('\r\n')
27 fed_dates:list[datetime.date] = [datetime.strptime(x, '%Y-%m-%d') for x in dates]
28 self.three_days_before_fed:list[datetime.date] = [(x - BDay(3)).date() for x in fed_dates]
29def on_data(self, data: Slice) -> None:
30 curr_date:datetime.date = self.time.date()
31 custom_data_last_update_date: Dict[Symbol, datetime.date] = LastDateHandler.get_last_update_date()
32 if all((self.securities[symbol].get_last_data() and self.time.date() > custom_data_last_update_date[symbol]) for symbol in [self.symbol, self.fed_fund_rate]):
33     self.liquidate()
34     return
35 # coming on daily basis
36 if self.fed_fund_futures in data and data[self.fed_fund_futures]:
37     fed_fund_futures_value = data[self.fed_fund_futures].value
38     self.last_target_rate = 100 - fed_fund_futures_value
39     self.last_target_rate_update = curr_date
40 # coming on monthly basis
41 if self.fed_fund_rate in data and data[self.fed_fund_rate]:
42     self.last_fed_rate = data[self.fed_fund_rate].Value
43 if self.last_target_rate_update != None and (curr_date - self.last_target_rate_update).days > self.max_missing_days_target_rate:
44     self.last_target_rate = None
45     self.last_target_rate_update = None
46 # Three days before fed calculate expected target rate and spread
47 if curr_date in self.three_days_before_fed and self.last_target_rate != None and self.last_fed_rate != None:
48     self.spread = self.last_target_rate - self.last_fed_rate
49 
50 if self.symbol in data and data[self.symbol]:
51     # Buy dollar future two days before fed. Emit insight and let it expire (liquidate) by itself.
52     if self.spread:
53         if self.spread >= self.spread_threshold:
54             self.set_holdings(self.symbol, 1)
55         elif self.spread <= -self.spread_threshold:
56             self.set_holdings(self.symbol, -1)
57     
58         self.spread = None
59 if self.portfolio.invested:
60     self.days_counter += 1
61 if self.days_counter >= self.holding_period:
62     self.days_counter = 0
63     self.liquidate()
64class NasdaqFedFuture(NasdaqDataLink):
65def __init__(self) -> None:
66 self.ValueColumnName = "Value"
67 
68class LastDateHandler():
69_last_update_date:Dict[Symbol, datetime.date] = {}
70
71@staticmethod
72def get_last_update_date() -> Dict[Symbol, datetime.date]:
73return LastDateHandler._last_update_date
74 
75# Quantpedia data.
76# NOTE: IMPORTANT: Data order must be ascending (datewise)
77class QuantpediaFutures(PythonData):
78def GetSource(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
79 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/futures/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
80def Reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData:
81 data = QuantpediaFutures()
82 data.Symbol = config.Symbol
83 
84 if not line[0].isdigit(): return None
85 split = line.split(';')
86 
87 data.Time = datetime.strptime(split[0], "%d.%m.%Y") + timedelta(days=1)
88 data['back_adjusted'] = float(split[1])
89 data['spliced'] = float(split[2])
90 data.Value = float(split[1])
91 if config.Symbol not in LastDateHandler._last_update_date:
92     LastDateHandler._last_update_date[config.Symbol] = datetime(1,1,1).date()
93 if data.Time.date() > LastDateHandler._last_update_date[config.Symbol]:
94     LastDateHandler._last_update_date[config.Symbol] = data.Time.date()
95 return data
96# Quantpedia monthly custom data.
97# NOTE: IMPORTANT: Data order must be ascending (datewise)
98class FedRate(PythonData):
99def GetSource(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
100 return SubscriptionDataSource(f'data.quantpedia.com/backtesting_data/economic/{config.Symbol.Value}.csv', SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
101def Reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData:
102 data = FedRate()
103 data.Symbol = config.Symbol
104 
105 if not line[0].isdigit(): return None
106 split: str = line.split(';')
107 
108 data.Time = datetime.strptime(split[0], "%Y-%m-%d") + relativedelta(months=1)
109 data.Value = float(split[1])
110 if config.Symbol not in LastDateHandler._last_update_date:
111     LastDateHandler._last_update_date[config.Symbol] = datetime(1,1,1).date()
112 if data.Time.date() > LastDateHandler._last_update_date[config.Symbol]:
113     LastDateHandler._last_update_date[config.Symbol] = data.Time.date()
114 return data
115# Custom fee model
116class CustomFeeModel(FeeModel):
117def GetOrderFee(self, parameters):
118 fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
119 return OrderFee(CashAmount(fee, "USD"))