158 lines
4.4 KiB
Python
158 lines
4.4 KiB
Python
"""
|
|
Various tool function for Freqtrade and scripts
|
|
"""
|
|
|
|
import gzip
|
|
import logging
|
|
import re
|
|
from datetime import datetime
|
|
from typing import Dict
|
|
|
|
from ccxt import Exchange
|
|
import numpy as np
|
|
from pandas import DataFrame
|
|
import rapidjson
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def shorten_date(_date: str) -> str:
|
|
"""
|
|
Trim the date so it fits on small screens
|
|
"""
|
|
new_date = re.sub('seconds?', 'sec', _date)
|
|
new_date = re.sub('minutes?', 'min', new_date)
|
|
new_date = re.sub('hours?', 'h', new_date)
|
|
new_date = re.sub('days?', 'd', new_date)
|
|
new_date = re.sub('^an?', '1', new_date)
|
|
return new_date
|
|
|
|
|
|
############################################
|
|
# Used by scripts #
|
|
# Matplotlib doesn't support ::datetime64, #
|
|
# so we need to convert it into ::datetime #
|
|
############################################
|
|
def datesarray_to_datetimearray(dates: np.ndarray) -> np.ndarray:
|
|
"""
|
|
Convert an pandas-array of timestamps into
|
|
An numpy-array of datetimes
|
|
:return: numpy-array of datetime
|
|
"""
|
|
return dates.dt.to_pydatetime()
|
|
|
|
|
|
def common_datearray(dfs: Dict[str, DataFrame]) -> np.ndarray:
|
|
"""
|
|
Return dates from Dataframe
|
|
:param dfs: Dict with format pair: pair_data
|
|
:return: List of dates
|
|
"""
|
|
alldates = {}
|
|
for pair, pair_data in dfs.items():
|
|
dates = datesarray_to_datetimearray(pair_data['date'])
|
|
for date in dates:
|
|
alldates[date] = 1
|
|
lst = []
|
|
for date, _ in alldates.items():
|
|
lst.append(date)
|
|
arr = np.array(lst)
|
|
return np.sort(arr, axis=0)
|
|
|
|
|
|
def file_dump_json(filename, data, is_zip=False) -> None:
|
|
"""
|
|
Dump JSON data into a file
|
|
:param filename: file to create
|
|
:param data: JSON Data to save
|
|
:return:
|
|
"""
|
|
logger.info(f'dumping json to "{filename}"')
|
|
|
|
if is_zip:
|
|
if not filename.endswith('.gz'):
|
|
filename = filename + '.gz'
|
|
with gzip.open(filename, 'w') as fp:
|
|
rapidjson.dump(data, fp, default=str, number_mode=rapidjson.NM_NATIVE)
|
|
else:
|
|
with open(filename, 'w') as fp:
|
|
rapidjson.dump(data, fp, default=str, number_mode=rapidjson.NM_NATIVE)
|
|
|
|
logger.debug(f'done json to "{filename}"')
|
|
|
|
|
|
def json_load(datafile):
|
|
"""
|
|
load data with rapidjson
|
|
Use this to have a consistent experience,
|
|
sete number_mode to "NM_NATIVE" for greatest speed
|
|
"""
|
|
return rapidjson.load(datafile, number_mode=rapidjson.NM_NATIVE)
|
|
|
|
|
|
def file_load_json(file):
|
|
|
|
gzipfile = file.with_suffix(file.suffix + '.gz')
|
|
|
|
# Try gzip file first, otherwise regular json file.
|
|
if gzipfile.is_file():
|
|
logger.debug('Loading ticker data from file %s', gzipfile)
|
|
with gzip.open(gzipfile) as tickerdata:
|
|
pairdata = json_load(tickerdata)
|
|
elif file.is_file():
|
|
logger.debug('Loading ticker data from file %s', file)
|
|
with open(file) as tickerdata:
|
|
pairdata = json_load(tickerdata)
|
|
else:
|
|
return None
|
|
return pairdata
|
|
|
|
|
|
def format_ms_time(date: int) -> str:
|
|
"""
|
|
convert MS date to readable format.
|
|
: epoch-string in ms
|
|
"""
|
|
return datetime.fromtimestamp(date/1000.0).strftime('%Y-%m-%dT%H:%M:%S')
|
|
|
|
|
|
def deep_merge_dicts(source, destination):
|
|
"""
|
|
>>> a = { 'first' : { 'rows' : { 'pass' : 'dog', 'number' : '1' } } }
|
|
>>> b = { 'first' : { 'rows' : { 'fail' : 'cat', 'number' : '5' } } }
|
|
>>> merge(b, a) == { 'first' : { 'rows' : { 'pass' : 'dog', 'fail' : 'cat', 'number' : '5' } } }
|
|
True
|
|
"""
|
|
for key, value in source.items():
|
|
if isinstance(value, dict):
|
|
# get node or create one
|
|
node = destination.setdefault(key, {})
|
|
deep_merge_dicts(value, node)
|
|
else:
|
|
destination[key] = value
|
|
|
|
return destination
|
|
|
|
|
|
def timeframe_to_seconds(ticker_interval: str) -> int:
|
|
"""
|
|
Translates the timeframe interval value written in the human readable
|
|
form ('1m', '5m', '1h', '1d', '1w', etc.) to the number
|
|
of seconds for one timeframe interval.
|
|
"""
|
|
return Exchange.parse_timeframe(ticker_interval)
|
|
|
|
|
|
def timeframe_to_minutes(ticker_interval: str) -> int:
|
|
"""
|
|
Same as above, but returns minutes.
|
|
"""
|
|
return Exchange.parse_timeframe(ticker_interval) // 60
|
|
|
|
|
|
def timeframe_to_msecs(ticker_interval: str) -> int:
|
|
"""
|
|
Same as above, but returns milliseconds.
|
|
"""
|
|
return Exchange.parse_timeframe(ticker_interval) * 1000
|