3.1 框架介绍

本篇重点：本篇讲策略由哪些文件组成、框架什么时候调用你的函数、传进来什么数据、要返回什么信号，最后从零写一个双均线策略。

本篇是第 3 部分「策略编写」的第一篇。读这篇之前，最好已经按 1.5 跑通第一个回测 跑通过一次回测，并看过 1.7 回测配置详解 和 1.8 触发方式与盘前盘后。本篇讲策略文件怎么写、框架怎么调用它，函数的完整清单放在 3.2 策略工具箱。

1 策略由哪两个文件组成

打开策略目录就能看到，自带的每个示例都是同名的一对文件：.kh 是回测配置，.py 是策略代码。

1.1 .py：策略代码

策略文件就是一个普通的 Python 文件。第一行通常是：

from khQuantImport import *

这一行把框架提供的工具函数（khGet、khPrice、khHas、khMA、khHistory、generate_signal 等）、MyTT 指标函数（MA、RSI、MACD 等）以及 logging、np、pd、类型注解用的 Dict、List 一次性导入。后面的例子都默认写了这一行。

文件里不需要写主程序，也不用自己循环行情。只要按约定的名字写好回调函数，框架会在合适的时候调用它们。

1.2 .kh：回测配置

.kh 是 JSON 格式的文本文件，保存一次回测的全部参数。在软件里点「保存配置」得到的就是它，一般不需要手改。和写策略关系最大的几项是：

各项在界面上对应哪里，见 1.7。V3 用户也可以不开界面，用命令行 kh run 配置.kh 直接回测，见 2.2 命令行模式与 AI 智能策略研究。

2 回调函数：框架什么时候调用你的代码

2.1 一次回测里的调用顺序

以日 K 线触发、打开了盘前盘后回调为例，一次回测的过程是：

1. 回测开始前：调用一次 init(stock_codes, init_data)。
2. 每个交易日：
   1. 先调用 khPreMarket(data)（默认 08:30:00）；
   2. 每到一个触发点调用一次 khHandlebar(data)，日线触发就是每天一次；
   3. 当天行情结束后调用 khPostMarket(data)（默认 15:30:00）。
3. 所有交易日跑完，生成回测结果。

2.2 init(stock_codes, init_data)：开始前调用一次

• stock_codes：股票池代码列表，例如 ['000001.SZ', '600000.SH']。
• init_data：包含 __account__（初始资金）、__positions__（此时为空）、__stock_list__、__framework__ 和 __current_time__。注意这里的 __current_time__ 是电脑当前的时间，不是回测开始日期，不要拿它做日期判断。
• 适合做的事：打印参数、初始化全局变量、读取外部文件、用 khAddExtraFields 登记要预加载的自定义指标（见 3.2）。

即使什么都不做，也要写上 init，函数体写 pass 即可。框架会直接调用它，缺了这个函数或者参数个数不对（例如写成 def init():），回测会在开始时报错退出。参数名可以自己起（内置案例里写成 stocks=None, data=None），但必须能接收这两个参数；本教程统一写 stock_codes, init_data。

2.3 khHandlebar(data)：每个触发点调用一次

这是写交易逻辑的地方。调用频率由触发方式决定：Tick 触发每笔 tick 一次，1 分钟/5 分钟触发每根 K 线一次，日 K 线触发每个交易日一次，自定义定时触发在每个设定时间点一次（详见 1.8）。

• 参数 data：当前触发点的时间、全部股票的行情、账户、持仓，结构见第 3 节。
• 返回值：信号列表 List[Dict]。没有操作就返回空列表 []，不要返回列表以外的类型。

某个触发点上如果股票池里所有股票都没有行情（例如全部停牌），框架会跳过这一次调用，并在日志里给出警告。

2.4 khPreMarket(data)：盘前回调

• 要在界面「盘前盘后触发设置」里勾选「触发盘前回调」（即 .kh 里 market_callback.pre_market_enabled 为 true）才会被调用，默认时间 08:30:00。
• data 里的行情是上一个交易日最后一根 K 线，看不到当天开盘后的价格，这样不会用到未来数据。回测第一天没有上一个交易日，这天的 data 里没有个股行情，只有时间、账户和持仓，代码里要判断。
• 返回的信号会转成当天有效的限价挂单（没写 order_type 时按 limit，没写 time_in_force 时按 day），开盘后行情满足价格才成交，当天收盘仍未成交就作废。
• 常见用途：每天选股、重置当天的计数器和状态变量。

