第十二章：回测框架初探：API与基本结构

如果说前十一章是对软件操作的指南，相对易于掌握，那么本章将深入到整个系统的核心——回测框架。本章是实现量化交易思想的关键，旨在为开发者提供一个关于”看海量化交易系统”策略框架的直观理解，后续将基于此框架，一步步构建属于自己的交易策略。我们将重点介绍策略文件的基本结构、核心回调函数，以及与框架交互的关键API。
需要注意的是，本节所讲到的API是全功能的，尤其是回调函数的入口参数使用起来较为复杂，对此我开发了一系列便于使用的函数，比如快速查询当前时间、快速查询当前账户资金、快速查询当前持仓情况、快速生成交易信号等等，这些函数工具将在后面章节中详细介绍。因此对于本章的内容，大家可以先大体浏览，对框架结构有一个直观的感知，后续章节中我将会对如何使用函数和API举一些应用案例，届时有不明白的可再行查阅。

---

策略框架是什么

对于每一位交易者而言，心中都有一套独特的交易逻辑。量化交易的本质，便是将这套逻辑转化为代码，让计算机去执行。然而，从零开始编写一套完整的交易程序是复杂且耗时的，需要处理数据获取、订单管理、风险控制、事件驱动等一系列繁琐的底层任务。

策略框架正是为了解决这一问题而生。可以将它理解为一个半成品的交易机器人，它已经为开发者搭建好了坚实的骨架，处理了所有与交易逻辑无关的技术细节。开发者只需要专注于策略本身，将交易思想填充到框架预留的接口中，即可快速构建出功能完备、稳定可靠的自动化交易策略。

为了贴近主流的量化实践和用户习惯，khQuant 定义了一套符合直觉的策略框架。它主要包含四个核心部分，分别对应策略生命周期中的不同阶段：

• 初始化 (init): 策略开始运行时，用于进行全局设置，仅执行一次。
• 盘前回调 (khPreMarket): 每个交易日开盘前，用于执行每日的准备工作。
• 主回调 (khHandlebar): 在盘中根据设定的频率被反复调用，是策略逻辑的核心。
• 盘后回调 (khPostMarket): 每个交易日收盘后，用于执行当日的复盘和清理任务。

这四个部分均以函数的形式存在于策略文件中。框架会在特定的时间点自动调用这些函数，并将包含当前市场行情、账户资金、持仓情况等所有必要信息的 context 对象作为参数传递给它们。在函数中完成逻辑判断后，只需返回标准的交易指令（我们称之为”信号”），框架便会自动执行后续的下单操作。这种设计极大地简化了策略编写的复杂度，使开发者可以聚焦于策略逻辑本身，而非底层实现。

---

12.1 策略框架一览

一个策略文件本质上是一个标准的Python脚本，通过实现框架预定义的一系列函数来完成策略逻辑。以下是一个策略文件的最简化结构，它包含了所有主要的回调函数，可以作为开始编写新策略的模板。

from typing import Dict, List

def init(stock_list, context):
    """
    策略初始化函数，在任务开始时仅执行一次。
    用于定义全局变量、加载外部数据等。
    """
    pass

def khHandlebar(context: Dict) -> List[Dict]:
    """
    策略核心逻辑，会被框架根据设定的频率反复调用。
    负责行情判断和生成交易信号。
    """
    signals = []
    return signals

def khPreMarket(context: Dict) -> List[Dict]:
    """
    盘前处理函数（可选）。
    在每日开盘前的指定时间点调用。
    """
    signals = []
    return signals

def khPostMarket(context: Dict) -> List[Dict]:
    """
    盘后处理函数（可选）。
    在每日收盘后的指定时间点调用。
    """
    signals = []
    return signals

---

12.2 核心回调函数详解

框架通过在特定时机调用策略文件中的特定函数来驱动策略运行。开发者需要根据需求实现这些函数。为了更好地理解，可以参考我们提供的 strategies/MACD.py 策略示例文件，它展示了如何实现一个完整的策略。

12.2.1 init(stock_list, context) – 初始化函数

