-
Notifications
You must be signed in to change notification settings - Fork 673
/
daily.py
169 lines (153 loc) · 5.35 KB
/
daily.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
import datetime
import json
import os
from dateutil.relativedelta import relativedelta
import pandas as pd
from pandas_datareader.base import _DailyBaseReader
# 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 IEXDailyReader(_DailyBaseReader):
"""
Returns DataFrame of historical stock prices
from symbols, over date range, start to end. To avoid being penalized by
IEX servers, pauses between downloading 'chunks' of symbols can be
specified.
Parameters
----------
symbols : string, array-like object (list, tuple, Series), or DataFrame
Single stock symbol (ticker), array-like object of symbols or
DataFrame with index containing stock symbols.
start : string, int, date, datetime, Timestamp
Starting date. Parses many different kind of date
representations (e.g., 'JAN-01-2010', '1/1/10', 'Jan, 1, 1980'). Defaults to
15 years before current date
end : string, int, date, datetime, Timestamp
Ending date
retry_count : int, default 3
Number of times to retry query request.
pause : int, default 0.1
Time, in seconds, to pause between consecutive queries of chunks. If
single value given for symbol, represents the pause between retries.
chunksize : int, default 25
Number of symbols to download consecutively before intiating pause.
session : Session, default None
requests.sessions.Session instance to be used
api_key: str
IEX Cloud Secret Token
"""
def __init__(
self,
symbols=None,
start=None,
end=None,
retry_count=3,
pause=0.1,
session=None,
chunksize=25,
api_key=None,
):
if api_key is None:
api_key = os.getenv("IEX_API_KEY")
if not api_key or not isinstance(api_key, str):
raise ValueError(
"The IEX Cloud API key must be provided either "
"through the api_key variable or through the "
" environment variable IEX_API_KEY"
)
# Support for sandbox environment (testing purposes)
if os.getenv("IEX_SANDBOX") == "enable":
self.sandbox = True
else:
self.sandbox = False
self.api_key = api_key
super(IEXDailyReader, self).__init__(
symbols=symbols,
start=start,
end=end,
retry_count=retry_count,
pause=pause,
session=session,
chunksize=chunksize,
)
@property
def default_start_date(self):
today = datetime.date.today()
return today - datetime.timedelta(days=365 * 15)
@property
def url(self):
"""API URL"""
if self.sandbox is True:
return "https://sandbox.iexapis.com/stable/stock/market/batch"
else:
return "https://cloud.iexapis.com/stable/stock/market/batch"
@property
def endpoint(self):
"""API endpoint"""
return "chart"
def _get_params(self, symbol):
chart_range = self._range_string_from_date()
if isinstance(symbol, list):
symbolList = ",".join(symbol)
else:
symbolList = symbol
params = {
"symbols": symbolList,
"types": self.endpoint,
"range": chart_range,
"token": self.api_key,
}
return params
def _range_string_from_date(self):
delta = relativedelta(self.start, datetime.datetime.now())
years = delta.years * -1
if 5 <= years <= 15:
return "max"
if 2 <= years < 5:
return "5y"
elif 1 <= years < 2:
return "2y"
elif 0 <= years < 1:
delta_days = (datetime.datetime.now() - self.start).days
if 0 <= delta_days < 6:
return "5d"
elif 6 <= delta_days < 28:
return "1m"
elif 28 <= delta_days < 84:
return "3m"
elif 84 <= delta_days < 168:
return "6m"
return "1y"
else:
raise ValueError("Invalid date specified. Must be within past 15 years.")
def read(self):
"""Read data"""
try:
return self._read_one_data(self.url, self._get_params(self.symbols))
finally:
self.close()
def _read_lines(self, out):
data = out.read()
json_data = json.loads(data)
result = {}
if type(self.symbols) is str:
syms = [self.symbols]
else:
syms = self.symbols
for symbol in syms:
d = json_data.pop(symbol)["chart"]
df = pd.DataFrame(d)
df.set_index("date", inplace=True)
values = ["open", "high", "low", "close", "volume"]
df = df[values]
sstart = self.start.strftime("%Y-%m-%d")
send = self.end.strftime("%Y-%m-%d")
df = df.loc[sstart:send]
result.update({symbol: df})
if len(result) > 1:
result = pd.concat(result).unstack(level=0)
result.columns.names = ["Attributes", "Symbols"]
return result
return result[self.symbols]