2.5 khPostMarket(data)：盘后回调

• 要勾选「触发盘后回调」（market_callback.post_market_enabled）才会被调用，默认时间 15:30:00。
• data 里的行情是当天最后一根 K 线，账户和持仓是当天交易后的状态。
• 返回的信号一律转成下一个交易日的限价挂单，有效到下一个交易日收盘。回测最后一天的盘后信号没有后续行情，不会成交。
• 常见用途：记录当天持仓和资产、计算第二天要用的数据、提前挂好次日的单。

3 data 字典里有什么

3.1 结构总览

khHandlebar、khPreMarket、khPostMarket 收到的 data 结构相同：

3.2 时间：__current_time__

date_num 常用作 khMA、khHistory 的截止日期参数，第 7 节的例子会用到。

分钟或 tick 触发时，可以按时间段控制交易：

def khHandlebar(data: Dict) -> List[Dict]:
    t = khGet(data, "time")              # 例如 "10:15:00"
    if not ("10:00:00" <= t <= "14:30:00"):
        return []                        # 这个时间段以外不交易
    # ……交易逻辑……
    return []

日 K 线触发时，time 显示为 09:30:00，只是日期标记，不代表真实时刻，不要用它做时段判断。

3.3 行情：以股票代码为键的 Series

每只股票一个 pandas.Series，包含配置里 data.fields 选的字段，默认有 open、high、low、close、volume、amount、preClose、suspendFlag 等。tick 数据的最新价字段是 lastPrice。

直接写 data["000001.SZ"]["close"] 可以取到，但股票当天没有数据时会出错。推荐用 khPrice：

price = khPrice(data, "000001.SZ", "open")   # 第三个参数不写时取 close
if price <= 0:
    return []                                # 取不到时返回 0.0

3.4 账户：__account__

market_value 和 total_asset 在每次记录结果时按最新价格更新；判断能不能买、能买多少，以 cash 为准。

3.5 持仓：__positions__

键是股票代码，值是这只股票的持仓信息。没有持仓的股票不在里面。常用字段：

判断有没有持仓用 khHas(data, code)，它在持仓数量大于 0 时返回 True。下面是一个按成本价止损的写法（用到的 generate_signal 见第 4 节），盈亏比例自己用当前价格算，不依赖 profit_ratio 是否已更新：

def khHandlebar(data: Dict) -> List[Dict]:
    signals = []
    for code, pos in khGet(data, "positions").items():
        price = khPrice(data, code, "open")
        cost = pos.get("avg_price", 0)
        if price <= 0 or cost <= 0:
            continue
        if price / cost - 1 < -0.05:                     # 亏损超过 5%
            signals.extend(generate_signal(data, code, price, 1.0, "sell", "亏损超过5%止损"))
    return signals

要知道一只股票现在能卖多少，读持仓里的 can_use_volume（可卖数量），例如 khGet(data, "positions")[code].get("can_use_volume", 0)。

3.6 常用取值函数一览

上面各小节用到的取值方法汇总如下，完整的函数清单和更多参数见 3.2 策略工具箱。

需要读回测配置时，通过 __framework__ 拿到配置字典：

def init(stock_codes, init_data):
    cfg = init_data["__framework__"].config.config_dict
    logging.info(f"基准: {cfg['backtest']['benchmark']}")
    logging.info(f"佣金率: {cfg['backtest']['trade_cost']['commission_rate']}")

4 交易信号

4.1 信号字典的字段

回调函数返回的列表里，每个元素是一条信号（一个字典）：

timestamp 不用自己填，框架会写入当前触发点的时间。止损单还要填 stop_price，这类进阶字段的用法见 3.2。

4.2 几条规则

