Orderflow Python Binding (orderflow-gregorian09)

Production-focused Python API for the Orderflow runtime.
This package wraps the stable of_ffi_c ABI via ctypes and provides a typed, high-level interface for lifecycle management, subscriptions, snapshots, and external feed ingestion. The package includes a PEP 561 py.typed marker so type checkers can consume the inline annotations shipped with the binding.

The binding also exposes an additive execution API through ExecutionEngine. Execution uses a separate native handle from analytics and returns typed execution events rather than JSON on the order path.

The README is intentionally API-complete so the PyPI page can be used as a single reference, similar to high-signal package pages such as TA-Lib and FastAPI.

What's New In 0.5.0

0.5.0 keeps the existing Python market-data and execution APIs intact while adding production persistence lifecycle, typed adapter diagnostics, config-driven signal replay validation, and deterministic TWAP parent control.

Highlights:

• additive simulated execution APIs: ExecutionEngine, ConcurrentExecutionEngine, OrderRequest, CancelRequest, AmendRequest, RiskLimits, RouteConfig, and execution event dataclasses
• multi-route execution construction for multi-symbol order flow
• bounded concurrent execution worker for many producers and one deterministic native order-state owner
• typed execution events and command reports instead of JSON on the order path
• route/account/symbol-scoped risk checks before adapter routing
• offline WAL and checkpoint-store diagnostics for recovery checks without opening an execution engine
• adapter inventory/status helpers for provider capability discovery before connecting a feed
• signal descriptor discovery through signal_descriptors() and Engine.signal_descriptors() for dashboard/configuration inventory
• signal explanation discovery through Engine.signal_explanation(symbol) for audit and dashboard diagnostics
• signal metrics through Engine.signal_metrics() for state counts, confidence, quality, and explanation coverage diagnostics
• offline config-driven signal validation through SignalConfig, validate_signal_config(), and validate_signal_replay() with parsed reports, retained samples, and replay-order warnings
• manifest/header-driven low-level ctypes signature generation with CI drift checks, while typed high-level wrappers remain manually designed
• analytics-to-execution examples in this README and the handbook
• continued PEP 561 py.typed support and bundled native library lookup

Version policy:

• Python package: 0.5.0
• compatible native of_ffi_c library/header: 0.5.0
• of_execution 0.2.0, of_execution_core 0.2.0, and their independently versioned execution companions behind the native ABI (of_execution_algos is 0.1.0, of_execution_adapters is 0.2.0, and of_fix remains 0.1.x)

Install the Python package and native runtime at the same release version. Execution users should pin of_execution_core and of_execution to compatible 0.2.x versions, and use the matching 0.1.x algorithm/adapter companion releases when they build custom native providers.

Execution quick start

from orderflow import (
    ConcurrentExecutionEngine, ExecutionEngine, ExecutionOrderType, ExecutionSide, ExecutionTimeInForce,
    OrderRequest, RiskLimits, RouteConfig,
)

limits = RiskLimits(False, 100, 1_000_000, 10, 10_000_000, 0)
routes = [
    RouteConfig("SIM", "ACC", "SIM", "ES", True, limits),
    RouteConfig("SIM", "ACC", "SIM", "NQ", True, limits),
]

with ExecutionEngine(routes) as execution:
    events = execution.submit_order(OrderRequest(
        "C1", "ACC", "SIM", "STRAT", "SIM", "ES",
        ExecutionSide.BUY, ExecutionOrderType.LIMIT, ExecutionTimeInForce.DAY,
        10, 5000,
    ))

Deterministic TWAP planning remains separate from OMS submission:

from orderflow import TwapConfig, TwapExecutionAlgo

config = TwapConfig(
    "parent-1", "ACC", "SIM", "TWAP", "SIM", "ES",
    ExecutionSide.BUY, ExecutionOrderType.LIMIT, ExecutionTimeInForce.DAY,
    100, 5000, 1_000, 11_000, 10, 25, 2_000,
)
with ExecutionEngine(routes) as execution, TwapExecutionAlgo(config) as twap:
    child = twap.plan(1_000, "child-1", "order-1", 1_001)
    if child is not None:
        events = execution.submit_order(child.request)
        twap.commit_pending()
        for event in events:
            twap.record_execution(event.last_qty, event.leaves_qty, event.order_status)

