Wingfoil

https://wingfoil.readthedocs.io/en/latest/
Admin
Wingfoil is a high-performance stream processing framework designed for latency-critical use cases...
Tokens:25,803
Snippets:178
Trust Score:9.6
Update:5 days ago
Show doc for...
Tokens:
Raw
### Quick Start: Expected log output

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Example output from the quick start snippet, showing INFO logs with timestamps and values.

```text
[INFO wingfoil] 0.000_092 >> hello, world 1
[INFO wingfoil] 1.008_038 >> hello, world 2
[INFO wingfoil] 2.012_219 >> hello, world 3
```

--------------------------------

### Create a KDB+ table for wingfoil integration

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Start a q process (q -p 5000) and create the target table for KDB+ write/read examples.

```q
test_trades:([]time:`timestamp$();sym:`symbol$();price:`float$();qty:`long$())
```

--------------------------------

### Install Wingfoil with pip

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Install the wingfoil package from PyPI. Wheels are available for Linux, macOS, and Windows on CPython 3.8+.

```bash
pip install wingfoil
```

--------------------------------

### PrometheusExporter.serve

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Starts the HTTP server for Prometheus metrics and returns the bound port.

```APIDOC
## PrometheusExporter.serve

### Description
Starts the HTTP server that serves a Prometheus-compatible GET /metrics endpoint. Returns the port that was bound.

### Method
serve

### Signature
serve()

### Returns
int - The port that was bound.
```

--------------------------------

### Stream.latency_report

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Installs a latency report sink.

```APIDOC
## Stream.latency_report

### Description
Installs a latency report sink. The stream must carry TracedBytes values. Per-stage delta statistics (count/min/mean/p50/p99/max) are printed on graph shutdown.

### Method
latency_report

### Signature
latency_report(stages, print_on_teardown=True)

### Parameters
- **stages** (list) - Required - Stage names in order (same list used for Latency).
- **print_on_teardown** (bool) - Optional - Whether to print the report on shutdown (default True).

### Returns
Node - A Node that drives the report sink.
```

--------------------------------

### Quick Start: ticker stream with count, map, and logged

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Defines a stream that ticks every second, counts ticks, maps to a formatted string, logs each value, and runs in real-time for 3 seconds. The run() call blocks until the stop condition is reached.

```python
from wingfoil import ticker

(
    ticker(1.0)                       # tick every second
        .count()                      # 1, 2, 3, ...
        .map(lambda n: f"hello, world {n}")
        .logged(">>")                 # INFO-log each value
        .run(realtime=True, duration=3.0)
)
```

--------------------------------

### iceoryx2 shared-memory pub/sub

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Zero-copy pub/sub over shared memory. Requires building wingfoil with the iceoryx2-beta feature (opt-in). Both ends accept variant (Ipc for cross-process, Local for same-process), history_size, and publisher-side initial_max_slice_len. The example uses Local variant and Signaled mode.

```python
from wingfoil import ticker, iceoryx2_sub, Iceoryx2ServiceVariant, Iceoryx2Mode, Graph

service = "wingfoil/demo"

sub = iceoryx2_sub(
    service,
    variant=Iceoryx2ServiceVariant.Local,   # or Ipc
    mode=Iceoryx2Mode.Signaled,             # Spin | Threaded | Signaled
)
sub = sub.inspect(lambda msgs: print("received:", msgs)).collect()

pub = (
    ticker(0.1).count()
        .map(lambda n: f"tick {n}".encode())
        .iceoryx2_pub(service, variant=Iceoryx2ServiceVariant.Local)
)

Graph([pub, sub]).run(realtime=True, duration=0.5)

```

--------------------------------

### Node.run

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Runs the node in realtime or with specified start time and duration.

```APIDOC
## Node.run

### Description
Runs the node. The `realtime` parameter controls whether the node runs in realtime mode. Optional `start`, `duration`, and `cycles` parameters allow for scheduled or bounded execution.

### Method
run

### Signature
run(realtime, start=None, duration=None, cycles=None)

### Parameters
- **realtime** (bool) - Required - Whether to run in realtime mode.
- **start** (optional) - Optional - Start time for the node.
- **duration** (optional) - Optional - Duration for the node run.
- **cycles** (optional) - Optional - Number of cycles to run.

### Returns
None
```

--------------------------------

### Build from source with maturin

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Builds wingfoil locally, e.g. to enable the iceoryx2-beta feature or develop against the bindings. Clones the repo, installs maturin, and runs maturin develop (optionally with --features iceoryx2-beta) and pytest.

```bash
git clone https://github.com/wingfoil-io/wingfoil
cd wingfoil/wingfoil-python
pip install maturin
maturin develop                               # or: maturin develop --features iceoryx2-beta
pytest

```

--------------------------------

### Stream.latency_report_if

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Like latency_report, but only installs the sink when enabled is True.