• 数量按股、买入整百。想买 5 手就写 "volume": 500。框架不会替手写的信号取整，A 股买入请自己保证是 100 的整数倍；用下面的 generate_signal 时会自动按整百计算。
• 卖出不能超过可卖数量。超出时这条信号被拒绝，日志里会提示「可用持仓不足」。T+1 下当天买的当天卖不掉；股票池全部是支持 T+0 的 ETF 时框架会自动切换为 T+0。
• 买入不能超过可用资金。资金不够（含手续费和滑点）时这条信号被拒绝，日志提示「资金不足」。
• 不写 order_type 就是市价单，在当前触发点立即成交。限价单在当根 K 线能成交时立即成交，否则进入挂单队列，按 time_in_force 默认当天收盘作废。
• 返回的是列表。一次可以返回多条信号；一条都没有时返回 []。

手写信号的例子：

# 以开盘价市价买入 500 股（5 手）
{"code": "000001.SZ", "action": "buy", "price": 10.50, "volume": 500, "reason": "示例买入"}

# 低于开盘价 1% 挂限价买单，当天没成交就作废
{"code": "000001.SZ", "action": "buy", "price": 10.40, "volume": 500,
 "order_type": "limit", "reason": "回落1%低吸"}

4.3 用 generate_signal 生成信号

自己算股数要考虑资金、手续费、滑点和整百，比较麻烦。一般用 generate_signal：

generate_signal(data, stock_code, price, ratio, action, reason="")

• 买入时，ratio 不大于 1 表示用当前可用资金的比例（1.0 为全部，0.5 为一半），函数扣除手续费和滑点后按整百算出股数；ratio 大于 1 表示直接指定股数，必须是 100 的整数倍，超过资金上限时自动减到最大可买量。
• 卖出时，ratio 是可卖数量的比例，结果向下取整到 100 股；1.0 表示全部可卖持仓，同样取整，零股会被舍掉。
• 返回值是一个列表（条件不满足时为空列表），所以多只股票合并信号时要用 signals.extend(...)，不能用 append。

4.4 挂单与撤单

限价单、止损单没成交时会留在挂单队列里。策略可以用 khGetPendingOrders(data) 查看当前挂单，用 khCancelOrder、khCancelOrdersByCode、khCancelAllOrders 撤单。这些函数 V2、V3 都有，参数和用法见 3.2。刚开始写策略用市价单就够了。

5 在策略里输出日志

框架已经配置好了 Python 自带的 logging，策略里直接调用即可（from khQuantImport import * 已经导入了它）。日志会显示在软件的运行日志面板，同时写进日志文件；用命令行回测时输出到终端。

消息里只要包含 [TRADE]，界面就会把它显示为 TRADE 类型并去掉这个标记，可以在日志面板用「TRADE」复选框单独筛出来。

logging.info(f"[TRADE] {code} 满足买入条件，均线差 {diff:.2f}")

logging.debug 的内容不会出现在界面的运行日志面板里，调试时请用 info。在 except 里记录错误时，加 exc_info=True 可以把完整报错堆栈写进日志文件：

try:
    ma = khMA(code, 20, end_time=khGet(data, "date_num"))
except Exception as e:
    logging.error(f"{code} 计算均线出错: {e}", exc_info=True)

每次触发都打印大量日志会明显拖慢回测，长区间回测时只在关键位置输出。日志面板的用法见 1.9。

6 最小模板

下面这个文件可以直接运行：每天在日志里打印第一只股票的开盘价，不交易。新建策略时可以从它开始改。

# coding: utf-8
from khQuantImport import *   # 一行导入：工具函数、logging、Dict/List 等


def init(stock_codes, init_data):
    """回测开始前调用一次"""
    logging.info(f"策略启动，股票池共 {len(stock_codes)} 只")


def khHandlebar(data: Dict) -> List[Dict]:
    """每次触发调用一次，返回信号列表"""
    signals = []
    stock = khGet(data, "first_stock")       # 股票池第一只
    price = khPrice(data, stock, "open")     # 当前这根K线的开盘价
    logging.info(f"{khGet(data, 'date')} {stock} 开盘价 {price:.2f}")
    return signals


def khPreMarket(data: Dict) -> List[Dict]:
    """盘前回调：配置里开启后才会被调用"""
    return []


def khPostMarket(data: Dict) -> List[Dict]:
    """盘后回调：配置里开启后才会被调用"""
    return []