plan() is retry-stable while a child is pending. Call commit_pending() only after OMS submission succeeds, or discard_pending() when submission did not occur. The planner itself cannot bypass OMS risk, journaling, or adapters.

ExecutionEngine(route) remains supported for single-symbol integrations. When you pass a route list, native risk accounting is scoped per route/account/symbol.

Use ConcurrentExecutionEngine(routes) when multiple producer threads need to queue commands into one deterministic native worker. Command methods return a sequence number; try_recv_report() returns completed command reports without blocking.

Binding manifest policy

Low-level native symbols are tracked in bindings/api_manifest.toml. The manifest controls membership, ordering, ownership metadata, and exposure. The validated orderflow.h supplies exact parameter and return types to tools/generate_binding_signatures.py, which commits orderflow/_generated_signatures.py. orderflow/_ffi.py defines structure layouts and library loading, then delegates function registration to that generated private module. CI runs the generator with --check and compiles the result.

Do not edit _generated_signatures.py directly. After an additive C ABI change, update the header and manifest, run:

python3 tools/check_api_manifest.py
python3 tools/generate_binding_signatures.py
python3 tools/test_generate_binding_signatures.py
python3 tools/check_binding_parity.py

Generation does not replace the hand-written Engine, ExecutionEngine, and ConcurrentExecutionEngine wrappers. Context managers, dataclasses, typed requests/events, automatic buffers, exceptions, and Python naming remain manual because they express user-facing ownership and ergonomics rather than C types.

Signal descriptor discovery

from orderflow import signal_descriptors

inventory = signal_descriptors()
for descriptor in inventory.get("signals", []):
    print(descriptor["id"], descriptor["required_inputs"], descriptor["output_semantics"])

The descriptor inventory is read-only metadata. It helps dashboards and config tools list built-in signals, required inputs, warmup, parameters, and output semantics without constructing a live strategy or submitting orders.

After a signal has evaluated for a symbol, Engine.signal_explanation(symbol) returns the latest explanation payload with reason code, observed inputs, thresholds, and confidence contributors. This is a diagnostics surface; order submission decisions should still flow through explicit strategy/risk/OMS code.

Engine.signal_metrics() returns a compact runtime summary of the current signal cache: state counts, directional count, average confidence, quality flagged signals, and explanation coverage.

Config-driven replay validation

Use the offline facade to validate a built-in signal without creating a live engine. Parameters are checked against the selected descriptor before native construction. Each observation is evaluated before its future markout is read, and optional exchange timestamps are checked for backward movement.

from orderflow import (
    SignalConfig,
    SignalConfigParameter,
    SignalValidationConfig,
    SignalValidationEvent,
    validate_signal_config,
    validate_signal_replay,
)

signal = SignalConfig(
    "delta_momentum_v1",
    (SignalConfigParameter("threshold", 10),),
)

config_result = validate_signal_config(signal)
if not config_result["valid"]:
    raise ValueError(config_result["error"])

events = [
    SignalValidationEvent(delta=20, last_price=100, ts_exchange_ns=1),
    SignalValidationEvent(delta=-20, last_price=90, ts_exchange_ns=2),
    SignalValidationEvent(delta=-20, last_price=80, ts_exchange_ns=3),
]
report = validate_signal_replay(
    signal,
    events,
    SignalValidationConfig(
        markout_horizon_events=1,
        flat_price_threshold=0,
        min_confidence_bps=0,
        store_samples=True,
        check_monotonic_timestamps=True,
    ),
)

print(report.directional_accuracy_bps, report.label_coverage_bps)
for sample in report.samples:
    print(sample["event_index"], sample["predicted_direction"], sample["correct"])
for warning in report.warnings:
    print(warning["code"])

SignalValidationReport.raw retains the full schema-versioned native document. An invalid registry id, duplicate/unknown parameter, type mismatch, or descriptor range violation raises OrderflowArgError from validate_signal_replay; use validate_signal_config first when a UI or config service needs a non-throwing validation result. This path is for research, replay review, and CI. It does not alter live Engine signal state or send OMS orders.

Architecture

Installation

pip install orderflow-gregorian09

