ZenQuant.Recorder.Replay (zen_quant v0.2.0)

Filtered, unwrapped replay of JSONL-recorded market data.

Builds on ZenQuant.Recorder.JSONL.stream/1 to provide:

Unwrapping — skips errors, yields plain entry maps
Filtering — by type and/or time range
Temporal delays — speed-adjusted Process.sleep/1 for realistic replay

Usage

# Lazy stream of filtered entries
Replay.stream("/tmp/session.jsonl", type: :ticker, from: ~U[2026-02-23 12:00:00Z])
|> Enum.each(&process/1)

# Callback-based replay with realtime delays
Replay.run("/tmp/session.jsonl", &process/1, speed: 1.0, type: :ticker)

# Accumulate a result
Replay.reduce("/tmp/session.jsonl", [], fn entry, acc -> [entry | acc] end, type: :greeks)

API Functions

Function	Arity	Description	Param Kinds
`reduce`	4	Replay entries through a reducer function, accumulating a result.	`path: value`, `initial: value`, `reducer: value`
`run`	3	Replay entries through a callback function, returning the count processed.	`path: value`, `callback: value`
`stream`	2	Stream filtered, unwrapped entries from a JSONL file.	`path: value`

Summary

Functions

reduce(path, initial, reducer, opts \\ [])

Replays entries through a reducer function, accumulating a result.

run(path, callback, opts \\ [])

Replays entries through a callback function, returning the count.

stream(path, opts \\ [])

Returns a lazy stream of entry maps from a JSONL file.

Functions

reduce(path, initial, reducer, opts \\ [])

@spec reduce(Path.t(), acc, (map(), acc -> acc), keyword()) ::
  {:ok, acc} | {:error, term()}
when acc: term()

Replays entries through a reducer function, accumulating a result.

Returns {:ok, final_acc} on success or {:error, reason} if the file is missing/unreadable. Exceptions from the reducer propagate directly.

Options

Same as run/3 (including speed:).

Examples

{:ok, prices} = Replay.reduce("/tmp/session.jsonl", [], fn entry, acc ->
  [entry.data["last"] | acc]
end, type: :ticker)

run(path, callback, opts \\ [])

@spec run(Path.t(), (map() -> any()), keyword()) ::
  {:ok, non_neg_integer()} | {:error, term()}

Replays entries through a callback function, returning the count.

Calls callback.(entry) for each matching entry. Returns {:ok, count} on success or {:error, reason} if the file is missing/unreadable.

Options

Same as stream/2, plus:

speed: — replay speed multiplier. nil (default) = instant, 1.0 = realtime, 2.0 = 2x faster, 0.5 = 2x slower. Must be positive; raises ArgumentError otherwise.

Examples

{:ok, count} = Replay.run("/tmp/session.jsonl", &process/1, speed: 1.0)

stream(path, opts \\ [])

@spec stream(
  Path.t(),
  keyword()
) :: Enumerable.t()

Returns a lazy stream of entry maps from a JSONL file.

Malformed lines are skipped silently. Each element is a plain map (not an {:ok, entry} tuple).

Raises File.Error if the file doesn't exist (same as JSONL.stream/1).

Options

type: — filter by entry type (atom or string). Atoms are auto-converted.
from: — DateTime.t(), inclusive start bound.
to: — DateTime.t(), inclusive end bound.

Examples

Replay.stream("/tmp/session.jsonl", type: :ticker)
|> Enum.take(10)