-
Notifications
You must be signed in to change notification settings - Fork 1
/
IBUtils.py
32 lines (30 loc) · 1.46 KB
/
IBUtils.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
import pandas as pd
import warnings
def filter_US_conids(search_results):
# Select the primary US conid from a list of lists, returning None for errors
US_exchanges = ['NYSE', 'NASDAQ', 'AMEX', 'ARCA', 'BATS']
results = [None if len(j) == 0 else j[0] for j in
[[el['conid'] for el in search_results[i] if el != 'error' and el['description'] in US_exchanges]
for i in range(len(search_results)) if search_results[i] is not None]]
errors = [search_results[i][0]['symbol'] for i, t in enumerate(results) if t is None]
if len(errors) > 0:
warnings.warn(f'None objects for tickers {errors}.')
return results
def data_bars_to_df(bars):
# Takes a market_data_history() response and converts it to a dataframe
# Remove any None responses and warn
errors = [i for i, t in enumerate(bars) if t is None]
if len(errors) > 0:
warnings.warn(f'None objects in locations {errors}.')
bars = [b for b in bars if b is not None]
if len(bars) == 0:
return None
# Create a dictionary that's almost the multi-indexed df we want
bar_dict = {(pd.to_datetime(j['t'], utc=True, unit='ms'), i['symbol']):
[j['o'], j['h'], j['l'], j['c'], j['v']]
for i in bars for j in i['data']}
# Put it into a df
idx = pd.MultiIndex.from_tuples(bar_dict.keys())
cols = ['Open', 'High', 'Low', 'Close', 'Volume']
df = pd.DataFrame(bar_dict.values(), index=idx, columns=cols)
return(df)