Python support

• Python 3.10+

Native runtime requirement

The Python package is a wrapper. A compatible libof_ffi_c shared library must be available at runtime. Binary wheels can bundle this library under orderflow/native/; source installs can still use an externally-built runtime.

Library resolution order:

1. library_path= passed to Engine(...)
2. ORDERFLOW_LIBRARY_PATH environment variable
3. bundled wheel library (orderflow/native/libof_ffi_c.*)
4. default local debug path (target/debug/libof_ffi_c.*)

export ORDERFLOW_LIBRARY_PATH=/absolute/path/to/libof_ffi_c.so

Quick Start

from orderflow import DataQualityFlags, Engine, EngineConfig, Symbol, StreamKind

with Engine(EngineConfig(instance_id="py-client")) as eng:
    sym = Symbol("CME", "ESM6", depth_levels=10)
    eng.subscribe(sym, StreamKind.ANALYTICS)
    eng.poll_once(DataQualityFlags.NONE)
    print("api_version", eng.api_version)
    print("build_info", eng.build_info)
    print("analytics", eng.analytics_snapshot(sym))
    print("derived", eng.derived_analytics_snapshot(sym))
    print("signal", eng.signal_snapshot(sym))
    print("metrics", eng.metrics())

Production normalized market-data WAL

from orderflow import (
    Engine,
    EngineConfig,
    MarketDataPersistenceFailureAction,
    MarketDataWalConfig,
    MarketDataWalSyncPolicy,
)

eng = Engine(EngineConfig(instance_id="capture"))
eng.configure_market_data_wal(MarketDataWalConfig(
    root_path="data/normalized-wal",
    queue_capacity=8192,
    max_queued_payload_bytes=128 * 1024 * 1024,
    sync_policy=MarketDataWalSyncPolicy.EVERY_RECORDS,
    sync_every_records=1000,
    failure_action=MarketDataPersistenceFailureAction.STOP_TRADING,
))
eng.start()
try:
    # subscribe/poll or configure external ingest here
    print(eng.market_data_persistence_health())
    eng.flush_market_data_wal()  # blocking control-plane barrier
finally:
    eng.stop()
    eng.shutdown_market_data_wal()
    eng.close()

Event admission is bounded and nonblocking; native worker-side encoding keeps filesystem work and JSON formatting off the calling thread. Flush and shutdown block and belong on a control-plane thread. The health payload exposes queue depth and high-water marks, event-time backlog, rejected/abandoned records, write/sync failures, and last written/durable sequence.

Complete End-To-End Example

This example uses deterministic external ingest and simulated execution. It is safe for documentation, CI smoke tests, and first user experiments because it does not connect to a broker.

from orderflow import (
    BookAction,
    DataQualityFlags,
    Engine,
    EngineConfig,
    ExecutionEngine,
    ExecutionOrderType,
    ExecutionSide,
    ExecutionTimeInForce,
    ExternalFeedPolicy,
    OrderRequest,
    RiskLimits,
    RouteConfig,
    Side,
    StreamKind,
    Symbol,
)


def signal_allows_long(analytics: dict, signal: dict) -> bool:
    quality = int(analytics.get("quality_flags", 0))
    delta = int(analytics.get("delta", 0))
    cumulative_delta = float(analytics.get("cumulative_delta", 0.0))
    confidence = float(signal.get("confidence", 0.0))
    return (
        quality == DataQualityFlags.NONE
        and delta > 0
        and cumulative_delta > 0.0
        and confidence >= 0.50
    )


sym = Symbol("SIM", "ES", 10)
limits = RiskLimits(False, 5, 1_000_000, 1, 1_000_000, 0)
routes = [RouteConfig("SIM", "ACC", "SIM", "ES", True, limits)]

