stable/freqtrade/data/datahandlers/jsondatahandler.py

111 lines
3.5 KiB
Python
Raw Normal View History

import re
from pathlib import Path
from typing import Dict, List, Optional
from pandas import DataFrame
from freqtrade import misc
from freqtrade.configuration import TimeRange
from .idatahandler import IDataHandler
class JsonDataHandler(IDataHandler):
_use_zip = False
@classmethod
def ohlcv_get_pairs(cls, datadir: Path, timeframe: str) -> List[str]:
"""
Returns a list of all pairs available in this datadir
"""
2019-12-25 09:21:30 +00:00
_tmp = [re.search(r'^(\S+)(?=\-' + timeframe + '.json)', p.name)
for p in datadir.glob(f"*{timeframe}.{cls._get_file_extension()}")]
2019-12-25 09:21:30 +00:00
# Check if regex found something and only return these results
2019-12-25 09:25:30 +00:00
return [match[0].replace('_', '/') for match in _tmp if match]
2019-12-25 09:30:35 +00:00
def ohlcv_store(self, timeframe: str, data: DataFrame) -> None:
"""
Store data
"""
2019-12-25 09:37:31 +00:00
filename = JsonDataHandler._pair_data_filename(self._datadir, self._pair, timeframe)
misc.file_dump_json(filename, data, is_zip=self._use_zip)
2019-12-25 09:30:35 +00:00
def ohlcv_append(self, timeframe: str, data: DataFrame) -> None:
"""
Append data to existing files
"""
raise NotImplementedError()
def ohlcv_load(self, timeframe: str, timerange: Optional[TimeRange] = None) -> DataFrame:
"""
Load data for one pair
:return: Dataframe
"""
2019-12-25 09:25:30 +00:00
filename = JsonDataHandler._pair_data_filename(self._datadir, self._pair, timeframe)
pairdata = misc.file_load_json(filename)
if not pairdata:
return []
if timerange:
pairdata = IDataHandler.trim_tickerlist(pairdata, timerange)
return pairdata
@classmethod
def trades_get_pairs(cls, datadir: Path) -> List[str]:
"""
Returns a list of all pairs available in this datadir
"""
2019-12-25 09:21:30 +00:00
_tmp = [re.search(r'^(\S+)(?=\-trades.json)', p.name)
for p in datadir.glob(f"*trades.{cls._get_file_extension()}")]
2019-12-25 09:21:30 +00:00
# Check if regex found something and only return these results to avoid exceptions.
2019-12-25 09:25:30 +00:00
return [match[0].replace('_', '/') for match in _tmp if match]
def trades_store(self, data: List[Dict]):
"""
Store data
"""
filename = self._pair_trades_filename(self._datadir, self._pair)
misc.file_dump_json(filename, data, is_zip=self._use_zip)
def trades_append(self, data: DataFrame):
"""
Append data to existing files
"""
raise NotImplementedError()
def trades_load(self, timerange: Optional[TimeRange] = None) -> List[Dict]:
"""
Load a pair from file, either .json.gz or .json
# TODO: validate timerange ...
:return: List of trades
"""
filename = self._pair_trades_filename(self._datadir, self._pair)
tradesdata = misc.file_load_json(filename)
if not tradesdata:
return []
return tradesdata
@classmethod
def _pair_data_filename(cls, datadir: Path, pair: str, timeframe: str) -> Path:
pair_s = pair.replace("/", "_")
filename = datadir.joinpath(f'{pair_s}-{timeframe}.{cls._get_file_extension()}')
return filename
@classmethod
def _get_file_extension(cls):
return "json.gz" if cls._use_zip else "json"
@classmethod
def _pair_trades_filename(cls, datadir: Path, pair: str) -> Path:
pair_s = pair.replace("/", "_")
filename = datadir.joinpath(f'{pair_s}-trades.{cls._get_file_extension()}')
return filename
class JsonGzDataHandler(JsonDataHandler):
_use_zip = True