```APIDOC
## Stream.latency_report_if

### Description
Like latency_report, but only installs the sink when enabled is True. When False, returns the upstream as a Node (no report sink).

### Method
latency_report_if

### Signature
latency_report_if(stages, enabled, print_on_teardown=True)

### Parameters
- **stages** (list) - Required - Stage names in order.
- **enabled** (bool) - Required - Whether to install the sink.
- **print_on_teardown** (bool) - Optional - Whether to print the report on shutdown (default True).

### Returns
Node
```

--------------------------------

### Publish and subscribe to etcd with etcd_pub and etcd_sub

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Start etcd (docker run --rm -p 2379:2379 gcr.io/etcd-development/etcd:v3.5.0). Publish each dict as {"key": str, "value": bytes} or a list of them per tick. Subscribe to snapshot + watch events under a prefix; each tick = list[event]. Event dicts have shape: {"kind": "put"|"delete", "key": str, "value": bytes, "revision": int}.

```python
from wingfoil import ticker, etcd_sub

ENDPOINT = "http://localhost:2379"

# Publish: each dict = {"key": str, "value": bytes}, or a list of them per tick.
(
    ticker(1.0).count().limit(3)
        .map(lambda i: {"key": f"/wf/item/{i}", "value": str(i).encode()})
        .etcd_pub(ENDPOINT, lease_ttl=30.0, force=True)
        .run(realtime=True)
)

# Subscribe: snapshot + watch events under a prefix; each tick = list[event].
events = etcd_sub(ENDPOINT, "/wf/").inspect(print)
events.run(realtime=True, duration=2.0)
```

--------------------------------

### Graph.run

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Runs the stream processing graph in realtime or with a specified start time and duration.

```APIDOC
## Graph.run

### Description
Runs the stream processing graph. The `realtime` parameter controls whether the graph runs in realtime mode. Optional `start`, `duration`, and `cycles` parameters allow for scheduled or bounded execution.

### Method
run

### Signature
run(realtime, start=None, duration=None, cycles=None)

### Parameters
- **realtime** (bool) - Required - Whether to run in realtime mode.
- **start** (optional) - Optional - Start time for the graph.
- **duration** (optional) - Optional - Duration for the graph run.
- **cycles** (optional) - Optional - Number of cycles to run.

### Returns
None
```

--------------------------------

### Run a stream in historical mode with realtime=False

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Pass realtime=False to drive the graph from source timestamps. Add start= for sources requiring a specific epoch start (e.g., kdb_read) and cap replay with duration= or cycles=. Historical mode is deterministic, suitable for unit tests and strategy backtests.

```python
from datetime import datetime, timezone
from wingfoil import ticker

stream = ticker(0.01).count().collect()
stream.run(
    realtime=False,
    start=datetime(2025, 1, 1, tzinfo=timezone.utc),
    cycles=5,
)
print(stream.peek_value())   # [1, 2, 3, 4, 5]
```

--------------------------------

### kdb_read() - Read from KDB+

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Reads from a KDB+ table using a time-sliced query, returning a stream of dicts. `start` and `duration` bound the replay window against the KDB time column.

```APIDOC
## kdb_read() - Read from KDB+

### Description
Reads from a KDB+ table using a time-sliced query, returning a stream of dicts. `start` and `duration` bound the replay window against the KDB time column.

### Method
kdb_read

### Parameters
- **host** (str) - Required - KDB+ host.
- **port** (int) - Required - KDB+ port.
- **query** (str) - Required - KDB+ query string.
- **time_col** (str) - Required - Name of the time column.
- **chunk_size** (int) - Required - Number of rows per chunk.

### Request Example
```python
from wingfoil import ticker, kdb_read

HOST, PORT, TABLE = "localhost", 5000, "test_trades"

rows = kdb_read(
    host=HOST, port=PORT,
    query=f"select from {TABLE}",
    time_col="time",
    chunk_size=10_000,
).collect()
rows.run(realtime=False, start=946684800.0, duration=86400.0)
print(rows.peek_value())
```

### Response
#### Success Response
- **peek_value()** (list) - List of dicts, each representing a row.

#### Response Example
No explicit response example provided.
```

--------------------------------

### Build a pandas DataFrame from multiple streams using build_dataframe

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Use wingfoil.build_dataframe to align several .dataframe() streams by graph time. The example creates two streams from a common source, runs them with realtime=False, and prints the resulting DataFrame.

```python
from wingfoil import ticker, Graph, build_dataframe

source = ticker(0.01).count().limit(5)
prices = source.map(lambda i: 100 + i).dataframe()
quantities = source.map(lambda _: 10).dataframe()

Graph([prices, quantities]).run(realtime=False)

df = build_dataframe({"price": prices, "qty": quantities})
print(df)
#       time  price  qty
# 0  0.0e+00    101   10
# 1  1.0e-02    102   10
# ...
```

--------------------------------

