跳转至

utils — 通用工具

utils.retry_utils

retry_utils

重试工具模块

提供带指数退避的智能重试装饰器和辅助函数 用于增强xtquant等数据源的稳定性

RetryStats 

RetryStats()

重试统计收集器

源代码位于： utils/retry_utils.py 
138
139
140
141
142
143
def __init__(self):
    self.total_requests = 0
    self.successful_first_try = 0
    self.successful_after_retry = 0
    self.failed_after_all_retries = 0
    self.total_retries = 0
record_success 
record_success(attempts: int)

记录成功请求

源代码位于： utils/retry_utils.py 
145
146
147
148
149
150
151
152
def record_success(self, attempts: int):
    """记录成功请求"""
    self.total_requests += 1
    if attempts == 1:
        self.successful_first_try += 1
    else:
        self.successful_after_retry += 1
        self.total_retries += attempts - 1
record_failure 
record_failure(attempts: int)

记录失败请求

源代码位于： utils/retry_utils.py 
154
155
156
157
158
def record_failure(self, attempts: int):
    """记录失败请求"""
    self.total_requests += 1
    self.failed_after_all_retries += 1
    self.total_retries += attempts
get_stats 
get_stats() -> dict

获取统计数据

源代码位于： utils/retry_utils.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def get_stats(self) -> dict:
    """获取统计数据"""
    stats = {
        'total_requests': self.total_requests,
        'successful_first_try': self.successful_first_try,
        'successful_after_retry': self.successful_after_retry,
        'failed_after_all_retries': self.failed_after_all_retries,
        'total_retries': self.total_retries
    }

    if self.total_requests > 0:
        stats['first_try_success_rate'] = (
            self.successful_first_try / self.total_requests * 100
        )
        stats['overall_success_rate'] = (
            (self.successful_first_try + self.successful_after_retry)
            / self.total_requests * 100
        )
        stats['avg_retries_per_request'] = (
            self.total_retries / self.total_requests
        )
    else:
        stats['first_try_success_rate'] = 0
        stats['overall_success_rate'] = 0
        stats['avg_retries_per_request'] = 0

    return stats
reset 
reset()

重置统计

源代码位于： utils/retry_utils.py 
188
189
190
def reset(self):
    """重置统计"""
    self.__init__()

retry_with_backoff 

retry_with_backoff(max_retries: int = 3, initial_delay: float = 0.5, max_delay: float = 5.0, backoff_factor: float = 2.0, retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,), logger_name: str = 'DataMaster.Retry') -> Callable

带指数退避的重试装饰器

参数：

名称 类型 描述 默认
max_retries int

最大重试次数

 3
initial_delay float

初始延迟（秒）

 0.5
max_delay float

最大延迟（秒）

 5.0
backoff_factor float

退避因子

 2.0
retryable_exceptions Tuple[Type[Exception], ...]

可重试的异常类型

 (Exception,)
logger_name str

日志记录器名称

 'DataMaster.Retry'

返回：

类型 描述
Callable

装饰器函数
Example

@retry_with_backoff(max_retries=3, initial_delay=0.5) def get_data(): # 可能失败的操作 pass

