-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathutility.py
206 lines (154 loc) · 5.44 KB
/
utility.py
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
# -*- coding:utf-8 -*-
from __future__ import division
import datetime
import json
import requests
from requests.adapters import HTTPAdapter
from urllib3.util import Retry
import gugu.config as cf
class Utility():
@staticmethod
def str2Dict(string):
string = eval(string, type('Dummy', (dict,), dict(__getitem__ = lambda s, n:n))())
string = json.dumps(string)
return json.loads(string)
@staticmethod
def fceil(number, ndigits=2):
if number == 0:
return number
tail_number = 1 / pow(10, ndigits)
ndigits_number = round(number, ndigits)
result = number - ndigits_number
if result == 0:
return number
elif result > 0:
return ndigits_number + tail_number
else:
return ndigits_number
@staticmethod
def checkQuarter(year, quarter):
if isinstance(year, str) or year < 1989 :
raise TypeError(cf.DATE_CHK_MSG)
elif quarter is None or isinstance(quarter, str) or quarter not in [1, 2, 3, 4]:
raise TypeError(cf.DATE_CHK_Q_MSG)
else:
return True
@staticmethod
def checkLhbInput(last):
if last not in [5, 10, 30, 60]:
raise TypeError(cf.LHB_MSG)
else:
return True
@staticmethod
def symbol(code):
"""
生成symbol代码标志
"""
if code in cf.INDEX_LABELS:
return cf.INDEX_LIST[code]
elif len(code) != 6 :
return ''
else:
return 'sh%s' % code if code[:1] in ['5', '6', '9'] else 'sz%s' % code
@staticmethod
def random(n=13):
from random import randint
start = 10**(n-1)
end = (10**n)-1
return str(randint(start, end))
@staticmethod
def getToday():
return str(datetime.datetime.today().date())
@staticmethod
def getHour():
return datetime.datetime.today().hour
@staticmethod
def getMonth():
return datetime.datetime.today().month
@staticmethod
def getYear():
return datetime.datetime.today().year
@staticmethod
def getTodayLastYear():
return str(Utility.getDateInterval(days=-365))
@staticmethod
def getTodayLastMonth():
return str(Utility.getDateInterval(days=-30))
@staticmethod
def getTodayLastWeek():
return str(Utility.getDateInterval(days=-7))
@staticmethod
def getDateInterval(date=str(datetime.datetime.today().date()), days=-1):
return datetime.datetime.strptime(date, '%Y-%m-%d').date() + datetime.timedelta(days)
@staticmethod
def diffDays(start=None, end=None):
d1 = datetime.datetime.strptime(end, '%Y-%m-%d')
d2 = datetime.datetime.strptime(start, '%Y-%m-%d')
delta = d1 - d2
return delta.days
@staticmethod
def ttDates(start='', end=''):
startyear = int(start[0:4])
endyear = int(end[0:4])
dates = [d for d in range(startyear, endyear+1, 2)]
return dates
@staticmethod
def lastTradeDate(date=str(datetime.datetime.today().date())):
lastdate = str(Utility.getDateInterval(date, -1))
if Utility.isTradeDay(lastdate):
return lastdate
else:
return Utility.lastTradeDate(lastdate)
@staticmethod
def nextTradeDate(date=str(datetime.datetime.today().date())):
nextdate = str(Utility.getDateInterval(date, 1))
if Utility.isTradeDay(nextdate):
return nextdate
else:
return Utility.nextTradeDate(nextdate)
@staticmethod
def isHoliday(date=str(datetime.datetime.today().date())):
"""
节假日判断
Parameters
------
date: string
查询日期 format:YYYY-MM-DD 为空时取当前日期
return
------
True or False
"""
session = requests.Session()
retry = Retry(connect=5, backoff_factor=1)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.keep_alive = False
try:
request = session.get( cf.HOLIDAY_URL % date, timeout=10 )
dataDict = json.loads(request.text)
if dataDict['code'] != 0:
raise IOError(cf.HOLIDAY_SERVE_ERR)
elif dataDict['holiday'] is None:
return False
else:
return True
except Exception as e:
print(e)
@staticmethod
def isTradeDay(date=str(datetime.datetime.today().date())):
"""
交易日判断
Parameters
------
date: string
查询日期 format:YYYY-MM-DD 为空时取当前日期
return
------
True or False
"""
date = datetime.datetime.strptime(date, '%Y-%m-%d')
w = int(date.strftime('%w'))
if w in [6, 0]:
return False
else:
return not Utility.isHoliday(date)