### Write and read KDB+ data with kdb_write and kdb_read

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Write each dict as a row to a KDB+ table using kdb_write, specifying non-time columns. Read with a time-sliced query using kdb_read, where start and duration bound the replay window against the KDB time column. Supported kdb_write column types: "symbol", "float", "long", "int", "bool".

```python
from wingfoil import ticker, kdb_read

HOST, PORT, TABLE = "localhost", 5000, "test_trades"

# Write: each dict becomes one row; "columns" names the non-time columns.
(
    ticker(1.0).count().limit(10)
        .map(lambda i: {"sym": "AAPL", "price": 100.0 + i, "qty": i * 10 + 1})
        .kdb_write(
            host=HOST, port=PORT, table=TABLE,
            columns=[("sym", "symbol"), ("price", "float"), ("qty", "long")],
        )
        .run(realtime=False)
)

# Read: time-sliced query; returns a stream of dicts.
# `start` and `duration` bound the replay window against the KDB time column.
rows = kdb_read(
    host=HOST, port=PORT,
    query=f"select from {TABLE}",
    time_col="time",
    chunk_size=10_000,
).collect()
rows.run(realtime=False, start=946684800.0, duration=86400.0)
print(rows.peek_value())
```

--------------------------------

### stream.run() - Historical Mode

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Runs a stream in historical mode using source timestamps instead of wall clock time. Supports optional start time, duration, or cycle limits for deterministic replay.

```APIDOC
## stream.run() - Historical Mode

### Description
Runs a stream in historical mode using source timestamps instead of wall clock time. Supports optional start time, duration, or cycle limits for deterministic replay.

### Method
run

### Parameters
- **realtime** (bool) - Required - Set to `False` to enable historical mode.
- **start** (datetime) - Optional - Epoch start for sources that require it (e.g., `kdb_read`).
- **duration** (float) - Optional - Cap the replay duration.
- **cycles** (int) - Optional - Cap the number of cycles.

### Request Example
```python
from datetime import datetime, timezone
from wingfoil import ticker

stream = ticker(0.01).count().collect()
stream.run(
    realtime=False,
    start=datetime(2025, 1, 1, tzinfo=timezone.utc),
    cycles=5,
)
print(stream.peek_value())   # [1, 2, 3, 4, 5]
```

### Response
#### Success Response
- **peek_value()** (list) - The collected values from the stream.

#### Response Example
```python
[1, 2, 3, 4, 5]
```
```

--------------------------------

### FIX TLS initiator with injector

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Connects to a FIX TLS endpoint (e.g. LMAX). Returns (data_stream, status_stream, injector). The injector sends outbound FIX messages on the session. Example sends a market data request (msg_type V) with fields (262, "req1"), (263, "1"), (264, "0").

```python
from wingfoil import fix_connect_tls

data, status, injector = fix_connect_tls(
    host="fix-marketdata.london-digital.lmax.com",
    port=443,
    sender_comp_id="USERNAME",
    target_comp_id="LMXBL",
    password="secret",
)

# Send a FIX message on the session:
injector.inject({
    "msg_type": "V",
    "fields": [(262, "req1"), (263, "1"), (264, "0")],
})

```

--------------------------------

### .run(realtime, start=, duration=, cycles=)

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Builds and runs a one-root graph.

```APIDOC
## .run(realtime, start=, duration=, cycles=)

### Description
Build and run a one-root graph.

### Parameters
- **realtime** (boolean) - Required - If True, tracks wall-clock time; if False, historical replay.
- **start** (number) - Optional - Start time for historical replay.
- **duration** (number) - Optional - Duration of the run.
- **cycles** (integer) - Optional - Number of cycles to run.

### Returns
- **None** - Runs the graph.

### Example
```python
stream.run(realtime=False, cycles=10)
```
```

--------------------------------

### Graph([...]).run(...)

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Builds and runs a multi-root graph.

```APIDOC
## Graph([...]).run(...)

### Description
Build and run a multi-root graph.

### Parameters
- **roots** (list of Stream/Node) - Required - The roots to run together.
- **realtime** (boolean) - Required - If True, tracks wall-clock time; if False, historical replay.
- **start** (number) - Optional - Start time for historical replay.
- **duration** (number) - Optional - Duration of the run.
- **cycles** (integer) - Optional - Number of cycles to run.

### Returns
- **None** - Runs the graph.

### Example
```python
from wingfoil import ticker, Graph
quotes = ticker(0.1).count().map(lambda i: 100 + i).logged("quote")
heartbeat = ticker(1.0).count().logged("heartbeat")
Graph([quotes, heartbeat]).run(realtime=True, duration=2.5)
```
```

--------------------------------

### WebServer Class

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Python-facing handle to a running WebServer. Mirrors the builder on the Rust side: construct with a bind address, optionally set the codec and static-file directory, then register publish / subscribe streams.