• 执行时机：在整个回测或交易任务开始时，被框架调用一次。
• 核心作用：用于执行策略的全局初始化任务，如设置参数、加载数据等。
• 参数说明：
  • stock_list (list): 框架传入的股票池列表，例如 ['000001.SZ', '600000.SH']。
  • context (dict): 一个包含初始化时刻上下文信息的字典。其内部详细结构如下：
  • __current_time__ (dict): 包含当前时间信息的字典。
    • timestamp: (int) 标准Unix时间戳。
    • datetime: (str) 格式为 “YYYY-MM-DD HH:MM:SS” 的日期时间字符串。
    • date: (str) 格式为 “YYYY-MM-DD” 的日期字符串。
    • time: (str) 格式为 “HH:MM:SS” 的时间字符串。
  • __account__ (dict): 包含账户资金信息的字典。
    • account_type: (str) 账户类型 (例如, ‘STOCK’)。
    • account_id: (str) 账户ID。
    • cash: (float) 当前可用资金。
    • frozen_cash: (float) 冻结资金。
    • market_value: (float) 持仓市值。
    • total_asset: (float) 总资产。
    • benchmark: (str) 基准指数代码。
  • __positions__ (dict): 包含持仓信息的字典，在初始化时通常为空 {}。
  • __framework__ (object): 框架核心类的实例。它包含了最全面的框架信息和功能接口。如果其他上下文参数不包含所需信息，可以尝试通过此对象获取。

12.2.2 khHandlebar(context) – 策略主逻辑函数

• 执行时机：根据主界面”运行驱动区”设置的触发方式，被框架反复、高频地调用。
• 核心作用：这是实现交易策略核心逻辑的地方，包括行情判断、信号生成、下单等。
• 参数说明：
  • context (dict): 包含当前时间点所有可用信息的字典，其结构为：
  • __current_time__ (dict): 包含当前时间信息的字典。
    • timestamp: (int) 标准Unix时间戳。
    • datetime: (str) 格式为 “YYYY-MM-DD HH:MM:SS” 的日期时间字符串。
    • date: (str) 格式为 “YYYY-MM-DD” 的日期字符串。
    • time: (str) 格式为 “HH:MM:SS” 的时间字符串。
  • __account__ (dict): 包含账户资金信息的字典。
    • account_type: (str) 账户类型。
    • account_id: (str) 账户ID。
    • cash: (float) 当前可用资金。
    • frozen_cash: (float) 冻结资金。
    • market_value: (float) 持仓市值。
    • total_asset: (float) 总资产。
    • benchmark: (str) 基准指数代码。
  • __positions__ (dict): 包含当前持仓信息的字典。其结构为 {股票代码: 持仓详情}。持仓详情是一个字典，包含了该股票的详细持仓信息，其内部字段如下：
    • account_type: (int) 账户类型。
    • account_id: (str) 账户ID。
    • stock_code: (str) 股票代码。
    • volume: (int) 持仓数量。
    • can_use_volume: (int) 可用数量（可卖出数量）。
    • open_price: (float) 当日开盘价。
    • market_value: (float) 持仓市值。
    • frozen_volume: (int) 冻结数量。
    • on_road_volume: (int) 在途数量。
    • yesterday_volume: (int) 昨日持仓数量。
    • avg_price: (float) 持仓成本价。
    • current_price: (float) 当前市价。
    • direction: (int) 持仓方向。
    • profit: (float) 持仓浮动盈亏。
    • profit_ratio: (float) 持仓盈亏率。
  • __framework__ (object): 框架核心类的实例。它包含了最全面的框架信息和功能接口。如果其他上下文参数不包含所需信息，可以尝试通过此对象获取。
  • [股票代码] (pandas.Series): 以股票代码（如'000001.SZ'）为键，值为一个Pandas Series对象，包含了该股票在当前时间点的所有行情字段（如open, high, low, close, volume等）。
• 返回值:
  • 该函数需要返回一个交易信号列表 (List[Dict])。框架在收到返回的列表后，会自动解析其中的每一条指令，并调用底层的交易接口去执行。如果列表为空，则框架认为当前时间点无任何操作。
  • 一个标准的交易信号字典包含以下键值对：

键 (Key)	类型 (Type)	必填	说明
code	str	是	股票代码，必须是标准的QMT格式，例如 '000001.SZ' 或 '600036.SH'。
action	str	是	交易动作。可选值为 ‘buy’ (买入) 或 ‘sell’ (卖出)。
price	float	是	交易价格。
volume	int	是	交易数量，必须是100的整数倍。
reason	str	否	交易原因或备注。此信息会显示在日志和交易记录中。
timestamp	int	否	信号生成时的时间戳。

• 关于信号字典的更详细说明，请参考 12.7 交易信号详解。

12.2.3 khPreMarket(context) – 盘前处理函数（可选）

• 执行时机：在每个交易日的指定盘前时间点（如09:00）被调用一次。该功能需要在主界面”盘前盘后触发设置”中勾选”触发盘前回调”才会生效。
• 参数 context：其结构与 khHandlebar 函数的 context 完全相同。它包含了当天第一个行情数据点的所有信息，包括账户、持仓、以及股票池内所有股票在该时刻的行情数据。
• 返回值：与 khHandlebar 类似，返回一个交易信号列表 (List[Dict])。
• 常见用途：
  • 每日选股、计算因子。
  • 重置当日的交易状态或计数器。
  • 提前下达一些集合竞价阶段的预埋单。

