Skip to content

kxi-toolkit

Standalone Python UDA Toolkit for kdb Insights Enterprise.

Tip

You can install this toolkit locally for IDE support — syntax highlighting, type hints, and autocompletion all work without a running Insights environment. UDAs will only execute inside kdb Insights Enterprise, where the underlying kdb Insights runtime is available.

Provides a decorator-based API for defining and describing User-Defined Analytics (UDAs) in Python, plus a convenience wrapper around .kxi.selectTable. Inside kdb Insights Enterprise, UDAs are executed via pykx — but it is the underlying kdb Insights runtime that drives execution, and that runtime is not available during local development. Installing pykx locally gives your IDE full type hints and introspection, but UDAs will not execute without the kdb Insights runtime.

Installation

For uv-managed projects — adds the dependency to pyproject.toml and updates your lockfile.

uv add kxi-toolkit --extra-index-url https://portal.dl.kx.com/assets/pypi/
uv add kxi-toolkit[pykx] --extra-index-url https://portal.dl.kx.com/assets/pypi/  # optional: local development only

For plain virtualenvs — installs into the current environment without modifying project files.

uv pip install kxi-toolkit --extra-index-url https://portal.dl.kx.com/assets/pypi/
uv pip install kxi-toolkit[pykx] --extra-index-url https://portal.dl.kx.com/assets/pypi/  # optional: local development only

If you are unable to use uv and must use vanilla pip.

pip install kxi-toolkit --extra-index-url https://portal.dl.kx.com/assets/pypi/
pip install kxi-toolkit[pykx] --extra-index-url https://portal.dl.kx.com/assets/pypi/  # optional: local development only

Defining a UDA

import pykx as kx
from kxi_toolkit import uda, register_udas, select_table, Symbol, Timestamp, Table

@uda.name("salesSummary")
@uda.description("Sales summary grouped by sym")
@uda.safe()
def sales_summary(table: Symbol, startTS: Timestamp = None, endTS: Timestamp = None) -> Table:
    raw = select_table(table, startTS, endTS)
    return kx.q(
        "{0!select cnt:count i, avg_price:avg price, total_vol:sum size by sym from x}",
        raw,
    )

@sales_summary.aggregation
def sales_summary_agg(partials) -> Table:
    combined = kx.q.raze([kx.Table(data=d) for d in partials])
    result = kx.q(
        "{select cnt:sum cnt, avg_price:avg avg_price, total_vol:sum total_vol by sym from x}",
        combined,
    )
    return kx.q(".kxi.response.ok", result)

register_udas()

Decorator reference

Decorator Description
@uda.name(str) q-style UDA name (required)
@uda.description(str) Human-readable description
@uda.param(name, type, ...) Override / annotate a parameter
@uda.returns(type=, description=) Override inferred return type
@uda.misc(safe=True, ...) Miscellaneous metadata
@uda.safe() Shorthand for @uda.misc(safe=True)
@uda.categories(*cats) Set UDA categories
@fn.aggregation Attach an aggregation function

Parameters and return type are inferred from Python type hints automatically — use @uda.param / @uda.returns to override.

Namespace

UDAs are registered under the pyuda namespace by default, so @uda.name("salesSummary") registers as .pyuda.salesSummary in q. To use a different namespace, pass a fully-qualified name starting with .:

@uda.name(".myns.salesSummary")

Type hints

Import KX type sentinels from kxi_toolkit to annotate UDA signatures:

from kxi_toolkit import (
    Boolean, BooleanVector,
    Char, Date,
    Float, FloatVector,
    Int, Long, LongVector,
    Real, String,
    Symbol, SymbolVector,
    Table, Dictionary,
    Timestamp, TimestampVector,
)

Querying tables

select_table wraps .kxi.selectTable (requires pykx):

from kxi_toolkit import select_table

result = select_table(
    table="trades",
    start=start_ts,
    end=end_ts,
    where=["sym=`AAPL"],
    agg={"totalQty": "sum qty"},
)
Back to top