```APIDOC
## class WebServer(_addr_, _codec=None_, _static_dir=None_, _historical=False_)

Bases: `object`

### Methods

#### codec_name()

Return the codec currently used on the wire ("bincode" or "json").

#### port()

The bound port (0 when historical=True).

#### sub(_topic_)

Subscribe to topic. Returns a source Stream of dicts / lists / primitives decoded from the clients' frames. To _publish_ a stream, use the fluent stream.web_pub(server, topic) method on the stream itself.
```

--------------------------------

### iceoryx2 Publisher and Subscriber

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Zero-copy pub/sub over shared memory using iceoryx2.

```APIDOC
## iceoryx2 Publisher and Subscriber

### Description
Zero-copy pub/sub over shared memory using iceoryx2.

### Method
N/A (Python API)

### Endpoint
N/A

### Parameters
- **service** (string) - Required - The service name.
- **variant** (Iceoryx2ServiceVariant) - Optional - `Local` or `Ipc` (default: `Local`).
- **mode** (Iceoryx2Mode) - Optional - `Spin`, `Threaded`, or `Signaled` (default: `Signaled`).
- **history_size** (int) - Optional - History size for the subscription.
- **initial_max_slice_len** (int) - Optional - Initial maximum slice length for the publisher.

### Request Example
```python
from wingfoil import ticker, iceoryx2_sub, Iceoryx2ServiceVariant, Iceoryx2Mode, Graph

service = "wingfoil/demo"

sub = iceoryx2_sub(
    service,
    variant=Iceoryx2ServiceVariant.Local,   # or Ipc
    mode=Iceoryx2Mode.Signaled,             # Spin | Threaded | Signaled
)
sub = sub.inspect(lambda msgs: print("received:", msgs)).collect()

pub = (
    ticker(0.1).count()
        .map(lambda n: f"tick {n}".encode())
        .iceoryx2_pub(service, variant=Iceoryx2ServiceVariant.Local)
)

Graph([pub, sub]).run(realtime=True, duration=0.5)
```

### Response
No response; runs indefinitely.
```

--------------------------------

### Graph: run multiple roots together with ticker and logged

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Creates two independent streams (quotes and heartbeat) and runs them together in a Graph with realtime=True for 2.5 seconds. Demonstrates multi-root execution.

```python
from wingfoil import ticker, Graph

quotes = ticker(0.1).count().map(lambda i: 100 + i).logged("quote")
heartbeat = ticker(1.0).count().logged("heartbeat")

Graph([quotes, heartbeat]).run(realtime=True, duration=2.5)

```

--------------------------------

### .buffer(n)

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Creates a tumbling window of size n.

```APIDOC
## .buffer(n)

### Description
Tumbling window of size `n`.

### Parameters
- **n** (integer) - Required - The window size.

### Returns
- **Stream** - A new Stream emitting buffers of size n.

### Example
```python
stream.buffer(3)
```
```

--------------------------------

### I/O adapters

Source: https://wingfoil.readthedocs.io/en/latest/api.html

Adapters for reading from and writing to external systems.

```APIDOC
## I/O Adapters

### CSV
- `csv_read()`
- `Stream.csv_write(path)`

### KDB+
- `kdb_read()`
- `kdb_write()`
- `Stream.kdb_write(...)`

### etcd
- `etcd_sub()`
- `Stream.etcd_pub(endpoint, lease_ttl=None, force=True)`

### ZeroMQ
- `zmq_sub()`
- `zmq_sub_etcd()`
- `Stream.zmq_pub(port)`
- `Stream.zmq_pub_etcd(name, port, endpoint)`
- `Stream.zmq_pub_etcd_on(name, address, port, endpoint)`

### iceoryx2 (feature-gated)
- `iceoryx2_sub()`
- `Iceoryx2ServiceVariant`
- `Iceoryx2Mode`
- `Stream.iceoryx2_pub(service_name, variant, history_size, initial_max_slice_len)`

### FIX 4.4
- `fix_connect()`
- `fix_connect_tls()`
- `fix_accept()`

### Prometheus
- `PrometheusExporter`

### OpenTelemetry OTLP
- `Stream.otlp_push(metric_name, endpoint, service_name)`

```

--------------------------------

### ZeroMQ etcd discovery publisher and subscriber

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Publishers register under a service name; subscribers look it up. Requires a running etcd. The publisher uses zmq_pub_etcd with service name "quotes", port 7779, and etcd endpoint http://127.0.0.1:2379. The subscriber uses zmq_sub_etcd with the same service name and endpoint.

```python
# publisher
wf.ticker(0.5).count().map(lambda n: str(n).encode()) \
    .zmq_pub_etcd("quotes", 7779, "http://127.0.0.1:2379") \
    .run(realtime=True)

# subscriber
data, status = wf.zmq_sub_etcd("quotes", "http://127.0.0.1:2379")

```

--------------------------------

### ZeroMQ Publisher with etcd Discovery (Multi-host)

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Publishes a stream to a ZeroMQ socket and registers with etcd, specifying a routable address for multi-host deployments.

