第十三章：策略工具箱：常用函数与技巧

为了让策略编写更加高效和便捷，看海量化（KHQuant）提供了一套丰富的内置函数工具箱。这些工具函数封装了常见的重复性任务，例如时间判断、数据查询、交易计算、指标分析等，使您可以将更多精力聚焦于策略逻辑本身，而不是底层的实现细节。

本章将详细介绍这些最核心、最常用的工具函数。您无需关心它们来自哪个具体的内部模块，只需知道可以通过 from khQuantImport import * 这一行简单的代码，将它们全部导入到您的策略文件中，然后便可直接调用。

13.1 便捷数据查询函数

这类函数的目标是以最简洁的方式，从 context 数据字典中提取出您最关心的数据。

`khGet(data, key)` – 万能数据获取函数

khGet 是一个功能强大的“万能”数据获取函数，它可以帮助您轻松地从 context 中提取各种嵌套的信息，而无需编写繁琐的字典访问代码。

函数签名：khGet(data: Dict, key: str) -> Any
功能：根据指定的 key，从 context 数据字典中安全地获取对应的值。
参数:
- data (dict): 即策略回调函数中收到的 context 对象。
- 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(context: Dict) -> List[Dict]:
    # 获取时间信息
    current_date = khGet(context, 'date_str') # "2024-07-15"
    current_time = khGet(context, 'time_str') # "09:31:00"

    # 获取账户信息
    available_cash = khGet(context, 'cash')

    # 获取股票池信息
    stock_list = khGet(context, 'stocks')
    first_stock = khGet(context, 'first_stock')

    logging.info(f"日期: {current_date}, 时间: {current_time}, 资金: {available_cash:.2f}")

    return []

`khPrice(data, stock_code, field)` – 获取行情价格

khPrice 是专门用于获取股票行情数据的函数，相比直接访问 context，它更加安全，能自动处理数据不存在或无效的情况。

函数签名：khPrice(data: Dict, stock_code: str, field: str = 'close') -> float
功能：获取指定股票在当前时间点的特定行情字段值。
参数:
- data (dict): 即 context 对象。
- stock_code (str): 股票代码，如 '000001.SZ'。
- field (str, 可选): 行情字段名，默认为 'close'。支持 'open', 'high', 'low', 'volume', 'amount' 等。
返回值：(float) 对应的行情数据。如果数据不存在或无效，安全地返回 0.0。

示例：

from khQuantImport import *

def khHandlebar(context: Dict) -> List[Dict]:
    stock = '000001.SZ'

    # 获取平安银行的当前开盘价和收盘价
    open_price = khPrice(context, stock, 'open')
    close_price = khPrice(context, stock, 'close') # 等同于 khPrice(context, stock)

    if open_price > 0:
        logging.info(f"{stock} 开盘价: {open_price}, 最新价: {close_price}")

    return []

`khHas(data, stock_code)` – 判断持仓

khHas 是一个简洁的函数，用于快速判断当前是否持有某只特定的股票。

函数签名：khHas(data: Dict, stock_code: str) -> bool
功能：检查账户当前是否持有指定的股票。
参数:
- data (dict): 即 context 对象。
- stock_code (str): 股票代码。
返回值：(bool) 如果持有该股票，返回 True；否则返回 False。

示例：

from khQuantImport import *

def khHandlebar(context: Dict) -> List[Dict]:
    stock = '600519.SH'

    # 检查是否持有贵州茅台
    if khHas(context, stock):
        logging.info(f"当前持有 {stock}。")
        # 在这里可以添加卖出逻辑
    else:
        logging.info(f"当前未持有 {stock}。")
        # 在这里可以添加买入逻辑

    return []

13.2 时间工具函数

在编写择时策略或进行时间相关的逻辑判断时，时间工具函数是必不可少的。

`is_trade_time()`

