28095393343

Committed 24 Jun 2026 11:31AM UTC coverage: 68.506% (-0.6%) from 69.114%

Build # 28095393343

Build Type

Pull #1018

github

Committed by

web-flow

Commit Message

Merge 218c805f6 into 533de06cd

Pull Request Pull Request #1018: refactor update bundle

Coverage Stats

115 of 358 new or added lines in 5 files covered. (32.12%)

24 existing lines in 1 file now uncovered.

7983 of 11653 relevant lines covered (68.51%)

5.43 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.82

/rqalpha/data/bundle/automatic_update.py

import datetime
import os
from itertools import chain
from typing import Callable, Optional, Union, List
from filelock import FileLock, Timeout

import h5py
import numpy as np
from rqalpha.utils.datetime_func import convert_date_to_date_int
from rqalpha.utils.i18n import gettext as _
from rqalpha.utils.functools import lru_cache
from rqalpha.environment import Environment
from rqalpha.model.instrument import Instrument
from rqalpha.data.bundle.utils import START_DATE


class AutomaticUpdateBundle(object):
    def __init__(self, path: str, filename: str, api: Callable, fields: List[str], end_date: datetime.date, start_date: Union[int, datetime.date] = START_DATE) -> None:
        if not os.path.exists(path):
            os.makedirs(path)
        self._file = os.path.join(path, filename)
        self._trading_dates = None
        self._filename = filename
        self._api = api
        self._fields = fields
        self._start_date = start_date
        self.updated = []
        self._env = Environment.get_instance()
        self._file_lock = FileLock(self._file + ".lock")

    def get_data(self, instrument: Instrument, dt: datetime.date) -> Optional[np.ndarray]:
        dt_int = convert_date_to_date_int(dt)
        data = self._get_data_all_time(instrument)
        if data is None:
            return data
        else:
            try:
                data = data[np.searchsorted(data['trading_dt'], dt_int)]
            except IndexError:
                data = None
            return data

    @lru_cache(128)
    def _get_data_all_time(self, instrument: Instrument) -> Optional[np.ndarray]:
        if instrument.order_book_id not in self.updated:
            self._auto_update_task(instrument)
            self.updated.append(instrument.order_book_id)
        with h5py.File(self._file, "r") as h5:
            data = h5[instrument.order_book_id][:]
            if len(data) == 0:
                return None
        return data
    
    def _auto_update_task(self, instrument: Instrument) -> None:
        """
        在 rqalpha 策略运行过程中自动更新所需的日线数据

        :param instrument: 合约对象
        :type instrument: `Instrument`
        """
        order_book_id = instrument.order_book_id
        start_date = self._start_date
        try:
            with self._file_lock.acquire():
                h5 = h5py.File(self._file, "a")
                if order_book_id in h5 and h5[order_book_id].dtype.names:
                    if 'trading_dt' in h5[order_book_id].dtype.names:
                        # 需要兼容此前的旧版数据，对字段名进行更新
                        if len(h5[order_book_id][:]) != 0:
                            last_date = datetime.datetime.strptime(str(h5[order_book_id][-1]['trading_dt']), "%Y%m%d").date()
                            if last_date >= self._env.config.base.end_date:
                                return
                            start_date = self._env.data_proxy._data_source.get_next_trading_date(last_date).date()
                            if start_date > self._env.config.base.end_date:
                                return
                    else:
                        del h5[order_book_id]
                
                arr = self._get_array(instrument, start_date)
                if arr is None:
                    if order_book_id not in h5:
                        arr = np.array([])
                        h5.create_dataset(order_book_id, data=arr)
                else:
                    if order_book_id in h5:
                        data = np.array(
                            [tuple(i) for i in chain(h5[order_book_id][:], arr)],
                            dtype=h5[order_book_id].dtype)
                        del h5[order_book_id]
                        h5.create_dataset(order_book_id, data=data)
                    else:
                        h5.create_dataset(order_book_id, data=arr)
        except (OSError, Timeout) as e:
            raise OSError(_("File {} update failed, if it is using, please update later, "
                          "or you can delete then update again".format(self._file))) from e
        finally:
            h5.close()
    
    def _get_array(self, instrument: Instrument, start_date: Union[datetime.date, int]) -> Optional[np.ndarray]:
        df = self._api(instrument.order_book_id, start_date, self._env.config.base.end_date, self._fields)
        if not (df is None or df.empty):
            df = df[self._fields].loc[instrument.order_book_id] # rqdatac.get_open_auction_info get Futures's data will auto add 'open_interest' and 'prev_settlement'
            record = df.iloc[0: 1].to_records()
            dtype = [('trading_dt', 'int')]
            for field in self._fields:
                dtype.append((field, record.dtype[field]))
            trading_dt = self._env.data_proxy._data_source.batch_get_trading_date(df.index)
            trading_dt = convert_date_to_date_int(trading_dt)
            arr = np.ones((trading_dt.shape[0], ), dtype=dtype)
            arr['trading_dt'] = trading_dt
            for field in self._fields:
                arr[field] = df[field].values
            return arr
        return None