```APIDOC
## ZeroMQ Publisher with etcd Discovery (Multi-host)

### Description
Publishes a stream to a ZeroMQ socket and registers with etcd, specifying a routable address for multi-host deployments.

### Method
N/A (Python API)

### Endpoint
N/A

### Parameters
- **name** (string) - Required - The service name to register in etcd.
- **address** (string) - Required - The routable address to advertise.
- **port** (int) - Required - The port to bind the ZeroMQ socket.
- **endpoint** (string) - Required - The etcd endpoint URL.

### Request Example
```python
import wingfoil as wf

wf.ticker(0.5).count().map(lambda n: str(n).encode()) \
    .zmq_pub_etcd_on("quotes", "192.168.1.10", 7779, "http://127.0.0.1:2379") \
    .run(realtime=True)
```

### Response
No response; runs indefinitely.
```

--------------------------------

### Sources: ticker(), constant()

Source: https://wingfoil.readthedocs.io/en/latest/api.html

Functions to create data sources. ticker() likely generates a ticker stream, constant() creates a constant value stream.

```APIDOC
## Sources

### ticker()

### constant()

```

--------------------------------

### Pipeline with most operators: count, filter, map, average, logged, run, peek_value

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Builds a stream from ticker(0.1), counts ticks, filters odd counts, maps to float, computes running average, logs each value, runs in historical mode for 10 cycles, and prints the last value via peek_value().

```python
from wingfoil import ticker

avg_of_odds = (
    ticker(0.1)
        .count()
        .filter(lambda x: x % 2 == 1)   # 1, 3, 5, ...
        .map(float)
        .average()                      # running mean
        .logged("avg")
)

avg_of_odds.run(realtime=False, cycles=10)
print("last:", avg_of_odds.peek_value())

```

--------------------------------

### py_zmq_sub_etcd(name, endpoint)

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Subscribes to a named ZMQ publisher via etcd service discovery. Looks up name in etcd and connects to the publisher at the stored address.

```APIDOC
## py_zmq_sub_etcd(name, endpoint)

### Description
Subscribe to a named ZMQ publisher via etcd service discovery.
Looks up name in etcd and connects to the publisher at the stored address. The lookup happens at call time — ensure the publisher is registered before calling this function.

### Parameters
- **name** (string) - Required - Publisher name / etcd key (e.g. "quotes")
- **endpoint** (string) - Required - etcd endpoint (e.g. "http://localhost:2379")

### Returns
Tuple of (data_stream, status_stream).

### Raises
- **RuntimeError** - if the key is absent or etcd is unreachable
```

--------------------------------

### FIX Initiator

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Connects to a FIX 4.4 server as an initiator and returns data and status streams.

```APIDOC
## FIX Initiator

### Description
Connects to a FIX 4.4 server as an initiator and returns data and status streams.

### Method
N/A (Python API)

### Endpoint
N/A

### Parameters
- **host** (string) - Required - The FIX server hostname.
- **port** (int) - Required - The FIX server port.
- **sender_comp_id** (string) - Required - The sender comp ID.
- **target_comp_id** (string) - Required - The target comp ID.

### Returns
- **data_stream** - Stream of `list[dict]` where each dict is `{"msg_type": str, "seq_num": int, "fields": [(tag, value), ...]}`.
- **status_stream** - Stream of `"disconnected" | "logging_in" | "logged_in"` or a dict `{"status": "logged_out"|"error", "reason"|"message": str}`.

### Request Example
```python
from wingfoil import fix_connect

data, status = fix_connect(
    host="fix.example.com",
    port=9876,
    sender_comp_id="MYCOMP",
    target_comp_id="BROKER",
)

messages = data.inspect(lambda msgs: [print("fix:", m) for m in msgs])
states = status.inspect(lambda ss: [print("session:", s) for s in ss])

import wingfoil as wf
wf.Graph([messages, states]).run(realtime=True, duration=10.0)
```

### Response
No response; runs indefinitely.
```

--------------------------------

### Pandas helpers: to_dataframe(), build_dataframe()

Source: https://wingfoil.readthedocs.io/en/latest/api.html

Helper functions to convert streams to pandas DataFrames.

```APIDOC
## Pandas Helpers

### to_dataframe()

### build_dataframe()

```

--------------------------------

### FIX 4.4 initiator

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Connects to a FIX 4.4 acceptor. Returns (data_stream, status_stream). Each data tick yields a list[dict] where every dict is {"msg_type": str, "seq_num": int, "fields": [(tag, value), ...]}. Status values are "disconnected" | "logging_in" | "logged_in" or a dict {"status": "logged_out"|"error", "reason"|"message": str}.

```python
from wingfoil import fix_connect

data, status = fix_connect(
    host="fix.example.com",
    port=9876,
    sender_comp_id="MYCOMP",
    target_comp_id="BROKER",
)

messages = data.inspect(lambda msgs: [print("fix:", m) for m in msgs])
states = status.inspect(lambda ss: [print("session:", s) for s in ss])

import wingfoil as wf
wf.Graph([messages, states]).run(realtime=True, duration=10.0)

```

