3.2 策略工具箱

本篇重点：本篇是策略里可用函数的完整清单：取数、历史数据、指标、下单与挂单、日内再触发和自定义指标。

1 先看全貌：按用途找函数

下表是本篇涉及的函数分组。表里没有的冷门函数，在第 11 节速查表里；第 10 节是一个把常用函数串起来的模板。

所有例子都默认写在策略文件里，开头已经有：

from khQuantImport import *

这一行除了导入 khQuant 自己的函数，还导入了 os、sys、json、logging、datetime（注意导入的是 datetime 类，不是模块，写 datetime.now()，不要写 datetime.datetime.now()）、dt、timedelta、np、pd，以及类型标注用的 Dict、List、Optional 等，不需要再单独 import。

2 取时间与行情

2.1 khGet：取时间、资金、股票池、持仓

khGet(data: Dict, key: str) -> Any

用途：从回调收到的 data 里按「别名」取值，省掉手写多层字典访问。key 支持下面这些别名：

返回值：按 key 返回不同类型。取不到时返回安全默认值：时间类返回空字符串，资金类返回 0。不在上表里的 key，会原样当作 data 的键去取，取不到返回 None。

def khHandlebar(data: Dict) -> List[Dict]:
    today = khGet(data, 'date_num')        # "20240715"
    now = khGet(data, 'datetime_str')      # "2024-07-15 09:31:00"
    cash = khGet(data, 'cash')
    stocks = khGet(data, 'stocks')
    logging.info(f"{now} 可用资金 {cash:.2f}，股票池 {len(stocks)} 只")
    return []

2.2 khPrice：取当前行情字段

khPrice(data: Dict, stock_code: str, field: str = 'close') -> float

用途：取某只股票在当前这根 K 线（或当前 tick）上的某个行情字段。

• field：默认 'close'，也可以是 'open'、'high'、'low'、'volume'、'amount' 等数据里有的字段。
• tick 数据没有有效的 close，传 'close' 时会自动改取 lastPrice（最新价），日线、分钟线和 tick 策略可以用同一种写法。
• 返回值：float。数据不存在、为 None、为 NaN 时都返回 0.0，不会抛异常。

price = khPrice(data, '000001.SZ')            # 当前价（K 线收盘价 / tick 最新价）
open_price = khPrice(data, '000001.SZ', 'open')
if price <= 0:
    return []    # 当前没有有效价格（停牌、数据缺失），跳过

2.3 khHas：是否持有某只股票

khHas(data: Dict, stock_code: str) -> bool

用途：判断当前是否持有该股票。持仓里有这只股票且数量大于 0 时返回 True，否则 False。

stock = '600519.SH'
if not khHas(data, stock):
    pass   # 未持仓：写买入逻辑
else:
    pass   # 已持仓：写卖出逻辑

注意它只看总持仓，不看能不能卖。T+1 下当天买入的股票 khHas 已经是 True，但当天卖不出去。要判断可卖数量，看第 4 节。

2.4 交易日工具

is_trade_day：判断某天是不是 A 股交易日。

is_trade_day(date_str: str = None) -> bool

date_str 支持 "YYYY-MM-DD" 和 "YYYYMMDD"，不传时判断电脑上的今天。查询优先读本地交易日历缓存，缓存没覆盖该年份时会联网补齐；都取不到时退化为「非周末即交易日」。

is_trade_day("20241001")    # False，国庆休市
is_trade_day(khGet(data, 'date_num'))

get_trade_days_count：数两个日期之间（含首尾）有多少个交易日。

get_trade_days_count(start_date: str, end_date: str) -> int

两个日期只接受 "YYYY-MM-DD" 格式，传 "20240101" 会记一条错误日志并返回 0。开始日期晚于结束日期也返回 0。

days = get_trade_days_count("2024-01-01", "2024-03-31")

需要拿到具体日期列表时，用速查表里的 get_trade_days_set(start_date, end_date)，它返回 "YYYYMMDD" 字符串组成的集合，两种日期格式都接受。

3 历史数据与指标

3.1 khHistory：取历史 K 线（最常用）

khHistory(symbol_list, fields, bar_count, fre_step,
          current_time=None, skip_paused=False, fq=None, force_download=False)