使用步骤：把它存为 .py 文件；在软件里选中这个策略文件，按 1.5 的方法设好回测区间、日线数据和日 K 线触发，股票池自行添加一只已经下载了日线的股票（随包配置默认留空）；点「开始运行」，运行日志里每天会多一行开盘价。确认能跑通后，点「保存配置」得到同名的 .kh。

V3 还可以在命令行里用 kh strategy validate 文件名.py 先检查结构：必需的 khHandlebar 在不在、init 写没写，两个盘前盘后回调是可选的，缺了只是提示：

它只查结构和语法，策略逻辑对不对还要看回测的交易记录（见 3.4 第 4 节）。

7 完整例子：从零写一个双均线策略

这一节把前面的内容串起来，写一个完整的策略。代码和 V3 自带案例「【1-MA策略案例】双均线精简_使用khMA函数」「【1-MA策略案例】双均线多股票_使用khMA函数」写法一致（V2 开源版的 strategies 目录里也有这两个文件）。例子里的参数和标的只用来演示写法，不构成任何投资建议。

7.1 策略思路

双均线策略用两条不同周期的移动平均线判断趋势：

• 短期均线（例如 5 日）对价格变化敏感，长期均线（例如 20 日）更平滑。
• 短期均线在长期均线上方，并且当前没有持仓：全仓买入。
• 短期均线在长期均线下方，并且当前有持仓：全部卖出。

这里比较的是两条线的上下位置，不是严格的「刚刚穿越」那一天。因为加了「有没有持仓」的条件，效果上就是在金叉后第一次触发时买入、死叉后第一次触发时卖出。

7.2 回测设置

1. 策略文件：选中下面的单股票版本 .py。
2. 回测区间：任选一段，例如一年。
3. 数据周期选 1d，触发方式选「日 K 线触发」，每个交易日调用一次 khHandlebar。日线策略用 tick 或分钟触发只会更慢，结果不会更准。
4. 复权方式建议选前复权，khMA 会跟随这里的设置。
5. 股票池：单股票版只加一只股票。
6. 确认这只股票在回测区间以及之前一段时间的日线已经下载到本地（数据管理见 1.4）。长期均线要用到回测开始日之前的 K 线，例如 20 日均线至少要前 20 个交易日的数据。

7.3 单股票版本

# coding: utf-8
from khQuantImport import *  # 导入所有量化工具

# ==================== 策略参数 ====================
params = {
    'ma_short': 5,    # 短期均线（内置案例里这里是随机数，研究时改成固定值）
    'ma_long': 20,    # 长期均线
}

def init(stock_codes, init_data):  # 策略初始化函数
    """策略初始化"""
    logging.info(f"=== 双均线策略启动 ===")
    logging.info(f"参数: 短期均线={params['ma_short']}日, 长期均线={params['ma_long']}日")

def khHandlebar(data: Dict) -> List[Dict]:  # 主策略函数
    """策略主逻辑，在每个K线或Tick数据到来时执行"""
    signals = []  # 初始化信号列表
    stock_code = khGet(data, "first_stock")  # 获取第一只股票代码
    current_price = khPrice(data, stock_code, "open")  # 获取当前开盘价
    current_date_str = khGet(data, "date_num")  # 获取当前日期数字格式

    ma_s = params['ma_short']
    ma_l = params['ma_long']

    try:
        ma_short = khMA(stock_code, ma_s, end_time=current_date_str)  # 计算短期均线
        ma_long = khMA(stock_code, ma_l, end_time=current_date_str)   # 计算长期均线
    except ValueError:
        # 上市不久或本地数据不够，历史K线不足以计算均线：今天不操作
        return signals

    has_position = khHas(data, stock_code)  # 检查是否持有该股票

    if ma_short > ma_long and not has_position:  # 金叉且无持仓
        signals = generate_signal(data, stock_code, current_price, 1.0, 'buy', f"{ma_s}日线({ma_short:.2f}) 上穿 {ma_l}日线({ma_long:.2f})，全仓买入")  # 生成买入信号

    elif ma_short < ma_long and has_position:  # 死叉且有持仓
        signals = generate_signal(data, stock_code, current_price, 1.0, 'sell', f"{ma_s}日线({ma_short:.2f}) 下穿 {ma_l}日线({ma_long:.2f})，全仓卖出")  # 生成卖出信号

    return signals  # 返回交易信号