--------------------------------

### bimap: fuse two streams with sample and constant

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Combines two streams (a count stream and a constant sampled by a ticker) using bimap to add their values, logs the sum, and runs in historical mode for 5 cycles.

```python
from wingfoil import ticker, constant, bimap

a = ticker(0.1).count()                       # 1, 2, 3, ...
b = constant(0.5).sample(ticker(0.1))         # 0.5 on every tick

(bimap(a, b, lambda x, y: x + y)
    .logged("sum")
    .run(realtime=False, cycles=5))

```

--------------------------------

### .sample(trigger)

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Re-emits the current value on each trigger tick.

```APIDOC
## .sample(trigger)

### Description
Re-emit the current value on each `trigger` tick.

### Parameters
- **trigger** (Stream) - Required - The stream that triggers sampling.

### Returns
- **Stream** - A new Stream that emits the latest value on each trigger tick.

### Example
```python
stream.sample(ticker(0.1))
```
```

--------------------------------

### py_kdb_read(host, port, query, time_col, chunk_size=3600)

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Reads data from a KDB+ database. Returns a PyStream where each tick is a dict with column names as keys.

```APIDOC
## py_kdb_read(host, port, query, time_col, chunk_size=3600)

### Description
Read data from a KDB+ database.
Returns a PyStream where each tick is a dict with column names as keys.

### Parameters
- **host** (string) - Required - KDB+ server hostname
- **port** (integer) - Required - KDB+ server port
- **query** (string) - Required - q query to execute (e.g. "select from trade")
- **time_col** (string) - Required - Name of the time column for time-slice filtering
- **chunk_size** (integer) - Optional - Duration of each time slice in seconds (default: 3600)

### Returns
Stream where each tick yields a dict of {column_name: value}.

### Notes
Requires RunMode::HistoricalFrom with a non-zero start time and RunFor::Duration.
```

--------------------------------

### py_fix_connect(host, port, sender_comp_id, target_comp_id)

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Connects to a FIX acceptor and returns a tuple of (data_stream, status_stream). The data_stream yields lists of message dicts and the status_stream yields session status strings.

```APIDOC
## py_fix_connect(host, port, sender_comp_id, target_comp_id)

### Description
Connect to a FIX acceptor and return (data_stream, status_stream).

### Parameters
- **host** (string) - Required - FIX acceptor hostname
- **port** (integer) - Required - FIX acceptor port
- **sender_comp_id** (string) - Required - SenderCompID for this session
- **target_comp_id** (string) - Required - TargetCompID of the counterparty

### Returns
A tuple of (data_stream, status_stream) where data_stream yields lists of message dicts and status_stream yields session status strings.
```

--------------------------------

### ZeroMQ Publisher with etcd Discovery

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Publishes a stream to a ZeroMQ socket and registers the service with etcd for discovery.

```APIDOC
## ZeroMQ Publisher with etcd Discovery

### Description
Publishes a stream to a ZeroMQ socket and registers the service with etcd for discovery.

### Method
N/A (Python API)

### Endpoint
N/A

### Parameters
- **service_name** (string) - Required - The service name to register in etcd.
- **port** (int) - Required - The port to bind the ZeroMQ socket.
- **etcd_endpoint** (string) - Required - The etcd endpoint URL.

### Request Example
```python
import wingfoil as wf

wf.ticker(0.5).count().map(lambda n: str(n).encode()) \
    .zmq_pub_etcd("quotes", 7779, "http://127.0.0.1:2379") \
    .run(realtime=True)
```

### Response
No response; runs indefinitely.
```

--------------------------------

### ZeroMQ Subscriber with etcd Discovery

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Subscribes to a ZeroMQ publisher discovered via etcd.

```APIDOC
## ZeroMQ Subscriber with etcd Discovery

### Description
Subscribes to a ZeroMQ publisher discovered via etcd.

### Method
N/A (Python API)

### Endpoint
N/A

### Parameters
- **service_name** (string) - Required - The service name to look up in etcd.
- **etcd_endpoint** (string) - Required - The etcd endpoint URL.

### Returns
- **data_stream** - Stream of `list[bytes]` messages received each cycle.
- **status_stream** - Stream of `"connected"` or `"disconnected"`.

### Request Example
```python
import wingfoil as wf

data, status = wf.zmq_sub_etcd("quotes", "http://127.0.0.1:2379")
```

### Response
No response; runs indefinitely.
```

--------------------------------

### ZeroMQ direct mode subscriber (zmq_sub.py)

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Subscribes to a ZeroMQ publisher at tcp://127.0.0.1:7779. Returns (data_stream, status_stream); data_stream yields list[bytes] of messages received that cycle, status_stream yields "connected" / "disconnected".

