07 Python 调试技巧与日志系统

资料推荐：做量化这么多年，我收集、整理了大量通达信选股公式，全部源码、无加密，如需购买，可在**小红书搜索「疯狂量化」**进店，也可直接点击小红书商品页下单。
开源代码：

https://github.com/z1041950008/deyide_quant— 张得意量化工具箱，脚本与示例随仓库更新。

📖 引言

实盘还没上，时间先耗在「为什么这里多了个 NaN」「为什么回测和实盘对不上」——十有七八是调试和日志没打好，不是策略有多深奥。

这篇就两件事：pdb怎么停在当下看一眼变量、logging怎么记得住哪天哪笔出的问题。别用打印一霸梭到天亮。

能带走就行

• 会设断点、会单步，别只会print(df.head())
• 知道logging比print强在哪儿（级别、输出到文件）
• 异常别乱except: pass，那等于把雷埋给以后的自己

🔧 核心概念

调试的四个层次

层次 1: Print 调试          → 快速但混乱
层次 2: Assert 断言         → 验证假设
层次 3: PDB 交互式调试      → 精确控制
层次 4: Logging 日志系统    → 专业可追溯

量化代码的常见错误类型

💻 代码实现

完整的调试与日志示例

"""
debug_examples.py - 量化代码调试与日志完整示例
"""
import logging
import pdb
import sys
from datetime import datetime
from pathlib import Path

# ============================================================================
# 第一部分：日志系统配置
# ============================================================================

def setup_logging(log_file: str = 'quant_trading.log', level: int = logging.INFO):
    """
    配置专业日志系统

    日志级别:
    - DEBUG: 详细调试信息
    - INFO: 一般信息
    - WARNING: 警告信息
    - ERROR: 错误信息
    - CRITICAL: 严重错误
    """
    # 创建日志目录
    Path('logs').mkdir(exist_ok=True)

    # 配置日志格式
    log_format = logging.Formatter(
        fmt='%(asctime)s | %(levelname)-8s | %(name)s | %(filename)s:%(lineno)d | %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )

    # 文件处理器
    file_handler = logging.FileHandler(f'logs/{log_file}', encoding='utf-8')
    file_handler.setFormatter(log_format)

    # 控制台处理器
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(log_format)

    # 配置根日志器
    logger = logging.getLogger()
    logger.setLevel(level)
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)

    return logger

# 初始化日志
logger = setup_logging()

# ============================================================================
# 第二部分：Print 调试的替代方案
# ============================================================================

def debug_with_logging():
    """使用日志代替 print"""
    logger.debug("开始执行函数")

    prices = [100, 102, 101, 105, 108]
    logger.debug(f"价格数据：{prices}")

    returns = [(prices[i] - prices[i-1]) / prices[i-1] for i in range(1, len(prices))]
    logger.info(f"计算完成，收益率：{returns}")

    avg_return = sum(returns) / len(returns)
    logger.info(f"平均收益率：{avg_return:.4%}")

    return returns

# ============================================================================
# 第三部分：Assert 断言验证
# ============================================================================

def calculate_portfolio_return(weights: list, returns: list) -> float:
    """
    计算组合收益率

    使用断言验证输入数据
    """
    # 数据验证
    assert len(weights) == len(returns), "权重和收益率长度必须一致"
    assert abs(sum(weights) - 1.0) < 0.01, "权重之和必须为 1"
    assert all(w >= 0 for w in weights), "权重不能为负"

    logger.debug(f"权重：{weights}, 收益率：{returns}")

    portfolio_return = sum(w * r for w, r in zip(weights, returns))

    # 结果验证
    assert isinstance(portfolio_return, (int, float)), "返回值必须是数字"

    return portfolio_return

# ============================================================================
# 第四部分：PDB 调试器实战
# ============================================================================

def debug_with_pdb():
    """演示 PDB 调试器使用"""
    logger.info("进入 PDB 调试模式")

    data = {'close': [100, 102, 101, 105, 108, 107, 110]}
    window = 3

    # 设置断点
    # 方法 1: 在代码中插入
    # pdb.set_trace()

    # 方法 2: 使用 breakpoint() (Python 3.7+)
    breakpoint()

    # PDB 常用命令:
    # n (next)     - 执行下一行
    # c (continue) - 继续执行到下一个断点
    # q (quit)     - 退出调试
    # p variable   - 打印变量
    # l (list)     - 显示代码
    # w (where)    - 显示调用栈
    # s (step)     - 进入函数内部

    ma = []
    for i in range(window - 1, len(data['close'])):
        window_data = data['close'][i-window+1:i+1]
        ma.append(sum(window_data) / window)

    return ma