# khPreMarket 和 khPostMarket 函数省略，本次策略未使用

7.4 逐段解释

参数放在 params 字典里。把周期写在文件开头，改参数时不用到逻辑里去找。内置案例里这两个值是这样写的：

import random
params = {
    'ma_short': random.randint(3, 10),   # 短期均线：在一定范围内随机
    'ma_long': random.randint(15, 30),   # 长期均线：在一定范围内随机
}

这样每次运行参数都会变，结果也会跟着变，目的是避免案例被当成推荐参数。自己研究时请像上面那样换成固定值，否则两次回测之间没法比较。实际用了哪组参数，init 里会打印到日志。

init 只打印参数。这个策略不需要初始化，但 init 仍然要写。

取数据。khGet(data, "first_stock") 取股票池第一只股票，单股票策略就是它；khPrice(data, stock_code, "open") 取当天开盘价作为下单价格；khGet(data, "date_num") 取 "20240102" 这样的日期，给均线函数当截止日期。

算均线。khMA(股票代码, 周期, end_time=截止日期) 从本地数据取最近 N 根日线的收盘价求平均。它取的数据不包含 end_time 当天，所以算出来的是截至前一个交易日的均线，不会用到当天收盘价。历史数据不足 N 根时 khMA 会抛出 ValueError，用 try/except 接住，这一天不操作。

判断持仓。khHas(data, stock_code) 返回是否持有这只股票，避免重复买入、空仓卖出。

生成信号。买入时 ratio=1.0 表示用全部可用资金，generate_signal 会扣掉手续费和滑点后按整百算股数；卖出时 ratio=1.0 表示卖出全部可卖持仓。最后一个参数是交易原因，用 f-string 把当时的均线值写进去，在交易记录里能看到每笔交易的依据。

返回信号。单股票版只会产生一条信号，所以直接把 generate_signal 的返回值赋给 signals。两个条件都不满足时返回空列表，这一天不操作。

7.5 改成多股票版本

把同样的逻辑放进循环，对股票池里每只股票各判断一次：

# coding: utf-8
from khQuantImport import *  # 导入所有量化工具

params = {
    'ma_short': 5,
    'ma_long': 20,
}

def init(stock_codes, init_data):  # 策略初始化函数
    """本策略不需初始化"""
    logging.info(f"=== 双均线策略启动 ===")
    logging.info(f"参数: 短期均线={params['ma_short']}日, 长期均线={params['ma_long']}日")

def khHandlebar(data: Dict) -> List[Dict]:  # 主策略函数
    """策略主逻辑，支持多只股票的双均线策略"""
    signals = []  # 初始化信号列表
    stock_list = khGet(data, "stocks")  # 获取股票池列表
    current_date_str = khGet(data, "date_num")  # 获取当前日期数字格式

    ma_s = params['ma_short']
    ma_l = params['ma_long']

    for stock_code in stock_list:  # 遍历每只股票
        current_price = khPrice(data, stock_code, "open")  # 获取当前开盘价
        try:
            ma_short = khMA(stock_code, ma_s, end_time=current_date_str)  # 计算短期均线
            ma_long = khMA(stock_code, ma_l, end_time=current_date_str)   # 计算长期均线
        except ValueError:
            # 上市不久或本地数据不够，历史K线不足以计算均线：今天先跳过这只股票
            continue

        has_position = khHas(data, stock_code)  # 检查是否持有该股票

        if ma_short > ma_long and not has_position:  # 金叉且无持仓
            signals.extend(generate_signal(data, stock_code, current_price, 0.5, 'buy', f"{stock_code[:6]} 金叉买入"))

        elif ma_short < ma_long and has_position:  # 死叉且有持仓
            signals.extend(generate_signal(data, stock_code, current_price, 1, 'sell', f"{stock_code[:6]} 死叉卖出"))

    return signals  # 返回交易信号

# khPreMarket 和 khPostMarket 函数省略，本次策略未使用

