跳转至

工具模块 API 参考

QKA 的工具模块，提供日志系统、颜色输出和各种实用工具函数。

qka.utils.logger

日志系统模块，提供结构化的日志记录功能。

JSONFormatter

Bases: Formatter

JSON格式的日志格式化器

Source code in qka/utils/logger.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class JSONFormatter(logging.Formatter):
    """JSON格式的日志格式化器"""

    def format(self, record):
        log_entry = {
            'timestamp': datetime.fromtimestamp(record.created).isoformat(),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno
        }

        # 添加异常信息
        if record.exc_info:
            log_entry['exception'] = self.formatException(record.exc_info)

        # 添加自定义字段
        if hasattr(record, 'extra_fields'):
            log_entry.update(record.extra_fields)

        return json.dumps(log_entry, ensure_ascii=False)

ColoredFormatter

Bases: Formatter

带颜色的控制台格式化器

Source code in qka/utils/logger.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class ColoredFormatter(logging.Formatter):
    """带颜色的控制台格式化器"""

    # ANSI颜色代码
    COLORS = {
        'DEBUG': '\033[36m',     # 青色
        'INFO': '\033[32m',      # 绿色
        'WARNING': '\033[33m',   # 黄色
        'ERROR': '\033[31m',     # 红色
        'CRITICAL': '\033[35m',  # 紫色
        'RESET': '\033[0m'       # 重置
    }

    def format(self, record):
        color = self.COLORS.get(record.levelname, self.COLORS['RESET'])
        reset = self.COLORS['RESET']

        # 格式化消息
        formatted = super().format(record)
        return f"{color}{formatted}{reset}"

RemoveAnsiEscapeCodes

Bases: Filter

移除ANSI转义码的过滤器

Source code in qka/utils/logger.py 
60
61
62
63
64
65
class RemoveAnsiEscapeCodes(logging.Filter):
    """移除ANSI转义码的过滤器"""

    def filter(self, record):
        record.msg = re.sub(r'\033\[[0-9;]*m', '', str(record.msg))
        return True

StructuredLogger

结构化日志记录器

Source code in qka/utils/logger.py 
 68
 69
 70
 71
 72
 73
 74
 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
class StructuredLogger:
    """结构化日志记录器"""

    def __init__(self, name: str = 'qka'):
        self.logger = logging.getLogger(name)
        self._lock = threading.Lock()

    def log(self, level: int, message: str, **kwargs):
        """记录结构化日志"""
        with self._lock:
            # 创建日志记录，附加额外字段
            if kwargs:
                record = self.logger.makeRecord(
                    self.logger.name, level, '', 0, message, (), None
                )
                record.extra_fields = kwargs
                self.logger.handle(record)
            else:
                self.logger.log(level, message)

    def debug(self, message: str, **kwargs):
        self.log(logging.DEBUG, message, **kwargs)

    def info(self, message: str, **kwargs):
        self.log(logging.INFO, message, **kwargs)

    def warning(self, message: str, **kwargs):
        self.log(logging.WARNING, message, **kwargs)

    def error(self, message: str, **kwargs):
        self.log(logging.ERROR, message, **kwargs)

    def critical(self, message: str, **kwargs):
        self.log(logging.CRITICAL, message, **kwargs)

log(level, message, **kwargs)

记录结构化日志

源代码位于： qka/utils/logger.py 
75
76
77
78
79
80
81
82
83
84
85
86
def log(self, level: int, message: str, **kwargs):
    """记录结构化日志"""
    with self._lock:
        # 创建日志记录，附加额外字段
        if kwargs:
            record = self.logger.makeRecord(
                self.logger.name, level, '', 0, message, (), None
            )
            record.extra_fields = kwargs
            self.logger.handle(record)
        else:
            self.logger.log(level, message)

create_logger(name='qka', level='INFO', console_output=True, file_output=True, log_dir='logs', max_file_size='10MB', backup_count=10, json_format=False, colored_console=True)

创建增强的日志记录器

参数：

名称 类型 描述 默认
name str

日志记录器名称

 'qka'
level str

日志级别

 'INFO'
console_output bool

是否输出到控制台

 True
file_output bool

是否输出到文件

 True
log_dir str

日志文件目录

 'logs'
max_file_size str

最大文件大小

 '10MB'
backup_count int

备份文件数量

 10
json_format bool

是否使用JSON格式

 False
colored_console bool

控制台是否使用颜色

 True
源代码位于： qka/utils/logger.py 
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
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
170
171
172
173
174
175
176
177
178
def create_logger(
    name: str = 'qka',
    level: str = 'INFO',
    console_output: bool = True,
    file_output: bool = True,
    log_dir: str = 'logs',
    max_file_size: str = '10MB',
    backup_count: int = 10,
    json_format: bool = False,
    colored_console: bool = True
):
    """
    创建增强的日志记录器

    Args:
        name: 日志记录器名称
        level: 日志级别
        console_output: 是否输出到控制台
        file_output: 是否输出到文件
        log_dir: 日志文件目录
        max_file_size: 最大文件大小
        backup_count: 备份文件数量
        json_format: 是否使用JSON格式
        colored_console: 控制台是否使用颜色
    """
    logger = logging.getLogger(name)
    logger.setLevel(getattr(logging, level.upper()))

    # 清除现有处理器
    logger.handlers.clear()

    # 创建格式化器
    if json_format:
        file_formatter = JSONFormatter()
        console_formatter = JSONFormatter()
    else:
        file_formatter = logging.Formatter(
            '[%(levelname)s][%(asctime)s][%(name)s] %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S'
        )
        console_formatter = ColoredFormatter(
            '[%(levelname)s][%(asctime)s] %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S'
        ) if colored_console else file_formatter

    # 控制台处理器
    if console_output:
        stream_handler = logging.StreamHandler()
        stream_handler.setFormatter(console_formatter)
        logger.addHandler(stream_handler)

    # 文件处理器
    if file_output:
        # 创建日志文件夹
        Path(log_dir).mkdir(parents=True, exist_ok=True)

        # 解析文件大小
        size_map = {'KB': 1024, 'MB': 1024*1024, 'GB': 1024*1024*1024}
        size_unit = max_file_size[-2:].upper()
        size_value = int(max_file_size[:-2])
        max_bytes = size_value * size_map.get(size_unit, 1024*1024)

        # 创建轮转文件处理器
        file_handler = RotatingFileHandler(
            f"{log_dir}/{date.today().strftime('%Y-%m-%d')}.log",
            maxBytes=max_bytes,
            backupCount=backup_count,
            encoding='utf-8'
        )
        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(file_formatter)
        file_handler.addFilter(RemoveAnsiEscapeCodes())
        logger.addHandler(file_handler)

    return logger

