fiml (Python bindings)

Python bindings for the fiml indicator engine. They run the exact Rust feature extractor — features are produced by replaying events through the same dispatch path the live Rust environment uses — so feature generation on historical data in Python and live computation in Rust produce identical output on identical data.

There is intentionally no reimplementation of the indicators in Python. Computing features twice (once in pandas/TA-Lib, once in Rust) drifts: EMA seeds its first value with the raw input, OBV buckets by timestamp, and float summation order matters. One implementation removes that whole class of train/serve skew.

Install from PyPI

Requires CPython 3.12 or newer. This guide targets 0.2.0; until that version is published to PyPI, use source installation.

python3 -m venv .venv

Activate it with source .venv/bin/activate on Linux or macOS, or .\.venv\Scripts\Activate.ps1 in Windows PowerShell, then install:

python -m pip install "fiml[pandas]>=0.2.0,<0.3"

Omit [pandas] for the NumPy-only API. To also fit supported scikit-learn transformations, install "fiml[pandas,sklearn]>=0.2.0,<0.3" instead. Installing a compatible wheel requires no Rust toolchain; building from source requires the tools listed below.

Install from source

To build the package from this repository, you need:

• a Rust toolchain (rustup — https://rustup.rs)
• Python ≥ 3.12

Into a fresh environment (recommended)

From the repository root:

python3 -m venv .venv
source .venv/bin/activate
pip install "./crates/fiml-python[pandas]"

pip invokes the maturin build backend declared in pyproject.toml, compiles the Rust extension, and installs the fiml package with its only runtime dependency (numpy). The pandas extra installs pandas ≥ 2.0 for the trade-DataFrame compute_features and time-horizon future_value APIs; low-level NumPy users can omit it.

Installing straight from git also works:

pip install "fiml @ git+https://github.com/pbezglasny/fiml.git#subdirectory=crates/fiml-python"

For development (editable)

Rebuild-and-reinstall in one step while hacking on the Rust side:

pip install maturin numpy
maturin develop -m crates/fiml-python/Cargo.toml --release

Very new Python? If your interpreter is newer than the pinned PyO3 release knows about, prefix either install command (pip install or maturin develop) with PYO3_USE_ABI3_FORWARD_COMPATIBILITY=1.

Verify the install:

python crates/fiml-python/examples/quickstart.py

Run inside a Jupyter notebook

Jupyter must run a kernel from the environment where fiml is installed. Two ways to get there:

A — install Jupyter into the same venv:

source .venv/bin/activate
python -m pip install "fiml[pandas]>=0.2.0,<0.3" jupyterlab
jupyter lab

B — register the venv as a kernel for an existing Jupyter:

source .venv/bin/activate
python -m pip install "fiml[pandas]>=0.2.0,<0.3" ipykernel
python -m ipykernel install --user --name fiml --display-name "Python (fiml)"

then pick the Python (fiml) kernel in the notebook UI. Alternatively, install directly from a notebook cell into whatever kernel is running:

%pip install "fiml[pandas]>=0.2.0,<0.3"

Note: fiml is a compiled extension module. After rebuilding the Rust code (maturin develop / pip install again), restart the kernel — importlib.reload cannot reload a native module.

A minimal notebook session:

import pandas as pd
import fiml

spec = (fiml.FeatureExtractorSpec()
      .simple_returns("BTCUSDT", [1, 5], source="trade_price")
      .log_returns("BTCUSDT", [1, 5], source="trade_price")
      .field("BTCUSDT", source="trade_price")
      .obv_timed("BTCUSDT", aggregation="1ms", windows=["30s", "60s"])
      .trade_volume_timed("BTCUSDT", aggregation="1ms", windows=["30s", "60s"])
      .vwap_timed("BTCUSDT", aggregation="1ms", windows=["30s", "60s"])
      .trade_count_timed("BTCUSDT", aggregation="1ms", window="60s")
      .day_of_week())

extractor = fiml.FeatureExtractor(spec, output_dtype="float32")

trades = pd.read_csv("trades.csv")     # symbol, ts, price, volume columns
feats = extractor.compute_features(trades)
feats.head()                            # one snapshot after every trade

feats preserves the input index and starts with copied symbol and ts columns, followed by extractor.feature_names(). The feature columns are ready to feed to lightgbm/xgboost/catboost/sklearn. Window indicators default to fiml.WarmupPolicy.FULL_WINDOW, so each cell remains NaN until its complete sample or time window is ready. Pass warmup=fiml.WarmupPolicy.FIRST_VALUE to a builder method when partial values are desired. Returns remain NaN until their exact sample lag exists. Simple returns also require a nonzero lagged value; log returns require positive current and lagged values. Gradient-boosting libraries handle NaN natively; drop or mask those rows for models that don't.

Column mappings remain configurable when a frame uses other names:

feats = extractor.compute_features(
    trades,
    symbol="ticker",
    time="timestamp",
    price="px",
    volume="qty",
    side="aggressor_side",
)

The input must have nondecreasing signed-int64 epoch-millisecond timestamps within each symbol; symbols may interleave in any arrival order. Symbols must be non-empty strings; prices and volumes must be finite and positive. The optional side column uses fiml.SIDE_AGGRESSOR_BUY and fiml.SIDE_AGGRESSOR_SELL; omit it when the input does not classify trade aggressors. The complete frame is validated before the extractor changes.

output_dtype accepts "float32", "float64", numpy.float32, or numpy.float64 and applies to values, transform, and feature DataFrame columns. Calculation state remains float64. The property can be changed until the first event is processed and is then locked.

Time-horizon training targets

fiml.future_value generates offline labels separately from feature replay. Pass one instrument/state stream, sorted by timestamp, at a time. For a row at t, the default direction="backward" selects the last observed value at or before t + horizon. Use direction="forward" explicitly for the first event at or after that time. These are elapsed-time lookups, not row shifts.

import numpy as np
import pandas as pd
import fiml

# Work within one chronological dataset partition and one instrument.
trades = pd.DataFrame({
    "symbol": ["BTCUSDT"] * 5,
    "ts": [0, 200, 490, 700, 1100],  # Unix milliseconds
    "price": [100.0, 100.1, 100.2, 100.4, 100.3],
    "volume": [1.0] * 5,
})
raw = fiml.FeatureExtractorSpec().field(
    "BTCUSDT", source="trade_price", id="price",
)
spec = fiml.PipelineSpec(raw).sma("price", window=2, output="sma2")
pipeline = fiml.ModelInputPipeline(spec)
features = pipeline.compute_features(trades)
future = fiml.future_value(
    trades, timestamp_column="ts", value_column="price", horizon="500ms",
)
target_change = future - trades["price"]
target_return = future / trades["price"] - 1
target_log_return = np.log(future / trades["price"])

epsilon = 0.001
target_class = (
    (target_return > epsilon).astype(float)
    - (target_return < -epsilon).astype(float)
).where(target_return.notna())  # Keep unavailable labels missing.

# Explicitly select model inputs; targets never enter the feature spec.
X = features[pipeline.feature_names()]
valid = target_return.notna() & X.notna().all(axis=1)
X_train, y_train = X.loc[valid], target_return.loc[valid]
# model.fit(X_train, y_train)

The returned Series is named future_value and preserves the original index, including duplicate index labels. The input is never modified. Unsorted or null timestamps are rejected. Timestamp columns may contain pandas datetimes (including timezones), or integer Unix timestamps with timestamp_unit="s", "ms" (default), "us", or "ns". Lookup retains nanosecond precision; floating-point/string timestamps are rejected. Horizons and tolerances accept nonnegative duration strings or timedelta objects. Out-of-range datetime arithmetic raises an error instead of wrapping.

For duplicate timestamps, backward lookup chooses the last input row and forward lookup chooses the first, including exact matches and zero horizons. Matched null values stay null. A target time beyond the stream's last timestamp is missing in both modes, even when backward lookup could reuse an older value. Use tolerance="100ms" to reject stale backward matches or distant forward matches. Observation coverage is defined by the input's first/last timestamps.

Inspect matches or generate several horizons with the same primitive:

details = fiml.future_value(
    trades, timestamp_column="ts", value_column="price", horizon="500ms",
    tolerance="100ms", return_details=True,
)
# Columns: timestamp (original), target_time, matched_timestamp, future_value.
# Lookup timestamps are datetimes; invalid matches have NaT and a missing value.
targets = pd.DataFrame({
    f"target_return_{horizon}": fiml.future_value(
        trades, timestamp_column="ts", value_column="price", horizon=horizon,
    ) / trades["price"] - 1
    for horizon in ["100ms", "500ms", "1s"]
})

Generate targets within each chronological training/validation/test partition so training labels cannot reach into a later partition. Fit preprocessing only on training rows. For multiple instruments, call the helper on each instrument's stream separately; it does not infer grouping from a symbol column.

Feature-vector specs

FeatureExtractorSpec is the versioned parity artifact shared by Python training and Rust serving. Author it fluently, serialize it once, and load that same JSON in either language:

spec = fiml.FeatureExtractorSpec(capacity=128, checksum="model-v7").field(
    "BTCUSDT", source="trade_price", id="price"
)
json_text = spec.to_json()

restored = fiml.FeatureExtractorSpec.from_json(json_text)
extractor = fiml.FeatureExtractor.from_json(json_text, output_dtype="float64")

Omitting capacity keeps it equal to the active output count. An explicit larger capacity reserves trailing model-input cells. Those cells are exposed as __reserved_<index> columns and remain NaN; adding outputs beyond the fixed capacity raises ValueError. n_features() and feature_names() cover the complete model width, while active_feature_count() excludes reserved cells. The optional checksum is opaque metadata and is round-tripped without being calculated or verified. Serialized specs also include a canonical required_events list of concrete symbol/event pairs that callers must feed to the extractor.

Builder methods: field, simple_returns, log_returns, cvd, volatility, sma_timed, obv_timed, volatility_timed, vpt, trade_count_timed, trade_volume_timed, vwap_timed, day_of_week, and time_since_first_event_of_day (fixed-offset tz, default "UTC"). CVD, timed SMA, timed OBV, timed trade volume, and timed VWAP accept ordered window lists; each list becomes one runtime indicator with adjacent output cells. Durations are strings ("500ms", "1s", "5m", "1h"). Every window builder accepts a keyword-only warmup enum; its default is fiml.WarmupPolicy.FULL_WINDOW. vpt consumes trade price and volume, starts at zero, and emits its cumulative value from the first trade.

Raw fields and timed SMA accept a keyword-only source of "price", "volume", "trade_price", or "trade_volume" (default "price"). Use a trade source with compute_features. Output names are generated canonically at compilation, from each structural feature key; arbitrary aliases are accepted only when loading JSON with an explicit output id. ASCII symbol identity is case-insensitive throughout the library and canonical names use lowercase symbols.

The process-wide symbol interner holds at most 512 distinct names, including the reserved __global__ name. Builders, JSON loading, and runtime symbol() calls raise ValueError when a new name would exceed this limit. Existing names and their ASCII case variants remain usable, and a rejected name does not change existing handles or feature state. The limit is shared across all extractors and pipelines; dropping a runtime does not free interned names.

Fitted model-input pipelines

Keep raw indicator extraction separate from the fitted scalar transformations consumed by a model. PipelineSpec clones its raw FeatureExtractorSpec, keeps transformations in authored order, and serializes the canonical artifact read by the Rust PipelineSpec:

raw_spec = fiml.FeatureExtractorSpec(checksum="raw-v1").field(
    "BTCUSDT", source="trade_price", id="price"
)
model_spec = fiml.PipelineSpec(raw_spec, checksum="model-v1")

# Transfer fitted arrays explicitly in the raw spec's canonical order.
for feature_id, mean, scale in zip(
    raw_spec.feature_ids(), scaler.mean_, scaler.scale_, strict=True
):
    model_spec.standard_scale(
        feature_id, mean=float(mean), scale=float(scale)
    )

pipeline = fiml.ModelInputPipeline(model_spec, output_dtype="float64")

Sample averages belong to PipelineSpec and ScalarStage:

raw = fiml.FeatureExtractorSpec().field("BTCUSDT", source="trade_price", id="price")
spec = (fiml.PipelineSpec(raw)
        .sma("price", window=20, output="sma20")
        .ema("price", window=10, output="ema10"))

Both methods accept warmup=fiml.WarmupPolicy.FULL_WINDOW and output=None. The default output ID is the input ID. Use successive scalar stages to compose scaling, averages, and lags. See the migration guide.

identity(input, *, output=None), lagged(input, *, lag_window, output=None), and standard_scale(input, *, mean, scale, output=None) append one final scalar and return the spec for fluent use. Omitting output reuses the input ID. Omitting model capacity tracks the transformation count; an explicit capacity stays fixed and creates trailing __reserved_<index> cells. feature_ids() and raw_feature_ids() exclude reserved cells. Raw and model checksums are independent opaque metadata.

lagged reads the raw value from lag_window accepted events earlier. The window must be between 1 and 10,000 inclusive; outputs remain NaN until enough history exists. Multiple lags of the same input share one history buffer.

ModelInputPipeline mirrors the extractor's stateful symbol, update, transform, and compute_features event-replay APIs. values() and feature_names() describe final model input; raw_values() and raw_feature_names() expose only the current diagnostic raw snapshot. A pipeline DataFrame contains copied symbol/time metadata followed by final model columns only. Final and raw calculations stay in float64, while returned arrays use output_dtype, which locks after the first accepted event.

See examples/model_input_pipeline.py for a focused runnable example. Indicator state is cumulative; call pipeline.reset() before replaying an independent stream. Reset retains fitted parameters and registered symbol handles.

Fitting sklearn stages

Install fiml[sklearn] (currently scikit-learn >=1.9,<1.10) for training. Append SimpleImputer, StandardScaler, RobustScaler, MinMaxScaler, MaxAbsScaler, PowerTransformer, QuantileTransformer, VarianceThreshold, PCA, or fiml.ScalarStage instances before fitting or replaying:

from sklearn.decomposition import PCA
from sklearn.feature_selection import VarianceThreshold
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import PowerTransformer, QuantileTransformer, RobustScaler

base_spec = fiml.PipelineSpec(raw_spec)
for feature_id in raw_spec.feature_ids():
    base_spec.identity(feature_id)

pipeline = (fiml.ModelInputPipeline(base_spec)
            .add_transformation(SimpleImputer(add_indicator=True), name="impute")
            .add_transformation(VarianceThreshold(), name="variance")
            .add_transformation(RobustScaler(unit_variance=True), name="scale")
            .add_transformation(PowerTransformer(), name="power")
            .add_transformation(QuantileTransformer(n_quantiles=256), name="quantile")
            .add_transformation(PCA(n_components=2, whiten=True), name="pca"))
pipeline.fit(kind, symbol, timestamp, price=prices, volume=volumes, fit_mask=ready)
X_train = pipeline.transform(kind, symbol, timestamp, price=prices, volume=volumes)
artifact = pipeline.to_json()

Scalar stages select, rename, scale, or lag values from the preceding stage:

pipeline = (fiml.ModelInputPipeline(base_spec)
            .add_transformation(StandardScaler(), name="scale")
            .add_transformation(PCA(n_components=2), name="pca")
            .add_transformation(fiml.ScalarStage()
                                .identity("pca__pc0", output="current")
                                .lagged("pca__pc0", lag_window=2, output="lag2"),
                                name="lags")
            .add_transformation(StandardScaler(), name="final_scale"))

All definitions within a ScalarStage read the same preceding layout. To scale lag2 with fixed parameters, append another stage such as fiml.ScalarStage().standard_scale("lag2", mean=0., scale=2., output="scaled_lag"). Only explicitly declared outputs survive a scalar stage; use identity to carry columns forward. Input IDs and parameters are validated when the stage is added to a spec or reached during fitting. The pipeline snapshots the builder when added. For inference with already known parameters, use spec.scalar_stage(stage); this also works without sklearn.

Register the event symbols with this pipeline before constructing the symbol array. fit takes the same event columns as transform and returns self. fit_mask is an optional boolean vector selecting training snapshots, for example after indicator and downstream lag warm-up; every event is still replayed to preserve history. Even excluded middle rows enter lag history. Selected inputs to each sklearn estimator and selected final outputs must be finite, except that SimpleImputer and PowerTransformer and QuantileTransformer accept NaNs and still reject infinities. A later imputer is required if selected final outputs retain NaNs. An earlier scalar selection can drop unneeded warm-up columns. Fitting replays each completed prefix through Rust, so its cost grows with the number of stages. Fit only the chronological training partition. Reserved cells are excluded from fitting and stay NaN in output.

Each stage consumes the preceding complete active vector; the base spec can include lagged features. Scaling preserves names; PCA generates pca__pc0, etc. VarianceThreshold preserves the IDs and order of retained inputs. Capacity defaults to the fitted final width, while an explicitly authored capacity applies to the final vector; intermediate training layouts may be wider. PCA emits all NaNs until every stage input is finite. fit_transform fits and replays, returning every event row; keep labels aligned and select usable rows when training the predictor.

Successful fit leaves cold event state. Failed refits preserve the previous artifact and runtime. transform never fits or resets. Use to_spec() for an independent fitted spec snapshot. from_json restores an inference-only pipeline without sklearn; it starts cold and requires the historical warm-up prefix. DataFrame and order-book replay still work for inference; fitting currently accepts the columnar event API only. Arbitrary estimators/subclasses, copy=False, RobustScaler(with_scaling=True, unit_variance=True) with equal quantiles or quantiles at 0 or 100, and sklearn GridSearchCV integration are unsupported. Those quantile ranges produce a non-finite or zero fitted scale that cannot enter the pipeline's finite numeric state.

MinMaxScaler supports every valid feature_range and both clip settings. Without clipping, future values outside the fitted data range can exceed the configured feature range, matching sklearn.

MaxAbsScaler preserves zero and sign, and supports both clip settings. Without clipping, future magnitudes can exceed 1; fitting is not robust to outliers.

SimpleImputer supports dense numeric input with missing_values=np.nan, copy=True, the mean, median, most_frequent, and constant strategies, and both keep_empty_features and add_indicator settings. Empty columns follow sklearn's fitted retained layout. Indicator columns use sklearn's missingindicator_<input> names and only cover columns missing during fitting; NaNs first seen during inference are still replaced but do not add new indicators. Filling indicator warm-up is an explicit modeling choice, not evidence that a feature is ready. Use fit_mask to exclude those rows when that distinction matters.

PowerTransformer supports exact sklearn estimators with copy=True, both yeo-johnson and box-cox, and standardize=True/False. FIML preserves NaNs. Box-Cox rejects nonpositive fitting values; during frozen inference, a nonpositive value produces NaN only in that output column instead of rejecting the event.

QuantileTransformer supports exact sklearn estimators with copy=True, dense input, and uniform or normal output. Python owns fitting options such as n_quantiles, subsample, and random_state; artifacts store only the effective quantile/reference tables and boundary constants. Inference preserves IDs and NaNs, averages both sides of repeated quantiles like sklearn, and saturates values outside the fitted range. Normal output uses a full-precision AS 241 inverse-normal approximation and sklearn's exported finite tail clipping. The nonlinear mapping changes linear correlations. All-NaN fitted columns are stored as explicit indices and keep producing NaN, allowing a following imputer to fill them without putting non-finite numbers in the JSON artifact.

VarianceThreshold supports the exact sklearn estimator with any finite, nonnegative threshold; it has no copy option. Fitting requires finite selected training inputs. Rust stores only sklearn's ordered retained-column indices, not variances or training statistics. Inference preserves NaNs in retained columns and ignores discarded columns. Positive thresholds depend on input scale; VarianceThreshold is neither correlation-based nor supervised feature selection.

JSON writers and readers use pipeline version 3.0 and nested extractor version 2.0. Earlier artifacts are rejected; see the 0.2.0 migration guide before rebuilding and re-exporting configurations. Scalar stages serialize as {"type": "scalar", "transformations": [...]} within the required model_input.stages array. They use the same transformation fields as the base layout. Rust consumers should enable serde_json's float_roundtrip feature to preserve every fitted f64 on load, as the Python bindings do. No sklearn or matrix-library dependency is needed in Rust.

Run examples/sklearn_pipeline.py for a complete training/export example that checks Rust output against sklearn. See the implementation design for the wire contract and numerical limits. Whitening nearly zero-variance components can amplify sklearn/Rust rounding differences substantially; retain fewer components or disable whitening when needed. Train the predictor on FIML's returned matrix so deployed inference uses the same arithmetic.

Low-level event API

For raw event arrays (mixed streams, custom sources), transform replays a full stream and returns one feature row per event; update steps a single event; values() reads the current vector.

import numpy as np

btc = extractor.symbol("BTCUSDT")       # integer handle for the symbol column

n = prices.shape[0]
kind      = np.full(n, fiml.KIND_PRICE, dtype=np.uint8)
symbol    = np.full(n, btc,             dtype=np.int64)
timestamp = ts.astype(np.int64)         # epoch milliseconds
features  = extractor.transform(kind, symbol, timestamp, price=prices)

kind, symbol and timestamp are required; payload columns are keyword-only and optional, and each row reads only the columns its kind needs:

A row whose kind needs a column you did not pass raises ValueError naming that column; any column you do pass must match the length of kind. All rows are validated before the first dispatch, so a bad row never leaves the extractor half-stepped. Rows must be nondecreasing by timestamp within each symbol and are dispatched in array order. update(...) takes the same keyword payloads as scalars. The obsolete KIND_ORDERBOOK placeholder raises a migration error; use OrderBookEvent for real snapshots and deltas.

For both extractors and pipelines, required price and volume payloads must be finite, even for unsubscribed symbols. NaN and ±infinity raise ValueError before changing raw/final snapshots, timed-feature state, or the timestamp watermark. transform checks every row before dispatch and includes row N: in the error, leaving the whole batch unapplied. A rejected first input does not lock output_dtype. Unused payload columns are ignored. These low-level APIs accept finite zero and negative values; DataFrame compute_features continues to require strictly positive values.

Order-book configuration and replay

Book features use the same raw extractor and model-input pipeline as trade features. Configure every required symbol before constructing a runtime:

raw = (fiml.FeatureExtractorSpec()
       .configure_order_book("BTCUSDT", update_policy="contiguous", buffer_size=8)
       .order_book_mid_price("BTCUSDT")
       .order_book_imbalance("BTCUSDT", [5, 1])
       .order_book_best_bid_size("BTCUSDT"))
extractor = fiml.FeatureExtractor.from_json(raw.to_json())
rows = extractor.transform_order_book([
    fiml.OrderBookEvent.snapshot("BTCUSDT", 1, 1,
        [("98", "2"), ("97", "6")], [("102", "2"), ("103", "6")]),
    fiml.OrderBookEvent.delta("BTCUSDT", 2, 2, [("bid", "98", "6")]),
])

configure_order_book(symbol, *, update_policy, buffer_size) serializes only construction parameters. Policies are contiguous (consecutive IDs) and monotonic (increasing IDs, gaps allowed). Duplicate/global symbols are rejected; missing books fail runtime construction. Rebuilding from JSON starts fresh with NaN outputs. Buffer capacity bounds retained deltas; zero capacity retains the core's behavior and raises a capacity error when a delta needs retention.

Builders are order_book_mid_price, order_book_spread, order_book_spread_bps, order_book_weighted_mid_price, order_book_microprice, and order_book_imbalance(symbol, n_levels). Imbalance accepts a list of one to sixteen positive depths in output order. Quote builders are order_book_best_bid_price, order_book_best_bid_size, order_book_best_ask_price, and order_book_best_ask_size.

The remaining query builders take (symbol, side, ...), with side bid or ask:

Prices and sizes in queries and events are decimal strings, never floats. Out-of-range precision and negative values raise errors without rounding. Snapshots accept (price, size) pairs; deltas accept (side, price, size) triples. Both lists and tuples are accepted. Zero size deletes a level. Event symbols are names, independent of numeric symbol() handles. Timestamps are signed int64 milliseconds and update IDs are unsigned uint64.

Both runtimes expose update_order_book(event) and transform_order_book(events). The batch method returns one raw or model row per accepted event, using the configured output dtype. Constructors validate payloads; batch processing validates all configured symbols and timestamp order before any event applies. These errors leave the entire batch unapplied. Sequence gaps, stale snapshots, and history-capacity errors depend on book state: the batch stops with row N: context, keeps earlier accepted rows, and performs the rejected row's core synchronization transition. Later rows do not apply. Inspect values() (and pipeline raw_values()) and send a snapshot to resynchronize; do not blindly replay the accepted prefix. Rejected rows preserve raw/model outputs and the timestamp watermark. Buffered and stale delta events leave book outputs unchanged; accepted events still advance pipeline lag history.

Python payload/output conversion and book storage may allocate. Rust book feature derivation and model transformations allocate no memory during dispatch. See examples/order_book_replay.py for a runnable serialized pipeline example; tests/fixtures/order_book_replay.json is replayed by both Rust and Python tests with exact float64 expectations, including errors and missing values.

Binance depth stream example

examples/binance_depth.py loads the included real BTCUSDT recording, converts messages to OrderBookEvent.snapshot, and builds a ModelInputPipeline with mid-price, spread, spread in basis points, microprice, imbalance at 1/5/20 levels, best bid size, and previous-snapshot imbalance. Run from the repository root; output is a CSV feature matrix:

uv run --with ./crates/fiml-python python crates/fiml-python/examples/binance_depth.py

To record 30 fresh messages and replay the saved file:

uv run --with ./crates/fiml-python --with websockets python crates/fiml-python/examples/binance_depth.py --capture 30 --file /tmp/depth.jsonl
uv run --with ./crates/fiml-python python crates/fiml-python/examples/binance_depth.py --file /tmp/depth.jsonl

Capture refuses to overwrite an existing file. websockets is needed only for capture. Each JSONL record preserves the source URL and raw payload plus received_at_ms (local receive time, since partial depth has no exchange time). Prices and quantities remain decimal strings. Replay keeps file order and skips repeated/older update IDs; the first lagged output is NaN.

The Binance partial-depth stream btcusdt@depth20@100ms replaces the top 20 bids and asks on each update. Features therefore describe only those levels. This example uses snapshots with no delta buffer. Full-book replay requires the separate diff-depth stream, a REST snapshot, and validation of Binance's U/u update ranges before applying deltas.

Determinism rules (read these)

To guarantee identical output between Python (batch) and Rust (live):

1. f64 calculation state on both sides. The extractor calculates in f64; choose output_dtype="float64" when comparing exact Python/Rust output.
2. Same FeatureExtractorSpec configuration — same periods, aggregation/window durations, warm-up policies, symbol names, and feature order.
3. Replay the full event stream in the same order with the same millisecond timestamps. Do not downsample or skip rows: timed indicators (SmaTimed, ObvTimed, TradeCountTimed, TradeVolumeTimed) bucket by timestamp.
4. Use the same trade-side classifications. CVD ignores trades without a side and uses positive volume for SIDE_AGGRESSOR_BUY, negative volume for SIDE_AGGRESSOR_SELL.
5. Intern the same symbol strings on both sides.

Verifying parity

• transform(...) over the whole stream equals stepping the same events one at a time with update(...) then reading values() — same code path.
• End-to-end: run a recorded dataset + one feature-vector spec through the live Rust extractor and through transform; the two float64 matrices must be exactly equal (not just approximately; NaN warmup cells compare with equal_nan=True).

See examples/quickstart.py.

Timestamps must be nondecreasing per symbol across every update, transform, and compute_features call on an extractor. Equal timestamps are processed in caller-provided arrival order. transform and compute_features validate the entire batch before changing extractor state.

Each symbol shares one ordering stream across all event kinds, including unsubscribed symbols. KIND_TIME belongs to Symbol::GLOBAL; its symbol handle is ignored. Timed SMA, OBV, and trade-count windows advance only on events for their configured symbol. Other kinds for that symbol advance expiration and warm-up without supplying a sample; other symbols and global Time ticks leave them unchanged. Global calendar/session features use the maximum accepted timestamp, so older events for another symbol cannot move their day backward. Pipeline transformations run in accepted-event arrival order.

Externally supplied context

Register precomputed values alongside live features. Context names must be nonempty; use "__global__" for shared context or a market symbol for per-market values. Custom IDs are optional; updates always address registered feature IDs.

raw = (fiml.FeatureExtractorSpec()
       .field("BTCUSDT", source="trade_price", id="price")
       .context("BTCUSDT", "previous_day_high", id="previous_day_high")
       .context("__global__", "previous_day_volume", id="previous_day_volume"))
extractor = fiml.FeatureExtractor(raw)
extractor.update_context({"previous_day_high": 105.0, "previous_day_volume": 2000.0})

FeatureExtractor and ModelInputPipeline both expose update_context(dict). Each call is an atomic partial replacement: omitted IDs retain their values, None clears a value to NaN, and an empty dictionary does nothing. Unknown IDs, non-context IDs, and non-finite numbers are rejected before any writes. Context starts as NaN and is retained across market events. A nonempty successful update locks runtime configuration just as event processing does.

Pipeline values refresh immediately through stateless transformations, including scaling, selection, imputation, and PCA. Context updates leave timestamps, order books, indicators, and stateful transformer outputs unchanged. Context never counts as an SMA/EMA observation. Event lags sample the held context on actual events; context updates themselves do not advance lags. Supply averages of daily context snapshots as additional precomputed context values.

For NumPy replay, supply an optional schedule to transform, transform_order_book, fit, or fit_transform:

context_updates = {
    0: {"previous_day_high": 105.0},
    2500: {"previous_day_high": 108.0},
}
# kind, symbol, timestamp contain at least 2501 event rows.
matrix = extractor.transform(kind, symbol, timestamp, price=price,
                             volume=volume, context_updates=context_updates)

Schedule keys are zero-based event-row indices within that call. Updates apply immediately before the corresponding event and produce no additional rows. The entire schedule is validated before replay. Fitting reapplies it in every cold replay, including rows excluded by fit_mask; failed refits preserve live state. Specs serialize definitions only. New runtimes and ModelInputPipeline.reset() clear context values.

For chunked DataFrame extraction, update context explicitly between chunks:

extractor.update_context({"previous_day_high": yesterday_high})
first = extractor.compute_features(first_chunk)
extractor.update_context({"previous_day_high": next_previous_day_high})
second = extractor.compute_features(second_chunk)
features = pd.concat([first, second])

Split chunks at context availability boundaries. Callers own availability timing, day/session boundaries, and replacement or clearing of stale values; there is no automatic expiry, timestamp join, or historical-data fetching.