```python
# zmq_sub.py
import wingfoil as wf

data, status = wf.zmq_sub("tcp://127.0.0.1:7779")
data_node = data.inspect(lambda msgs: [print("msg:", m) for m in msgs])
status_node = status.inspect(lambda s: print("status:", s))
wf.Graph([data_node, status_node]).run(realtime=True)

```

--------------------------------

### ZeroMQ direct mode publisher (zmq_pub.py)

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Publishes a stream of encoded tick counts to a ZeroMQ socket on port 7779. Uses hard-coded address, no discovery infrastructure. The publisher runs in realtime mode.

```python
# zmq_pub.py
import wingfoil as wf

(
    wf.ticker(0.5).count()
        .map(lambda n: str(n).encode())
        .zmq_pub(port=7779)
        .run(realtime=True)
)

```

--------------------------------

### ZeroMQ Subscriber (Direct Mode)

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Subscribes to a ZeroMQ publisher and returns two streams: data and status.

```APIDOC
## ZeroMQ Subscriber (Direct Mode)

### Description
Subscribes to a ZeroMQ publisher and returns two streams: data and status.

### Method
N/A (Python API)

### Endpoint
N/A

### Parameters
- **address** (string) - Required - The ZeroMQ address to connect to (e.g., "tcp://127.0.0.1:7779").

### Returns
- **data_stream** - Stream of `list[bytes]` messages received each cycle.
- **status_stream** - Stream of `"connected"` or `"disconnected"`.

### Request Example
```python
import wingfoil as wf

data, status = wf.zmq_sub("tcp://127.0.0.1:7779")
data_node = data.inspect(lambda msgs: [print("msg:", m) for m in msgs])
status_node = status.inspect(lambda s: print("status:", s))
wf.Graph([data_node, status_node]).run(realtime=True)
```

### Response
No response; runs indefinitely.
```

--------------------------------

### Module Functions

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Utility functions for stream creation and data handling.

```APIDOC
## Module Functions

### bimap(_a_, _b_, _func_)

Maps streams a and b into a new stream using func (e.g lambda a, b: a + b).

### build_dataframe(_stream_dict: Dict[str, Any]_) → DataFrame

Builds a combined DataFrame from a dict of streams.

### constant(_val_)

A stream that ticks once, on first engine cycle.

### py_csv_read(_path_, _time_column_)

Read a CSV file into a stream of dicts. Each tick yields a dict where keys are column headers and values are strings. Rows sharing the same timestamp are collapsed to one-per-tick.

**Parameters:**
- **path** – Path to the CSV file (headers required).
- **time_column** – Name of the column containing the timestamp (parsed as integer nanoseconds since epoch).

### py_etcd_sub(_endpoint_, _prefix_)

Subscribe to etcd keys matching a prefix. Emits a snapshot of all existing keys under the prefix, then delivers live watch events as they arrive. Each tick yields a list of event dicts: `{"kind": "put" | "delete", "key": str, "value": bytes, "revision": int}`.

**Parameters:**
- **endpoint** – etcd endpoint, e.g. "http://localhost:2379".
- **prefix** – key prefix to watch, e.g. "/myapp/".

### py_fix_accept(_port_, _sender_comp_id_, _target_comp_id_)

Bind a FIX acceptor and return (data_stream, status_stream).
```

--------------------------------

### py_zmq_sub(address)

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Subscribes to a ZMQ PUB socket. Returns a (data_stream, status_stream) tuple.

```APIDOC
## py_zmq_sub(address)

### Description
Subscribe to a ZMQ PUB socket.
Returns a (data_stream, status_stream) tuple.
Each tick of data_stream yields a list[bytes] containing all messages received in that cycle. Each tick of status_stream yields the string "connected" or "disconnected".

### Parameters
- **address** (string) - Required - ZMQ endpoint to connect to (e.g. "tcp://localhost:5556")

### Returns
Tuple of (data_stream, status_stream).
```

--------------------------------

### FIX TLS Initiator

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Connects to a FIX 4.4 server over TLS and returns data, status, and an injector for sending messages.

```APIDOC
## FIX TLS Initiator

### Description
Connects to a FIX 4.4 server over TLS and returns data, status, and an injector for sending messages.

### Method
N/A (Python API)

### Endpoint
N/A

### Parameters
- **host** (string) - Required - The FIX server hostname.
- **port** (int) - Required - The FIX server port.
- **sender_comp_id** (string) - Required - The sender comp ID.
- **target_comp_id** (string) - Required - The target comp ID.
- **password** (string) - Required - The password for authentication.

### Returns
- **data_stream** - Stream of `list[dict]` where each dict is `{"msg_type": str, "seq_num": int, "fields": [(tag, value), ...]}`.
- **status_stream** - Stream of `"disconnected" | "logging_in" | "logged_in"` or a dict `{"status": "logged_out"|"error", "reason"|"message": str}`.
- **injector** - Object with an `inject` method to send FIX messages.

### Request Example
```python
from wingfoil import fix_connect_tls