用途：取一只或多只股票在某个时间点之前的若干根 K 线，是算指标的基础。

返回值：字典 {股票代码: DataFrame}。DataFrame 有一列 time 加上你要的字段，按时间升序排列，最后一行是离截止时间最近的那根。参数写错（股票列表或字段为空、bar_count 不大于 0、时间格式不对）会抛出 ValueError；取数过程中出错返回空字典 {}；数据不足时返回的行数会少于 bar_count，并在日志里提示。

current_time 的格式和「不含」规则

日线周期下即使传了带时分秒的时间，也只按日期比较，所以日线策略直接传 khGet(data, 'datetime_str') 也能得到「截至昨天」的数据。日期和时分秒之间要有空格，"20240115103000" 这种连写的格式解析不了，会报错。

为什么要「不含」：策略在这根 K 线上做决定时，这根 K 线本身的数据通过 khPrice 已经能看到；历史数据只给它之前的部分，算均线、算指标时就不会把「当前」和「过去」混在一起，也不会用到未来数据。

例 1：日线策略取最近 20 个交易日收盘价

def khHandlebar(data: Dict) -> List[Dict]:
    stock = '000001.SZ'
    hist = khHistory(stock, ['close'], 20, '1d',
                     current_time=khGet(data, 'date_num'))
    if stock not in hist or len(hist[stock]) < 20:
        return []                      # 数据不够，先不做判断
    closes = hist[stock]['close'].values
    ma20 = closes.mean()
    return []

例 2：分钟策略取最近 60 根 1 分钟线

now = khGet(data, 'datetime_str')     # "2024-01-15 14:00:00"
hist = khHistory(['000001.SZ', '600000.SH'], ['close', 'volume'], 60, '1m',
                 current_time=now)
for code, df in hist.items():
    if df.empty:
        continue
    last_close = df['close'].iloc[-1]  # 13:59 那根的收盘价

例 3：在回测外查某个历史时点（复盘、调试）

hist = khHistory('000001.SZ', ['close', 'volume'], 10, '1d',
                 current_time='2023-01-21')

3.2 khMA：一行算均线

khMA(stock_code: str, period: int, field: str = 'close', fre_step: str = '1d',
     end_time: str = None, fq: str = None, data: Dict = None) -> float

用途：直接算某只股票的 N 周期简单均线，内部就是 khHistory 取 period 根再求平均。

• end_time：截止时间，规则同 khHistory 的 current_time（不含该时间点）。回测中不传时自动用当前 K 线时间，所以日线上 khMA 用的是截至前一天的数据，不含当天。
• fq：同 khHistory，默认跟随回测的复权方式。
• data：可选，传入后按框架设置的价格精度取整。
• 返回值：按价格精度四舍五入后的 float（股票 2 位小数，ETF 等场内基金 3 位）。
• 数据不足 period 根时会抛出 ValueError，不是返回 0。

try:
    ma5 = khMA('000001.SZ', 5, data=data)
    ma20 = khMA('000001.SZ', 20, data=data)
except ValueError as e:
    logging.warning(f"均线数据不足: {e}")
    return []

只要一两条均线时 khMA 最省事。要同时算多条均线或别的指标，一次 khHistory 取够数据再交给 MyTT 计算会更快，避免重复取数。

3.3 khKline：任意周期 K 线

khKline(symbol_list, period: str, bar_count: int, fields: List[str] = None,
        end_time: str = None, fq: str = None, force_download: bool = False)
        -> Dict[str, pd.DataFrame]

用途：把日线或分钟线拼成任意周期，例如 '15m'、'30m'、'2h'、'3d'，常用于缠论等多级别分析。

• period：数字加单位，单位是 m（分钟）、h（小时）、d（天）。
• fields：默认 ['open', 'high', 'low', 'close', 'volume']。
• end_time：支持 '2024-12-01 14:30:00'、'2024-12-01 14:30'、'20241201 1430'、'20241201' 等，秒数会被忽略。回测中不传时用当前 K 线时间。
• 多日周期（2d、3d…）每年都以第一个交易日为起点分组；年初根数不够时，会接着往前一年取。
• 返回值：{股票代码: DataFrame}，出错时返回空字典。