和单股票版相比改了四处：

1. 取整个股票池。khGet(data, "stocks") 返回股票池全部代码，代替 first_stock。
2. 循环。取价格、算均线、判断持仓都移到 for 循环里；某只股票数据不够时用 continue 跳过它，不影响其他股票。
3. 买入比例改成 0.5。如果每只都全仓买，第一只买完就没钱了。0.5 是「当前可用资金」的一半，同一天有多只股票满足条件时，第一只用一半，第二只用剩下的一半，依此类推。想更平均地分配资金，可以按股票数量计算比例，或者用 ratio 大于 1 的写法直接指定股数。
4. 用 extend 合并信号。generate_signal 返回的是列表，循环里要用 signals.extend(...) 把里面的信号逐条加进总列表。用 append 会变成列表套列表，框架无法识别。

回测设置上，多股票版只需要把股票池换成多只股票（例如导入一个指数成分股列表），并确认这些股票的日线都已下载。多股票版只放一只股票也能运行，区别只是每次买入用一半资金。

7.6 运行和查看结果

点「开始运行」后，运行日志里会先打印参数，之后每次产生信号都有一行「生成买入信号」或「生成卖出信号」，后面附带数量、价格和原因。回测结束后在结果窗口查看收益曲线、交易记录和每日统计，读法见 1.10。交易记录里的「原因」就是 generate_signal 最后一个参数写的内容。

回测结果不代表未来收益，不构成投资建议。

用命令行跑时（V3），结果直接打印在终端里，HTML 报告的位置也会给出：

7.7 接下来

• 试着换参数（例如 10 日和 30 日），比较结果有什么不同；把 khMA 换成 khHistory 加 MyTT 的 EMA、MACD 等指标。函数清单见 3.2 策略工具箱。
• 想让 AI 帮你写策略、跑回测验证，见 3.4 使用 AI 编写和回测验证策略。
• V3 用户可以用命令行加 AI 做批量参数研究，见 2.2 命令行模式与 AI 智能策略研究。
• V3 自带的其他案例（RSI、T+0 网格、自定义指标、日内多次触发等）见 3.5 V3 内置策略一览。

常见问题

问：策略里没写 init，为什么一运行就报错？
答：框架在回测开始时直接调用 init(股票池, data)。缺这个函数会报「has no attribute ‘init’」，写成不带参数的 def init(): 会报「takes 0 positional arguments but 2 were given」。写成 def init(stock_codes, init_data): pass 即可。

问：写了 khPreMarket，但它从来没被调用？
答：盘前、盘后回调默认关闭，要在界面「盘前盘后触发设置」里勾选，或在 .kh 的 market_callback 里把 pre_market_enabled、post_market_enabled 设为 true。只开开关、策略里没写对应函数时，日志会提示「策略模块未实现」。

问：返回了信号，交易记录里却没有成交？
答：先看运行日志里的警告。常见原因：资金不足；卖出数量超过可卖数量（T+1 下当天买入的不能当天卖）；限价单价格没有达到，挂单到期作废；多股票循环里用了 append 而不是 extend。另外 generate_signal 在资金不够买一手或没有可卖持仓时会返回空列表，并在日志里说明原因。

问：volume 写 5 是 5 手吗？
答：不是，是 5 股。数量单位一律是股，5 手要写 500。

问：双均线案例每次跑的结果都不一样？
答：内置案例的均线周期是用 random.randint 随机取的，每次运行都不同，日志开头会打印本次用的参数。研究时把 params 里的值改成固定数字。

问：khMA 报「数据量不足」？
答：本地日线不够 N 根。回测开始日之前也需要有足够的历史数据，例如 20 日均线需要回测开始前至少 20 个交易日的日线。补下载更早的数据，或者像案例那样用 try/except 跳过数据不足的日子。

问：日线触发时，能不能用当天的收盘价判断、当天买入？
答：程序上能取到，但这是用了未来数据：开盘时并不知道收盘价。按开盘决策就用开盘价和前几天的数据；想用收盘价判断，可以在盘后回调里判断，信号会转成下一个交易日的挂单。

本篇基于 V3.4.1.10 和开源 V2 编写。