源代码位于： utils/retry_utils.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def retry_with_backoff(
    max_retries: int = 3,
    initial_delay: float = 0.5,
    max_delay: float = 5.0,
    backoff_factor: float = 2.0,
    retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,),
    logger_name: str = "DataMaster.Retry"
) -> Callable:
    """
    带指数退避的重试装饰器

    Args:
        max_retries: 最大重试次数
        initial_delay: 初始延迟（秒）
        max_delay: 最大延迟（秒）
        backoff_factor: 退避因子
        retryable_exceptions: 可重试的异常类型
        logger_name: 日志记录器名称

    Returns:
        装饰器函数

    Example:
        @retry_with_backoff(max_retries=3, initial_delay=0.5)
        def get_data():
            # 可能失败的操作
            pass
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            logger = logging.getLogger(logger_name)
            delay = initial_delay
            last_exception = None

            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except retryable_exceptions as e:
                    last_exception = e
                    if attempt < max_retries:
                        logger.debug(
                            f"{func.__name__} 重试 {attempt+1}/{max_retries}, "
                            f"延迟 {delay:.2f}s, 错误: {e}"
                        )
                        time.sleep(delay)
                        delay = min(delay * backoff_factor, max_delay)
                    else:
                        logger.warning(
                            f"{func.__name__} 重试{max_retries}次后失败: {e}"
                        )

            # 所有重试失败
            if last_exception:
                raise last_exception
            return None

        return wrapper
    return decorator

retry_on_failure 

retry_on_failure(func: Callable, max_retries: int = 3, delay: float = 0.5, backoff: bool = True, on_retry: Callable = None, logger: Logger = None) -> Any

执行带重试的函数调用

参数：

名称 类型 描述 默认
func Callable

要执行的函数

 必需
max_retries int

最大重试次数

 3
delay float

重试延迟（秒）

 0.5
backoff bool

是否使用指数退避

 True
on_retry Callable

重试时的回调函数

 None
logger Logger

日志记录器

 None

返回：

类型 描述
Any

函数执行结果
Example

result = retry_on_failure( lambda: api.get_data(), max_retries=3, delay=0.5 )

源代码位于： utils/retry_utils.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def retry_on_failure(
    func: Callable,
    max_retries: int = 3,
    delay: float = 0.5,
    backoff: bool = True,
    on_retry: Callable = None,
    logger: logging.Logger = None
) -> Any:
    """
    执行带重试的函数调用

    Args:
        func: 要执行的函数
        max_retries: 最大重试次数
        delay: 重试延迟（秒）
        backoff: 是否使用指数退避
        on_retry: 重试时的回调函数
        logger: 日志记录器

    Returns:
        函数执行结果

    Example:
        result = retry_on_failure(
            lambda: api.get_data(),
            max_retries=3,
            delay=0.5
        )
    """
    if logger is None:
        logger = logging.getLogger("DataMaster.Retry")

    current_delay = delay
    last_exception = None

    for attempt in range(max_retries + 1):
        try:
            return func()
        except Exception as e:
            last_exception = e

            if attempt < max_retries:
                if on_retry:
                    on_retry(attempt + 1, e)

                logger.debug(
                    f"重试 {attempt+1}/{max_retries}, "
                    f"延迟 {current_delay:.2f}s, 错误: {e}"
                )

                time.sleep(current_delay)

                if backoff:
                    current_delay = min(current_delay * 2, 10.0)
            else:
                logger.warning(f"重试{max_retries}次后失败: {e}")

    return None

utils.lib_loader

lib_loader

动态库加载管理器

支持内置库和系统库的智能加载和切换

LibLoader 

LibLoader(config: dict = None)

库加载管理器

初始化库加载器

参数：

名称 类型 描述 默认
config dict

配置字典,包含use_builtin_libs等选项

 None
源代码位于： utils/lib_loader.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(self, config: dict = None):
    """
    初始化库加载器

    Args:
        config: 配置字典,包含use_builtin_libs等选项
    """
    self.config = config or {}
    self.logger = logging.getLogger("LibLoader")

    # 获取StockDataMaster根目录
    self.root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    self.lib_dir = os.path.join(self.root_dir, 'lib')

    # 是否使用内置库(默认优先使用内置库)
    self.use_builtin = self.config.get('use_builtin_libs', True)

    # 已加载的库缓存
    self._loaded_libs = {}
load_library 
load_library(lib_name: str, fallback: bool = True) -> Optional[Any]

智能加载库(内置库或系统库)

参数：

名称 类型 描述 默认
lib_name str

库名称 (mootdx/baostock/tushare/pandas等)

 必需
fallback bool

是否在失败时降级尝试另一种方式

 True

返回：

类型 描述
Optional[Any]

加载的库模块,失败返回None
源代码位于： utils/lib_loader.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def load_library(self, lib_name: str, fallback: bool = True) -> Optional[Any]:
    """
    智能加载库(内置库或系统库)

    Args:
        lib_name: 库名称 (mootdx/baostock/tushare/pandas等)
        fallback: 是否在失败时降级尝试另一种方式

    Returns:
        加载的库模块,失败返回None
    """
    # 检查缓存
    if lib_name in self._loaded_libs:
        return self._loaded_libs[lib_name]

    lib_module = None

    if self.use_builtin:
        # 优先尝试内置库
        lib_module = self._load_builtin_lib(lib_name)

        if lib_module is None and fallback:
            # 内置库失败,降级到系统库
            self.logger.warning(f"内置库{lib_name}加载失败,尝试系统库")
            lib_module = self._load_system_lib(lib_name)
    else:
        # 优先尝试系统库
        lib_module = self._load_system_lib(lib_name)

        if lib_module is None and fallback:
            # 系统库失败,降级到内置库
            self.logger.warning(f"系统库{lib_name}加载失败,尝试内置库")
            lib_module = self._load_builtin_lib(lib_name)

    # 缓存结果
    if lib_module is not None:
        self._loaded_libs[lib_name] = lib_module
        self.logger.info(f"库{lib_name}加载成功")
    else:
        self.logger.error(f"库{lib_name}加载失败(内置和系统库均不可用)")

    return lib_module
check_library 
check_library(lib_name: str) -> dict

检查库的可用性

参数：

名称 类型 描述 默认
lib_name str

库名称

 必需

返回：

类型 描述
dict

检查结果字典
源代码位于： utils/lib_loader.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def check_library(self, lib_name: str) -> dict:
    """
    检查库的可用性

    Args:
        lib_name: 库名称

    Returns:
        检查结果字典
    """
    result = {
        'name': lib_name,
        'builtin_available': False,
        'system_available': False,
        'builtin_path': None,
        'system_version': None,
        'loaded_from': None
    }

    # 检查内置库
    builtin_path = os.path.join(self.lib_dir, lib_name)
    if os.path.exists(builtin_path):
        result['builtin_available'] = True
        result['builtin_path'] = builtin_path

    # 检查系统库
    try:
        module = importlib.import_module(lib_name)
        result['system_available'] = True
        result['system_version'] = getattr(module, '__version__', 'unknown')
    except ImportError:
        pass

    # 检查当前加载状态
    if lib_name in self._loaded_libs:
        result['loaded_from'] = 'cached'

    return result
get_library_status 
get_library_status() -> dict

获取所有库的加载状态

返回：

类型 描述
dict

状态字典
源代码位于： utils/lib_loader.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def get_library_status(self) -> dict:
    """
    获取所有库的加载状态

    Returns:
        状态字典
    """
    libs = ['mootdx', 'baostock', 'tushare', 'pandas']

    status = {
        'use_builtin': self.use_builtin,
        'lib_dir': self.lib_dir,
        'libraries': {}
    }

    for lib in libs:
        status['libraries'][lib] = self.check_library(lib)

    return status
reload_library 
reload_library(lib_name: str) -> bool

重新加载库

参数：

名称 类型 描述 默认
lib_name str

库名称

 必需

返回：

类型 描述
bool

是否成功
源代码位于： utils/lib_loader.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def reload_library(self, lib_name: str) -> bool:
    """
    重新加载库

    Args:
        lib_name: 库名称

    Returns:
        是否成功
    """
    if lib_name in self._loaded_libs:
        del self._loaded_libs[lib_name]

    lib = self.load_library(lib_name)
    return lib is not None

get_lib_loader 

get_lib_loader(config: dict = None) -> LibLoader

获取全局库加载器实例(单例)

参数：

名称 类型 描述 默认
config dict

配置字典

 None

返回：

类型 描述
LibLoader

LibLoader实例
源代码位于： utils/lib_loader.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def get_lib_loader(config: dict = None) -> LibLoader:
    """
    获取全局库加载器实例(单例)

    Args:
        config: 配置字典

    Returns:
        LibLoader实例
    """
    global _global_loader
    if _global_loader is None:
        _global_loader = LibLoader(config)
    return _global_loader