12.2.4 khPostMarket(context) – 盘后处理函数（可选）

• 执行时机：在每个交易日的指定盘后时间点（如15:30）被调用一次。该功能需要在主界面”盘前盘后触发设置”中勾选”触发盘后回调”才会生效。
• 参数 context：其结构与 khHandlebar 函数的 context 完全相同。它包含了当天最后一个行情数据点的所有信息，包括账户、持仓、以及股票池内所有股票在该时刻的行情数据。
• 返回值：与 khHandlebar 类似，返回一个交易信号列表 (List[Dict])。
• 常见用途：
  • 当日交易复盘、业绩归因分析。
  • 保存当日的策略状态或数据到本地文件。
  • 清理当日持仓，或为下一个交易日做准备。

---

12.3 获取时间数据

在策略中，所有时间相关的信息都储存在 context['__current_time__'] 这个字典中。通过访问它，可以获取到策略当前执行点的精确时间。

• context['__current_time__']['timestamp']: 返回一个整数形式的Unix时间戳。
• context['__current_time__']['datetime']: 返回 YYYY-MM-DD HH:MM:SS 格式的字符串，最常用。
• context['__current_time__']['date']: 返回 YYYY-MM-DD 格式的日期字符串。
• context['__current_time__']['time']: 返回 HH:MM:SS 格式的时间字符串。

示例：实现简单的择时逻辑

def khHandlebar(context: Dict) -> List[Dict]:
    time_info = context['__current_time__']

    # 获取当前时间字符串，如 '09:31:00'
    current_time = time_info['time']

    # 简单的交易时间控制：只在上午10点后、下午2点半前进行交易判断
    if "10:00:00" < current_time < "14:30:00":
        # 在这里编写主要的策略逻辑...
        print("处于交易时间段，执行策略。")
    else:
        # 非交易时间段，不执行任何操作
        print("非交易时间段，跳过。")

    return []

---

12.4 获取账户数据

账户的资金状况信息储存在 context['__account__'] 字典中，它提供了账户资产的实时快照。

• context['__account__']['cash']: (float) 当前可用于交易的现金。
• context['__account__']['market_value']: (float) 所有持仓按当前市价计算的总市值。
• context['__account__']['total_asset']: (float) 总资产，即 cash + market_value。
• context['__account__']['frozen_cash']: (float) 因挂单而冻结的资金。

示例：根据资金动态计算买入量

def khHandlebar(context: Dict) -> List[Dict]:
    account = context['__account__']

    # 获取当前可用资金
    available_cash = account['cash']

    # 设定一个目标仓位：使用当前可用资金的20%
    cash_to_use = available_cash * 0.2

    # 获取 '000001.SZ' 的当前价格
    stock_data = context.get('000001.SZ')
    if stock_data is not None and not stock_data.empty:
        stock_price = stock_data['close']

        # 计算理论上可以买入多少股，并向下取整到100的倍数
        if stock_price > 0:
            volume_to_buy = int(cash_to_use / stock_price / 100) * 100
            if volume_to_buy > 0:
                print(f"资金充足，计划以 {stock_price} 元的价格买入 {volume_to_buy} 股 000001.SZ")
                # 此处可以构建并返回一个买入信号

    return []

---

12.5 获取持仓数据

持仓信息 context['__positions__'] 是一个字典，键为股票代码，值为该股票的详细持仓信息字典。这使得检查特定持仓、获取持仓细节变得非常方便。

示例：实现持仓股票的止盈止损

def khHandlebar(context: Dict) -> List[Dict]:
    positions = context['__positions__']
    signals = []

    # 检查是否持有 '600519.SH' (贵州茅台)
    if '600519.SH' in positions:
        # 获取该持仓的详细信息
        pos_info = positions['600519.SH']

        volume = pos_info['volume']
        profit_ratio = pos_info['profit_ratio'] # 获取持仓盈亏率

        print(f"持有 {volume} 股 600519.SH，当前盈利 {profit_ratio*100:.2f}%")

        # 止盈逻辑：如果盈利超过10%，就全部卖出
        if profit_ratio > 0.10:
            sell_signal = {
                'action': 'sell',
                'code': '600519.SH',
                'volume': volume, # 卖出全部持仓
                'remark': '盈利超过10%，止盈'
            }
            signals.append(sell_signal)

        # 止损逻辑：如果亏损超过5%，就全部卖出
        elif profit_ratio < -0.05:
            sell_signal = {
                'action': 'sell',
                'code': '600519.SH',
                'volume': volume, # 卖出全部持仓
                'remark': '亏损超过5%，止损'
            }
            signals.append(sell_signal)

    return signals

---

12.6 获取当前运行配置

