1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
from abc import ABC
import pandas as pd
from dateutil.relativedelta import relativedelta
from py_jftech import get_config, autowired, component
from api import PortfoliosRisk, SignalType, Navs
from rebalance.base_signal import BaseRebalanceSignal
from rebalance.dao import robo_rebalance_signal as rrs
class CrisisSignal(BaseRebalanceSignal, ABC):
@autowired
def __init__(self, navs: Navs = None):
super().__init__()
self._navs = navs
self._config = get_config(__name__)
@property
def exp_init(self):
return pd.to_datetime(self._config['exp-init']) if 'exp-init' in self._config else None
@property
def exp_years(self):
return self._config['exp-years'] if 'exp-years' in self._config else 1
@property
def inversion_years(self):
return self._config['inversion-years'] if 'inversion-years' in self._config else 1
@property
def inversion_threshold(self):
return self._config['inversion-threshold'] if 'inversion-threshold' in self._config else 0.3
def get_exp_start_date(self, day, risk: PortfoliosRisk):
assert day, "get crisis exp start date, day can not be none"
assert risk, "get crisis exp start date, PortfoliosRisk can not be none"
exp_date = day - relativedelta(years=self.exp_years)
if self.exp_init and self.exp_init >= exp_date:
return self.exp_init
exp_signal = rrs.get_first_after(type=SignalType.CRISIS_EXP, risk=risk, min_date=exp_date)
if not exp_signal:
inversion_date = day - relativedelta(years=self.inversion_years)
ten_before = self._navs.get_last_index_close(max_date=inversion_date, ticker='USGG10YR Index')
ten_today = self._navs.get_last_index_close(max_date=day, ticker='USGG10YR Index')
two_before = self._navs.get_last_index_close(max_date=inversion_date, ticker='USGG2YR Index')
two_today = self._navs.get_last_index_close(max_date=day, ticker='USGG2YR Index')
if ten_today['close'] - two_today['close'] <= ten_before['close'] - two_before['close'] and \
ten_today['close'] - two_today['close'] <= self.inversion_threshold:
rrs.insert({
'date': day,
'type': SignalType.CRISIS_EXP,
'risk': risk,
})
exp_signal = rrs.get_first_after(type=SignalType.CRISIS_EXP, risk=risk, min_date=exp_date)
return exp_signal['date'] if exp_signal else None
@component(bean_name='crisis_one')
class CrisisOneSignal(CrisisSignal, BaseRebalanceSignal):
@property
def consecut_days(self):
return self._config['crisis-1']['consecut-days']
@property
def mean_count(self):
return self._config['crisis-1']['mean-count']
@property
def signal_type(self):
return SignalType.CRISIS_ONE
def is_trigger(self, day, risk: PortfoliosRisk) -> bool:
exp_date = self.get_exp_start_date(day, risk)
if exp_date:
crisis_one = rrs.get_first_after(type=SignalType.CRISIS_ONE, risk=risk, min_date=exp_date)
if not crisis_one:
spx = self._navs.get_last_index_close(max_date=day, ticker='SPX Index', count=self.mean_count)
spx_ma850 = pd.DataFrame(spx).close.mean()
return len([x for x in spx[0:5] if x['close'] > spx_ma850]) == 0
return False
@component(bean_name='crisis_two')
class CrisisTwoSignal(CrisisSignal, BaseRebalanceSignal):
@property
def negative_growth_years(self):
return self._config['crisis-2']['negative-growth']
@property
def fed_months(self):
return self._config['crisis-2']['fed-months']
@property
def fed_threshold(self):
return self._config['crisis-2']['fed-threshold']
@property
def signal_type(self):
return SignalType.CRISIS_TWO
def is_trigger(self, day, risk: PortfoliosRisk) -> bool:
exp_date = self.get_exp_start_date(day, risk)
if exp_date:
crisis_two = rrs.get_first_after(type=SignalType.CRISIS_TWO, risk=risk, min_date=exp_date)
if not crisis_two:
ng_date = day - relativedelta(years=self.negative_growth_years)
ten_today = self._navs.get_last_index_close(max_date=day, ticker='USGG10YR Index')
cpi_today = self._navs.get_last_eco_values(max_date=day, ticker='CPI YOY Index', by_release_date=True)
ten_before = self._navs.get_last_index_close(max_date=ng_date, ticker='USGG10YR Index')
cpi_before = self._navs.get_last_eco_values(max_date=ng_date, ticker='CPI YOY Index', by_release_date=True)
before = ten_before['close'] - cpi_before['indicator']
today = ten_today['close'] - cpi_today['indicator']
fed_today = self._navs.get_last_eco_values(max_date=day, ticker='FDTR Index', by_release_date=True)
fed_before = self._navs.get_last_eco_values(max_date=day - relativedelta(months=self.fed_months), ticker='FDTR Index', by_release_date=True)
return today <= before and fed_today['indicator'] - fed_before['indicator'] < self.fed_threshold
return False