k15 = khKline('000001.SZ', '15m', 50)
df = k15.get('000001.SZ')
if df is not None and len(df) >= 3:
    logging.info(f"最新一根 15 分钟线收盘 {df['close'].iloc[-1]}")

3.4 MyTT 常用指标

khQuant 内置 MyTT 指标库，函数名和常见行情软件的公式写法一致，输入是 numpy 数组（或 pandas Series 的 .values），输出是同样长度的数组，取 [-1] 就是最新值。用法固定为两步：先用 khHistory 取数据，再交给 MyTT 计算。

stock = '000001.SZ'
hist = khHistory(stock, ['close', 'high', 'low'], 100, '1d',
                 current_time=khGet(data, 'date_num'))
if stock not in hist or len(hist[stock]) < 60:
    return []
C = hist[stock]['close'].values
H = hist[stock]['high'].values
L = hist[stock]['low'].values

DIF, DEA, MACD_BAR = MACD(C)            # 默认 12, 26, 9
if DIF[-1] > DEA[-1] and DIF[-2] <= DEA[-2]:
    logging.info(f"{stock} MACD 金叉")
K, D, J = KDJ(C, H, L)
UPPER, MID, LOWER = BOLL(C)
rsi14 = RSI(C, 14)[-1]                  # 想要 14 日 RSI 要显式写 14

常用指标的签名和默认参数（以源码 MyTT.py 为准）：

写条件时常用的基础函数：REF(S, N=1) 取 N 周期前的值；HHV(S, N) / LLV(S, N) 取 N 周期最高 / 最低；CROSS(S1, S2) 判断上穿；COUNT(S, N)、EVERY(S, N)、EXIST(S, N) 统计 N 周期内条件成立的次数、是否全部成立、是否有一次成立；BARSLAST(S) 上次条件成立到现在的周期数。其余 PSY、MTM、ROC、KTN、VR、EMV、MFI、CR、BRAR、ASI、DPO 等指标也都能直接调用，名字和参数见 MyTT.py。

3.5 缠论工具

六个函数都以一段 K 线 DataFrame 为输入，DataFrame 至少要有 high、low 列（判断趋势还要 close）。数据通常来自 khKline 或 khHistory。

stock = '000001.SZ'
df = khKline(stock, '15m', 50).get(stock)
if df is not None and len(df) >= 20:
    if is_bottom_fractal(df, index=-2) and get_trend_direction(df) != 'down':
        stop_price = get_fractal_price(df, -2, 'bottom')
        logging.info(f"{stock} 15 分钟底分型确立，分型低点 {stop_price}")

注意上一小节说过的：khKline 的最后一根可能还没收线，用 index=-2 判断时，右边那根就是这根未收线的 K 线。如果要求分型完全确认，可以再往前错一根，或只在周期收线的时刻判断。

4 账户与持仓

4.1 账户资金

资金类数据直接用 khGet 取：khGet(data, 'cash') 可用资金、khGet(data, 'market_value') 持仓市值、khGet(data, 'total_asset') 总资产。在同一个回调里返回的信号还没执行，资金不会在你生成信号时立刻扣减，多只股票同时买入时要自己分配比例（见第 5 节）。

4.2 持仓字典

khGet(data, 'positions') 返回 {股票代码: 持仓信息}，每只股票常用的字段：

for code, pos in khGet(data, 'positions').items():
    cost = pos.get('avg_price', 0)
    price = khPrice(data, code)
    if cost > 0 and price > 0 and price / cost - 1 < -0.08:
        logging.info(f"{code} 浮亏超过 8%，可卖 {pos.get('can_use_volume', 0)} 股")

4.3 PositionParser：持仓解析器

不想手写字典访问时，可以用解析器：

pp = PositionParser(data)
pp.has('000001.SZ')         # 是否持有，等同 khHas
pp.get_volume('000001.SZ')  # 总持仓，未持有返回 0
pp.get_cost('000001.SZ')    # 持仓均价，未持有返回 0
pp.get_all()                # 全部持仓的副本

要取可卖数量，读持仓里的 can_use_volume：khGet(data, 'positions')[code].get('can_use_volume', 0)。

5 下单与仓位

5.1 信号字典

khHandlebar 返回一个信号列表，每个信号是一个字典。必填四个字段，其余可选：

