mirror of
https://github.com/microsoft/qlib.git
synced 2026-06-06 14:01:28 +08:00
21f0b394e7
* change_url * fix_CI * fix_CI_2 * fix_CI_3 * fix_CI_4 * fix_CI_5 * fix_CI_6 * fix_CI_7 * fix_CI_8 * fix_CI_9 * fix_CI_10 * fix_CI_11 * fix_CI_12 * fix_CI_13 * fix_CI_13 * fix_CI_14 * fix_CI_15 * fix_CI_16 * fix_CI_17 * fix_CI_18 * fix_CI_19 * fix_CI_20 * fix_CI_21 * fix_CI_22 * fix_CI_23 * fix_CI_24 * fix_CI_25 * fix_CI_26 * fix_CI_27 * fix_get_data_error * fix_get_data_error2 * modify_get_data * modify_get_data2 * modify_get_data3 * modify_get_data4 * fix_CI_28 * fix_CI_29 * fix_CI_30 --------- Co-authored-by: Linlang <v-linlanglv@microsoft.com>
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
# Copyright (c) Microsoft Corporation.
|
|
# Licensed under the MIT License.
|
|
|
|
|
|
import sys
|
|
import shutil
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import qlib
|
|
import numpy as np
|
|
import pandas as pd
|
|
from qlib.data import D
|
|
|
|
sys.path.append(str(Path(__file__).resolve().parent.parent.joinpath("scripts")))
|
|
from get_data import GetData
|
|
from dump_bin import DumpDataAll, DumpDataFix
|
|
|
|
|
|
DATA_DIR = Path(__file__).parent.joinpath("test_dump_data")
|
|
SOURCE_DIR = DATA_DIR.joinpath("source")
|
|
SOURCE_DIR.mkdir(exist_ok=True, parents=True)
|
|
QLIB_DIR = DATA_DIR.joinpath("qlib")
|
|
QLIB_DIR.mkdir(exist_ok=True, parents=True)
|
|
|
|
|
|
class TestDumpData(unittest.TestCase):
|
|
FIELDS = "open,close,high,low,volume".split(",")
|
|
QLIB_FIELDS = list(map(lambda x: f"${x}", FIELDS))
|
|
DUMP_DATA = None
|
|
STOCK_NAMES = None
|
|
|
|
# simpe data
|
|
SIMPLE_DATA = None
|
|
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
GetData().download_data(file_name="csv_data_cn.zip", target_dir=SOURCE_DIR)
|
|
TestDumpData.DUMP_DATA = DumpDataAll(csv_path=SOURCE_DIR, qlib_dir=QLIB_DIR, include_fields=cls.FIELDS)
|
|
TestDumpData.STOCK_NAMES = list(map(lambda x: x.name[:-4].upper(), SOURCE_DIR.glob("*.csv")))
|
|
provider_uri = str(QLIB_DIR.resolve())
|
|
qlib.init(
|
|
provider_uri=provider_uri,
|
|
expression_cache=None,
|
|
dataset_cache=None,
|
|
)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls) -> None:
|
|
shutil.rmtree(str(DATA_DIR.resolve()))
|
|
|
|
def test_0_dump_bin(self):
|
|
self.DUMP_DATA.dump()
|
|
|
|
def test_1_dump_calendars(self):
|
|
ori_calendars = set(
|
|
map(
|
|
pd.Timestamp,
|
|
pd.read_csv(QLIB_DIR.joinpath("calendars", "day.txt"), header=None).loc[:, 0].values,
|
|
)
|
|
)
|
|
res_calendars = set(D.calendar())
|
|
assert len(ori_calendars - res_calendars) == len(res_calendars - ori_calendars) == 0, "dump calendars failed"
|
|
|
|
def test_2_dump_instruments(self):
|
|
ori_ins = set(map(lambda x: x.name[:-4].upper(), SOURCE_DIR.glob("*.csv")))
|
|
res_ins = set(D.list_instruments(D.instruments("all"), as_list=True))
|
|
assert len(ori_ins - res_ins) == len(ori_ins - res_ins) == 0, "dump instruments failed"
|
|
|
|
def test_3_dump_features(self):
|
|
df = D.features(self.STOCK_NAMES, self.QLIB_FIELDS)
|
|
TestDumpData.SIMPLE_DATA = df.loc(axis=0)[self.STOCK_NAMES[0], :]
|
|
self.assertFalse(df.dropna().empty, "features data failed")
|
|
self.assertListEqual(list(df.columns), self.QLIB_FIELDS, "features columns failed")
|
|
|
|
def test_4_dump_features_simple(self):
|
|
stock = self.STOCK_NAMES[0]
|
|
dump_data = DumpDataFix(
|
|
csv_path=SOURCE_DIR.joinpath(f"{stock.lower()}.csv"), qlib_dir=QLIB_DIR, include_fields=self.FIELDS
|
|
)
|
|
dump_data.dump()
|
|
|
|
df = D.features([stock], self.QLIB_FIELDS)
|
|
|
|
self.assertEqual(len(df), len(TestDumpData.SIMPLE_DATA), "dump features simple failed")
|
|
self.assertTrue(np.isclose(df.dropna(), self.SIMPLE_DATA.dropna()).all(), "dump features simple failed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|