# ============================================================================
# 第五部分：异常处理最佳实践
# ============================================================================

class DataValidationError(Exception):
    """数据验证异常"""
    pass

class CalculationError(Exception):
    """计算异常"""
    pass

def safe_calculate_return(prices: list) -> list:
    """
    安全计算收益率

    包含完整的异常处理
    """
    try:
        # 输入验证
        if not prices:
            raise DataValidationError("价格列表不能为空")

        if len(prices) < 2:
            raise DataValidationError("至少需要两个价格")

        if any(p <= 0 for p in prices):
            raise DataValidationError("价格必须为正数")

        logger.debug(f"输入验证通过，价格数量：{len(prices)}")

        # 计算收益率
        returns = []
        for i in range(1, len(prices)):
            if prices[i-1] == 0:
                raise CalculationError(f"第{i-1}个价格为 0，无法计算收益率")

            ret = (prices[i] - prices[i-1]) / prices[i-1]

            # 检查异常收益率
            if abs(ret) > 0.5:  # 单日涨跌超过 50%
                logger.warning(f"第{i}天收益率异常：{ret:.2%}")

            returns.append(ret)

        logger.info(f"成功计算 {len(returns)} 个收益率")
        return returns

    except DataValidationError as e:
        logger.error(f"数据验证失败：{e}")
        raise
    except CalculationError as e:
        logger.error(f"计算错误：{e}")
        raise
    except Exception as e:
        logger.exception(f"未知错误：{e}")
        raise

# ============================================================================
# 第六部分：装饰器日志
# ============================================================================

def log_function_call(func):
    """
    函数调用日志装饰器

    自动记录函数调用参数、返回值、执行时间
    """
    import functools
    import time

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        logger.debug(f"调用 {func.__name__}, 参数：args={args}, kwargs={kwargs}")

        try:
            result = func(*args, **kwargs)
            elapsed = time.time() - start_time
            logger.debug(f"{func.__name__} 完成，耗时：{elapsed:.4f}秒，返回：{result}")
            return result
        except Exception as e:
            elapsed = time.time() - start_time
            logger.error(f"{func.__name__} 异常，耗时：{elapsed:.4f}秒，错误：{e}")
            raise

    return wrapper

@log_function_call
def calculate_sma(prices: list, window: int) -> list:
    """计算简单移动平均"""
    if len(prices) < window:
        raise ValueError("数据长度不足")

    ma = []
    for i in range(window - 1, len(prices)):
        ma.append(sum(prices[i-window+1:i+1]) / window)

    return ma

# ============================================================================
# 第七部分：量化策略调试实例
# ============================================================================

class TradingStrategyDebugger:
    """
    交易策略调试器

    记录策略执行的每个步骤，便于排查问题
    """

    def __init__(self, strategy_name: str):
        self.strategy_name = strategy_name
        self.signal_log = []
        self.error_log = []

    def generate_signal(self, df: dict) -> int:
        """
        生成交易信号

        记录每个判断步骤
        """
        try:
            logger.info(f"[{self.strategy_name}] 开始生成信号")

            # 步骤 1: 数据检查
            required_cols = ['close', 'ma20', 'ma60']
            for col in required_cols:
                if col not in df:
                    raise KeyError(f"缺少必要列：{col}")

            logger.debug(f"[{self.strategy_name}] 数据检查通过")

            # 步骤 2: 获取最新数据
            latest_close = df['close'].iloc[-1]
            latest_ma20 = df['ma20'].iloc[-1]
            latest_ma60 = df['ma60'].iloc[-1]

            logger.debug(f"[{self.strategy_name}] 最新数据：close={latest_close:.2f}, "
                        f"ma20={latest_ma20:.2f}, ma60={latest_ma60:.2f}")

            # 步骤 3: 信号判断
            if latest_ma20 > latest_ma60:
                signal = 1
                reason = "MA20 上穿 MA60"
            elif latest_ma20 < latest_ma60:
                signal = -1
                reason = "MA20 下穿 MA60"
            else:
                signal = 0
                reason = "无明确信号"

            # 记录信号
            self.signal_log.append({
                'timestamp': datetime.now(),
                'signal': signal,
                'reason': reason,
                'data': {
                    'close': latest_close,
                    'ma20': latest_ma20,
                    'ma60': latest_ma60
                }
            })

            logger.info(f"[{self.strategy_name}] 信号：{signal}, 原因：{reason}")

            return signal

        except Exception as e:
            self.error_log.append({
                'timestamp': datetime.now(),
                'error': str(e)
            })
            logger.exception(f"[{self.strategy_name}] 信号生成失败：{e}")
            return 0

    def print_debug_report(self):
        """打印调试报告"""
        print("=" * 60)
        print(f"策略调试报告 - {self.strategy_name}")
        print("=" * 60)

        print(f"\n信号记录数量：{len(self.signal_log)}")
        if self.signal_log:
            print("\n最近 5 个信号:")
            for log in self.signal_log[-5:]:
                print(f"  {log['timestamp']}: signal={log['signal']}, reason={log['reason']}")

        print(f"\n错误记录数量：{len(self.error_log)}")
        if self.error_log:
            print("\n错误列表:")
            for log in self.error_log:
                print(f"  {log['timestamp']}: {log['error']}")