signals = [{
    "code": "000001.SZ",
    "action": "buy",
    "price": khPrice(data, "000001.SZ"),
    "volume": 1000,           # 1000 股，即 10 手
    "reason": "手写信号示例",
}]
return signals

5.2 generate_signal：按比例生成信号（推荐）

generate_signal(data: Dict, stock_code: str, price: float, ratio: float,
                action: str, reason: str = "") -> List[Dict]

用途：只给方向和比例，由它算好股数、拼好信号。

• ratio，买入时：小于等于 1 表示用可用资金的比例（0.5 即一半资金）；大于 1 表示买入的股数，必须是 100 的整数倍，否则返回空列表；股数超过资金能买的上限时，自动降到上限。
• ratio，卖出时：可卖数量（can_use_volume）的比例，1.0 即全部可卖部分。结果向下取整到 100 股。
• price：按价格精度四舍五入后写入信号。
• 返回值：只含一个信号的列表；资金不足、没有持仓、算出来股数为 0 时返回空列表，并在日志里说明原因。

def khHandlebar(data: Dict) -> List[Dict]:
    signals = []
    stock = '000001.SZ'
    price = khPrice(data, stock)
    if price <= 0:
        return signals
    if not khHas(data, stock):
        signals.extend(generate_signal(data, stock, price, 0.5, 'buy', "用一半资金买入"))
    else:
        signals.extend(generate_signal(data, stock, price, 1.0, 'sell', "全部卖出"))
    return signals

同时买多只股票时，ratio 每次都是按「当前可用资金」算的，而同一轮里前一个信号还没成交、不会扣资金。所以想三只股票各用三分之一资金，每只都传 1/3，不要依次传 1/3、1/2、1。

5.3 calculate_max_buy_volume：最多能买多少股

calculate_max_buy_volume(data: Dict, stock_code: str, price: float,
                         cash_ratio: float = 1.0) -> int

用途：按可用资金乘以 cash_ratio，在扣除佣金、印花税、过户费和滑点之后，算出最多能买的股数，结果是 100 的整数倍。不足 100 股或价格无效时返回 0。generate_signal 买入时内部就是调用它。

price = khPrice(data, '000001.SZ')
vol = calculate_max_buy_volume(data, '000001.SZ', price, cash_ratio=0.3)
if vol > 0:
    signals.append({"code": "000001.SZ", "action": "buy",
                    "price": price, "volume": vol, "reason": "30% 资金"})

手写信号时用 append 加单个字典是对的；只有 generate_signal 这种返回列表的函数才用 extend。

5.4 价格精度

股票报价两位小数，ETF、LOF 等场内基金三位小数。自己算出来的价格（比如「现价下浮 1% 挂单」）要先取整再填进信号：

round_price(price: float, decimals: int = None, data: Dict = None, stock_code: str = None) -> float
get_price_decimals(data: Dict = None, stock_code: str = None) -> int

decimals 不传时按 stock_code 判断：场内基金 3 位、普通 A 股代码 2 位；也不传 stock_code 时，用框架按股票池设置的精度（纯股票池 2 位，含基金 3 位）。

code = '510300.SH'
limit_price = round_price(khPrice(data, code) * 0.99, data=data, stock_code=code)

框架在处理信号时也会按精度再取整一次，所以就算忘了取整也不会报错，但提前取整能让日志里看到的价格和实际委托价一致。

5.5 订单类型与有效期

撮合引擎在 V2 和 V3 中都始终开启，信号会按 order_type 撮合：

所有成交都计入滑点和交易费用；限价单加滑点后的成交价不会超过限价。一字涨停买不进、一字跌停卖不出。如果在软件设置里开启了成交量限制，单根成交量不够时会部分成交，剩余数量继续挂单。

有效期由 time_in_force 决定：

# 现价下方 2% 挂一张限价买单，最多等到 3 天后
code = '000001.SZ'
price = khPrice(data, code)
limit_price = round_price(price * 0.98, data=data, stock_code=code)
expire = (khGet(data, 'datetime_obj') + timedelta(days=3)).strftime('%Y-%m-%d')
signals.append({
    "code": code, "action": "buy", "price": limit_price, "volume": 500,
    "order_type": "limit", "time_in_force": "gtd", "expire_date": expire,
    "reason": "回调 2% 挂单",
})

# 持仓设一个止损：跌到成本价的 92% 时卖出
pos = khGet(data, 'positions').get(code)
if pos and pos.get('can_use_volume', 0) > 0:
    stop = round_price(pos['avg_price'] * 0.92, data=data, stock_code=code)
    signals.append({
        "code": code, "action": "sell", "price": stop, "stop_price": stop,
        "volume": pos['can_use_volume'], "order_type": "stop",
        "reason": "止损",
    })

上面的止损单默认当日有效，每天都要重新下；也可以用 gtd 把有效期拉长，但要注意不要每天重复下单。下一节讲怎么先查挂单再决定是否下单。

6 挂单管理：查询与撤单

限价单、止损单没成交时会留在挂单队列里。下面六个函数用来查看和撤销挂单，两版都有，参数里的 data 就是回调收到的 data。

khGetPendingOrders 返回的每笔挂单，常用字段有：order_id（单号，撤单时用）、code、action、price（委托价）、volume（委托数量）、remaining_volume（剩余未成交数量）、order_type、time_in_force、expire_date、create_date（下单日期）。返回的是副本，改它不会影响真实挂单。

def khHandlebar(data: Dict) -> List[Dict]:
    signals = []
    code = '000001.SZ'
    price = khPrice(data, code)
    if price <= 0:
        return signals

    pending = khGetPendingOrders(data, code)
    # 已有买单挂着，但价格已经比挂单价高出 3%：撤掉旧单，按新价重挂
    for order in pending:
        if order['action'] == 'buy' and price > order['price'] * 1.03:
            if khCancelOrder(data, order['order_id']):
                logging.info(f"撤单 {order['order_id']}，原价 {order['price']}")
                pending = []

    if not pending and not khHas(data, code):
        signals.append({
            "code": code, "action": "buy", "volume": 500,
            "price": round_price(price * 0.99, data=data, stock_code=code),
            "order_type": "limit", "reason": "低 1% 挂单",
        })

    summary = khGetPendingSummary(data)
    if summary.get('total_count', 0) > 0:
        logging.info(f"当前挂单 {summary['total_count']} 笔")
    return signals

7 日内多次触发：khRequestNextDailyTrigger

khRequestNextDailyTrigger() -> bool

用途：日线策略默认每天只调用一次 khHandlebar。在 khHandlebar 里调用这个函数，可以申请当天处理完本轮信号后再调用一次。典型用法是先卖后买：第一轮卖出，卖出成交后资金回到账户，第二轮再用这笔资金买入。

前提：触发方式是日线（1d），并在界面的「日内触发上限」（配置里的 daily_trigger_cap）设为大于 1 的数。上限是每天最多调用 khHandlebar 的总次数，包含第一次。

返回 True 表示申请已登记。下面几种情况返回 False：

• 触发方式不是日线（分钟、tick、自定义时间点都不支持）；
• 当天已调用次数达到「日内触发上限」；
• 不在策略回调执行过程中调用（比如在模块顶层调用）；
• 框架内部出错。

再次触发时，行情数据和上一轮完全一样（同一天、同一根日线），但 __account__、__positions__ 已经反映了上一轮的成交结果。策略要自己用全局变量记住「今天是第几轮」：

STATE = {"date": None, "round": 0}

def init(stock_codes, init_data):
    STATE["date"], STATE["round"] = None, 0

def khHandlebar(data: Dict) -> List[Dict]:
    signals = []
    today = khGet(data, 'date')
    if STATE["date"] != today:
        STATE["date"], STATE["round"] = today, 1
    else:
        STATE["round"] += 1

    if STATE["round"] == 1:
        # 第 1 轮：只做卖出
        for code in khGet(data, 'positions'):
            price = khPrice(data, code)
            if price > 0 and should_sell(code, data):   # should_sell 为你自己的卖出条件
                signals.extend(generate_signal(data, code, price, 1.0, 'sell', "第 1 轮卖出"))
        if khRequestNextDailyTrigger():
            logging.info(f"{today} 已申请第 2 轮")
    elif STATE["round"] == 2:
        # 第 2 轮：用释放出来的资金买入
        logging.info(f"{today} 第 2 轮，可用资金 {khGet(data, 'cash'):.2f}")
        # ……买入逻辑……
    return signals

完整可运行的案例见 3.5 V3 内置策略一览 · 日内多次触发。

8 自定义指标：khDuckWrite、khDuckDB、khAddExtraFields、khIndex

回测几百只股票、好几年时，在 khHandlebar 里每根 K 线都取一遍历史数据再算指标，会很慢。更快的做法是把指标提前算好、写进本地行情库，回测时框架和行情一起一次性加载进内存，策略里直接读。分三步：

1. 预先写库：在一个独立脚本里（不是策略文件），用 khDuckDB 读行情、算指标，再用 khDuckWrite 写回库里。
2. 注册字段：策略的 init 里用 khAddExtraFields 告诉框架要多加载哪几列。
3. 读取：khHandlebar 里用 khIndex 读当前 K 线上的指标值。

这四个函数 V2 和 V3 都有。

8.1 khDuckDB：直接读本地行情库

khDuckDB(stock_list, period, fields=None, start_time=None, end_time=None,
         dividend_type='none', duckdb_path=None, return_format='dict')

• period：'1d'、'1m'、'5m'、'tick'。
• fields：要的列，None 返回全部列（包括你写进去的自定义列）。
• start_time / end_time：'YYYYMMDD' 或 'YYYYMMDDHHmmss'。只写日期的 end_time 会取到当天 23:59:59，包含当天。
• dividend_type：'none'（默认）、'front'、'back'、'front_ratio'、'back_ratio'。注意默认是不复权，和 khHistory 默认跟随界面不同。
• duckdb_path：数据目录，不传时用当前回测的数据目录，再退到软件设置里的数据目录。
• 返回值：{股票代码: DataFrame}，读不到的股票对应空 DataFrame。

res = khDuckDB(['000001.SZ'], '1d', fields=['time', 'close'],
               start_time='20230101', end_time='20231231')
df = res['000001.SZ']

8.2 khDuckWrite：把指标写进库

khDuckWrite(stock_list, period, data, fields=None, time_col="time",
            duckdb_path=None, field_types=None, insert_missing=False,
            update_time=True, allow_builtin_fields=False)

• data：{股票代码: DataFrame}，或单只股票时直接传 DataFrame。DataFrame 必须有时间列（默认叫 time）。
• fields：要写入的列名，None 时写入除时间列外的全部列。库里没有的列会自动新增。
• 日线按交易日匹配，时间列写纯日期即可。
• insert_missing：默认 False，只更新库里已有时间的行；时间对不上的行不会写入，并在日志里提示多少行没对上。
• allow_builtin_fields：默认 False，不允许改写 open、close、volume 等内置行情列和复权列。自定义指标请换个名字。
• 列名不区分大小写，dif 和 DIF 是同一列。
• 返回值：{股票代码: {'updated': 更新行数, 'inserted': 插入行数, 'rows': 输入行数, 'fields': 写入的列}}。

# 独立脚本：给股票池算 MACD 并写库（运行前关掉占用数据目录的回测和下载任务）
from khQTTools import khDuckDB, khDuckWrite
import pandas as pd

for code in ['000001.SZ', '600000.SH']:
    df = khDuckDB([code], '1d', fields=['time', 'close'])[code]
    if df.empty:
        continue
    fast = df['close'].ewm(span=12, adjust=False).mean()
    slow = df['close'].ewm(span=26, adjust=False).mean()
    df['DIF'] = fast - slow
    df['DEA'] = df['DIF'].ewm(span=9, adjust=False).mean()
    df['MACD'] = (df['DIF'] - df['DEA']) * 2
    print(khDuckWrite([code], '1d', df[['time', 'DIF', 'DEA', 'MACD']],
                      fields=['DIF', 'DEA', 'MACD']))

8.3 khAddExtraFields：在 init 里注册字段

khAddExtraFields(context: Dict, fields: List[str]) -> None

用途：告诉框架回测加载行情时，把这些列也一起读进来。context 传 init 收到的 data。只能在 init 里调用，到了 khHandlebar 再调用就晚了。

它不会计算指标，只负责从库里读出已有的列。列名要和写库时完全一致。库里没有这一列的股票，整只股票的行情都读不出来（日志里报一条错误「查询失败 … Referenced column “DIF” not found」）：这只股票在回测里等于没有数据，khIndex 读到 0，也取不到价格、不会交易；其他股票照常回测。所有股票都缺这一列时，回测报「没有找到任何有效的时间点」并停止。

8.4 khIndex：读当前 K 线上的指标值

khIndex(data: Dict, stock_code: str, field: str) -> float

用途：读当前 K 线上某个自定义指标的值。底层和 khPrice 相同，只是语义上区分「行情」和「指标」。field 没有默认值，必须写。取不到时返回 0.0。

PREV = {}

def init(stock_codes, init_data):
    khAddExtraFields(init_data, ["DIF", "DEA", "MACD"])

def khHandlebar(data: Dict) -> List[Dict]:
    signals = []
    for code in khGet(data, 'stocks'):
        price = khPrice(data, code)
        if price <= 0:
            continue
        dif, dea = khIndex(data, code, "DIF"), khIndex(data, code, "DEA")
        last = PREV.get(code)
        PREV[code] = (dif, dea)
        if last is None:
            continue
        if last[0] < last[1] and dif > dea and not khHas(data, code):
            signals.extend(generate_signal(data, code, price, 0.1, 'buy', "DIF 上穿 DEA"))
    return signals

写库脚本、注册和读取的完整案例，见 3.5 V3 内置策略一览 · 自定义指标。

9 日志

策略里直接用 Python 标准库 logging（已随 khQuantImport 导入），输出会显示在界面的日志区，也会写进本次回测的日志文件。

logging.info("普通信息")
logging.warning("警告：数据不足，跳过")
logging.error("出错了")
logging.info(f"[TRADE] {code} 满足买入条件，价格 {price}")   # 带 [TRADE] 的消息在界面里高亮

消息里带 [TRADE] 标签时，界面会把它当作交易类日志高亮显示（标签本身不显示），适合标出关键决策点。日志在界面上怎么筛选、在哪里找日志文件，见 1.9 节。

两点建议：只在关键决策处写 info，循环里每根 K 线都打日志会拖慢回测；用 try / except 包住单只股票的处理，在 except 里 logging.error，一只股票出错不至于让整个回测停下。

10 把工具串起来：一个多股票模板

下面这个模板把前面几节最常用的函数串在一起：遍历股票池，一次 khHistory 取够数据，用 MyTT 算两条均线，先处理卖出、再处理买入，买入资金按股票数平均分配，还挂着的单先撤掉。它不是一个推荐的交易策略，只是演示各个工具在一个 khHandlebar 里怎么配合。

from khQuantImport import *

SHORT, LONG = 5, 20          # 演示参数

def init(stock_codes, init_data):
    logging.info(f"股票池 {len(stock_codes)} 只，均线 {SHORT}/{LONG}")

def khHandlebar(data: Dict) -> List[Dict]:
    signals = []
    stocks = khGet(data, 'stocks')
    if not stocks:
        return signals

    # 1. 一次取够所有股票的历史数据（不含当天）
    hist = khHistory(stocks, ['close'], LONG + 5, '1d',
                     current_time=khGet(data, 'date_num'))

    # 2. 还挂着没成交的单（如 gtd 单、盘后单）先撤掉，避免重复下单
    if khGetPendingSummary(data).get('total_count', 0) > 0:
        n = khCancelAllOrders(data)
        logging.info(f"撤销未成交挂单 {n} 笔")

    buy_list = []
    for code in stocks:
        try:
            price = khPrice(data, code)
            df = hist.get(code)
            if price <= 0 or df is None or len(df) < LONG + 1:
                continue
            closes = df['close'].values
            ma_s, ma_l = MA(closes, SHORT), MA(closes, LONG)
            golden = ma_s[-2] <= ma_l[-2] and ma_s[-1] > ma_l[-1]
            dead = ma_s[-2] >= ma_l[-2] and ma_s[-1] < ma_l[-1]

            # 3. 先卖：死叉且有持仓
            if dead and khHas(data, code):
                signals.extend(generate_signal(data, code, price, 1.0, 'sell',
                                               f"MA{SHORT} 下穿 MA{LONG}"))
            # 记下要买的，稍后统一分配资金
            elif golden and not khHas(data, code):
                buy_list.append((code, price))
        except Exception as e:
            logging.error(f"{code} 处理出错: {e}")

    # 4. 再买：每只股票用同样比例的当前可用资金
    if buy_list:
        ratio = min(1.0 / len(buy_list), 0.3)     # 单只最多 30%
        for code, price in buy_list:
            sig = generate_signal(data, code, price, ratio, 'buy',
                                  f"MA{SHORT} 上穿 MA{LONG}")
            if sig:
                logging.info(f"[TRADE] 买入 {code} {sig[0]['volume']} 股")
            signals.extend(sig)
    return signals

