Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion qlib/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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",
},
}


Expand Down
1 change: 1 addition & 0 deletions qlib/constant.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
19 changes: 18 additions & 1 deletion qlib/utils/time.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"),
Expand All @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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}")

Expand Down Expand Up @@ -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")

Expand Down
76 changes: 72 additions & 4 deletions tests/misc/test_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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):
Expand Down
Loading