1	import datetime	8✔
2	import os	8✔
3	from itertools import chain	8✔
4	from typing import Callable, Optional, Union, List	8✔
5	from filelock import FileLock, Timeout	8✔
6
7	import h5py	8✔
8	import numpy as np	8✔
9	from rqalpha.utils.datetime_func import convert_date_to_date_int	8✔
10	from rqalpha.utils.i18n import gettext as _	8✔
11	from rqalpha.utils.functools import lru_cache	8✔
12	from rqalpha.environment import Environment	8✔
13	from rqalpha.model.instrument import Instrument	8✔
14	from rqalpha.data.bundle.utils import START_DATE	8✔
15
16
17	class AutomaticUpdateBundle(object):	8✔
18	def __init__(self, path: str, filename: str, api: Callable, fields: List[str], end_date: datetime.date, start_date: Union[int, datetime.date] = START_DATE) -> None:	8✔
19	if not os.path.exists(path):	8✔
20	os.makedirs(path)	8✔
21	self._file = os.path.join(path, filename)	8✔
22	self._trading_dates = None	8✔
23	self._filename = filename	8✔
24	self._api = api	8✔
25	self._fields = fields	8✔
26	self._start_date = start_date	8✔
27	self.updated = []	8✔
28	self._env = Environment.get_instance()	8✔
29	self._file_lock = FileLock(self._file + ".lock")	8✔
30
31	def get_data(self, instrument: Instrument, dt: datetime.date) -> Optional[np.ndarray]:	8✔
32	dt_int = convert_date_to_date_int(dt)	8✔
33	data = self._get_data_all_time(instrument)	8✔
34	if data is None:	8✔
NEW 35	return data	×
36	else:
37	try:	8✔
38	data = data[np.searchsorted(data['trading_dt'], dt_int)]	8✔
NEW 39	except IndexError:	×
NEW 40	data = None	×
41	return data	8✔
42
43	@lru_cache(128)	8✔
44	def _get_data_all_time(self, instrument: Instrument) -> Optional[np.ndarray]:	8✔
45	if instrument.order_book_id not in self.updated:	8✔
46	self._auto_update_task(instrument)	8✔
47	self.updated.append(instrument.order_book_id)	8✔
48	with h5py.File(self._file, "r") as h5:	8✔
49	data = h5[instrument.order_book_id][:]	8✔
50	if len(data) == 0:	8✔
NEW 51	return None	×
52	return data	8✔
53
54	def _auto_update_task(self, instrument: Instrument) -> None:	8✔
55	"""
56	在 rqalpha 策略运行过程中自动更新所需的日线数据
57
58	:param instrument: 合约对象
59	:type instrument: `Instrument`
60	"""
61	order_book_id = instrument.order_book_id	8✔
62	start_date = self._start_date	8✔
63	try:	8✔
64	with self._file_lock.acquire():	8✔
65	h5 = h5py.File(self._file, "a")	8✔
66	if order_book_id in h5 and h5[order_book_id].dtype.names:	8✔
NEW 67	if 'trading_dt' in h5[order_book_id].dtype.names:	×
68	# 需要兼容此前的旧版数据，对字段名进行更新
NEW 69	if len(h5[order_book_id][:]) != 0:	×
NEW 70	last_date = datetime.datetime.strptime(str(h5[order_book_id][-1]['trading_dt']), "%Y%m%d").date()	×
NEW 71	if last_date >= self._env.config.base.end_date:	×
NEW 72	return	×
NEW 73	start_date = self._env.data_proxy._data_source.get_next_trading_date(last_date).date()	×
NEW 74	if start_date > self._env.config.base.end_date:	×
NEW 75	return	×
76	else:
NEW 77	del h5[order_book_id]	×
78
79	arr = self._get_array(instrument, start_date)	8✔
80	if arr is None:	8✔
NEW 81	if order_book_id not in h5:	×
NEW 82	arr = np.array([])	×
NEW 83	h5.create_dataset(order_book_id, data=arr)	×
84	else:
85	if order_book_id in h5:	8✔
NEW 86	data = np.array(	×
87	[tuple(i) for i in chain(h5[order_book_id][:], arr)],
88	dtype=h5[order_book_id].dtype)
NEW 89	del h5[order_book_id]	×
NEW 90	h5.create_dataset(order_book_id, data=data)	×
91	else:
92	h5.create_dataset(order_book_id, data=arr)	8✔
NEW 93	except (OSError, Timeout) as e:	×
NEW 94	raise OSError(_("File {} update failed, if it is using, please update later, "	×
95	"or you can delete then update again".format(self._file))) from e
96	finally:
97	h5.close()	8✔
98
99	def _get_array(self, instrument: Instrument, start_date: Union[datetime.date, int]) -> Optional[np.ndarray]:	8✔
100	df = self._api(instrument.order_book_id, start_date, self._env.config.base.end_date, self._fields)	8✔
101	if not (df is None or df.empty):	8✔
102	df = df[self._fields].loc[instrument.order_book_id] # rqdatac.get_open_auction_info get Futures's data will auto add 'open_interest' and 'prev_settlement'	8✔
103	record = df.iloc[0: 1].to_records()	8✔
104	dtype = [('trading_dt', 'int')]	8✔
105	for field in self._fields:	8✔
106	dtype.append((field, record.dtype[field]))	8✔
107	trading_dt = self._env.data_proxy._data_source.batch_get_trading_date(df.index)	8✔
108	trading_dt = convert_date_to_date_int(trading_dt)	8✔
109	arr = np.ones((trading_dt.shape[0], ), dtype=dtype)	8✔
110	arr['trading_dt'] = trading_dt	8✔
111	for field in self._fields:	8✔
112	arr[field] = df[field].values	8✔
113	return arr	8✔
NEW 114	return None	×

ricequant / rqalpha / 28095393343

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous