为了让策略编写更加高效和便捷,看海量化(KHQuant)提供了一套丰富的内置函数工具箱。这些工具函数封装了常见的重复性任务,例如时间判断、数据查询、交易计算、指标分析等,使您可以将更多精力聚焦于策略逻辑本身,而不是底层的实现细节。
本章将详细介绍这些最核心、最常用的工具函数。您无需关心它们来自哪个具体的内部模块,只需知道可以通过 from khQuantImport import * 这一行简单的代码,将它们全部导入到您的策略文件中,然后便可直接调用。
1 便捷数据查询函数
这类函数的目标是以最简洁的方式,从 data 数据字典中提取出您最关心的数据。
khGet(data, key) – 万能数据获取函数 (V2/V3 通用)
khGet 是一个功能强大的“万能”数据获取函数,它可以帮助您轻松地从 data 中提取各种嵌套的信息,而无需编写繁琐的字典访问代码。
- 函数签名:
khGet(data: Dict, key: str) -> Any - 功能:根据指定的
key,从data数据字典中安全地获取对应的值。 - 参数:
data(dict): 即策略回调函数中收到的data对象。key(str): 您希望获取的数据的“别名”。支持的别名包括:'date_str':返回"YYYY-MM-DD"格式的日期字符串。'date_num':返回"YYYYMMDD"格式的日期字符串。'time_str':返回"HH:MM:SS"格式的时间字符串。'datetime_str':返回"YYYY-MM-DD HH:MM:SS"格式的日期时间字符串。'datetime_obj':返回 Python 的datetime对象。'timestamp':返回当前时间的 Unix 时间戳。'cash':返回当前可用资金 (float)。'market_value':返回当前持仓总市值 (float)。'total_asset':返回当前总资产 (float)。'stocks':返回当前股票池的完整列表 (List[str])。'first_stock':返回股票池中的第一支股票代码 (str)。'positions':返回完整的持仓字典 (Dict)。
- 返回值:根据
key返回不同类型的数据。如果键不存在或数据无效,会返回一个安全的默认值(如None,0.0,[])。
示例:
from khQuantImport import *
def khHandlebar(data: Dict) -> List[Dict]:
# 获取时间信息
current_date = khGet(data, 'date_str') # "2024-07-15"
current_time = khGet(data, 'time_str') # "09:31:00"
# 获取账户信息
available_cash = khGet(data, 'cash')
# 获取股票池信息
stock_list = khGet(data, 'stocks')
first_stock = khGet(data, 'first_stock')
logging.info(f"日期: {current_date}, 时间: {current_time}, 资金: {available_cash:.2f}")
return []
khPrice(data, stock_code, field) – 获取行情价格 (V2/V3 通用)
khPrice 是专门用于获取股票行情数据的函数,相比直接访问 data,它更加安全,能自动处理数据不存在或无效的情况。
- 函数签名:
khPrice(data: Dict, stock_code: str, field: str = 'close') -> float - 功能:获取指定股票在当前时间点的特定行情字段值。
- 参数:
data(dict): 即data对象。stock_code(str): 股票代码,如'000001.SZ'。field(str, 可选): 行情字段名,默认为'close'。支持'open','high','low','volume','amount'等。
- 返回值:(float) 对应的行情数据。如果数据不存在或无效,安全地返回
0.0。
示例:
from khQuantImport import *
def khHandlebar(data: Dict) -> List[Dict]:
stock = '000001.SZ'
# 获取平安银行的当前开盘价和收盘价
open_price = khPrice(data, stock, 'open')
close_price = khPrice(data, stock, 'close') # 等同于 khPrice(data, stock)
if open_price > 0:
logging.info(f"{stock} 开盘价: {open_price}, 最新价: {close_price}")
return []
khIndex(data, stock_code, field) – 获取技术指标 (V3 专用)
khIndex 是专门用于获取技术指标数据的函数。虽然底层实现与 khPrice 类似,但在语义上区分了“基础行情”和“衍生指标”,使代码更具可读性。
- 函数签名:
khIndex(data: Dict, stock_code: str, field: str) -> float - 功能:获取指定股票在当前时间点的特定技术指标值。
- 参数:
data(dict): 即data对象。stock_code(str): 股票代码。field(str): 指标字段名,如'MA_5','RSI_14','MACD'等。注意:这些指标需要在init函数中通过khAddExtraFields预先注册才会自动计算。
- 返回值:(float) 对应的指标数据。如果数据不存在或无效,返回
0.0。
示例:
# 在 init 中注册指标
def init(stock_list, data):
khAddExtraFields(data, ['MA_5', 'RSI_14'])
# 在 khHandlebar 中获取指标
def khHandlebar(data: Dict) -> List[Dict]:
stock = '000001.SZ'
ma5 = khIndex(data, stock, 'MA_5')
rsi = khIndex(data, stock, 'RSI_14')
# ...
khHas(data, stock_code) – 判断持仓 (V2/V3 通用)
khHas 是一个简洁的函数,用于快速判断当前是否持有某只特定的股票。
- 函数签名:
khHas(data: Dict, stock_code: str) -> bool - 功能:检查账户当前是否持有指定的股票。
- 参数:
data(dict): 即data对象。stock_code(str): 股票代码。
- 返回值:(bool) 如果持有该股票,返回
True;否则返回False。
示例:
from khQuantImport import *
def khHandlebar(data: Dict) -> List[Dict]:
stock = '600519.SH'
# 检查是否持有贵州茅台
if khHas(data, stock):
logging.info(f"当前持有 {stock}。")
# 在这里可以添加卖出逻辑
else:
logging.info(f"当前未持有 {stock}。")
# 在这里可以添加买入逻辑
return []
2 策略配置与数据预加载 (进阶)
为了提高回测效率,KHQuant 提供了数据预加载机制,允许您在策略初始化阶段声明需要用到的额外数据字段。
khAddExtraFields(data, fields) – 注册额外数据字段 (V3 专用)
这是一个非常重要的优化函数。默认情况下,框架只加载基础行情数据(OHLCV)。如果您在策略中需要使用技术指标(如 MA, RSI, MACD 等),必须在 init 函数中调用此函数进行注册。框架会自动计算并在 khHandlebar 中提供这些指标数据。
- 函数签名:
khAddExtraFields(data: Dict, fields: List[str]) -> None - 功能:动态配置需要额外加载的数据字段。
- 参数:
data(dict): 策略的data字典(在init函数中传入)。fields(list): 需要额外加载的字段列表,如['MA_5', 'RSI_14', 'MACD', 'KDJ']。
- 使用场景:必须在
init函数中调用。
示例:
def init(stock_list, data):
# 告诉框架:我需要用到 5日均线、14日RSI 和 MACD 指标
# 框架会在回测开始前自动计算好这些数据
khAddExtraFields(data, ['MA_5', 'RSI_14', 'MACD'])
def khHandlebar(data: Dict) -> List[Dict]:
# 直接获取计算好的指标,速度极快
ma5 = khIndex(data, '000001.SZ', 'MA_5')
pass
3 时间与日期工具
在编写择时策略或进行时间相关的逻辑判断时,时间工具函数是必不可少的。
is_trade_time() (V2/V3 通用)
- 函数签名:
is_trade_time() -> bool - 功能:判断当前时间是否处于A股的常规交易时间段内(
09:30-11:30,13:00-15:00)。 - 返回值:(bool) 是则返回
True,否则返回False。 - 使用场景:确保交易逻辑只在开盘时段执行,避免在午休或非交易时段产生无效信号。
示例:
if is_trade_time():
# 只在交易时段内执行核心策略
pass
is_trade_day(date_str) (V2/V3 通用)
- 函数签名:
is_trade_day(date_str: str = None) -> bool - 功能:判断指定日期是否为A股的交易日(会自动剔除周末和中国的法定节假日)。
- 参数:
date_str(str, 可选): 日期字符串,支持"YYYY-MM-DD"和"YYYYMMDD"两种格式。如果留空,则默认判断当天。
- 返回值:(bool) 是交易日则返回
True,否则返回False。
示例:
# 判断特定日期是否为交易日
is_trading_day = is_trade_day("20241001") # 返回 False (国庆节)
# 结合 khGet 使用
current_date = khGet(data, 'date_num')
if is_trade_day(current_date):
pass
get_trade_days_count(start_date, end_date) (V2/V3 通用)
- 函数签名:
get_trade_days_count(start_date: str, end_date: str) -> int - 功能:计算两个日期之间的A股交易日天数。
- 参数:
start_date(str):"YYYY-MM-DD"格式的开始日期。end_date(str):"YYYY-MM-DD"格式的结束日期。
- 返回值:(int) 两个日期之间的交易日总数。
示例:
# 计算2024年第一季度的交易日天数
days = get_trade_days_count("2024-01-01", "2024-03-31")
logging.info(f"2024年Q1共有 {days} 个交易日。")
4 历史数据与技术指标 (核心)
这类函数专注于历史数据的获取和常用技术指标的计算。
khHistory(symbol_list, fields, bar_count, fre_step, ...) – 历史数据获取 (V2/V3 通用)
khHistory 是策略中最核心的功能之一,用于获取指定证券的历史K线数据,是计算各种技术指标的基础。
- 函数签名:
khHistory(symbol_list, fields, bar_count, fre_step, current_time=None, skip_paused=False, fq='pre', force_download=False) - 功能:获取历史K线数据。
- 参数:
symbol_list(list/str): 一个或多个股票代码。fields(list): 希望获取的行情字段,如['open', 'high', 'low', 'close']。bar_count(int): 希望获取的K线数量。fre_step(str): K线周期,支持'1m','5m','1d'等。current_time(str, 可选): 获取历史数据的截止时间点(不包含)。如果为None,则数据截止到当前最新时间。在回测中,为了避免用到未来数据,通常应传入当前时间点。skip_paused(bool, 可选): 是否跳过停牌日,默认为False。fq(str, 可选): 复权类型,默认为'pre'(前复权)。force_download(bool, 可选): 是否强制下载新数据,默认为False。
- 返回值: (dict) 一个字典,键为股票代码,值为包含历史数据的
pandas.DataFrame。
current_time 参数详解
current_time 参数定义了数据获取的“观察点”,函数将返回该观察点之前的历史数据,严格不包含 current_time 这一时刻的数据。这是为了在回测中模拟真实的交易场景:在做决策时,我们只能看到“当前时刻”之前发生的行情。
该参数支持多种时间格式,系统会根据 fre_step 自动适配过滤逻辑:
- 日期格式 (
YYYY-MM-DD或YYYYMMDD)- 适用场景:主要用于日线 (
1d) 回测。 - 过滤逻辑:返回日期 小于
current_time的数据。 - 示例:若
current_time="2024-01-15",则返回2024-01-14及之前的 K 线,不包含 15 号当天的数据(因为 15 号收盘后才能看到当日 K 线)。
- 适用场景:主要用于日线 (
- 时间格式 (
YYYY-MM-DD HH:MM:SS或YYYYMMDD HHMMSS)- 适用场景:主要用于分钟线 (
1m,5m) 或 Tick 数据回测。 - 过滤逻辑:返回时间戳 小于
current_time的数据。 - 示例:若
current_time="2024-01-15 10:30:00",则返回10:30:00之前的所有分钟线(如10:29及其之前的 K 线),不包含10:30这一时刻结束的 K 线。
- 适用场景:主要用于分钟线 (
使用案例
案例 1:在日线策略中获取历史数据
def khHandlebar(data: Dict) -> List[Dict]:
# 假设当前回测日期为 2024-01-15
current_date = khGet(data, 'date_num') # "20240115"
# 获取截止到昨日的最近20根日线
history_data = khHistory(
symbol_list='000001.SZ',
fields=['close'],
bar_count=20,
fre_step='1d',
current_time=current_date # 传入当天日期
)
# 结果:history_data 中包含 2024-01-14 及之前的20天数据
# 不包含 2024-01-15 的数据(避免未来函数)
pass
案例 2:在分钟策略中获取历史数据
def khHandlebar(data: Dict) -> List[Dict]:
# 假设当前回测时间为 2024-01-15 14:00:00
current_time = khGet(data, 'datetime_str') # "2024-01-15 14:00:00"
# 获取截止到当前时刻前的最近60根1分钟K线
history_data = khHistory(
symbol_list='000001.SZ',
fields=['close'],
bar_count=60,
fre_step='1m',
current_time=current_time
)
# 结果:history_data 中最后一条数据的时间应为 13:59:00
# 不包含 14:00:00 的数据(假设此时刚到14点,K线尚未走完或刚走完)
pass
案例 3:获取指定历史时刻的数据(用于复盘)
# 想要查看 2023年春节前(2023-01-20)最后10天的行情
history_data = khHistory(
symbol_list='000001.SZ',
fields=['close', 'volume'],
bar_count=10,
fre_step='1d',
current_time='2023-01-21' # 设置为春节后第一天或春节期间
)
MyTT 指标库速查 (V2/V3 通用)
KHQuant 内置了功能强大的 MyTT 指标库,您可以直接在策略中调用这些经典的通达信/同花顺风格指标函数。无需自己编写复杂的公式,直接使用即可。
以下是支持的核心指标函数一览:
基础运算类
REF(S, N): 引用N周期前的数据(如REF(CLOSE, 1)为昨收)。ABS(S): 绝对值。MAX(S1, S2)/MIN(S1, S2): 序列最大/最小值。HHV(S, N)/LLV(S, N): N周期内的最高值/最低值。MA(S, N): 简单移动平均。EMA(S, N): 指数移动平均。SMA(S, N, M): 加权移动平均(中国式)。STD(S, N): 样本标准差。SUM(S, N): N周期累加和。CONST(S): 取序列最后一个值为常数。
核心指标类
MACD(CLOSE, SHORT=12, LONG=26, M=9): MACD指标。返回(DIF, DEA, MACD)。RSI(CLOSE, N=14): 相对强弱指标。KDJ(CLOSE, HIGH, LOW, N=9, M1=3, M2=3): KDJ指标。返回(K, D, J)。BOLL(CLOSE, N=20, P=2): 布林带。返回(UPPER, MID, LOWER)。BIAS(CLOSE, L1=6, L2=12, L3=24): 乖离率。WR(CLOSE, HIGH, LOW, N=10, N1=6): 威廉指标。CCI(CLOSE, HIGH, LOW, N=14): 顺势指标。ATR(CLOSE, HIGH, LOW, N=20): 真实波幅。TRIX(CLOSE): 三重指数平滑平均线。
使用示例:计算 MACD 金叉
from khQuantImport import *
def khHandlebar(data: Dict) -> List[Dict]:
stock = '000001.SZ'
# 1. 获取历史数据
h_data = khHistory(stock, ['close'], 100, '1d', current_time=khGet(data, 'date_num'))
if stock not in h_data: return []
# 2. 提取收盘价序列 (转换为 numpy array)
close_prices = h_data[stock]['close'].values
# 3. 使用 MyTT 计算 MACD
DIF, DEA, MACD_VAL = MACD(close_prices)
# 4. 判断金叉 (DIF 上穿 DEA)
if DIF[-1] > DEA[-1] and DIF[-2] <= DEA[-2]:
logging.info(f"{stock} 出现 MACD 金叉!")
return []
缠论工具箱 (KhChanLunTools) (V2/V3 通用)
对于高阶用户,KHQuant 提供了专门的缠论形态分析工具,支持分型识别、趋势判断和分型突破检测。这些工具对于构建基于形态的策略非常有帮助。
1. 分型识别
分型是缠论中最基本的形态单元,分为顶分型和底分型。
is_bottom_fractal(kline_data, index=-1): 判断是否为底分型。- 定义: 中间K线的高点和低点都低于左右两根K线。
- 参数:
kline_data(pd.DataFrame): K线数据,必须包含high和low列。index(int): 要判断的K线位置,默认为-1(最新一根)。注意:index 指向的是分型中间那根K线的位置。
- 返回值: (bool) 是底分型返回
True,否则返回False。
is_top_fractal(kline_data, index=-1): 判断是否为顶分型。- 定义: 中间K线的高点和低点都高于左右两根K线。
- 参数: 同上。
- 返回值: (bool) 是顶分型返回
True,否则返回False。
find_all_fractals(kline_data): 批量查找所有分型。- 功能: 扫描整段K线数据,找出所有的顶分型和底分型位置。
- 返回值:
(List[int], List[int]),分别返回顶分型和底分型的索引列表。
2. 趋势与价格分析
get_trend_direction(kline_data, period=20): 判断K线趋势方向。- 原理: 综合均线方向、高低点序列和收盘价位置进行判断。
- 参数:
kline_data: K线数据。period(int): 判断周期,默认20。
- 返回值: (str) 返回
'up'(上升),'down'(下降), 或'sideways'(震荡)。
get_fractal_price(kline_data, index, fractal_type): 获取分型关键价格。- 参数:
index: 分型位置索引。fractal_type:'top'(顶分型) 或'bottom'(底分型)。
- 返回值: (float) 顶分型返回最高价,底分型返回最低价。
- 参数:
check_fractal_break(kline_data, fractal_index, fractal_type, current_index=-1): 检查分型是否被突破。- 功能: 判断后续行情是否突破了指定分型的极值点(顶分型被向上突破,底分型被向下突破)。
- 返回值: (bool) 突破返回
True。
3. 高级K线数据获取
khKline(symbol_list, period, bar_count, ...): 高级K线获取函数。- 特点:
- 支持任意自定义周期(如
3d,2h,15m)。 - 年对齐: 多日周期(如3日线)以每年第一个交易日为基准对齐,避免跨年数据错位。
- 实时快照: 在交易时段内,未收线的K线会包含当前时刻的快照数据,适合实时监控。
- 支持任意自定义周期(如
- 参数:
symbol_list: 股票代码或列表。period: 周期字符串,如'15m','2h','3d'。bar_count: 获取数量。end_time: 结束时间(可选),用于历史回测验证。
- 特点:
综合示例:缠论底分型买入策略
from khQuantImport import *
def khHandlebar(data: Dict) -> List[Dict]:
stock = '000001.SZ'
# 1. 获取 15分钟 K线数据(缠论常用小周期)
# 获取50根K线,确保有足够数据判断趋势和分型
kline_dict = khKline(stock, '15m', 50)
if stock in kline_dict:
df = kline_dict[stock]
# 2. 判断最新 K 线(倒数第二根为中间K线,所以检查 index=-2 或 -1)
# 注意:实盘中通常等待分型确认(右侧K线收盘),或者在右侧K线运行中预判
# 这里演示判断刚刚完成的一组分型(假设当前时刻是右侧K线的结束或下一根的开始)
# 判断倒数第二根是否构成底分型的中间K线(即分型已确立)
if is_bottom_fractal(df, index=-2):
# 3. 结合趋势过滤:只有在上升趋势或震荡中才买入
trend = get_trend_direction(df)
if trend != 'down':
logging.info(f"{stock} 15分钟级别确立底分型,趋势: {trend}")
# 4. 获取分型最低点作为止损位
stop_loss_price = get_fractal_price(df, -2, 'bottom')
logging.info(f"建议止损位: {stop_loss_price}")
# 生成买入信号...
return []
5 交易信号与资金管理
这类函数用于简化交易信号的生成和仓位计算,能极大提高策略编写效率。
calculate_max_buy_volume(data, stock_code, price, cash_ratio) (V2/V3 通用)
- 函数签名:
calculate_max_buy_volume(data: Dict, stock_code: str, price: float, cash_ratio: float = 1.0) -> int - 功能:根据可用资金、股价和交易成本,精确计算出理论上可以买入的最大股票数量(会自动向下取整到100的倍数)。
- 参数:
data(dict):data对象。stock_code(str): 股票代码。price(float): 计划买入的价格。cash_ratio(float, 可选): 计划使用的资金比例,默认为1.0(全部可用资金)。
- 返回值:(int) 可以买入的最大股数。
示例:
# 计划用50%的资金买入平安银行
current_price = khPrice(data, '000001.SZ', 'close')
max_volume = calculate_max_buy_volume(data, '000001.SZ', current_price, cash_ratio=0.5)
if max_volume > 0:
logging.info(f"最多可买入 {max_volume} 股")
generate_signal(data, stock_code, price, ratio, action, reason) – 智能信号生成 (V2/V3 通用)
generate_signal 是一个强烈推荐使用的高级信号生成函数。它封装了繁琐的仓位计算逻辑,让您只需关注交易方向和比例即可。
- 函数签名:
generate_signal(data: Dict, stock_code: str, price: float, ratio: float, action: str, reason: str = "") -> List[Dict] - 功能:智能生成标准交易信号。
- 参数:
data(dict):data对象。stock_code(str): 股票代码。price(float): 交易价格。ratio(float): 核心参数。- 当
action为'buy'时,ratio代表资金使用比例(如0.5代表使用50%的可用资金)。 - 当
action为'sell'时,ratio代表持仓卖出比例(如1.0代表全部卖出)。 - 特殊用法: 当
action为'buy'且ratio> 1 时,ratio会被视为固定股数进行买入(必须是100的整数倍)。
- 当
action(str):'buy'或'sell'。reason(str, 可选): 交易原因,默认为空字符串。
- 返回值:
List[Dict],一个包含单个交易信号的列表。如果条件不满足(如资金不足或无仓可卖),则返回空列表。
示例:
from khQuantImport import *
def khHandlebar(data: Dict) -> List[Dict]:
signals = []
stock = '000001.SZ'
price = khPrice(data, stock, 'close')
if price == 0: return []
# 逻辑1:如果未持仓,则使用50%的资金买入
if not khHas(data, stock):
# generate_signal会自动计算股数并返回信号列表
buy_signals = generate_signal(data, stock, price, 0.5, 'buy', "首次建仓")
signals.extend(buy_signals)
# 逻辑2:如果已持仓,且满足某个条件,则全部卖出
elif should_sell_condition:
# ratio=1.0 代表全仓卖出
sell_signals = generate_signal(data, stock, price, 1.0, 'sell', "止盈卖出")
signals.extend(sell_signals)
return signals
round_price(price, decimals) – 价格规整 (V2/V3 通用)
在实盘交易中,申报价格必须符合交易所的最小变动价位(tick size)。此函数用于将计算出的理论价格规整为合法的申报价格。
- 函数签名:
round_price(price: float, decimals: int = None, data: Dict = None) -> float - 功能:对价格进行四舍五入。
- 参数:
price: 原始价格。decimals: 保留的小数位数。如果为None,则尝试从data中的框架配置自动获取,或根据默认规则(股票2位,ETF 3位)处理。data: 策略上下文数据(可选,用于获取配置)。
- 返回值:规整后的价格。
6 ETF 专属工具
针对 ETF 交易,框架提供了专门的辅助函数。
is_etf(stock_code)(V2/V3 通用): 判断指定代码是否为 ETF 基金。- 示例:
is_etf('510300.SH')返回True。
- 示例:
is_t0_etf(stock_code)(V3 专用): 判断指定 ETF 是否支持 T+0 交易(当日买入当日可卖)。- 示例:
is_t0_etf('159915.SZ')返回True(创业板ETF支持T+0)。
- 示例:
7 其他实用工具
get_stock_names(stock_codes, stock_list_file) (V2/V3 通用)
- 函数签名:
get_stock_names(stock_codes: Union[List[str], str], stock_list_file: str) -> Dict[str, str] - 功能:根据股票代码列表,查询并返回对应的股票名称。
- 参数:
stock_codes(list/str): 一个或多个股票代码。stock_list_file(str): 包含股票代码和名称映射关系的CSV文件路径。通常是data/stock_list/沪深A股_股票列表.csv。
- 返回值:一个字典,键为股票代码,值为股票名称。
示例:
stock_names = get_stock_names(['000001.SZ', '600519.SH'], "data/stock_list/沪深A股_股票列表.csv")
# 返回: {'000001.SZ': '平安银行', '600519.SH': '贵州茅台'}
print(stock_names)
normalize_stock_code(stock_code) (V2/V3 通用)
- 函数签名:
normalize_stock_code(stock_code: str) -> str - 功能:将各种格式的股票代码统一转换为标准格式(
XXXXXX.SZ或XXXXXX.SH)。 - 支持格式:
"sh.600000"->"600000.SH""600000"(需结合上下文,此函数主要处理带后缀的情况)"sz000001"->"000001.SZ"