diff --git a/qlib/config.py b/qlib/config.py index ae05037e2ff..0e0a656ac2e 100644 --- a/qlib/config.py +++ b/qlib/config.py @@ -23,7 +23,7 @@ from typing import Callable, Optional, Union from typing import TYPE_CHECKING -from qlib.constant import REG_CN, REG_US, REG_TW +from qlib.constant import REG_CN, REG_US, REG_TW, REG_GB if TYPE_CHECKING: from qlib.utils.time import Freq @@ -309,6 +309,11 @@ def register_from_C(config, skip_register=True): "limit_threshold": 0.1, "deal_price": "close", }, + REG_GB: { + "trade_unit": 1, + "limit_threshold": None, + "deal_price": "close", + }, } diff --git a/qlib/constant.py b/qlib/constant.py index ac6c76ae22c..ba48a3e9629 100644 --- a/qlib/constant.py +++ b/qlib/constant.py @@ -10,6 +10,7 @@ REG_CN = "cn" REG_US = "us" REG_TW = "tw" +REG_GB = "gb" # Epsilon for avoiding division by zero. EPS = 1e-12 diff --git a/qlib/utils/time.py b/qlib/utils/time.py index b052f6ab9f7..533d5f4cc8a 100644 --- a/qlib/utils/time.py +++ b/qlib/utils/time.py @@ -13,7 +13,7 @@ import pandas as pd from qlib.config import C -from qlib.constant import REG_CN, REG_TW, REG_US +from qlib.constant import REG_CN, REG_TW, REG_US, REG_GB CN_TIME = [ datetime.strptime("9:30", "%H:%M"), @@ -26,6 +26,7 @@ datetime.strptime("9:00", "%H:%M"), datetime.strptime("13:30", "%H:%M"), ] +GB_TIME = [datetime.strptime("8:00", "%H:%M"), datetime.strptime("16:30", "%H:%M")] @functools.lru_cache(maxsize=240) @@ -65,6 +66,11 @@ def get_min_cal(shift: int = 0, region: str = REG_CN) -> List[time]: pd.date_range(US_TIME[0], US_TIME[1] - timedelta(minutes=1), freq="1min") - pd.Timedelta(minutes=shift) ): cal.append(ts.time()) + elif region == REG_GB: + for ts in list( + pd.date_range(GB_TIME[0], GB_TIME[1] - timedelta(minutes=1), freq="1min") - pd.Timedelta(minutes=shift) + ): + cal.append(ts.time()) else: raise ValueError(f"{region} is not supported") return cal @@ -107,6 +113,12 @@ def is_single_value(start_time, end_time, freq, region: str = REG_CN): if start_time.hour == 15 and start_time.minute == 59 and start_time.second == 0: return True return False + elif region == REG_GB: + if end_time - start_time < freq: + return True + if start_time.hour == 16 and start_time.minute == 29 and start_time.second == 0: + return True + return False else: raise NotImplementedError(f"please implement the is_single_value func for {region}") @@ -276,6 +288,11 @@ def time_to_day_index(time_obj: Union[str, datetime], region: str = REG_CN): return int((time_obj - TW_TIME[0]).total_seconds() / 60) else: raise ValueError(f"{time_obj} is not the opening time of the {region} stock market") + elif region == REG_GB: + if GB_TIME[0] <= time_obj < GB_TIME[1]: + return int((time_obj - GB_TIME[0]).total_seconds() / 60) + else: + raise ValueError(f"{time_obj} is not the opening time of the {region} stock market") else: raise ValueError(f"{region} is not supported") diff --git a/tests/misc/test_utils.py b/tests/misc/test_utils.py index db5b0724886..72aaac60c7b 100644 --- a/tests/misc/test_utils.py +++ b/tests/misc/test_utils.py @@ -7,11 +7,20 @@ from qlib import init from qlib.config import C from qlib.log import TimeInspector -from qlib.constant import REG_CN, REG_US, REG_TW -from qlib.utils.time import cal_sam_minute as cal_sam_minute_new, get_min_cal, CN_TIME, US_TIME, TW_TIME +from qlib.constant import REG_CN, REG_US, REG_TW, REG_GB +from qlib.utils.time import ( + cal_sam_minute as cal_sam_minute_new, + get_min_cal, + is_single_value, + time_to_day_index, + CN_TIME, + US_TIME, + TW_TIME, + GB_TIME, +) from qlib.utils.data import guess_horizon -REG_MAP = {REG_CN: CN_TIME, REG_US: US_TIME, REG_TW: TW_TIME} +REG_MAP = {REG_CN: CN_TIME, REG_US: US_TIME, REG_TW: TW_TIME, REG_GB: GB_TIME} def cal_sam_minute(x: pd.Timestamp, sam_minutes: int, region: str): @@ -77,7 +86,7 @@ def setUpClass(cls): def test_cal_sam_minute(self): # test the correctness of the code random_n = 1000 - regions = [REG_CN, REG_US, REG_TW] + regions = [REG_CN, REG_US, REG_TW, REG_GB] def gen_args(cal: List): for time in np.random.choice(cal, size=random_n, replace=True): @@ -113,6 +122,65 @@ def gen_args(cal: List): cal_sam_minute_new(*args, region=region) +class GBTimeUtils(TestCase): + """Tests for GB (London Stock Exchange) region support in time utils.""" + + def test_get_min_cal_gb_count(self): + # LSE trades 08:00–16:29 inclusive = 510 one-minute bars + cal = get_min_cal(region=REG_GB) + self.assertEqual(len(cal), 510) + + def test_get_min_cal_gb_open(self): + cal = get_min_cal(region=REG_GB) + self.assertEqual(cal[0].hour, 8) + self.assertEqual(cal[0].minute, 0) + + def test_get_min_cal_gb_close(self): + cal = get_min_cal(region=REG_GB) + self.assertEqual(cal[-1].hour, 16) + self.assertEqual(cal[-1].minute, 29) + + def test_is_single_value_gb_freq_too_small(self): + # window smaller than freq → single value + start = pd.Timestamp("2024-01-02 10:00:00") + end = pd.Timestamp("2024-01-02 10:00:00") + freq = pd.Timedelta("1min") + self.assertTrue(is_single_value(start, end, freq, region=REG_GB)) + + def test_is_single_value_gb_last_bar(self): + # 16:29 is the last bar of the day → single value + start = pd.Timestamp("2024-01-02 16:29:00") + end = pd.Timestamp("2024-01-02 16:30:00") + freq = pd.Timedelta("1min") + self.assertTrue(is_single_value(start, end, freq, region=REG_GB)) + + def test_is_single_value_gb_mid_session(self): + # mid-session bar spanning a full minute → not single value + start = pd.Timestamp("2024-01-02 12:00:00") + end = pd.Timestamp("2024-01-02 12:01:00") + freq = pd.Timedelta("1min") + self.assertFalse(is_single_value(start, end, freq, region=REG_GB)) + + def test_time_to_day_index_gb_open(self): + # 08:00 is index 0 + self.assertEqual(time_to_day_index("8:00", region=REG_GB), 0) + + def test_time_to_day_index_gb_mid(self): + # 12:00 = 240 minutes after 08:00 + self.assertEqual(time_to_day_index("12:00", region=REG_GB), 240) + + def test_time_to_day_index_gb_last(self): + # 16:29 = 509 minutes after 08:00 + self.assertEqual(time_to_day_index("16:29", region=REG_GB), 509) + + def test_time_to_day_index_gb_out_of_range(self): + # outside trading hours should raise + with self.assertRaises(ValueError): + time_to_day_index("7:59", region=REG_GB) + with self.assertRaises(ValueError): + time_to_day_index("16:30", region=REG_GB) + + class DataUtils(TestCase): @classmethod def setUpClass(cls):