-
Notifications
You must be signed in to change notification settings - Fork 673
/
stats.py
144 lines (110 loc) · 4.87 KB
/
stats.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
from datetime import datetime, timedelta
import pandas as pd
from pandas_datareader.exceptions import UnstableAPIWarning
from pandas_datareader.iex import IEX
# Data provided for free by IEX
# Data is furnished in compliance with the guidelines promulgated in the IEX
# API terms of service and manual
# See https://iextrading.com/api-exhibit-a/ for additional information
# and conditions of use
class DailySummaryReader(IEX):
def __init__(self, symbols=None, start=None, end=None, retry_count=3,
pause=0.001, session=None):
import warnings
warnings.warn('Daily statistics is not working due to issues with the IEX API',
UnstableAPIWarning)
self.curr_date = start
super(DailySummaryReader, self).__init__(symbols=symbols,
start=start, end=end,
retry_count=retry_count,
pause=pause, session=session)
@property
def service(self):
return "stats/historical/daily"
def _get_params(self, symbols):
p = {}
if self.curr_date is not None:
p['date'] = self.curr_date.strftime('%Y%m%d')
return p
def read(self):
"""Unfortunately, IEX's API can only retrieve data one day or one month
at a time. Rather than specifying a date range, we will have to run
the read function for each date provided.
:return: DataFrame
"""
tlen = self.end - self.start
dfs = []
for date in (self.start + timedelta(n) for n in range(tlen.days)):
self.curr_date = date
tdf = super(IEX, self).read()
dfs.append(tdf)
return pd.concat(dfs)
class MonthlySummaryReader(IEX):
def __init__(self, symbols=None, start=None, end=None, retry_count=3,
pause=0.001, session=None):
self.curr_date = start
self.date_format = '%Y%m'
super(MonthlySummaryReader, self).__init__(symbols=symbols,
start=start, end=end,
retry_count=retry_count,
pause=pause,
session=session)
@property
def service(self):
return "stats/historical"
def _get_params(self, symbols):
p = {}
if self.curr_date is not None:
p['date'] = self.curr_date.strftime(self.date_format)
return p
def read(self):
"""Unfortunately, IEX's API can only retrieve data one day or one month
at a time. Rather than specifying a date range, we will have to run
the read function for each date provided.
:return: DataFrame
"""
tlen = self.end - self.start
dfs = []
# Build list of all dates within the given range
lrange = [x for x in (self.start + timedelta(n)
for n in range(tlen.days))]
mrange = []
for dt in lrange:
if datetime(dt.year, dt.month, 1) not in mrange:
mrange.append(datetime(dt.year, dt.month, 1))
lrange = mrange
for date in lrange:
self.curr_date = date
tdf = super(IEX, self).read()
# We may not return data if this was a weekend/holiday:
if not tdf.empty:
tdf['date'] = date.strftime(self.date_format)
dfs.append(tdf)
# We may not return any data if we failed to specify useful parameters:
return pd.concat(dfs) if len(dfs) > 0 else pd.DataFrame()
class RecordsReader(IEX):
def __init__(self, symbols=None, start=None, end=None, retry_count=3,
pause=0.001, session=None):
super(RecordsReader, self).__init__(symbols=symbols,
start=start, end=end,
retry_count=retry_count,
pause=pause, session=session)
@property
def service(self):
return "stats/records"
def _get_params(self, symbols):
# Record Stats API does not take any parameters, returning empty dict
return {}
class RecentReader(IEX):
def __init__(self, symbols=None, start=None, end=None, retry_count=3,
pause=0.001, session=None):
super(RecentReader, self).__init__(symbols=symbols,
start=start, end=end,
retry_count=retry_count,
pause=pause, session=session)
@property
def service(self):
return "stats/recent"
def _get_params(self, symbols):
# Record Stats API does not take any parameters, returning empty dict
return {}