# ============================================================================
# 主程序 - 演示所有调试功能
# ============================================================================

if __name__ == "__main__":
    logger.info("=" * 60)
    logger.info("调试与日志系统演示")
    logger.info("=" * 60)

    # 1. 日志调试
    print("\n【1. 日志调试示例】")
    debug_with_logging()

    # 2. 断言验证
    print("\n【2. 断言验证示例】")
    try:
        ret = calculate_portfolio_return([0.5, 0.5], [0.1, 0.2])
        logger.info(f"组合收益率：{ret:.2%}")
    except AssertionError as e:
        logger.error(f"断言失败：{e}")

    # 3. 异常处理
    print("\n【3. 异常处理示例】")
    try:
        returns = safe_calculate_return([100, 102, 101, 105])
        logger.info(f"收益率：{returns}")
    except Exception as e:
        logger.error(f"计算失败：{e}")

    # 4. 装饰器日志
    print("\n【4. 装饰器日志示例】")
    sma = calculate_sma([100, 102, 101, 105, 108], 3)
    logger.info(f"SMA 结果：{sma}")

    # 5. 策略调试器
    print("\n【5. 策略调试器示例】")
    import pandas as pd
    import numpy as np

    # 生成测试数据
    dates = pd.date_range('2023-01-01', periods=100)
    close = pd.Series(100 * np.cumprod(1 + np.random.normal(0.001, 0.02, 100)), index=dates)
    df = pd.DataFrame({
        'close': close,
        'ma20': close.rolling(20).mean(),
        'ma60': close.rolling(60).mean()
    })

    debugger = TradingStrategyDebugger("DualMA")
    for i in range(20, 100, 10):
        debugger.generate_signal(df.iloc[:i])

    debugger.print_debug_report()

    logger.info("=" * 60)
    logger.info("演示完成")
    logger.info("=" * 60)

🎯 实战应用

日志输出示例

2024-01-15 10:30:15 | INFO     | root | debug_examples.py:45 | 开始执行函数
2024-01-15 10:30:15 | DEBUG    | root | debug_examples.py:48 | 价格数据：[100, 102, 101, 105, 108]
2024-01-15 10:30:15 | INFO     | root | debug_examples.py:51 | 计算完成，收益率：[0.02, -0.0098, 0.0396, 0.0286]
2024-01-15 10:30:15 | INFO     | root | debug_examples.py:54 | 平均收益率：1.9604%

PDB 调试会话示例

> /path/to/debug_examples.py(78)debug_with_pdb()
     76     # 方法 2: 使用 breakpoint() (Python 3.7+)
     77     breakpoint()
     78 
     79     ma = []
     80     for i in range(window - 1, len(data['close'])):

(Pdb) p data
{'close': [100, 102, 101, 105, 108, 107, 110]}

(Pdb) p window
3

(Pdb) n
> /path/to/debug_examples.py(80)debug_with_pdb()

(Pdb) c

❓ 常见问题

Q1: 生产环境应该用什么日志级别？

答案:

• 开发环境：DEBUG
• 测试环境：INFO
• 生产环境：WARNING 或 ERROR

Q2: 如何处理大量日志文件？

解决方案: 使用日志轮转

from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler(
    'app.log',
    maxBytes=10*1024*1024,  # 10MB
    backupCount=5  # 保留 5 个备份
)

Q3: 调试时如何避免修改代码？

解决方案: 使用条件断点

# 只在特定条件下触发
if some_condition:
    pdb.set_trace()

# 或使用 pdb.set_trace() 的 condition 参数

📝 小结

线上排错靠日志，本地抠逻辑靠断点。print能应急，别当长期方案；异常写清楚堆栈比你加一百句「出错了」有用。策略调试器那套，有了更好，没有也别停在手工猜上。

仓库链接：https://github.com/z1041950008/deyide_quant