Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def merge_arrays(arrays):
parsed_data = DataCollectorParser.get_empty_parsed_data()
ohlcv_data = parsed_data[BACKTESTING_DATA_OHLCV]
for time_frame in arrays:
data = arrays[time_frame]
ohlcv_data[time_frame] = []
for i in range(len(data[PriceIndexes.IND_PRICE_TIME.value])):
ohlcv_data[time_frame].insert(i, [None]*len(PriceIndexes))
ohlcv_data[time_frame][i][PriceIndexes.IND_PRICE_CLOSE.value] = \
data[PriceIndexes.IND_PRICE_CLOSE.value][i]
ohlcv_data[time_frame][i][PriceIndexes.IND_PRICE_OPEN.value] = \
data[PriceIndexes.IND_PRICE_OPEN.value][i]
ohlcv_data[time_frame][i][PriceIndexes.IND_PRICE_HIGH.value] = \
data[PriceIndexes.IND_PRICE_HIGH.value][i]
ohlcv_data[time_frame][i][PriceIndexes.IND_PRICE_LOW.value] = \
data[PriceIndexes.IND_PRICE_LOW.value][i]
ohlcv_data[time_frame][i][PriceIndexes.IND_PRICE_TIME.value] = \
data[PriceIndexes.IND_PRICE_TIME.value][i]
def parse(file):
if os.path.isfile(CONFIG_DATA_COLLECTOR_PATH + file):
file_content = DataCollectorParser.get_file_content(CONFIG_DATA_COLLECTOR_PATH + file)
else:
file_content = DataCollectorParser.get_file_content(file)
return file_content
def get_file_content(file_name):
file_content = read_data_file(file_name)
data_type = get_data_type(file_name)
if data_type == BacktestingDataFormats.REGULAR_COLLECTOR_DATA:
return DataCollectorParser.merge_arrays(file_content)
else:
raise BacktestingDataFileException(file_name)