with Engine(EngineConfig(instance_id="py-end-to-end")) as market, ExecutionEngine(routes) as execution:
    market.configure_external_feed(ExternalFeedPolicy(2_000, True))
    market.subscribe(sym, StreamKind.ANALYTICS)
    market.subscribe(sym, StreamKind.SIGNALS)

    market.ingest_book(sym, Side.BID, 0, 500_000, 100, BookAction.UPSERT, sequence=1)
    market.ingest_book(sym, Side.ASK, 0, 500_025, 120, BookAction.UPSERT, sequence=2)
    market.ingest_trade(sym, 500_025, 2, Side.ASK, sequence=3)
    market.poll_once(DataQualityFlags.NONE)

    analytics = market.analytics_snapshot(sym)
    signal = market.signal_snapshot(sym)

    if signal_allows_long(analytics, signal):
        events = execution.submit_order(OrderRequest(
            "PY-0001",
            "ACC",
            "SIM",
            "DOCS",
            "SIM",
            "ES",
            ExecutionSide.BUY,
            ExecutionOrderType.LIMIT,
            ExecutionTimeInForce.DAY,
            1,
            500_025,
        ))
        print("events", events)
        print("state", execution.order_state("PY-0001"))
        print("execution metrics", execution.execution_metrics())
    else:
        print("blocked", {"analytics": analytics, "signal": signal})

Production applications add durable persistence, execution journaling, provider adapter ownership, reconnect/recovery policy, and monitoring around this shape.

Public API Reference

Constants

StreamKind

Side

BookAction

DataQualityFlags

Exceptions

Data Classes

Symbol(venue: str, symbol: str, depth_levels: int = 10)

• venue/instrument descriptor used by subscribe/snapshot/ingest APIs.

EngineConfig(...)

EngineConfig fields:

ExternalFeedPolicy(stale_after_ms: int = 15000, enforce_sequence: bool = True)

• external ingest supervision policy for stale and sequence validation.

Engine API

Constructor and properties

Lifecycle and session

Subscription and polling

Adapter discovery

Adapter inventory records include provider metadata and additive capability flags such as supports_backpressure, supports_raw_capture, supports_fixture_replay, supports_stale_detection, and supports_latency_metrics when exposed by the native runtime.

engine.adapter_status() also returns additive operational fields: mode, connection_state, endpoint_redacted, app_name, reconnect_attempt, subscription_count, sorted subscribed_symbols, queue depth/capacity, drop/gap counters, stale state, raw-capture utilization, and optional activity ages. Endpoint output contains only scheme and authority; user information, paths, queries, and fragments are omitted. Consumers should tolerate future additive keys and treat null as unavailable rather than zero.

External feed supervision

External ingest

Snapshots and metrics

WAL lifecycle methods are configure_market_data_wal(config), flush_market_data_wal(), and shutdown_market_data_wal().

book_snapshot(symbol) returns a dictionary with:

• venue
• symbol
• bids
• asks
• last_sequence
• ts_exchange_ns
• ts_recv_ns

The Python wrapper retries automatically if the native snapshot payload is larger than the initial buffer.

session_candle_snapshot(symbol) returns a dictionary with:

• open
• high
• low
• close
• trade_count
• first_ts_exchange_ns
• last_ts_exchange_ns

interval_candle_snapshot(symbol, window_ns) returns a dictionary with:

• window_ns
• open
• high
• low
• close
• trade_count
• total_volume
• vwap
• first_ts_exchange_ns
• last_ts_exchange_ns

Execution API Reference

Execution objects use typed dataclasses and separate native handles from the analytics runtime.

Execution constants

Execution dataclasses

TwapExecutionAlgo

ExecutionEngine

ConcurrentExecutionEngine

Top-level execution helpers

Recovery integrity diagnostics

Use inspect_execution_wal() and inspect_execution_segmented_wal() before recovery drills, after crash restart, or in an operations health check. Both helpers read bytes outside the order path and return counts, byte position, optional sequence range, checksum/sequence failure counts, and validity flags. Use the segmented helper for production rotated WAL roots.

Use inspect_execution_checkpoint_store() with the same restart workflow to validate checkpoint files before selecting a restart point. It reports discovered, valid, and invalid checkpoint counts, total checkpoint bytes, and the latest valid checkpoint id, covered WAL sequence, and creation timestamp. Corrupt checkpoint files do not raise when the root can be listed; they return a report with valid == False so operators can inspect the failure and fall back to the latest valid checkpoint. Missing or unreadable roots raise the mapped native I/O error.