data, status, injector = fix_connect_tls(
    host="fix-marketdata.london-digital.lmax.com",
    port=443,
    sender_comp_id="USERNAME",
    target_comp_id="LMXBL",
    password="secret",
)

# Send a FIX message on the session:
injector.inject({
    "msg_type": "V",
    "fields": [(262, "req1"), (263, "1"), (264, "0")],
})
```

### Response
No response; runs indefinitely.
```

--------------------------------

### .logged("label")

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Logs each value at INFO level and passes it through.

```APIDOC
## .logged("label")

### Description
`INFO`-log each value and pass it through.

### Parameters
- **label** (string) - Required - The log label.

### Returns
- **Stream** - A new Stream that logs values and passes them through.

### Example
```python
stream.logged("my_stream")
```
```

--------------------------------

### etcd_pub() - Publish to etcd

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Publishes a stream of dicts to etcd. Each dict must have `key` (str) and `value` (bytes), or a list of such dicts per tick.

```APIDOC
## etcd_pub() - Publish to etcd

### Description
Publishes a stream of dicts to etcd. Each dict must have `key` (str) and `value` (bytes), or a list of such dicts per tick.

### Method
etcd_pub

### Parameters
- **endpoint** (str) - Required - etcd endpoint URL.
- **lease_ttl** (float) - Optional - Lease TTL in seconds.
- **force** (bool) - Optional - Force write.

### Request Example
```python
from wingfoil import ticker, etcd_sub

ENDPOINT = "http://localhost:2379"

(
    ticker(1.0).count().limit(3)
        .map(lambda i: {"key": f"/wf/item/{i}", "value": str(i).encode()})
        .etcd_pub(ENDPOINT, lease_ttl=30.0, force=True)
        .run(realtime=True)
)
```

### Response
#### Success Response
- **Node** - The publish operation returns a Node that can be run.

#### Response Example
No explicit response example provided.
```

--------------------------------

### .sum()

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Computes the running sum of numeric values.

```APIDOC
## .sum()

### Description
Running sum (values must be numeric).

### Returns
- **Stream** - A new Stream emitting the running sum.

### Example
```python
stream.sum()
```
```

--------------------------------

### Stream.etcd_pub

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Publishes this stream of dicts to etcd via PUT.

```APIDOC
## Stream.etcd_pub

### Description
Publishes this stream of dicts to etcd via PUT. Stream values must be dicts with "key" (str) and "value" (bytes), or lists of such dicts for multiple writes per tick.

### Method
etcd_pub

### Signature
etcd_pub(endpoint, lease_ttl=None, force=True)

### Parameters
- **endpoint** (str) - Required - etcd endpoint, e.g. "http://localhost:2379".
- **lease_ttl** (int, optional) - Optional - Lease TTL in seconds; keys expire after this duration and vanish immediately on clean shutdown. Pass None for persistent keys (default).
- **force** (bool) - Optional - If True (default), silently overwrite existing keys. If False, fail if any key already exists.

### Returns
Node - A Node that drives the write operation.
```

--------------------------------

### .count()

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

Emits the tick count: 1, 2, 3, ...

```APIDOC
## .count()

### Description
Emit tick count: 1, 2, 3, …

### Returns
- **Stream** - A new Stream emitting the running count.

### Example
```python
stream.count()
```
```

--------------------------------

### .peek_value()

Source: https://wingfoil.readthedocs.io/en/latest/readme.html

After run(), inspects the last emitted value.

```APIDOC
## .peek_value()

### Description
After `run()`, inspect the last emitted value.

### Returns
- **any** - The last value emitted by the stream.

### Example
```python
stream.run(realtime=False, cycles=10)
last = stream.peek_value()
```
```

--------------------------------

### Composition: bimap(), Graph, CustomStream

Source: https://wingfoil.readthedocs.io/en/latest/api.html

Composition utilities for building stream processing graphs.

```APIDOC
## Composition

### bimap()

### Graph

### CustomStream

```

--------------------------------

### Stream.inspect

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Inspects each element of the stream.

```APIDOC
## Stream.inspect

### Description
Inspects each element of the stream.

### Method
inspect

### Signature
inspect(func)

### Parameters
- **func** (callable) - Required - Function to inspect each element.

### Returns
Stream
```

--------------------------------

### Stream.otlp_push

Source: https://wingfoil.readthedocs.io/en/latest/api_generated/wingfoil.wingfoil.html

Pushes this stream as an OTLP gauge metric.

```APIDOC
## Stream.otlp_push

### Description
Pushes this stream as an OTLP gauge metric.

### Method
otlp_push

### Signature
otlp_push(metric_name, endpoint, service_name)

### Parameters
- **metric_name** (str) - Required - Name of the metric to report.
- **endpoint** (str) - Required - OTLP HTTP endpoint, e.g. "http://localhost:4318".
- **service_name** (str) - Required - Service name reported in OTLP resource attributes.

### Returns
Stream
```
Tokens:
Raw