setup_logging_from_config(config_dict)

从配置字典设置日志

源代码位于： qka/utils/logger.py 
181
182
183
def setup_logging_from_config(config_dict: Dict[str, Any]) -> logging.Logger:
    """从配置字典设置日志"""
    return create_logger(**config_dict)

get_structured_logger(name='qka')

获取结构化日志记录器

源代码位于： qka/utils/logger.py 
186
187
188
def get_structured_logger(name: str = 'qka') -> StructuredLogger:
    """获取结构化日志记录器"""
    return StructuredLogger(name)

add_wechat_handler(logger_instance, webhook_url, level='ERROR')

为日志记录器添加微信通知处理器

参数：

名称 类型 描述 默认
logger_instance Logger

日志记录器实例

 必需
webhook_url str

企业微信机器人webhook地址

 必需
level str

通知级别

 'ERROR'
源代码位于： qka/utils/logger.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def add_wechat_handler(logger_instance: logging.Logger, webhook_url: str, level: str = 'ERROR'):
    """
    为日志记录器添加微信通知处理器

    Args:
        logger_instance: 日志记录器实例
        webhook_url: 企业微信机器人webhook地址
        level: 通知级别
    """
    wechat_handler = WeChatHandler(webhook_url)
    wechat_handler.setLevel(getattr(logging, level.upper()))

    # 创建简单格式化器用于微信消息
    formatter = logging.Formatter(
        '[%(levelname)s][%(asctime)s] %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    wechat_handler.setFormatter(formatter)
    wechat_handler.addFilter(RemoveAnsiEscapeCodes())
    logger_instance.addHandler(wechat_handler)

主要功能

create_logger 函数

创建增强的日志记录器，支持控制台和文件输出。

StructuredLogger 类

结构化日志记录器，支持附加额外字段。

WeChatHandler 类

微信消息处理器，支持通过企业微信机器人发送日志。

使用示例

from qka.utils.logger import create_logger

# 创建日志记录器
logger = create_logger(
    name='my_strategy',
    level='DEBUG',
    console_output=True,
    file_output=True,
    log_dir='my_logs'
)

# 记录日志
logger.info("策略开始运行")
logger.error("交易失败", symbol='000001.SZ', price=10.5)

qka.utils.util

通用工具函数模块。

add_stock_suffix(stock_code)

为给定的股票代码添加相应的后缀。

源代码位于： qka/utils/util.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def add_stock_suffix(stock_code):
    """
    为给定的股票代码添加相应的后缀。
    """
    # 检查股票代码是否为6位数字
    if len(stock_code) != 6 or not stock_code.isdigit():
        raise ValueError("股票代码必须是6位数字")

    # 根据股票代码的前缀添加相应的后缀
    if stock_code.startswith("00") or stock_code.startswith("30") or stock_code.startswith("15") or stock_code.startswith("16") or stock_code.startswith("18") or stock_code.startswith("12"):
        return f"{stock_code}.SZ"  # 深圳证券交易所
    elif stock_code.startswith("60") or stock_code.startswith("68") or stock_code.startswith("11"):
        return f"{stock_code}.SH"  # 上海证券交易所
    elif stock_code.startswith("83") or stock_code.startswith("43"):
        return f"{stock_code}.BJ"  # 北京证券交易所

    return f"{stock_code}.SH"

timestamp_to_datetime_string(timestamp)

将时间戳转换为时间字符串。

:param timestamp: 时间戳（秒级） :return: 格式化的时间字符串 'YYYY-MM-DD HH:MM:SS'

源代码位于： qka/utils/util.py 
23
24
25
26
27
28
29
30
31
32
def timestamp_to_datetime_string(timestamp):
    """
    将时间戳转换为时间字符串。

    :param timestamp: 时间戳（秒级）
    :return: 格式化的时间字符串 'YYYY-MM-DD HH:MM:SS'
    """
    dt_object = datetime.fromtimestamp(timestamp)
    time_string = dt_object.strftime('%Y-%m-%d %H:%M:%S')
    return time_string

qka.utils.anis

ANSI 颜色代码工具，提供带颜色的控制台输出。

使用示例

from qka.utils.anis import RED, GREEN, BLUE, RESET

print(f"{RED}错误信息{RESET}")
print(f"{GREEN}成功信息{RESET}")
print(f"{BLUE}提示信息{RESET}")

模块导入方式

工具模块需要从子模块导入：

# 日志系统
from qka.utils.logger import create_logger, StructuredLogger

# 工具函数
from qka.utils.util import timestamp_to_datetime_string

# 颜色输出
from qka.utils.anis import RED, GREEN, BLUE, RESET

相关链接