Time series

Market-shaped data — a symbol column, a time column, prices and sizes — is what the engine is tuned hardest for. This tutorial covers the temporal kinds, asof joins, per-symbol windows, and time bucketing.

Temporal kinds

Dates (days since 1970-01-01), timestamps (nanoseconds since the epoch), and times (nanoseconds since midnight) are first-class column kinds. They parse straight from CSV, ride the integer encodings (a year of daily dates packs to a few bits per row), and group, sort, and join like any key. Arithmetic works in their native unit: timestamp ± int shifts nanoseconds, date ± int shifts days, and temporal − temporal is an integer difference.

import keyten as kt

with open("ticks.csv", "w") as f:
    f.write("sym,ts,px\nev,09:30:00,10.0\nev,09:30:05,10.2\nod,09:30:01,20.0\n")

ticks = kt.scan_csv("ticks.csv").collect()
assert ticks.column("ts").dtype == "time"

A plain datetime.time (or date, or naive datetime) works directly wherever a literal is expected, matched against the corresponding column kind:

from datetime import time

late = kt.scan_csv("ticks.csv").filter(kt.col("ts") >= time(9, 30, 1)).collect()
assert late.shape[0] == 2

time_lit(ns) builds the same kind of literal from raw nanoseconds since midnight — useful when the value is already an integer offset rather than a wall-clock time:

late = kt.scan_csv("ticks.csv").filter(
    kt.col("ts") >= kt.time_lit(((9 * 60 + 30) * 60 + 1) * 1_000_000_000)
).collect()
assert late.shape[0] == 2

The asof join

Pricing trades against the prevailing quote is one call. Each trade takes the latest quote at or before its timestamp, per symbol; trades with no earlier quote survive with nulls (the join is always left-outer shaped).

import keyten as kt

trades = kt.DataFrame([
    kt.Series.string("sym", ["ev", "ev", "od"]),
    kt.Series.int("ts", [100, 205, 150]),
    kt.Series.int("qty", [10, 20, 30]),
])
quotes = kt.DataFrame([
    kt.Series.string("sym", ["ev", "ev", "od"]),
    kt.Series.int("ts", [90, 200, 160]),
    kt.Series.float("bid", [9.9, 10.1, 19.8]),
])
priced = trades.lazy().join_asof(
    quotes.lazy(), ("ts", "ts"), by=[("sym", "sym")]
).collect()
assert priced.column("bid").to_list() == [9.9, 10.1, None]

strategy picks a different match rule than the "latest at or before" default: "forward" takes the earliest right row at or after the left row, and "nearest" takes whichever candidate is closer (ties going backward). tolerance bounds how stale a match may be — a duration string for Timestamp keys, a plain int for Int keys — nulling (not dropping) any match further away than that.

quotes2 = kt.DataFrame([
    kt.Series.string("sym", ["ev", "ev"]),
    kt.Series.int("ts", [100, 200]),
    kt.Series.float("bid", [9.9, 10.1]),
])
trades2 = kt.DataFrame([
    kt.Series.string("sym", ["ev", "ev"]),
    kt.Series.int("ts", [130, 400]),
])
nearest = trades2.lazy().join_asof(
    quotes2.lazy(), ("ts", "ts"), by=[("sym", "sym")], strategy="nearest"
).collect()
bounded = trades2.lazy().join_asof(
    quotes2.lazy(), ("ts", "ts"), by=[("sym", "sym")], tolerance=100
).collect()
assert nearest.column("bid").to_list() == [9.9, 10.1]   # 130 is closer to 100 than to 200
assert bounded.column("bid").to_list() == [9.9, None]   # 400 is 200 past its backward quote

Per-symbol windows

over evaluates an expression independently per partition. With an aggregate inside, every row sees its group's value; with a window operation inside, the window restarts at each symbol.

quotes = kt.DataFrame([
    kt.Series.string("sym", ["ev", "ev", "ev", "od", "od"]),
    kt.Series.float("mid", [10.0, 10.25, 10.125, 20.0, 20.5]),
])
out = quotes.lazy().with_columns([
    kt.col("mid").rolling_mean(2, 1).over("sym").alias("smooth"),
    kt.col("mid").diff().over("sym").alias("chg"),
    (kt.col("mid") - kt.col("mid").mean().over("sym")).alias("dev"),
]).collect()
assert out.column("chg").to_list() == [None, 0.25, -0.125, None, 0.5]
assert out.column("smooth").to_list()[:2] == [10.0, 10.125]