After the integrity checks, call inspect_execution_recovery() to prove that the selected checkpoint and WAL tail can reconstruct deterministic OMS state. The default requires a valid checkpoint. Pass require_checkpoint=False only for a deliberate full-WAL replay; legacy command-only frames cannot safely recreate missing orders and therefore fail closed. The helper performs no writes, opens no venue session, and returns no order identifiers. A successful report still requires venue reconciliation and keeps submissions disabled.

from orderflow import (
    inspect_execution_checkpoint_store,
    inspect_execution_recovery,
    inspect_execution_segmented_wal,
    inspect_execution_wal,
)

single = inspect_execution_wal("execution-wal/wal-000000000001.ofwal")
segmented = inspect_execution_segmented_wal("execution-wal")
checkpoints = inspect_execution_checkpoint_store("execution-checkpoints")
if not single.valid or not segmented.valid or not checkpoints.valid:
    raise RuntimeError("unsafe execution recovery inputs")
if checkpoints.latest_checkpoint_id is None:
    raise RuntimeError("no valid checkpoint available")

recovery = inspect_execution_recovery(
    "execution-wal",
    checkpoint_root="execution-checkpoints",
)
if not recovery.venue_reconciliation_required or recovery.submissions_enabled:
    raise RuntimeError("invalid fail-closed recovery gates")
# Reconcile recovery.open_orders against venue/drop-copy truth before the host
# creates or resumes any live execution engine.

Usage Patterns

Poll-only flow (no callback)

from orderflow import DataQualityFlags, Engine, EngineConfig, Symbol, StreamKind

with Engine(EngineConfig(instance_id="poll-only")) as eng:
    sym = Symbol("CME", "ESM6", 10)
    eng.subscribe(sym, StreamKind.ANALYTICS, callback=None)
    eng.poll_once(DataQualityFlags.NONE)
    snap = eng.analytics_snapshot(sym)
    print("delta", snap.get("delta"))

Callback flow

from orderflow import Engine, EngineConfig, Symbol, StreamKind

def on_analytics(ev: dict) -> None:
    print("analytics event:", ev)

with Engine(EngineConfig(instance_id="cb-flow")) as eng:
    sym = Symbol("CME", "ESM6", 10)
    eng.subscribe(sym, StreamKind.ANALYTICS, callback=on_analytics)
    eng.poll_once()

External ingest + quality gating

from orderflow import (
    BookAction,
    DataQualityFlags,
    Engine,
    EngineConfig,
    ExternalFeedPolicy,
    Side,
    Symbol,
    StreamKind,
)

sym = Symbol("BINANCE", "BTCUSDT", depth_levels=20)

with Engine(EngineConfig(instance_id="external-ingest")) as eng:
    eng.configure_external_feed(
        ExternalFeedPolicy(stale_after_ms=2_000, enforce_sequence=True)
    )
    eng.subscribe(sym, StreamKind.HEALTH, callback=lambda ev: print("health:", ev))

    eng.ingest_book(sym, Side.BID, 0, 62500000, 1000, BookAction.UPSERT, sequence=1)
    eng.ingest_trade(sym, 62510000, 200, Side.ASK, sequence=2)
    eng.poll_once(DataQualityFlags.NONE)

Operational Notes

• callbacks fire during poll_once(...) and ingest_* calls.
• callback handlers should remain non-blocking.
• snapshot APIs decode runtime JSON and return Python dict.
• OrderflowStateError("engine is closed") means close() was already called.

Troubleshooting

FileNotFoundError: Orderflow shared library not found

• build native runtime (cargo build -p of_ffi_c) or provide explicit path.
• verify ORDERFLOW_LIBRARY_PATH points to the correct platform library.

OrderflowArgError from subscribe/ingest

• validate symbol fields (venue, symbol not empty).
• validate enum-like integer constants (Side, BookAction, StreamKind).

No callback events

• ensure subscription callback is not None.
• call poll_once(...) regularly if using adapter-driven mode.

Documentation and Links

• Project docs: https://github.com/gregorian-09/orderflow/tree/main/docs
• Binding guide: https://github.com/gregorian-09/orderflow/tree/main/docs/bindings/python.md
• Handbook: https://github.com/gregorian-09/orderflow/tree/main/docs/handbook
• Changelog: https://github.com/gregorian-09/orderflow/blob/main/CHANGELOG.md