From d395c904f2821f7b3b59359a12a2e4195f7d8fb1 Mon Sep 17 00:00:00 2001 From: zhupr Date: Fri, 26 Mar 2021 16:14:45 +0800 Subject: [PATCH] Add FileStorage --- qlib/data/storage/__init__.py | 4 + qlib/data/storage/file_storage.py | 91 ++++++++++++++++++ qlib/data/storage/storage.py | 135 ++++++++++++++++++++++++++ qlib/storage/__init__.py | 0 qlib/storage/storage.py | 154 ------------------------------ 5 files changed, 230 insertions(+), 154 deletions(-) create mode 100644 qlib/data/storage/__init__.py create mode 100644 qlib/data/storage/file_storage.py create mode 100644 qlib/data/storage/storage.py delete mode 100644 qlib/storage/__init__.py delete mode 100644 qlib/storage/storage.py diff --git a/qlib/data/storage/__init__.py b/qlib/data/storage/__init__.py new file mode 100644 index 000000000..eb513714b --- /dev/null +++ b/qlib/data/storage/__init__.py @@ -0,0 +1,4 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +from .storage import CalendarStorage, InstrumentStorage, FeatureStorage \ No newline at end of file diff --git a/qlib/data/storage/file_storage.py b/qlib/data/storage/file_storage.py new file mode 100644 index 000000000..9d98545ce --- /dev/null +++ b/qlib/data/storage/file_storage.py @@ -0,0 +1,91 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +from pathlib import Path +from typing import Iterator, Iterable, Type, List, Tuple, Text, Union + +from data.storage.storage import FeatureVT + +import numpy as np +import pandas as pd +from qlib.data.storage import CalendarStorage, InstrumentStorage, FeatureStorage + + +CalVT = Type[pd.Timestamp] +# instrument value +InstVT = List[Tuple[CalVT, CalVT]] +# instrument key +InstKT = Text + + +class FileCalendarStorage(CalendarStorage): + def __init__(self, uri: str): + super(FileCalendarStorage, self).__init__(uri=uri) + with open(uri) as f: + self._data = [pd.Timestamp(x.strip()) for x in f] + + def __getitem__(self, i: Union[int, slice]) -> Union[CalVT, Iterable[CalVT]]: + if isinstance(i, (int, slice)): + return self._data[i] + else: + raise TypeError(f"type(i) = {type(i)}") + + def __len__(self) -> int: + return len(self._data) + + +class FileInstrumentStorage(InstrumentStorage): + def __init__(self, uri: str): + super(FileInstrumentStorage, self).__init__(uri=uri) + self._data = self._load_data() + + def _load_data(self): + _instruments = dict() + df = pd.read_csv( + self._uri, + sep="\t", + usecols=[0, 1, 2], + names=["inst", "start_datetime", "end_datetime"], + dtype={"inst": str}, + parse_dates=["start_datetime", "end_datetime"], + ) + for row in df.itertuples(index=False): + _instruments.setdefault(row[0], []).append((row[1], row[2])) + return _instruments + + def __getitem__(self, k: InstKT) -> InstVT: + return self._data[k] + + def __len__(self) -> int: + return len(self._data) + + def __iter__(self) -> Iterator[InstKT]: + return self._data.__iter__() + + +class FileFeatureStorage(FeatureStorage): + def __getitem__(self, i: Union[int, slice]) -> Union[FeatureVT, Iterable[FeatureVT]]: + with open(self._uri, "rb") as fp: + ref_start_index = int(np.frombuffer(fp.read(4), dtype=" i: + raise IndexError(f"{i}") + fp.seek(4 * (i - ref_start_index) + 4) + return i, float(fp.read(4)) + elif isinstance(i, slice): + start_index = i.start + end_index = i.stop - 1 + si = max(ref_start_index, start_index) + if si > end_index: + return [] + fp.seek(4 * (si - ref_start_index) + 4) + # read n bytes + count = end_index - si + 1 + data = np.frombuffer(fp.read(4 * count), dtype=" int: + return Path(self._uri).stat().st_size // 4 - 1 diff --git a/qlib/data/storage/storage.py b/qlib/data/storage/storage.py new file mode 100644 index 000000000..7848c243f --- /dev/null +++ b/qlib/data/storage/storage.py @@ -0,0 +1,135 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. +from abc import abstractmethod +from collections.abc import MutableSequence, MutableMapping, Sequence +from typing import Iterable, overload, TypeVar, Tuple, List, Text, Iterator + + +# calendar value type +CalVT = TypeVar("CalVT") + +# instrument value +InstVT = List[Tuple[CalVT, CalVT]] +# instrument key +InstKT = Text + + +FeatureVT = Tuple[int, float] + + +class CalendarStorage(MutableSequence): + def __init__(self, uri: str): + self._uri = uri + + def insert(self, index: int, o: CalVT) -> None: + raise NotImplementedError("Subclass of CalendarStorage must implement `insert` method") + + @overload + def __setitem__(self, i: int, o: CalVT) -> None: + """x.__setitem__(i, o) <==> x[i] = o""" + ... + + @overload + def __setitem__(self, s: slice, o: Iterable[CalVT]) -> None: + """x.__setitem__(s, o) <==> x[s] = o""" + ... + + def __setitem__(self, i, o) -> None: + raise NotImplementedError( + "Subclass of CalendarStorage must implement `__setitem__(i: int, o: CalVT)`/`__setitem__(s: slice, o: Iterable[CalVT])` method" + ) + + @overload + def __delitem__(self, i: int) -> None: + """x.__delitem__(i) <==> del x[i]""" + ... + + @overload + def __delitem__(self, i: slice) -> None: + """x.__delitem__(slice(start: int, stop: int, step: int)) <==> del x[start:stop:step]""" + ... + + def __delitem__(self, i) -> None: + raise NotImplementedError( + "Subclass of CalendarStorage must implement `__delitem__(i: int)`/`__delitem__(s: slice)` method" + ) + + @overload + def __getitem__(self, s: slice) -> Iterable[CalVT]: + """x.__getitem__(slice(start: int, stop: int, step: int)) <==> x[start:stop:step]""" + ... + + @overload + def __getitem__(self, i: int) -> CalVT: + """x.__getitem__(i) <==> x[i]""" + ... + + def __getitem__(self, i) -> CalVT: + raise NotImplementedError( + "Subclass of CalendarStorage must implement `__getitem__(i: int)`/`__getitem__(s: slice)` method" + ) + + def __len__(self) -> int: + """x.__len__() <==> len(x)""" + raise NotImplementedError("Subclass of CalendarStorage must implement `__len__` method") + + +class InstrumentStorage(MutableMapping): + def __init__(self, uri: str): + self._uri = uri + + def __setitem__(self, k: InstKT, v: InstVT) -> None: + """ Set self[key] to value. """ + raise NotImplementedError("Subclass of InstrumentStorage must implement `__setitem__` method") + + def __delitem__(self, k: InstKT) -> None: + """ Delete self[key]. """ + raise NotImplementedError("Subclass of InstrumentStorage must implement `__delitem__` method") + + def __getitem__(self, k: InstKT) -> InstVT: + """ x.__getitem__(k) <==> x[k] """ + raise NotImplementedError("Subclass of InstrumentStorage must implement `__getitem__` method") + + def __len__(self) -> int: + """ Return len(self). """ + raise NotImplementedError("Subclass of InstrumentStorage must implement `__len__` method") + + def __iter__(self) -> Iterator[InstKT]: + """ Return iter(self). """ + raise NotImplementedError("Subclass of InstrumentStorage must implement `__iter__` method") + + +class FeatureStorage(Sequence): + def __init__(self, uri: str): + self._uri = uri + + def append(self, obj: FeatureVT) -> None: + """ Append object to the end of the FeatureStorage. """ + raise NotImplementedError("Subclass of FeatureStorage must implement `append` method") + + def clear(self): + """ Remove all items from FeatureStorage. """ + raise NotImplementedError("Subclass of FeatureStorage must implement `clear` method") + + def extend(self, iterable: Iterable[FeatureVT]): + """ Extend list by appending elements from the iterable. """ + raise NotImplementedError("Subclass of FeatureStorage must implement `extend` method") + + @overload + def __getitem__(self, s: slice) -> Iterable[FeatureVT]: + """x.__getitem__(slice(start: int, stop: int, step: int)) <==> x[start:stop:step]""" + ... + + @overload + def __getitem__(self, i: int) -> float: + """x.__getitem__(y) <==> x[y]""" + ... + + def __getitem__(self, i) -> float: + """x.__getitem__(y) <==> x[y]""" + raise NotImplementedError( + "Subclass of FeatureStorage must implement `__getitem__(i: int)`/`__getitem__(s: slice)` method" + ) + + def __len__(self) -> int: + raise NotImplementedError("Subclass of FeatureStorage must implement `__len__` method") diff --git a/qlib/storage/__init__.py b/qlib/storage/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/qlib/storage/storage.py b/qlib/storage/storage.py deleted file mode 100644 index dac0e167d..000000000 --- a/qlib/storage/storage.py +++ /dev/null @@ -1,154 +0,0 @@ -# Copyright (c) Microsoft Corporation. -# Licensed under the MIT License. - - -import abc - -from typing import ( - Iterable, - overload, - TypeVar, - Tuple, - List, - Text, - Optional, - AbstractSet, - Mapping, - Iterator, -) - - -# calendar value type -CalVT = TypeVar("CalVT") - -# instrument value -InstVT = List[Tuple[CalVT, CalVT]] -# instrument key -InstKT = Text - - -FeatureVT = Tuple[int, float] - - -class CalendarStorage: - def __init__(self, uri: str): - self._uri = uri - - def append(self, obj: CalVT) -> None: - """ Append object to the end of the CalendarStorage. """ - raise NotImplementedError("Subclass of CalendarStorage must implement `append` method") - - def clear(self): - """ Remove all items from CalendarStorage. """ - raise NotImplementedError("Subclass of CalendarStorage must implement `clear` method") - - def extend(self, iterable: Iterable[CalVT]): - """ Extend list by appending elements from the iterable. """ - raise NotImplementedError("Subclass of CalendarStorage must implement `extend` method") - - @overload - @abc.abstractmethod - def __getitem__(self, s: slice) -> Iterable[CalVT]: - """x.__getitem__(slice(start: int, stop: int, step: int)) <==> x[start:stop:step]""" - raise NotImplementedError("Subclass of CalendarStorage must implement `__getitem__(s: slice)` method") - - @abc.abstractmethod - def __getitem__(self, i: int) -> CalVT: - """x.__getitem__(y) <==> x[y]""" - - raise NotImplementedError("Subclass of CalendarStorage must implement `__getitem__(i: int)` method") - - @abc.abstractmethod - def __iter__(self) -> Iterator[CalVT]: - """ Implement iter(self). """ - raise NotImplementedError("Subclass of CalendarStorage must implement `__iter__` method") - - def __len__(self) -> int: - raise NotImplementedError("Subclass of CalendarStorage must implement `__len__` method") - - -class InstrumentStorage: - def __init__(self, uri: str): - self._uri = uri - - def clear(self) -> None: - """ D.clear() -> None. Remove all items from D. """ - raise NotImplementedError("Subclass of InstrumentStorage must implement `clear` method") - - @abc.abstractmethod - def get(self, k: InstKT) -> Optional[InstVT]: - """D.get(k) -> InstV or None""" - raise NotImplementedError("Subclass of InstrumentStorage must implement `get` method") - - @abc.abstractmethod - def items(self) -> AbstractSet[Tuple[InstKT, InstVT]]: - """ D.items() -> a set-like object providing a view on D's items """ - raise NotImplementedError("Subclass of InstrumentStorage must implement `items` method") - - @abc.abstractmethod - def keys(self) -> AbstractSet[InstKT]: - """ D.keys() -> a set-like object providing a view on D's keys """ - raise NotImplementedError("Subclass of InstrumentStorage must implement `keys` method") - - def update(self, e: Mapping[InstKT, InstVT] = None, **f: InstVT) -> None: - """ - D.update([e, ]**f) -> None. Update D from dict/iterable e and f. - If e is present and has a .keys() method, then does: for k in e: D[k] = e[k] - If e is present and lacks a .keys() method, then does: for k, v in e: D[k] = v - In either case, this is followed by: for k in f: D[k] = f[k] - """ - raise NotImplementedError("Subclass of InstrumentStorage must implement `update` method") - - def __setitem__(self, k: InstKT, v: InstVT) -> None: - """ Set self[key] to value. """ - raise NotImplementedError("Subclass of InstrumentStorage must implement `__setitem__` method") - - def __delitem__(self, k: InstKT) -> None: - """ Delete self[key]. """ - raise NotImplementedError("Subclass of InstrumentStorage must implement `__delitem__` method") - - @abc.abstractmethod - def __getitem__(self, k: InstKT) -> InstVT: - """ x.__getitem__(y) <==> x[y] """ - raise NotImplementedError("Subclass of InstrumentStorage must implement `__getitem__` method") - - def __len__(self) -> int: - """ Return len(self). """ - raise NotImplementedError("Subclass of InstrumentStorage must implement `__len__` method") - - -class FeatureStorage: - def __init__(self, uri: str): - self._uri = uri - - def append(self, obj: FeatureVT) -> None: - """ Append object to the end of the FeatureStorage. """ - raise NotImplementedError("Subclass of FeatureStorage must implement `append` method") - - def clear(self): - """ Remove all items from FeatureStorage. """ - raise NotImplementedError("Subclass of FeatureStorage must implement `clear` method") - - def extend(self, iterable: Iterable[FeatureVT]): - """ Extend list by appending elements from the iterable. """ - raise NotImplementedError("Subclass of FeatureStorage must implement `extend` method") - - @overload - @abc.abstractmethod - def __getitem__(self, s: slice) -> Iterable[FeatureVT]: - """x.__getitem__(slice(start: int, stop: int, step: int)) <==> x[start:stop:step]""" - raise NotImplementedError("Subclass of FeatureStorage must implement `__getitem__(s: slice)` method") - - @abc.abstractmethod - def __getitem__(self, i: int) -> float: - """x.__getitem__(y) <==> x[y]""" - - raise NotImplementedError("Subclass of FeatureStorage must implement `__getitem__(i: int)` method") - - def __len__(self) -> int: - raise NotImplementedError("Subclass of FeatureStorage must implement `__len__` method") - - @abc.abstractmethod - def __iter__(self) -> Iterator[FeatureVT]: - """ Implement iter(self). """ - raise NotImplementedError("Subclass of FeatureStorage must implement `__iter__` method")