When the frame is sorted by symbol — the natural at-rest order for market data — the engine recognizes each symbol as one contiguous run and evaluates partitions without building any hash table at all. The same recognition powers grouping: group_by("sym") over symbol-sorted data folds runs in place.

Bucketing time

truncate(every) floors a temporal value to a multiple of a duration — every is <positive-int><unit> with unit one of ns/us/ms/s/m/h/d (e.g. "1m", "5s"), and it floors toward -inf so pre-epoch values bucket down, not toward zero. group_by_bar(time_col, every) fuses that with group_by: bucket time_col, then group by the bucketed value, in one call.

import keyten as kt

MIN_NS = 60_000_000_000
ticks = kt.DataFrame([
    kt.Series.string("sym", ["ev", "ev", "ev"]),
    kt.Series.time("ts", [t * MIN_NS // 2 for t in range(3)]),
    kt.Series.float("qty", [5.0, 7.0, 9.0]),
])
bars = (
    ticks.lazy()
    .group_by_bar("ts", "1m")
    .agg(kt.col("qty").sum().alias("vol"))
    .sort("ts")
    .collect()
)
assert bars.column("vol").to_list() == [12.0, 9.0]

Under the hood this is equivalent to truncating by hand with integer arithmetic and grouping on the result — t = col("ts").cast("int"), then (t - t % MIN_NS).cast("time") — which is why % is total on temporal kinds (never raises, never nulls) even at the type's extremes.

The same shape builds OHLCV bars — first/max/min/last/sum over price and size per bucket — and corr computes correlations per symbol in the same agg list as everything else. A VWAP (sum(px * qty) / sum(qty)) combines two aggregates arithmetically, which agg() accepts directly:

bars2 = kt.DataFrame([
    kt.Series.time("ts", [t * MIN_NS // 2 for t in range(3)]),
    kt.Series.float("px", [10.0, 10.2, 10.4]),
    kt.Series.float("qty", [5.0, 7.0, 9.0]),
])
vwap = (
    bars2.lazy()
    .group_by_bar("ts", "1m")
    .agg([((kt.col("px") * kt.col("qty")).sum() / kt.col("qty").sum()).alias("vwap")])
    .sort("ts")
    .collect()
)
assert abs(vwap.column("vwap").to_list()[0] - (10.0 * 5.0 + 10.2 * 7.0) / 12.0) < 1e-9

The split form — sum the numerator and denominator in agg, then divide in a following with_columns — still works, and is worth reaching for when the intermediate sums (notional, volume) are useful on their own rather than just the ratio; see LazyFrame and GroupBy.

Log returns on the resulting closes read the same way, with no cast hacks: close.log() - close.log().shift(1) (see Math).

Realized volatility

Bars, log returns, and a rolling window compose into realized volatility with no new machinery — reuse the bars computed above, add a close per bucket, take log returns, then feed those into rolling_std and ewm_std (see Rolling and exponentially weighted windows).

import math
import keyten as kt

MIN_NS = 60_000_000_000
n = 25
closes = [100.0 + 5.0 * math.sin(i / 3.0) for i in range(n)]  # one tick per bar, no randomness
ticks = kt.DataFrame([
    kt.Series.string("sym", ["ev"] * n),
    kt.Series.time("ts", [t * MIN_NS for t in range(n)]),
    kt.Series.float("px", closes),
])
bars = (
    ticks.lazy()
    .group_by_bar("ts", "1m")
    .agg(kt.col("px").last().alias("close"))
    .sort("ts")
    .collect()
)
vol = (
    bars.lazy()
    .with_columns((kt.col("close").log() - kt.col("close").log().shift(1)).alias("ret"))
    .with_columns([
        kt.col("ret").rolling_std(20).alias("vol_rolling"),
        kt.col("ret").ewm_std(span=20).alias("vol_ewm"),
    ])
    .collect()
)
vol_rolling = vol.column("vol_rolling").to_list()
vol_ewm = vol.column("vol_ewm").to_list()
assert vol.column("ret").to_list()[0] is None      # shift(1) leads with a null
assert vol_rolling[:20] == [None] * 20             # rolling_std(20) needs a full window (default min_samples)
assert vol_rolling[20] is not None
assert vol_ewm[0] is None and vol_ewm[1] is None   # first two rows: at most one valid return seen
assert vol_ewm[2] is not None                      # ewm_std needs only two valid returns, not a full window

The full pipeline

Every piece above composes into one script: ticks to bars, bars to returns, returns to realized vol, vol merged against a quote snapshot with an asof tolerance, and a cross-sectional rank of that vol at each timestamp — the shape a real signal pipeline takes, start to finish.

import math
import keyten as kt
from datetime import datetime, timedelta

EPOCH = datetime(1970, 1, 1)


def ns(dt):
    return int((dt - EPOCH) / timedelta(microseconds=1)) * 1000


base = datetime(2024, 1, 1, 9, 30)

# Ticks for two symbols, four trades per minute over three minutes.
tick_sym, tick_ts, tick_px, tick_qty = [], [], [], []
for sym, base_px, amp in (("AAA", 100.0, 1.0), ("BBB", 50.0, 0.4)):
    for m in range(3):
        for s in range(4):
            tick_sym.append(sym)
            tick_ts.append(ns(base + timedelta(minutes=m, seconds=s * 15)))
            tick_px.append(base_px + amp * math.sin((m * 4 + s) / 2.0))
            tick_qty.append(10.0 + s)

ticks = kt.DataFrame([
    kt.Series.string("sym", tick_sym),
    kt.Series.int("ts", tick_ts),
    kt.Series.float("px", tick_px),
    kt.Series.float("qty", tick_qty),
]).lazy().with_columns(kt.col("ts").cast("timestamp")).collect()

# 1. Ticks -> 1-minute bars, VWAP computed directly inside agg().
bars = (
    ticks.lazy()
    .with_columns(kt.col("ts").truncate("1m").alias("bar"))
    .group_by(["sym", "bar"])
    .agg([((kt.col("px") * kt.col("qty")).sum() / kt.col("qty").sum()).alias("vwap")])
    .sort(["sym", "bar"])
    .collect()
)

# 2. Bars -> log returns, per symbol.
rets = bars.lazy().with_columns(
    (kt.col("vwap").log() - kt.col("vwap").log().shift(1)).over("sym").alias("ret")
).collect()

# 3. Returns -> realized vol, both flavors, per symbol.
vol = rets.lazy().with_columns([
    kt.col("ret").rolling_std(2).over("sym").alias("vol_rolling"),
    kt.col("ret").ewm_std(span=2).over("sym").alias("vol_ewm"),
]).collect()

# 4. Merge a quote snapshot onto each bar, asof, with a tolerance —
#    a quote more than 10s stale nulls rather than matching.
quote_sym, quote_ts, quote_bid = [], [], []
for sym, base_bid in (("AAA", 100.0), ("BBB", 50.0)):
    for m in range(3):
        offset = -5 if m < 2 else -15  # the third quote is stale, outside tolerance
        quote_sym.append(sym)
        quote_ts.append(ns(base + timedelta(minutes=m, seconds=offset)))
        quote_bid.append(base_bid + 0.1 * m)

quotes = kt.DataFrame([
    kt.Series.string("sym", quote_sym),
    kt.Series.int("ts", quote_ts),
    kt.Series.float("bid", quote_bid),
]).lazy().with_columns(kt.col("ts").cast("timestamp")).collect()

merged = (
    vol.lazy()
    .join_asof(quotes.lazy(), ("bar", "ts"), by=[("sym", "sym")], tolerance="10s")
    .sort(["sym", "bar"])
    .collect()
)

# 5. Per-timestamp cross-sectional rank of realized vol across symbols.
ranked = (
    merged.lazy()
    .with_columns(kt.col("vol_rolling").rank().over("bar").alias("vol_rank"))
    .sort(["bar", "sym"])
    .collect()
    .to_dict()
)

assert ranked["bid"][-2:] == [None, None]           # both third-bar quotes were stale
assert ranked["vol_rank"][:4] == [None, None, None, None]  # not enough returns yet for a vol
assert ranked["vol_rank"][4:] == [2.0, 1.0]          # AAA's larger swing ranks its vol higher

rolling_std answers "volatility over the last N bars" with a hard cutoff at the window edge; ewm_std answers the same question with a decaying weight instead of a cliff, so a single outlier bar fades out gradually rather than dropping off all at once when it exits the window. Both are sample-variance flavored (ddof = 1) and both need at least two valid returns before they emit anything but null — annualizing either is a multiply by sqrt(periods_per_year) outside the expression, not part of it. Same float-accumulation caveat as the rest of the engine's running-total scans: over very long windows of large-magnitude values, rolling_std's subtract-on-exit running sum can lose precision to cancellation, which is why the kernel clamps variance to >= 0 before the square root instead of ever emitting a NaN from a small negative rounding residue.

For the same pieces assembled into a full session on a generated day of ticks — ingest, market-hours filtering, VWAP bars, vol, an NBBO merge, cross-sectional ranks, and top movers — see the tick analytics guide.