几个值得注意的写法：历史数据在循环外一次性取，避免每只股票各取一次；均线用 [-1] 和 [-2] 判断交叉，二者都来自截至昨天的数据；卖出信号和买入信号在同一轮返回时，卖出的资金在这一轮还没回到账户，所以买入比例只按当前可用资金算。如果想让卖出的钱当天就能用来买入，就要用第 7 节的日内多次触发，把买入放到第二轮。

想把这样的思路交给 AI 去写、去回测，可以看 3.4 使用 AI 编写和回测验证策略；想看更多现成写法，可以看 3.5 V3 内置策略一览。

11 冷门函数速查表

常见问题

问：khPrice 一直返回 0 是怎么回事？
答：0 表示没取到数据。常见原因：代码没写交易所后缀（要写 000001.SZ，不是 000001）；这只股票不在股票池里，data 里没有它的行情；当天停牌或数据缺失；字段名写错。先 logging.info(khGet(data, 'stocks')) 确认股票池，再检查数据是否下载。

问：khHistory 返回空字典或者行数不够？
答：先看日志里的错误或「历史数据不足」提示。最常见的是本地没有这段时间的数据（回测开始日期往前还需要 bar_count 根，数据要多下载一段），其次是 fre_step 和下载的数据周期不一致。数据下载见 1.4 数据管理模块。

问：同一根日线，khHistory 算出来的均线和 khKline 算出来的不一样？
答：两者截止规则相反。khHistory 不含当天，khKline 含当天。用 khKline 时去掉最后一根再算，就和 khHistory 对得上。

问：generate_signal 卖出时为什么少卖了几十股？
答：卖出数量按「可卖数量 × ratio」向下取整到 100 股。如果持仓里有不足 100 股的零股（比如送股后），全部卖出时这部分会被舍掉。要清掉零股，可以手写信号，把 volume 填成 can_use_volume。

问：为什么今天买的股票，generate_signal 卖出返回空列表？
答：A 股是 T+1，当天买入的股票不计入可卖数量，generate_signal 卖出看的是可卖数量。只有股票池全部是 T+0 品种时，框架才按 T+0 回测。

问：限价单一直没成交，也没看到报错？
答：没满足成交条件的限价单会进挂单队列，默认当天收盘后过期，不算错误。用 khGetPendingOrders(data) 看它还在不在，日志里也会有「已加入挂单队列」「挂单过期」的记录。另外，委托价超出当日涨跌停范围会直接拒单，日志里有拒单原因。

问：khRequestNextDailyTrigger 一直返回 False？
答：依次检查：触发方式是不是日线；「日内触发上限」是不是大于 1；当天是不是已经用满次数；是不是在 khHandlebar 里调用的。

问：khIndex 读出来全是 0？
答：多半是库里没有这一列（这时这只股票整只没有行情，日志里有「Referenced column … not found」），或者列名和 khAddExtraFields 里写的不一致，或者 khAddExtraFields 没写在 init 里。先用 khDuckDB 查一下这一列在不在。

问：V2 能用 khIndex、khAddExtraFields、挂单管理和日内多次触发吗？
答：能。V2 和 V3 的策略接口完全相同，这些用法也相同。V3 多出来的是命令行、Web 工作台、更多数据源这些软件功能，不在策略接口上。

问：能不能让 AI 帮我查这些函数怎么用？
答：可以。khQuant 的 Skill 里带有完整的接口说明，AI 写策略时会按它来调用这些函数，见 3.3 khQuant 的 Skill 介绍 和 3.4 使用 AI 编写和回测验证策略。V3 用户也可以在命令行模式里让 AI 直接写策略、跑回测。

本篇的代码片段只为演示函数写法，其中的标的、参数和条件都不构成投资建议；据此跑出的回测结果不代表未来收益。

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