函数签名：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)`

函数签名：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(context, 'date_num')
if is_trade_day(current_date):
    pass

`get_trade_days_count(start_date, end_date)`

函数签名：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} 个交易日。")

13.3 数据与指标计算函数

这类函数专注于历史数据的获取和常用技术指标的计算。

`khHistory(symbol_list, fields, bar_count, fre_step, ...)` – 历史数据获取

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。

示例：

from khQuantImport import *

def khHandlebar(context: Dict) -> List[Dict]:
    stock = khGet(context, 'first_stock')
    current_date = khGet(context, 'date_num')

    # 获取截至当前日期（不含）的过去20根日线数据
    history_data = khHistory(
        symbol_list=stock,
        fields=['close'],
        bar_count=20,
        fre_step='1d',
        current_time=current_date # 关键：确保数据截止到当前
    )

    if history_data and stock in history_data:
        df = history_data[stock]
        # 在这里可以基于df计算指标
        pass

    return []

`MA(stock_code, period, ...)` – 移动平均线

MA 是一个计算移动平均线（Moving Average）的便捷函数。

函数签名：MA(stock_code: str, period: int, field: str = 'close', fre_step: str = '1d', end_time: Optional[str] = None, fq: str = 'pre') -> float
功能：计算指定周期和字段的移动平均值。
参数:
- stock_code (str): 股票代码。
- period (int): 周期长度，如 5 代表5日均线。
- field (str, 可选): 计算字段，默认为 'close'。
- fre_step (str, 可选): 时间频率，默认为 '1d'。
- end_time (str, 可选): 均线计算的截止时间点。在回测中，强烈建议传入当前时间点，以避免未来数据。
- fq (str, 可选): 复权方式，默认为 'pre'。
返回值：(float) 计算出的均线值。如果数据不足，会安全地返回 0.0。

示例：

from khQuantImport import *

def khHandlebar(context: Dict) -> List[Dict]:
    stock = khGet(context, 'first_stock')
    current_date = khGet(context, 'date_num')

    # 计算5日和20日收盘价均线
    ma5 = MA(stock, 5, end_time=current_date)
    ma20 = MA(stock, 20, end_time=current_date)

    if ma5 > 0 and ma20 > 0:
        logging.info(f"{stock} - MA5: {ma5:.2f}, MA20: {ma20:.2f}")
        if ma5 > ma20:
            # 金叉逻辑
            pass

    return []

13.4 交易辅助函数

这类函数用于简化交易信号的生成和仓位计算，能极大提高策略编写效率。

`calculate_max_buy_volume(data, stock_code, price, cash_ratio)`

函数签名：calculate_max_buy_volume(data: Dict, stock_code: str, price: float, cash_ratio: float = 1.0) -> int
功能：根据可用资金、股价和交易成本，精确计算出理论上可以买入的最大股票数量（会自动向下取整到100的倍数）。
参数：
- data (dict): context 对象。
- stock_code (str): 股票代码。
- price (float): 计划买入的价格。
- cash_ratio (float, 可选): 计划使用的资金比例，默认为 1.0 (全部可用资金)。
返回值：(int) 可以买入的最大股数。

示例：

# 计划用50%的资金买入平安银行
current_price = khPrice(context, '000001.SZ', 'close')
max_volume = calculate_max_buy_volume(context, '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)` – 智能信号生成

generate_signal 是一个强烈推荐使用的高级信号生成函数。它封装了繁琐的仓位计算逻辑，让您只需关注交易方向和比例即可。

函数签名：generate_signal(data: Dict, stock_code: str, price: float, ratio: float, action: str, reason: str = "") -> List[Dict]
功能：智能生成标准交易信号。
参数：
- data (dict): context 对象。
- 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(context: Dict) -> List[Dict]:
    signals = []
    stock = '000001.SZ'
    price = khPrice(context, stock, 'close')

    if price == 0: return []

    # 逻辑1：如果未持仓，则使用50%的资金买入
    if not khHas(context, stock):
        # generate_signal会自动计算股数并返回信号列表
        buy_signals = generate_signal(context, stock, price, 0.5, 'buy', "首次建仓")
        signals.extend(buy_signals)

    # 逻辑2：如果已持仓，且满足某个条件，则全部卖出
    elif should_sell_condition:
        # ratio=1.0 代表全仓卖出
        sell_signals = generate_signal(context, stock, price, 1.0, 'sell', "止盈卖出")
        signals.extend(sell_signals)

    return signals

`get_stock_names(stock_codes, stock_list_file)`

函数签名：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)

13.1 便捷数据查询函数

khGet(data, key) – 万能数据获取函数

khPrice(data, stock_code, field) – 获取行情价格

khHas(data, stock_code) – 判断持仓

13.2 时间工具函数

is_trade_time()

is_trade_day(date_str)

get_trade_days_count(start_date, end_date)

13.3 数据与指标计算函数

khHistory(symbol_list, fields, bar_count, fre_step, ...) – 历史数据获取

MA(stock_code, period, ...) – 移动平均线

13.4 交易辅助函数

calculate_max_buy_volume(data, stock_code, price, cash_ratio)

generate_signal(data, stock_code, price, ratio, action, reason) – 智能信号生成

get_stock_names(stock_codes, stock_list_file)

`khGet(data, key)` – 万能数据获取函数

`khPrice(data, stock_code, field)` – 获取行情价格

`khHas(data, stock_code)` – 判断持仓

`is_trade_time()`

`is_trade_day(date_str)`

`get_trade_days_count(start_date, end_date)`

`khHistory(symbol_list, fields, bar_count, fre_step, ...)` – 历史数据获取

`MA(stock_code, period, ...)` – 移动平均线

`calculate_max_buy_volume(data, stock_code, price, cash_ratio)`

`generate_signal(data, stock_code, price, ratio, action, reason)` – 智能信号生成

`get_stock_names(stock_codes, stock_list_file)`