在某些复杂的策略场景中，可能需要在策略逻辑内部获取在GUI界面上配置的参数，例如获取设置的基准合约代码，或者根据不同的手续费配置调整交易行为。

这些配置信息可以通过 context 中的 __framework__ 对象进行访问。

• context['__framework__'].config: 这是一个 khConfig.KhConfig 类的实例，持有本次运行的所有配置信息。

示例：在策略中获取基准合约和手续费配置

def init(stock_list, context):
    # 从框架对象中获取配置实例
    config = context['__framework__'].config

    # 读取回测配置中的 'benchmark' 参数
    benchmark_code = config.get_config('backtest.benchmark')
    print(f"当前配置的基准合约为: {benchmark_code}")

    # 读取交易成本中的佣金比例
    commission_rate = config.get_config('backtest.trade_cost.commission_rate')
    print(f"当前配置的佣金比例为: {commission_rate}")

---

12.7 交易信号详解

交易信号是策略与交易执行模块沟通的唯一方式。它是一个包含了所有交易必要信息的Python字典。khHandlebar、khPreMarket、khPostMarket的返回值以及__framework__.trade()的参数都是交易信号。

一个标准的交易信号字典包含以下键值对：

键 (Key)	类型 (Type)	必填	说明
code	str	是	股票代码，必须是标准的QMT格式，例如'000001.SZ' 或 '600036.SH'。
action	str	是	交易动作。可选值为 ‘buy’ (买入) 或 ‘sell’ (卖出)。
price	float	是	交易价格。
volume	int	是	交易数量，必须是100的整数倍。
reason	str	否	交易原因或备注。此信息会显示在日志和交易记录中。
timestamp	int	否	信号生成时的时间戳。

示例：

# 以10.5元的价格买入100股平安银行
signal_1 = {
    'code': '000001.SZ',
    'action': 'buy',
    'price': 10.5,
    'volume': 100,
    'reason': 'MACD金叉买入'
}

# 以18.5元的价格卖出200股贵州茅台
signal_2 = {
    'code': '600519.SH',
    'action': 'sell',
    'price': 18.50,
    'volume': 200,
    'reason': '达到止盈位卖出'
}

12.8 在策略中使用日志

为了方便调试和监控策略的内部状态，可以在策略代码中直接调用日志输出功能，将信息发送到看海量化交易系统的日志系统（包括界面和文件）。

核心机制：框架已经对Python内置的 logging 模块进行了配置。因此，开发者无需关心日志的底层实现，只需在策略代码中导入 logging 模块，然后直接调用其标准函数即可。

使用方法：

1. 在策略文件顶部添加 import logging。
2. 在代码中调用 logging.info(), logging.warning(), logging.error() 等函数。

日志级别与界面上显示的颜色直接对应：

• 普通信息 (INFO): 使用 logging.info("进入长仓条件判断") 输出常规的流程信息或变量状态。这对应界面上的 白色 文本。
• 调试信息 (DEBUG): 如果需要输出更详细的、仅在调试时关心的变量值或中间计算结果，可以使用 logging.debug(f"当前ATR值: {atr_value}")。这对应界面上的 浅紫色 文本。（注意：默认配置下，DEBUG级别的日志可能不会显示，需在设置中调整）
• 警告信息 (WARNING): 当策略遇到一些非致命但需要注意的情况时，比如某个数据获取失败但有备用方案，可以使用 logging.warning("无法获取最新行情，使用上一周期数据代替")。这对应界面上的 橙色 文本，比较醒目。
• 错误信息 (ERROR): 当策略发生严重错误，可能导致后续逻辑无法正常执行时，应使用 logging.error("计算指标时出现除零错误") 。这对应界面上最醒目的 红色 文本，强烈提示需要检查问题。

示例：

import logging
from typing import Dict, List

def khHandlebar(context: Dict) -> List[Dict]:
    # 检查账户现金
    cash = context['__account__']['cash']
    logging.info(f"当前可用资金: {cash}")

    if cash < 10000:
        logging.warning("可用资金不足1万元，跳过本次交易机会。")
        return []

    # ...后续策略逻辑...
    try:
        # 可能会出错的代码
        risky_value = 1 / 0
    except Exception as e:
        logging.error(f"计算风险值时发生严重错误: {e}", exc_info=True) # exc_info=True会记录完整的错误堆栈

    return []

如何输出醒目的内容？

如果希望某条日志信息在界面上特别突出，最直接的方式是使用 logging.warning() 或 logging.error()。ERROR 级别（红色）最为醒目，通常用于指示发生了必须处理的问题。WARNING 级别（橙色）也比较突出，适合用于提示潜在风险或需要关注的状态。请根据信息的重要性和紧急程度，审慎选择合适的级别进行输出。