import requests import json from datetime import datetime, timedelta, timezone import pytz # helper function to pull most recent chart data on failure def pull_last_from_file(): fd = open('last_chart_data.json', 'r') d = json.loads(fd.read()) fd.close() return d def update_last_file(data): fd = open('last_chart_data.json', 'w') fd.truncate(0) fd.write(json.dumps(data)) fd.close() """ Given the parameters, fetches the data for the corresponding chart using yahoo finance. If bad request, returns the previous chart. """ def fetch_chart_data(ticker, period='5d', interval='1m'): params = { 'interval' : interval, # 1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 4h, 1d, 5d, 1wk, 1mo, 3mo 'range' : period, # "1d","5d","1mo","3mo","6mo","1y","2y","5y","10y","ytd","max" 'events' : 'div|split|earn', 'includePrePost' : 'false' } headers = {'User-agent' : 'fin-backtesting-proj'} r = requests.get("https://query1.finance.yahoo.com/v8/finance/chart/" + ticker, headers=headers, params=params) print(r.url) print("status_code:\t", r.status_code) # decode the JSON response data into a Python object try: r.raise_for_status() # raises if error before parsing except: last_data = pull_last_from_file() last_data['error'] = True return last_data data_obj = r.json() # get the specific data we want if 'timestamp' not in data_obj['chart']['result'][0]: last_data = pull_last_from_file() last_data['error'] = True return last_data timestamps = data_obj['chart']['result'][0]['timestamp'] close_prices = data_obj['chart']['result'][0]['indicators']['quote'][0]['close'] # clean out null's and 0s from the data i = 0 while i < len(timestamps): if close_prices[i] == None or close_prices[i] == 0: del close_prices[i] del timestamps[i] i -= 1 i += 1 name = data_obj['chart']['result'][0]['meta']['longName'] data = {'timestamps': timestamps, 'prices': close_prices, 'name': name, 'error': False} # update_last_file(data) # save data to file in case necessary return data """ Given the parameters, fetches the data for the corresponding chart using yahoo finance. Cosumes period_length as a timedelta and period_end_date as a date If bad request, returns the previous chart. """ def fetch_chart_data_backtest(ticker='XRP/USD', interval='1Min', period_end_date=None, period_length=None): if period_end_date == None: period_end_date = datetime.now() if period_length == None: period_length = timedelta(days=5) period_end_date = period_end_date.astimezone(pytz.timezone('UTC')) # cast to int to truncate the decimal period2 = datetime.isoformat(period_end_date) # find the first period via subtracting the period length period1 = datetime.isoformat(period_end_date - period_length) print(period2, period1) params = { 'symbols' : ticker, 'timeframe' : interval, 'start' : period1, 'end' : period2, 'limit': 10000, 'sort' : 'asc', } headers = {"accept": "application/json"} url = "https://data.alpaca.markets/v1beta3/crypto/us/bars" # r = requests.get(url, headers=headers, params=params) r = requests.get(url, params=params, headers=headers) print(r.url) print("status_code:\t", r.status_code) # print(r.text) # decode the JSON response data into a Python object try: r.raise_for_status() # raises if error before parsing except: print(r.text) last_data = pull_last_from_file() last_data['error'] = True return last_data data_obj = r.json() res_headers = r.headers print(res_headers) timestamps = [] close_prices = [] assert ticker in data_obj['bars'] chart_data = data_obj['bars'][ticker] for tmp in chart_data: # t, o, h, l, c, v, n, vw (keys for data) timezone_adjusted = datetime.fromisoformat(tmp['t'].replace("Z", "")).astimezone(pytz.timezone('US/Eastern')) timestamps.append(timezone_adjusted.timestamp()) close_prices.append(tmp['c']) # clean out null's and 0s from the data # i = 0 # while i < len(timestamps): # if close_prices[i] == None or close_prices[i] == 0: # del close_prices[i] # del timestamps[i] # i -= 1 # i += 1 name = ticker data = {'timestamps': timestamps, 'prices': close_prices, 'name': name, 'error': False} update_last_file(data) # save data to file in case necessary return data def fetch_chart_data_yahoo(ticker='ADA-USD', interval='1m', period_end_date=None, period_length=None): if period_end_date == None: period_end_date = datetime.now() if period_length == None: period_length = timedelta(days=5) # cast to int to truncate the decimal period2 = int(datetime.timestamp(period_end_date)) # find the first period via subtracting the period length period1 = int(datetime.timestamp(period_end_date - period_length)) print(datetime.isoformat(datetime.fromtimestamp(period2)), datetime.isoformat(datetime.fromtimestamp(period1))) params = { 'period1' : period1, # the start date (in epoch time) 'period2' : period2, # the end time (in epoch time) 'interval' : interval, # 1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 4h, 1d, 5d, 1wk, 1mo, 3mo 'events' : 'div|split|earn', 'includePrePost' : 'false', 'lang' : 'en-US', 'region' : 'US' } headers = {'User-agent' : 'fin-backtesting-proj'} r = requests.get("https://query2.finance.yahoo.com/v8/finance/chart/" + ticker, headers=headers, params=params) print(r.url) print("status_code:\t", r.status_code) # decode the JSON response data into a Python object try: r.raise_for_status() # raises if error before parsing except: print(r.text) last_data = pull_last_from_file() last_data['error'] = True return last_data data_obj = r.json() # get the specific data we want if 'timestamp' not in data_obj['chart']['result'][0]: last_data = pull_last_from_file() last_data['error'] = True return last_data timestamps = data_obj['chart']['result'][0]['timestamp'] close_prices = data_obj['chart']['result'][0]['indicators']['quote'][0]['close'] # clean out null's and 0s from the data i = 0 while i < len(timestamps): if close_prices[i] == None or close_prices[i] == 0: del close_prices[i] del timestamps[i] i -= 1 i += 1 name = data_obj['chart']['result'][0]['meta']['longName'] data = {'timestamps': timestamps, 'prices': close_prices, 'name': name, 'error': False} # save data to file in case necessary return data def test(): fetch_chart_data_backtest() test()