diff --git a/CMakeLists.txt b/CMakeLists.txt
index 36503e5..57209dc 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -85,6 +85,7 @@ add_library(pineforge STATIC
     src/engine_risk.cpp
     src/engine_run.cpp
     src/engine_security.cpp
+    src/engine_stream.cpp
     src/engine_strategy_commands.cpp
     src/engine_trade_accessors.cpp
     src/magnifier.cpp
diff --git a/README.md b/README.md
index 91138ec..9d50b83 100644
--- a/README.md
+++ b/README.md
@@ -120,7 +120,7 @@ for the full tool catalog, request schemas, and env vars (`PINEFORGE_ALLOW_ANYWH
 
 - 🎯 **TradingView-exact.** 251 of 252 reference strategies match TV trade-for-trade. The lone outlier is a stress probe at the 1× margin boundary where TV's broker emulator is non-deterministic — engine is correct. **100 of 100** PineForge excellent vs PyneCore + PineTS on the public three-way benchmark (~167,000 TV trades; PyneCore: 85 of 100; PineTS indicator-only).
 - ⚡ **Microsecond-class.** Median **162× faster than PyneCore** across 99 commonly-timed strategies (full 41,307-bar OHLCV, magnifier-on hot loop; see [benchmarks/results/speed.md](benchmarks/results/speed.md)). Parameter sweeps load one `.so` and re-run with new inputs — no recompile, no fork, no IPC.
-- 🔒 **Stable C ABI.** 10 functions, 6 POD types, one header (`<pineforge/pineforge.h>`). Append-only across minor versions, `static_assert`-pinned struct layouts, hidden-visibility hygiene. Drop a strategy `.so` in any harness; it just runs.
+- 🔒 **Stable C ABI.** 26 functions, one header (`<pineforge/pineforge.h>`). Append-only across minor versions, `static_assert`-pinned struct layouts, hidden-visibility hygiene. Drop a strategy `.so` in any harness; it just runs.
 - 🧪 **Reproducible to the bit.** Deterministic float ordering, deterministic bar magnifier, no internal RNG seeded from time. Two runs with the same inputs produce bit-identical trade lists.
 - 🧰 **FFI-friendly.** Call from Python (`ctypes`), Rust (`libloading`), Go (`cgo`), Node, Julia. Worked examples for [pure C](https://cdocs.pineforge.dev/examples_c.html), [Python sweep](https://cdocs.pineforge.dev/examples_python_sweep.html), [Rust](https://cdocs.pineforge.dev/examples_rust.html), [multi-strategy harness](https://cdocs.pineforge.dev/examples_multi.html), and [magnifier A/B](https://cdocs.pineforge.dev/examples_magnifier.html) ship in the docs.
 - 🌍 **Cross-platform CI.** Linux + macOS × Release + Debug. Universal mac binary. Static library, no runtime DSO surprises at deploy time.
@@ -129,7 +129,7 @@ for the full tool catalog, request schemas, and env vars (`PINEFORGE_ALLOW_ANYWH
 
 ## For developers: embed the runtime directly
 
-PineForge ships as a static C library (`libpineforge.a`) with a stable 10-symbol C ABI. Call from C, Python, Rust, Go, Node, Julia — one harness, swap strategies forever.
+PineForge ships as a static C library (`libpineforge.a`) with a stable 26-symbol C ABI. Call from C, Python, Rust, Go, Node, Julia — one harness, swap strategies forever.
 
 ### See it in 30 seconds
 
@@ -151,13 +151,15 @@ int main(void) {
 }
 ```
 
-That's the entire integration. Every PineForge-compiled strategy `.so` exports the same 10 symbols — write your harness once, swap strategies forever.
+That's the entire batch integration. Every PineForge-compiled strategy `.so`
+exports the same stable ABI — write your harness once, swap strategies forever.
 
 ```bash
 cmake -B build -DCMAKE_BUILD_TYPE=Release
 cmake --build build -j
 ctest --test-dir build --output-on-failure   # 39 tests, ~1 s
 bash tutorial/run.sh                          # MACD backtest end-to-end
+python3 tutorial/run_stream.py                # OHLCV warmup → realtime trades
 ```
 
 ## Documentation
@@ -167,6 +169,7 @@ bash tutorial/run.sh                          # MACD backtest end-to-end
 | 📖 **[cdocs.pineforge.dev](https://cdocs.pineforge.dev)** | Full C ABI reference, lifecycle, report schema, configuration knobs, magnifier, FFI bindings, ABI stability contract |
 | 🚀 **[Getting Started](https://cdocs.pineforge.dev/getting_started.html)** | 60-second build + install + smoke test |
 | 🧪 **[Tutorial: MACD on BTC/USDT](https://cdocs.pineforge.dev/tutorial_macd.html)** | End-to-end annotated walkthrough |
+| 📡 **[Historical → realtime streaming](https://cdocs.pineforge.dev/streaming.html)** | Preserve strategy and broker state while switching from OHLCV to ordered trades |
 | 📊 **[Trading metrics reference](https://cdocs.pineforge.dev/metrics.html)** | Every report metric (ABI v2): definitions, units, NaN rules, TV / quant-library validation status |
 | 🔌 **[FFI from Python](https://cdocs.pineforge.dev/ffi_python.html)** | Complete `ctypes` mirror — paste-ready |
 | 🦀 **[Calling from Rust](https://cdocs.pineforge.dev/examples_rust.html)** | Idiomatic `libloading` wrapper |
@@ -319,11 +322,26 @@ int main(void) {
 | `strategy_set_override`                  | Override a `strategy(...)` declaration param |
 | `strategy_set_magnifier_volume_weighted` | Toggle volume-weighted magnifier             |
 | `strategy_set_trace_enabled`             | Toggle per-bar trace recording               |
+| `strategy_set_trade_start_time`          | Suppress historical order placement          |
+| `strategy_stream_begin`                  | Warm on OHLCV, then enter realtime mode       |
+| `strategy_stream_push_tick`              | Push one normalized ordered trade             |
+| `strategy_stream_push_ticks`             | Push a raw-trade batch                        |
+| `strategy_stream_advance_time`           | Confirm bars / materialize quiet intervals    |
+| `strategy_stream_end`                    | End a realtime stream                         |
+| `strategy_stream_fill_report`            | Snapshot cumulative warmup + realtime state   |
+| `strategy_set_chart_timezone`            | Set Pine chart wall-clock timezone            |
+| `strategy_set_syminfo_timezone`          | Set symbol/exchange timezone                  |
+| `strategy_set_syminfo_session`           | Set symbol trading session                    |
+| `strategy_set_syminfo_mintick`           | Set symbol tick size                          |
+| `strategy_set_syminfo_pointvalue`        | Set symbol point value                        |
+| `strategy_set_syminfo_metadata`          | Inject numeric symbol metadata                |
+| `strategy_get_last_error`                | Read the latest runtime error                 |
 | `pf_version_get`                         | Runtime version                              |
 | `pf_abi_version`                         | Struct-layout version (`PF_ABI_VERSION`)     |
+| `pf_version_string`                      | Full runtime version string                   |
 
 
-POD types (`pf_bar_t`, `pf_trade_t`, `pf_report_t`, `pf_security_diag_t`, `pf_trace_entry_t`, `pf_version_t`, and — since ABI v2 — `pf_trade_stats_t`, `pf_equity_stats_t`, `pf_metrics_t`, `pf_equity_point_t`) and the `pf_magnifier_distribution_t` enum complete the surface. ABI v2 (`PF_ABI_VERSION == 2`, exported as `pf_abi_version()`) appends computed trading metrics and a per-script-bar equity curve to `pf_report_t`; callers must verify the version before running since the report struct is caller-allocated.
+POD types (`pf_bar_t`, `pf_trade_tick_t`, `pf_trade_t`, `pf_report_t`, `pf_security_diag_t`, `pf_trace_entry_t`, `pf_version_t`, and — since ABI v2 — `pf_trade_stats_t`, `pf_equity_stats_t`, `pf_metrics_t`, `pf_equity_point_t`) and the `pf_magnifier_distribution_t` enum complete the surface. ABI v2 (`PF_ABI_VERSION == 2`, exported as `pf_abi_version()`) appends computed trading metrics and a per-script-bar equity curve to `pf_report_t`; callers must verify the version before running since the report struct is caller-allocated.
 
 **Stability guarantee:** within the same `PINEFORGE_VERSION_MAJOR`, struct layouts and `extern "C"` signatures are append-only. New fields may be appended; existing fields are never reordered, removed, or retyped. New functions may be added; existing functions are never removed or signature-changed. Compile-time `static_assert`s in `src/c_abi.cpp` pin the layouts against drift.
 
@@ -367,6 +385,7 @@ benchmarks/             - three-way comparison harness vs PyneCore + PineTS
   └── run_all.sh        - one-shot reproducer: cmake build + run all 3 engines + diff (zero API keys)
 scripts/                - reproducibility tooling
   ├── run_strategy.py   - load any strategy.so via ctypes, write engine_trades.csv
+  ├── run_stream_corpus.py - slice OHLCV + raw trades and test live handoffs
   ├── run_corpus.sh     - one-shot: build all 252 .so + run + verify
   └── verify_corpus.py  - diff each engine_trades.csv against its tv_trades.csv
 cmake/                  - PineForgeConfig.cmake.in for downstream find_package()
@@ -376,7 +395,7 @@ cmake/smoke_consumer/   - Minimal find_package(PineForge) CI smoke project
 
 ## Visibility hygiene
 
-Every compiled strategy `.so` that statically links `libpineforge.a` exports **exactly the 10 documented C ABI symbols** and zero internal C++ symbols. This is enforced at the library level:
+Every compiled strategy `.so` that statically links `libpineforge.a` exports **exactly the 26 documented C ABI symbols** and zero internal C++ symbols. This is enforced at the library level:
 
 - `libpineforge.a` is built with `-fvisibility=hidden -fvisibility-inlines-hidden`
 - Public symbols are tagged `PF_API` (visibility=default)
@@ -478,4 +497,3 @@ See [CONTRIBUTING.md](CONTRIBUTING.md) (includes the **Apache-2.0** contribution
 1. Every contribution must keep the parity test green.
 2. Public-API changes (anything exported from `<pineforge/pineforge.h>`) require a major-version bump.
 3. Internal C++ helpers can change freely as long as the ABI surface stays put.
-
diff --git a/docs/Doxyfile b/docs/Doxyfile
index 128493e..9d65d03 100644
--- a/docs/Doxyfile
+++ b/docs/Doxyfile
@@ -90,6 +90,7 @@ INPUT                  = ../include/pineforge/pineforge.h \
                          pages/install.md \
                          pages/integration-cmake.md \
                          pages/lifecycle.md \
+                         pages/streaming.md \
                          pages/report-schema.md \
                          pages/metrics.md \
                          pages/configuration.md \
diff --git a/docs/README.md b/docs/README.md
index 75311cf..65036a0 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -17,6 +17,7 @@ docs/
 │   ├── install.md
 │   ├── integration-cmake.md
 │   ├── lifecycle.md
+│   ├── streaming.md
 │   ├── report-schema.md
 │   ├── configuration.md
 │   ├── magnifier.md
diff --git a/docs/cheatsheet-runtime-and-execution.md b/docs/cheatsheet-runtime-and-execution.md
index a350887..38f1acb 100644
--- a/docs/cheatsheet-runtime-and-execution.md
+++ b/docs/cheatsheet-runtime-and-execution.md
@@ -409,15 +409,14 @@ Audited gaps a forward/real-time executor must know (beyond per-order fills).
   are C++-accessor-only. Derive equity/profit-factor/Sharpe from the trades.
 - `pf_version_get()` / `pf_version_string()` for ABI gating.
 
-## 3.10 Multi-run state (reused handle)
-
-- **Reset per run:** `max_equity_`/`min_equity_`→`initial_capital_`, security
-  feed counters, aggregator state.
-- **NOT reset:** `trades_`, `net_profit_sum_`, win/loss/even counts,
-  `cons_loss_day_count_`, `risk_halted_`, `intraday_pnl_`, position/pyramid
-  state, `pending_orders_`, `trail_best_price_`, `consumed_partial_exit_ids_`,
-  trace buffer. Walk-forward on one handle **accumulates** — recreate the handle
-  per independent run.
+## 3.10 Multi-run state and streams
+
+- **One-shot `run_backtest*`:** resets broker, position, trades, equity,
+  series/TA, risk, trace, security, and aggregator state before each run.
+  Input/strategy/syminfo configuration remains attached to the handle.
+- **`strategy_stream_*`:** performs one historical warmup and deliberately
+  preserves that exact state while ordered trades arrive. Use this lifecycle,
+  not repeated one-shot calls, for continuous sessions.
 
 ## 3.11 Harness foot-gun
 
diff --git a/docs/coverage.md b/docs/coverage.md
index f7046bb..b954a94 100644
--- a/docs/coverage.md
+++ b/docs/coverage.md
@@ -24,16 +24,16 @@
 > "no runtime module — Pine surface still supported" bucket below calls
 > it out.
 >
-> **Out of scope.** Visual / charting / alert APIs are not implemented
-> by this runtime regardless of consumer (PineForge is an offline
-> backtesting engine, not a renderer).
+> **Out of scope today.** Visual / charting / alert APIs are not implemented
+> by this runtime regardless of consumer. The runtime now accepts continuous
+> ordered-trade streams, but it does not yet emit alert events or render charts.
 
 ## Coverage summary
 
 
 | Category                    | Runtime status                                                             | What `libpineforge.a` owns                                                                                                                                                                                                                                                                                                                                          |
 | --------------------------- | -------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| Engine / strategy lifecycle | Supported                                                                  | `BacktestEngine`, three `run(...)` overloads, bar loop, `on_bar(...)` hook, `ReportC` / `SecurityDiagC` reporting.                                                                                                                                                                                                                                                  |
+| Engine / strategy lifecycle | Supported                                                                  | `BacktestEngine`, one-shot `run(...)` overloads, continuous historical-to-realtime streams, bar loop, raw-trade broker passes, `on_bar(...)` hook, and cumulative reporting.                                                                                                                                                                                       |
 | Strategy orders             | Supported                                                                  | `strategy_entry / order / exit / close / close_all / cancel / cancel_all` with OHLC-path fill resolution, OCA, pyramiding, slippage, commissions, margin gates, partial / FIFO-vs-ANY closes, trailing stops, and TV deferred-flip carry handling.                                                                                                                  |
 | Strategy state / accessors  | Supported                                                                  | Position state, equity / drawdown / runup tracking, win / loss counts, full closed- and open-trade accessor methods, intraday fill counter.                                                                                                                                                                                                                         |
 | Strategy risk               | Supported                                                                  | All six `strategy.risk.*` gates are wired (`engine_risk.cpp` + the fill/order gates): `allow_entry_in` direction allow-list, `max_position_size`, `max_drawdown` (abs / % of peak equity), `max_intraday_loss` (abs / % of equity), `max_cons_loss_days`, and `max_intraday_filled_orders` (latch-till-day-rollover cap-close).                                      |
@@ -58,7 +58,7 @@
 ## Public C ABI
 
 `<pineforge/pineforge.h>` is the **single canonical consumer header**.
-Every compiled PineForge strategy `.so` exports exactly the 19 symbols
+Every compiled PineForge strategy `.so` exports exactly the 26 symbols
 declared there:
 
 
@@ -74,6 +74,12 @@ declared there:
 | `strategy_set_magnifier_volume_weighted` | Toggle volume-weighted magnifier                                  |
 | `strategy_set_trace_enabled`             | Toggle per-bar trace recording                                    |
 | `strategy_set_trade_start_time`          | Earliest Unix-ms at which order commands may fire                 |
+| `strategy_stream_begin`                  | Warm on confirmed OHLCV and enter realtime mode                   |
+| `strategy_stream_push_tick`              | Push one normalized ordered trade                                |
+| `strategy_stream_push_ticks`             | Push one contiguous ordered-trade array                           |
+| `strategy_stream_advance_time`            | Confirm elapsed bars and materialize quiet intervals              |
+| `strategy_stream_end`                    | End the realtime lifecycle                                       |
+| `strategy_stream_fill_report`            | Snapshot cumulative warmup + realtime state                       |
 | `strategy_set_chart_timezone`            | Chart display TZ (intraday-day rollover gates)                    |
 | `strategy_set_syminfo_timezone`          | Exchange TZ (`syminfo.timezone`)                                  |
 | `strategy_set_syminfo_session`           | Trading session string (`syminfo.session`)                        |
@@ -82,11 +88,13 @@ declared there:
 | `strategy_set_syminfo_metadata`          | Inject fundamental / exchange metadata by Pine member name        |
 | `strategy_get_last_error`                | Error message from the most recent failed run                     |
 | `pf_version_get`                         | Runtime version (struct)                                          |
+| `pf_abi_version`                         | Caller-allocated POD layout version                               |
 | `pf_version_string`                      | Runtime version (string)                                          |
 
 
-POD types (`pf_bar_t`, `pf_trade_t`, `pf_report_t`, `pf_security_diag_t`,
-`pf_trace_entry_t`, `pf_version_t`) and the `pf_magnifier_distribution_t`
+POD types (`pf_bar_t`, `pf_trade_tick_t`, `pf_trade_t`, `pf_report_t`,
+the metrics/equity structs, `pf_security_diag_t`, `pf_trace_entry_t`,
+`pf_version_t`) and the `pf_magnifier_distribution_t`
 enum complete the surface. **Stability:** within the same
 `PINEFORGE_VERSION_MAJOR`, struct layouts and `extern "C"` signatures are
 append-only. New fields may be appended; existing fields are never
@@ -109,8 +117,8 @@ single `.hpp`):
 
 | Module             | Header                   | Source                                                                                                                                                                                                                                   | Pine-facing role                                                                                                                                              |
 | ------------------ | ------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| Public C ABI       | `pineforge.h`            | `c_abi.cpp` (+ layout `static_assert`s)                                                                                                                                                                                                  | The 19 documented C symbols every compiled strategy `.so` exports.                                                                                            |
-| Engine             | `engine.hpp`             | `engine_run.cpp`, `engine_orders.cpp`, `engine_fills.cpp`, `engine_path_resolve.cpp`, `engine_strategy_commands.cpp`, `engine_trade_accessors.cpp`, `engine_security.cpp`, `engine_lower_tf.cpp`, `engine_risk.cpp`, `engine_report.cpp` | Strategy lifecycle, orders, fills, risk gating, reports, inputs / syminfo, magnifier loop, TF aggregation, `request.security` plumbing.                       |
+| Public C ABI       | `pineforge.h`            | `c_abi.cpp` (+ layout `static_assert`s)                                                                                                                                                                                                                       | The 26 documented C symbols every compiled strategy `.so` exports.                                                                                                           |
+| Engine             | `engine.hpp`             | `engine_run.cpp`, `engine_stream.cpp`, `engine_orders.cpp`, `engine_fills.cpp`, `engine_path_resolve.cpp`, `engine_strategy_commands.cpp`, `engine_trade_accessors.cpp`, `engine_security.cpp`, `engine_lower_tf.cpp`, `engine_risk.cpp`, `engine_report.cpp` | One-shot and continuous lifecycle, orders, raw-trade/bar fills, risk, reports, inputs / syminfo, magnifier, TF aggregation, and `request.security` plumbing.                  |
 | Engine internals   | `engine_internal.hpp`    | (private cross-TU header)                                                                                                                                                                                                                | `pineforge::internal::`* types and helpers shared between engine `.cpp` partitions; not part of the public ABI.                                               |
 | Technical analysis | `ta.hpp`                 | `ta_moving_averages.cpp`, `ta_oscillators.cpp`, `ta_volatility_trend.cpp`, `ta_extremes_volume.cpp`, `ta_misc.cpp`                                                                                                                       | Official `ta.`* functions and series variables backed by stateful runtime classes with `compute` / `recompute`, plus `pivot_point_levels(...)` free function. |
 | Math               | `math.hpp`               | `math.cpp`                                                                                                                                                                                                                               | Inline `pine_random(...)` PRNG and rolling `math::Sum` class.                                                                                                 |
@@ -767,20 +775,20 @@ carries a forward-looking assessment using these buckets:
 `fill`, `hline`, `bgcolor`, `barcolor`, `label.`*, `line.`*, `box.*`,
 `table.*`, `polyline.*`, `linefill.*`, `alert(...)`, `alertcondition(...)`.
 
-- **Feasibility:** *Feasible* for plotting primitives (capture series + style metadata into the `ReportC` extension or a side-channel CSV / JSON for an external renderer); *Out of scope structurally* for live `alert(...)` because PineForge produces no realtime stream.
-- **Future story:** A "report-as-data" path is the obvious target — `plot(...)` and friends would write tagged time-series rows into a new diagnostics array on `ReportC`, and a Python harness would render them with Plotly / matplotlib. `alertcondition(...)` results could be returned as a list of `(bar_time, message)` triples. The graphics primitives (`label.new`, `line.new`, `box.new`) would need a small annotation runtime — straightforward but high surface area. Webhook / push-notification alerts stay out of scope.
+- **Feasibility:** *Feasible* for plotting primitives (capture series + style metadata into the `ReportC` extension or a side-channel CSV / JSON for an external renderer). Realtime `alert(...)` is now also feasible because the engine has an explicit realtime lifecycle, but it still needs alert frequency/dedup state, an event ABI, and delivery plumbing.
+- **Future story:** A "report-as-data" path is the obvious target — `plot(...)` and friends would write tagged time-series rows into a new diagnostics array on `ReportC`, and a Python harness would render them with Plotly / matplotlib. `alertcondition(...)` results could be returned as a list of `(bar_time, message)` triples. Alert delivery should drain deterministic engine events into an external JSON-only webhook adapter so network retries never mutate engine state.
 - **Why not done yet:** Backtests already produce `TradeC[]` and per-bar diagnostics; visual plotting has not been the unblocker for any user-facing strategy validation. It is a UX feature, not a correctness feature.
 
 #### `varip` and realtime tick semantics
 
 `varip` (persists across realtime ticks; resets on bar close in TV);
 `barstate.isrealtime`, `barstate.isnew` in realtime, `calc_on_every_tick`,
-`calc_on_order_fills`, live tick streams.
+`calc_on_order_fills`.
 
 - **Feasibility:**
   - `varip` itself: *Feasible*. With the bar magnifier already simulating intrabar samples, `varip` could map onto sub-bar persistence (do not reset between magnifier ticks; reset only on bar close).
-  - Realtime barstate flags + `calc_on_every_tick` + `calc_on_order_fills`: *Out of scope structurally* — they require a live data feed PineForge does not have.
-- **Future story:** `varip` mapping to magnifier sub-bar state is the right design; currently the support checker rejects `varip` outright but that is conservative rather than fundamental. Realtime semantics would require a streaming runtime that PineForge is not built for and does not aim to be.
+  - Realtime barstate flags + `calc_on_every_tick` + `calc_on_order_fills`: *Feasible*. Ordered trades now reach the broker, but script recalculation still needs TradingView-style rollback, commit, and post-fill scheduling.
+- **Future story:** `varip` mapping to live per-tick persistence is the right design; currently the support checker rejects `varip` outright but that is conservative rather than fundamental. The new streaming lifecycle supplies the missing data/event boundary, while rollback and codegen support remain future work.
 - **Why not done yet:** `varip` use cases overlap heavily with what `var` already covers in batch mode. The realtime distinction TV makes is meaningful only when the data feed is live.
 
 #### Import / export / library system
@@ -858,4 +866,4 @@ and [PineTS](https://github.com/LuxAlgo/PineTS), exercising the same
 surface across three independent engines (PineForge hits canonical
 *excellent* tier on 48 / 50 strategies vs PyneCore's 45 / 50; the 3
 PyneCore-only outliers all involve bracket / trail / partial exits, see
-`[benchmarks/results/summary.md](../benchmarks/results/summary.md)`).
\ No newline at end of file
+`[benchmarks/results/summary.md](../benchmarks/results/summary.md)`).
diff --git a/docs/pages/abi-stability.md b/docs/pages/abi-stability.md
index 5b98805..278e3c6 100644
--- a/docs/pages/abi-stability.md
+++ b/docs/pages/abi-stability.md
@@ -54,7 +54,7 @@ Three layers:
 
 ## Symbol inventory
 
-A compiled strategy `.so` exports **exactly these 10 C symbols** and
+A compiled strategy `.so` exports **exactly these 26 C symbols** and
 zero internal C++ symbols:
 
 | Symbol | Group |
@@ -68,12 +68,28 @@ zero internal C++ symbols:
 | `strategy_set_override` | @ref pf_config |
 | `strategy_set_magnifier_volume_weighted` | @ref pf_config |
 | `strategy_set_trace_enabled` | @ref pf_config |
+| `strategy_set_trade_start_time` | @ref pf_config |
+| `strategy_stream_begin` | @ref pf_streaming |
+| `strategy_stream_push_tick` | @ref pf_streaming |
+| `strategy_stream_push_ticks` | @ref pf_streaming |
+| `strategy_stream_advance_time` | @ref pf_streaming |
+| `strategy_stream_end` | @ref pf_streaming |
+| `strategy_stream_fill_report` | @ref pf_streaming |
+| `strategy_set_chart_timezone` | @ref pf_config |
+| `strategy_set_syminfo_timezone` | @ref pf_config |
+| `strategy_set_syminfo_session` | @ref pf_config |
+| `strategy_set_syminfo_mintick` | @ref pf_config |
+| `strategy_set_syminfo_pointvalue` | @ref pf_config |
+| `strategy_set_syminfo_metadata` | @ref pf_config |
+| `strategy_get_last_error` | Diagnostics |
 | `pf_version_get` | @ref pf_version |
+| `pf_abi_version` | @ref pf_version |
+| `pf_version_string` | @ref pf_version |
 
-Plus the runtime-only export `pf_version_string` (string descriptor)
-and the configuration knob `strategy_set_trade_start_time`. The 10 above
-are the historical canonical surface; the additional symbols are
-append-only additions covered by the same minor-version guarantee.
+The five strategy-lifecycle functions are emitted by codegen. Runtime exports
+are force-linked into each strategy library, so consumers resolve the same
+complete ABI from the strategy `.so`. All additions remain covered by the
+minor-version append-only guarantee.
 
 You can verify this against any strategy `.so`:
 
diff --git a/docs/pages/configuration.md b/docs/pages/configuration.md
index b02d777..43907d1 100644
--- a/docs/pages/configuration.md
+++ b/docs/pages/configuration.md
@@ -102,8 +102,9 @@ Configuration values stick to the **handle**, not the run. They apply
 to every #run_backtest on that handle until overwritten or
 #strategy_free is called.
 
-For parameter sweeps, create a fresh handle per run — a handle's
-trade history accumulates across runs (see [Lifecycle](@ref lifecycle)):
+One-shot backtests reset broker and report state before every run, while these
+configuration values persist on the handle. Fresh handles remain a clear
+default for parameter sweeps (see [Lifecycle](@ref lifecycle)):
 
 ```c
 for (int len = 10; len <= 30; len += 2) {
diff --git a/docs/pages/examples-python-sweep.md b/docs/pages/examples-python-sweep.md
index 4c9aaca..0419d40 100644
--- a/docs/pages/examples-python-sweep.md
+++ b/docs/pages/examples-python-sweep.md
@@ -29,20 +29,16 @@ top 3 by net pnl:
 Numbers depend on the OHLCV snapshot — refresh with
 `python3 tutorial/data/fetch_btcusdt.py` for current Binance bars.
 
-## The fresh-handle rule
+## Fresh handles keep sweeps isolated
 
-A `pf_strategy_t` carries trade history, equity curve, and position
-state. Calling #run_backtest twice on the same handle accumulates
-trades from both runs into the second report — the engine never
-rewinds.
+#run_backtest resets broker, trade, equity, series, and report state before
+each one-shot run. Configuration overrides persist on the handle. The example
+still creates a fresh handle per grid point so every configuration is explicit
+and the same pattern works safely with concurrent workers.
 
-**For sweeps and walk-forward windows, always create a fresh handle
-per run.** Configuration overrides set on a handle apply to all runs
-on that handle, but trade state does not reset.
-
-@warning Reusing a handle across runs in a sweep is the most common
-PineForge integration bug. The runtime will not warn you — the second
-report's `trades_len` simply grows.
+Do not use repeated one-shot calls to model a continuous session: use the
+[streaming lifecycle](@ref streaming), which deliberately preserves the
+warmed strategy and broker state across the data-source handoff.
 
 ## Source: sweep.py
 
@@ -174,7 +170,7 @@ the engine.
 ## See also
 
 - [Tutorial: MACD](@ref tutorial_macd) — the single-run baseline
-- [Lifecycle](@ref lifecycle) — why one handle per run
+- [Lifecycle](@ref lifecycle) — one-shot reset and continuous-stream ownership
 - [Configuration](@ref configuration) — every override key
 - [`tutorial/run_advanced.py`](https://github.com/pineforge-4pass/pineforge-engine/blob/main/tutorial/run_advanced.py)
   — the shipping reference implementation this example mirrors
diff --git a/docs/pages/ffi-python.md b/docs/pages/ffi-python.md
index d3462ce..bca0275 100644
--- a/docs/pages/ffi-python.md
+++ b/docs/pages/ffi-python.md
@@ -2,7 +2,7 @@
 
 @tableofcontents
 
-The C ABI is FFI-friendly by design: a handful of functions, 10 POD
+The C ABI is FFI-friendly by design: a compact function set, 11 POD
 structs, one enum, no callbacks, no opaque types except `pf_strategy_t`
 (which is `void*`). This page shows the canonical `ctypes` wiring for Python; any
 language with a C-FFI (Rust `libc`, Go `cgo`, Node `ffi-napi`,
@@ -25,6 +25,14 @@ class pf_bar_t(ctypes.Structure):
         ("timestamp", ctypes.c_int64),
     ]
 
+class pf_trade_tick_t(ctypes.Structure):
+    _fields_ = [
+        ("timestamp", ctypes.c_int64),
+        ("sequence",  ctypes.c_uint64),
+        ("price",     ctypes.c_double),
+        ("quantity",  ctypes.c_double),
+    ]
+
 class pf_trade_t(ctypes.Structure):
     _fields_ = [
         ("entry_time",   ctypes.c_int64),
@@ -198,6 +206,34 @@ lib.run_backtest_full.argtypes = [
 ]
 lib.run_backtest_full.restype = None
 
+lib.strategy_stream_begin.argtypes = [
+    ctypes.c_void_p,
+    ctypes.POINTER(pf_bar_t), ctypes.c_int,
+    ctypes.c_char_p, ctypes.c_char_p,
+]
+lib.strategy_stream_begin.restype = ctypes.c_int
+
+lib.strategy_stream_push_tick.argtypes = [
+    ctypes.c_void_p, ctypes.POINTER(pf_trade_tick_t)]
+lib.strategy_stream_push_tick.restype = ctypes.c_int
+
+lib.strategy_stream_push_ticks.argtypes = [
+    ctypes.c_void_p, ctypes.POINTER(pf_trade_tick_t), ctypes.c_int]
+lib.strategy_stream_push_ticks.restype = ctypes.c_int
+
+lib.strategy_stream_advance_time.argtypes = [ctypes.c_void_p, ctypes.c_int64]
+lib.strategy_stream_advance_time.restype = ctypes.c_int
+
+lib.strategy_stream_end.argtypes = [ctypes.c_void_p, ctypes.c_int]
+lib.strategy_stream_end.restype = ctypes.c_int
+
+lib.strategy_stream_fill_report.argtypes = [
+    ctypes.c_void_p, ctypes.POINTER(pf_report_t)]
+lib.strategy_stream_fill_report.restype = ctypes.c_int
+
+lib.strategy_get_last_error.argtypes = [ctypes.c_void_p]
+lib.strategy_get_last_error.restype = ctypes.c_char_p
+
 lib.report_free.argtypes      = [ctypes.POINTER(pf_report_t)]
 lib.report_free.restype       = None
 
@@ -210,6 +246,39 @@ lib.strategy_set_override.argtypes = [ctypes.c_void_p,
 lib.strategy_set_override.restype  = None
 ```
 
+## Historical to realtime stream
+
+Warmup and realtime calls operate on the same handle. The plural push accepts
+one contiguous array and is useful for replay; it does not create chunks or
+session boundaries inside the strategy.
+
+```python
+state = lib.strategy_create(None)
+
+if lib.strategy_stream_begin(state, history, history_n, b"1", b"1") != 0:
+    raise RuntimeError(lib.strategy_get_last_error(state).decode())
+
+if lib.strategy_stream_push_ticks(state, ticks, len(ticks)) != 0:
+    raise RuntimeError(lib.strategy_get_last_error(state).decode())
+
+if lib.strategy_stream_advance_time(state, session_end_ms) != 0:
+    raise RuntimeError(lib.strategy_get_last_error(state).decode())
+if lib.strategy_stream_end(state, 0) != 0:
+    raise RuntimeError(lib.strategy_get_last_error(state).decode())
+
+report = pf_report_t()
+if lib.strategy_stream_fill_report(state, ctypes.byref(report)) != 0:
+    raise RuntimeError(lib.strategy_get_last_error(state).decode())
+
+lib.report_free(ctypes.byref(report))
+lib.strategy_free(state)
+```
+
+The complete runnable version is
+[`tutorial/run_stream.py`](https://github.com/pineforge-4pass/pineforge-engine/blob/main/tutorial/run_stream.py).
+See [Historical to realtime streaming](@ref streaming) for validation,
+clock advancement, and partial-bar rules.
+
 ## End-to-end run
 
 ```python
diff --git a/docs/pages/getting-started.md b/docs/pages/getting-started.md
index 7879c67..3bf4b42 100644
--- a/docs/pages/getting-started.md
+++ b/docs/pages/getting-started.md
@@ -22,9 +22,8 @@ cmake --build build -j
 ctest --test-dir build --output-on-failure
 ```
 
-Expect 30 tests to pass. The largest (`test_integration`,
-`test_request_security`) take a few hundred milliseconds; everything
-else completes faster.
+All configured tests should pass. The suite includes integration, ABI,
+timeframe, broker, and continuous-stream lifecycle coverage.
 
 ## Install
 
@@ -70,5 +69,6 @@ cc hello.c -lpineforge -lstdc++ -lm -o hello
 That's it — you have a working install. Next steps:
 
 - **[Lifecycle](@ref lifecycle)** — handle ownership and report freeing
+- **[Historical to realtime streaming](@ref streaming)** — preserve state while switching feeds
 - **[Tutorial: MACD](@ref tutorial_macd)** — full backtest walkthrough
 - **[FFI from Python](@ref ffi_python)** — calling the runtime via ctypes
diff --git a/docs/pages/index.md b/docs/pages/index.md
index 1fa7a2b..f205504 100644
--- a/docs/pages/index.md
+++ b/docs/pages/index.md
@@ -41,6 +41,11 @@ consumption.
   **[Pure C](@ref examples_c)** or **[Rust](@ref examples_rust)** worked
   examples.
 
+- <b class="tab-title">I'm connecting a realtime feed</b>
+  Start with **[Historical to realtime streaming](@ref streaming)** for
+  the warmup, ordered-trade, clock, and report lifecycle, then run
+  `tutorial/run_stream.py` against the bundled MACD strategy.
+
 - <b class="tab-title">I'm analysing backtest results</b>
   Read the **[Trading metrics reference](@ref metrics)** — every
   `pf_metrics_t` field with units, NaN rules, and TV / quant-library
@@ -58,7 +63,7 @@ consumption.
 
 ## Worked examples
 
-Five end-to-end, runnable examples that go beyond the MACD tutorial:
+End-to-end, runnable examples that go beyond the MACD tutorial:
 
 | Example | Use case |
 | --- | --- |
@@ -68,22 +73,25 @@ Five end-to-end, runnable examples that go beyond the MACD tutorial:
 | [Multi-strategy harness](@ref examples_multi) | Load N `.so` files; rank by net PnL; thread-pool execution. |
 | [Magnifier on vs off](@ref examples_magnifier) | A/B comparison with all six distribution modes. |
 | [Multi-timeframe (MTF)](@ref mtf) | `script_tf` switching, `request.security`, and lower-TF sub-bar synthesis. |
+| [Historical to realtime streaming](@ref streaming) | Warm on confirmed OHLCV and continue the same strategy on ordered trades. |
 | [Calling from Rust](@ref examples_rust) | Idiomatic `libloading` wrapper with safe Rust types. |
 
 ---
 
 ## API at a glance
 
-The entire public surface fits in **one header** and **10 functions**:
+The entire public surface fits in **one header** and **26 functions**:
 
 | Group | Symbols | Reference |
 | --- | --- | --- |
 | Lifecycle | `strategy_create`, `strategy_free`, `run_backtest`, `run_backtest_full`, `report_free` | @ref pf_lifecycle |
-| Configuration | `strategy_set_input`, `strategy_set_override`, `strategy_set_magnifier_volume_weighted`, `strategy_set_trace_enabled`, `strategy_set_trade_start_time` | @ref pf_config |
-| Version | `pf_version_get`, `pf_version_string` | @ref pf_version |
-| Types | `pf_bar_t`, `pf_trade_t`, `pf_report_t`, `pf_security_diag_t`, `pf_trace_entry_t`, `pf_version_t`, `pf_magnifier_distribution_t` | @ref pf_types |
+| Streaming | `strategy_stream_begin`, `strategy_stream_push_tick`, `strategy_stream_push_ticks`, `strategy_stream_advance_time`, `strategy_stream_end`, `strategy_stream_fill_report` | @ref pf_streaming |
+| Configuration | Inputs, strategy overrides, tracing, trade start, chart / symbol timezone, session, tick size, point value, and numeric metadata | @ref pf_config |
+| Diagnostics | `strategy_get_last_error` | #strategy_get_last_error |
+| Version | `pf_version_get`, `pf_abi_version`, `pf_version_string` | @ref pf_version |
+| Types | `pf_bar_t`, `pf_trade_tick_t`, `pf_trade_t`, `pf_report_t`, metrics, diagnostics, trace, equity, version, and `pf_magnifier_distribution_t` | @ref pf_types |
 
-Every PineForge-generated strategy `.so` exports exactly these symbols
+Every PineForge-generated strategy `.so` exports exactly these 26 symbols
 and zero internal C++ symbols — see
 **[ABI stability](@ref abi_stability)** for the full guarantee.
 
diff --git a/docs/pages/lifecycle.md b/docs/pages/lifecycle.md
index e834a31..31cd5c1 100644
--- a/docs/pages/lifecycle.md
+++ b/docs/pages/lifecycle.md
@@ -64,6 +64,52 @@ Or for the common case of "auto-detect timeframe, no magnifier":
 run_backtest(s, bars, n, &r);
 ```
 
+### Continue from historical bars into realtime trades
+
+Use the stream lifecycle when the data source changes but the strategy
+instance must not. `strategy_stream_begin()` executes the confirmed OHLCV
+warmup once. Every later operation advances the same broker, equity, pending
+orders, Pine variables, TA objects, `request.security()` evaluators, and any
+partially formed higher-timeframe candle.
+
+```c
+pf_bar_t *history = load_confirmed_ohlcv(&history_n);
+
+if (strategy_stream_begin(s, history, history_n, "1", "1") != 0)
+    fail(strategy_get_last_error(s));
+
+for (;;) {
+    pf_trade_tick_t tick = next_normalized_trade();
+    if (strategy_stream_push_tick(s, &tick) != 0)
+        fail(strategy_get_last_error(s));
+
+    /* Call on wall-clock boundaries too, including quiet markets. */
+    strategy_stream_advance_time(s, current_time_ms());
+    if (should_stop()) break;
+}
+
+strategy_stream_end(s, 0);  /* partial input bar remains unconfirmed */
+
+pf_report_t r = {0};
+strategy_stream_fill_report(s, &r);
+```
+
+The default strategy cadence remains close-only. Resting broker orders are
+checked on every normalized trade, so stop/limit fills use the observed source path
+and a market order from the preceding close fills on the first subsequent
+trade. `strategy_stream_push_ticks()` is the batch equivalent for replay and
+reduces FFI overhead without changing tick semantics.
+
+The warmup's last bar must be confirmed. Begin normalized trades at or after
+the next input-bar open. Call `strategy_stream_advance_time()` at confirmed boundaries;
+it closes elapsed bars, creates zero-volume carry-forward bars for quiet
+in-session intervals, and skips configured out-of-session intervals. Normally
+end with `finalize_partial_input_bar = 0` to avoid treating an open bar as confirmed.
+
+See [Historical to realtime streaming](@ref streaming) for the complete tick
+validation rules, a contiguous-replay example, and the runnable Python
+tutorial.
+
 The runtime fills `r` in place — the `pf_report_t` struct itself is
 caller-owned (typically stack-allocated), but the arrays it points to
 (`trades`, `security_diag`, `trace`, `trace_names`) are heap-allocated
@@ -99,12 +145,11 @@ strategy_free(s);  /* releases the handle */
 leaves dangling pointers — the trace name string table lives on the
 strategy, not the report.
 
-## One handle per run
+## Handle reuse and continuous streams
 
-A `pf_strategy_t` carries closed-trade history, equity curve, and
-position state. Calling #run_backtest twice on the same handle
-**accumulates** trades from both runs into the second report — the
-state machine never rewinds.
+Calling #run_backtest or #run_backtest_full starts a new backtest and resets
+per-run broker/report state. The `strategy_stream_*` lifecycle is the explicit
+way to preserve and continue state across historical and realtime sources.
 
 For parameter sweeps, walk-forward windows, or any A/B comparison,
 **create a fresh handle per run**:
@@ -131,16 +176,13 @@ the annotated walkthrough.
 
 ## Errors and partial state
 
-The C ABI does not return error codes. Instead:
+The one-shot `run_backtest*` calls return through
+#strategy_get_last_error. Stream lifecycle calls additionally return `0` on
+success and `-1` on failure. In both cases, read
+#strategy_get_last_error for the detailed message.
 
 - **Allocation failure** in `strategy_create` returns `NULL`. Always check.
-- **Bad input timeframe** falls back to auto-detection.
 - **Empty bar feed** (`n == 0`) is valid — the report is filled with zero
   counts and an empty trade list.
-- **Internal runtime errors** (e.g. invalid `request.security` symbol)
-  emit a `pine_runtime_error` log line and either skip the offending
-  evaluation or terminate the strategy bar — never the host process.
-
-If you need finer-grained error reporting, attach a logger via the
-runtime's internal log hook (internal API; not part of the
-public ABI).
+- **Invalid timeframes, reordered ticks, and replayed trade ids** are rejected
+  without unwinding a C++ exception across the C boundary.
diff --git a/docs/pages/streaming.md b/docs/pages/streaming.md
new file mode 100644
index 0000000..a679d74
--- /dev/null
+++ b/docs/pages/streaming.md
@@ -0,0 +1,131 @@
+# Historical to realtime streaming {#streaming}
+
+@tableofcontents
+
+Use the streaming lifecycle when a strategy must warm on confirmed OHLCV and
+then continue on normalized ordered trades **without replacing its instance**.
+The handoff preserves broker state, position and equity, pending orders, Pine
+series and variables, TA objects, `request.security()` evaluators, and a
+partially formed higher-timeframe candle.
+
+This is the runtime model used by a continuously running strategy:
+
+1. Call #strategy_stream_begin with every confirmed historical input bar.
+2. Feed normalized trades with #strategy_stream_push_tick, or pass one contiguous
+   replay tape to #strategy_stream_push_ticks.
+3. Call #strategy_stream_advance_time at wall-clock boundaries, including
+   quiet periods where no trade arrived.
+4. Call #strategy_stream_end, then snapshot cumulative state with
+   #strategy_stream_fill_report.
+5. Release report arrays with #report_free and the instance with
+   #strategy_free.
+
+## Run the tutorial
+
+The repository ships a complete Python `ctypes` example. It uses the frozen
+MACD tutorial strategy, warms on the first 640 confirmed candles, and converts
+the remaining 32 candles into a deterministic ordered-trade replay:
+
+```bash
+bash tutorial/run.sh
+python3 tutorial/run_stream.py
+```
+
+Expected output has this shape (trade and P&L values follow the frozen data):
+
+```text
+MACD historical -> realtime stream
+  warmup:     640 confirmed 15m bars
+  realtime:   32 bars from 128 ordered trades
+  handoff:    2026-05-06 10:15 UTC
+  processed:  672 input / 672 script bars
+```
+
+The tutorial's OHLC expansion keeps it self-contained. A production service
+should normalize any provider payload into #pf_trade_tick_t before calling the
+engine. Authentication, transport, symbol mapping, and provider-only metadata
+remain outside the runtime.
+
+## C lifecycle
+
+```c
+pf_strategy_t strategy = strategy_create(NULL);
+
+if (strategy_stream_begin(strategy, history, history_n, "1", "1") != 0)
+    fail(strategy_get_last_error(strategy));
+
+/* A live service normally calls this once for every received trade. */
+pf_trade_tick_t tick = {
+    .timestamp = 1743206400123LL,
+    .sequence = 1234567,
+    .price = 1823.45,
+    .quantity = 0.125,
+};
+if (strategy_stream_push_tick(strategy, &tick) != 0)
+    fail(strategy_get_last_error(strategy));
+
+/* Confirm elapsed bars even if the market was quiet. */
+if (strategy_stream_advance_time(strategy, confirmed_boundary_ms) != 0)
+    fail(strategy_get_last_error(strategy));
+if (strategy_stream_end(strategy, 0) != 0)
+    fail(strategy_get_last_error(strategy));
+
+pf_report_t report = {0};
+if (strategy_stream_fill_report(strategy, &report) != 0)
+    fail(strategy_get_last_error(strategy));
+consume(&report);
+report_free(&report);
+strategy_free(strategy);
+```
+
+Every stream function returns `0` on success and `-1` on failure. Read
+#strategy_get_last_error immediately after a failure.
+
+## Tick and bar semantics
+
+- Warmup bars must be strictly increasing, confirmed, and use a fixed-duration
+  input timeframe. The first normalized trade belongs at or after the next
+  input-bar open.
+- Timestamps may be equal but cannot move backwards. Non-zero normalized
+  sequence values must increase strictly; use zero if the source has no stable
+  ordering key.
+- Price must be finite and positive. Quantity must be finite and non-negative;
+  quantities accumulate into the forming bar's volume.
+- The default Pine strategy cadence remains close-only. Strategy code runs
+  when a realtime bar closes, while broker orders resting from an earlier
+  calculation evaluate against every normalized trade.
+- A market order created at the preceding close fills at the first subsequent
+  trade. Stops and limits see the observed trade path, not an inferred OHLC
+  traversal.
+- #strategy_stream_advance_time materializes elapsed quiet in-session intervals
+  as zero-volume carry-forward bars. Intervals outside the configured syminfo
+  session are skipped, so closed markets do not acquire synthetic bars.
+
+## One tick versus a contiguous replay
+
+#strategy_stream_push_tick is the natural live-feed API. The plural
+#strategy_stream_push_ticks is semantically identical and exists to avoid FFI
+overhead during replay. One call can cover an entire memory-mapped tick tape;
+it does not end a bar, checkpoint the strategy, or introduce a session
+boundary between records.
+
+## Ending and reports
+
+Normally advance to a confirmed boundary and call
+`strategy_stream_end(strategy, 0)`. Passing a non-zero
+`finalize_partial_input_bar` explicitly treats the currently forming input bar
+as complete, which is rarely appropriate for a stopped live service.
+
+#strategy_stream_fill_report may be used after ending to return the cumulative
+warmup plus realtime result. The report follows the normal ownership rules in
+[Report schema](@ref report_schema).
+
+## Current calculation scope
+
+The first streaming release implements TradingView's default close-only
+strategy cadence and raw-trade broker fills. `calc_on_every_tick`,
+`calc_on_order_fills`, realtime rollback/`varip`, and alert delivery are
+separate surfaces and are not implied by using this lifecycle.
+
+See [Lifecycle](@ref lifecycle) for handle ownership and
+[FFI from Python](@ref ffi_python) for the complete POD mirrors.
diff --git a/docs/pages/tutorial-macd.md b/docs/pages/tutorial-macd.md
index 906d1b3..4fb1ba2 100644
--- a/docs/pages/tutorial-macd.md
+++ b/docs/pages/tutorial-macd.md
@@ -13,7 +13,7 @@ minute. Source: [`tutorial/`](https://github.com/pineforge-4pass/pineforge-engin
 | 2 | Load `strategy.so` from Python via `ctypes`. |
 | 3 | Push an OHLCV feed and call #run_backtest_full. |
 | 4 | Read every interesting field of `pf_report_t`. |
-| 5 | Re-run the same handle with different parameters — no recompile. |
+| 5 | Re-run the same compiled strategy with different parameters — no recompile. |
 | 6 | Sweep a 2-D parameter grid in parallel using one `.so` per worker. |
 
 By the end you'll have the canonical patterns for ad-hoc backtests,
@@ -30,6 +30,7 @@ tutorial/
 │   ├── btcusdt_15m_7d.csv  # 672 frozen bars (Binance)
 │   └── fetch_btcusdt.py    # refresh from Binance public API
 ├── run.py                  # ctypes harness
+├── run_stream.py           # historical OHLCV → realtime trade stream
 ├── run_advanced.py         # parameter sweep using ABI overrides
 ├── run.sh                  # one-shot: cmake build + run.py
 └── CMakeLists.txt
@@ -191,6 +192,7 @@ is a self-contained, runnable example targeting a specific use case.
 | Example | What it shows |
 | --- | --- |
 | [Pure C example](@ref examples_c) | Same MACD run, no Python. End-to-end C code with `gcc` build. |
+| [Historical to realtime streaming](@ref streaming) | Warm the same MACD instance on OHLCV, then continue it on ordered trade ticks. |
 | [Parameter sweep in Python](@ref examples_python_sweep) | Re-run one `.so` with a 2-D MACD grid. No recompile per run. |
 | [Multi-strategy harness](@ref examples_multi) | Load N `.so` files, run them in parallel against the same feed. |
 | [Magnifier on vs off](@ref examples_magnifier) | A/B comparison showing how intra-bar fills change the trade list. |
diff --git a/include/pineforge/bar.hpp b/include/pineforge/bar.hpp
index e888bc0..836f73d 100644
--- a/include/pineforge/bar.hpp
+++ b/include/pineforge/bar.hpp
@@ -8,4 +8,15 @@ struct Bar {
     int64_t timestamp;  // Unix milliseconds
 };
 
+// One provider-neutral executed-trade update consumed by BacktestEngine's
+// realtime stream. `sequence` is assigned by the normalized data source: zero
+// means unavailable, while non-zero values must increase strictly so callers
+// cannot silently replay or reorder records.
+struct TradeTick {
+    int64_t timestamp;       // Unix milliseconds
+    uint64_t sequence;
+    double price;
+    double quantity;         // normalized size, accumulated into bar volume
+};
+
 } // namespace pineforge
diff --git a/include/pineforge/engine.hpp b/include/pineforge/engine.hpp
index 07e88d6..5aa92a8 100644
--- a/include/pineforge/engine.hpp
+++ b/include/pineforge/engine.hpp
@@ -1239,6 +1239,27 @@ class BacktestEngine {
     TimeframeAggregator script_tf_agg_;
     int64_t prev_bar_timestamp_ = 0;
 
+    // --- Historical -> realtime stream lifecycle ---
+    // stream_begin() executes the historical warmup through the normal run()
+    // path exactly once, then these fields carry the SAME broker, Pine series,
+    // TA and timeframe-aggregator state forward while normalized trades arrive.
+    enum class StreamPhase { IDLE, REALTIME, ENDED };
+    StreamPhase stream_phase_ = StreamPhase::IDLE;
+    bool stream_warmup_mode_ = false;
+    int64_t stream_input_tf_ms_ = 0;
+    int64_t stream_next_input_open_ms_ = 0;
+    int64_t stream_clock_ms_ = 0;
+    int64_t stream_last_tick_ms_ = 0;
+    uint64_t stream_last_sequence_ = 0;
+    bool stream_seen_sequence_ = false;
+    bool stream_has_input_bar_ = false;
+    Bar stream_input_bar_{};
+    double stream_last_price_ = 0.0;
+    bool stream_has_last_price_ = false;
+    int stream_next_script_bar_index_ = 0;
+    bool stream_script_bar_had_tick_ = false;
+    bool stream_script_tick_seen_ = false;
+
     // --- request.security state ---
     struct SecurityEvalState {
         int sec_id = 0;
@@ -1780,6 +1801,9 @@ class BacktestEngine {
     void run_simple_bar_loop(const Bar* input_bars, int n_input);
     void run_aggregation_bar_loop(const Bar* input_bars, int n_input,
                                   bool bar_magnifier, int expected_script_bars);
+    bool stream_finalize_until(int64_t timestamp_ms);
+    void stream_feed_input_bar(const Bar& bar, bool had_tick);
+    void stream_dispatch_script_bar(const Bar& bar, bool had_tick);
 
     // fill_report helpers (defined in engine_report.cpp).
     void fill_trades_section(ReportC* out) const;
@@ -1800,6 +1824,21 @@ class BacktestEngine {
              int magnifier_samples = 4,
              MagnifierDistribution magnifier_dist = MagnifierDistribution::ENDPOINTS);
 
+    // Execute confirmed historical bars, then keep this exact instance alive
+    // for realtime trade updates. The warmup feed must contain at least one
+    // complete input-timeframe bar. Normalized ticks begin at or after the next
+    // input bar's open; in-session gaps are materialized as zero-volume
+    // carry-forward bars when a later tick or stream_advance_time() crosses
+    // their close boundary. Configured out-of-session intervals are skipped.
+    bool stream_begin(const Bar* warmup_bars, int n_warmup,
+                      const std::string& input_tf,
+                      const std::string& script_tf = "");
+    bool stream_push_tick(const TradeTick& tick);
+    bool stream_push_ticks(const TradeTick* ticks, int n);
+    bool stream_advance_time(int64_t timestamp_ms);
+    bool stream_end(bool finalize_partial_input_bar = false);
+    bool stream_is_realtime() const { return stream_phase_ == StreamPhase::REALTIME; }
+
     void run(const Bar* input_bars, int n_input,
              const std::string& input_tf,
              const std::string& script_tf,
diff --git a/include/pineforge/pineforge.h b/include/pineforge/pineforge.h
index 3865c18..a7c4ed8 100644
--- a/include/pineforge/pineforge.h
+++ b/include/pineforge/pineforge.h
@@ -113,6 +113,20 @@ typedef struct pf_bar_s {
     int64_t timestamp;  /**< Bar open time, Unix milliseconds. */
 } pf_bar_t;
 
+/** One provider-neutral realtime executed-trade update.
+ *
+ *  `sequence` is optional: pass 0 when the normalized source has no stable
+ *  ordering key. Non-zero values must increase strictly within a stream.
+ *  `quantity` is expressed in the configured symbol's volume units and is
+ *  accumulated into the input bar's volume. The source adapter owns all
+ *  provider-specific fields and normalization. */
+typedef struct pf_trade_tick_s {
+    int64_t  timestamp;  /**< Source event time, Unix milliseconds. */
+    uint64_t sequence;   /**< Normalized per-stream sequence, or 0. */
+    double   price;      /**< Executed trade price (> 0). */
+    double   quantity;   /**< Traded quantity in symbol volume units (>= 0). */
+} pf_trade_tick_t;
+
 /** Closed-trade record returned in pf_report_t::trades.
  *
  *  Layout-compatible with internal `pineforge::TradeC`. */
@@ -490,6 +504,60 @@ PF_API void strategy_set_trace_enabled(pf_strategy_t s, int on);
  *  `strategy.entry/order/exit/close` commands are ignored. */
 PF_API void strategy_set_trade_start_time(pf_strategy_t s, int64_t timestamp_ms);
 
+/** @} */ /* end of pf_config */
+
+/** @defgroup pf_streaming Historical to realtime streaming
+ *  @brief Warm on confirmed OHLCV and continue the same strategy instance on
+ *         normalized ordered trades from any data source.
+ *  @{
+ */
+
+/** Warm a strategy with confirmed OHLCV, then switch the same instance to a
+ *  realtime trade stream without resetting position, equity, pending orders,
+ *  Pine variables, TA state, request.security state, or timeframe aggregation.
+ *
+ *  The warmup must contain at least one complete fixed-duration input bar.
+ *  Normalized ticks start at or after the next input bar's open. This
+ *  lifecycle uses close-only strategy calculation (the Pine strategy default)
+ *  while resting broker orders are evaluated on every normalized trade.
+ *
+ *  @return 0 on success, -1 on failure. Inspect #strategy_get_last_error. */
+PF_API int strategy_stream_begin(pf_strategy_t s,
+                                 const pf_bar_t* warmup_bars,
+                                 int n_warmup,
+                                 const char* input_tf,
+                                 const char* script_tf);
+
+/** Push one normalized realtime trade. Returns 0 on success, -1 on failure. */
+PF_API int strategy_stream_push_tick(pf_strategy_t s,
+                                     const pf_trade_tick_t* tick);
+
+/** Push an ordered batch of realtime trades. Semantically identical to
+ *  repeated #strategy_stream_push_tick calls, with lower FFI overhead. */
+PF_API int strategy_stream_push_ticks(pf_strategy_t s,
+                                      const pf_trade_tick_t* ticks,
+                                      int n);
+
+/** Advance the stream clock and close every input bar whose end is <= the
+ *  supplied time. Quiet in-session intervals become zero-volume carry-forward
+ *  bars; intervals outside the configured syminfo session are skipped. */
+PF_API int strategy_stream_advance_time(pf_strategy_t s, int64_t timestamp_ms);
+
+/** End a realtime stream. When @p finalize_partial_input_bar is non-zero, the
+ *  currently forming input bar is dispatched before ending; normally callers
+ *  should first advance to a confirmed boundary and pass zero here. */
+PF_API int strategy_stream_end(pf_strategy_t s, int finalize_partial_input_bar);
+
+/** Snapshot the cumulative warmup + realtime report. The embedded arrays are
+ *  caller-owned after return and must be released with #report_free. */
+PF_API int strategy_stream_fill_report(pf_strategy_t s, pf_report_t* out);
+
+/** @} */ /* end of pf_streaming */
+
+/** @addtogroup pf_config
+ *  @{
+ */
+
 /** Set the strategy's chart timezone (IANA / POSIX TZ string).
  *
  *  Pine builtins ``hour``, ``minute``, ``second``, ``dayofmonth``,
diff --git a/scripts/check_c_abi_runtime.py b/scripts/check_c_abi_runtime.py
index 3df2ace..6631588 100644
--- a/scripts/check_c_abi_runtime.py
+++ b/scripts/check_c_abi_runtime.py
@@ -25,6 +25,12 @@
     "strategy_set_syminfo_pointvalue",
     "strategy_set_syminfo_metadata",
     "strategy_get_last_error",
+    "strategy_stream_begin",
+    "strategy_stream_push_tick",
+    "strategy_stream_push_ticks",
+    "strategy_stream_advance_time",
+    "strategy_stream_end",
+    "strategy_stream_fill_report",
     "pf_version_get",
     "pf_version_string",
     "pf_abi_version",
diff --git a/scripts/run_strategy.py b/scripts/run_strategy.py
index 413005f..255a25d 100644
--- a/scripts/run_strategy.py
+++ b/scripts/run_strategy.py
@@ -310,6 +310,16 @@ class BarC(ctypes.Structure):
     ]
 
 
+class TradeTickC(ctypes.Structure):
+    """Mirror of pf_trade_tick_t for historical -> realtime handoff."""
+    _fields_ = [
+        ("timestamp", ctypes.c_int64),
+        ("sequence", ctypes.c_uint64),
+        ("price", ctypes.c_double),
+        ("quantity", ctypes.c_double),
+    ]
+
+
 class TradeC(ctypes.Structure):
     _fields_ = [
         ("entry_time", ctypes.c_int64),
@@ -606,6 +616,25 @@ def _setup_signatures(self) -> None:
         if hasattr(L, "strategy_get_last_error"):
             L.strategy_get_last_error.argtypes = [ctypes.c_void_p]
             L.strategy_get_last_error.restype = ctypes.c_char_p
+        if hasattr(L, "strategy_stream_begin"):
+            L.strategy_stream_begin.argtypes = [
+                ctypes.c_void_p, ctypes.POINTER(BarC), ctypes.c_int,
+                ctypes.c_char_p, ctypes.c_char_p]
+            L.strategy_stream_begin.restype = ctypes.c_int
+            L.strategy_stream_push_tick.argtypes = [
+                ctypes.c_void_p, ctypes.POINTER(TradeTickC)]
+            L.strategy_stream_push_tick.restype = ctypes.c_int
+            L.strategy_stream_push_ticks.argtypes = [
+                ctypes.c_void_p, ctypes.POINTER(TradeTickC), ctypes.c_int]
+            L.strategy_stream_push_ticks.restype = ctypes.c_int
+            L.strategy_stream_advance_time.argtypes = [
+                ctypes.c_void_p, ctypes.c_int64]
+            L.strategy_stream_advance_time.restype = ctypes.c_int
+            L.strategy_stream_end.argtypes = [ctypes.c_void_p, ctypes.c_int]
+            L.strategy_stream_end.restype = ctypes.c_int
+            L.strategy_stream_fill_report.argtypes = [
+                ctypes.c_void_p, ctypes.POINTER(ReportC)]
+            L.strategy_stream_fill_report.restype = ctypes.c_int
         if hasattr(L, "strategy_set_input"):
             L.strategy_set_input.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p]
         if hasattr(L, "strategy_set_override"):
diff --git a/scripts/run_stream_corpus.py b/scripts/run_stream_corpus.py
new file mode 100644
index 0000000..9ca487f
--- /dev/null
+++ b/scripts/run_stream_corpus.py
@@ -0,0 +1,413 @@
+#!/usr/bin/env python3
+"""Compare bar-only and historical->tick-stream runs across corpus strategies.
+
+Every experiment slices the canonical OHLCV CSV and the raw Binance daily
+trade ZIP(s) at runtime. No pre-trimmed fixture is used. The same strategy
+configuration is run twice:
+
+  1. confirmed 1-minute OHLCV for the complete interval;
+  2. confirmed OHLCV strictly before handoff, then raw trades until end.
+
+The second run is cumulative: positions, equity, pending orders, Pine/TA state,
+request.security state and partial timeframe aggregation all cross the handoff.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import ctypes
+import difflib
+import io
+import json
+import random
+import sys
+import time
+import zipfile
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "scripts"))
+
+from run_strategy import (  # noqa: E402
+    BarC, ReportC, Strategy, TradeTickC, _report_to_dict,
+    _VALIDATION_META_KEYS,
+)
+
+DEFAULT_DATA_ROOT = Path("/Volumes/PineforgeData/binance_ethusdtp_1y")
+DEFAULT_OHLCV = "ETHUSDT.P_1m_OHLCV_canonical_2025-01-01_2026-07-08.csv"
+RANGE_START = datetime(2025, 2, 1, tzinfo=timezone.utc)
+RANGE_END = datetime(2025, 4, 1, tzinfo=timezone.utc)
+
+
+def utc_ms(dt: datetime) -> int:
+    return int(dt.timestamp() * 1000)
+
+
+def parse_utc(text: str) -> datetime:
+    value = text.strip().replace("Z", "+00:00")
+    dt = datetime.fromisoformat(value)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)
+
+
+def choose_handoff(seed: int, duration_minutes: int) -> datetime:
+    latest = RANGE_END - timedelta(minutes=duration_minutes)
+    total_minutes = int((latest - RANGE_START).total_seconds() // 60)
+    offset = random.Random(seed).randrange(total_minutes)
+    return RANGE_START + timedelta(minutes=offset)
+
+
+def iter_days(start: datetime, end: datetime):
+    day = start.date()
+    while datetime.combine(day, datetime.min.time(), tzinfo=timezone.utc) < end:
+        yield day
+        day += timedelta(days=1)
+
+
+def load_ohlcv_slice(csv_path: Path, start_ms: int | None = None,
+                     end_ms: int | None = None) -> tuple[ctypes.Array, int]:
+    """Slice canonical dataset OHLCV directly from its source CSV."""
+    rows: list[tuple[float, float, float, float, float, int]] = []
+    with csv_path.open(newline="", encoding="utf-8") as handle:
+        reader = csv.DictReader(handle)
+        timestamp_key = "timestamp" if "timestamp" in (reader.fieldnames or []) else "open_time"
+        for row in reader:
+            timestamp = int(row[timestamp_key])
+            if start_ms is not None and timestamp < start_ms:
+                continue
+            if end_ms is not None and timestamp > end_ms:
+                break
+            rows.append((
+                float(row["open"]), float(row["high"]), float(row["low"]),
+                float(row["close"]), float(row["volume"]), timestamp))
+    bars = (BarC * len(rows))()
+    for i, values in enumerate(rows):
+        bars[i] = BarC(*values)
+    return bars, len(rows)
+
+
+def load_tick_slice(data_root: Path, start_ms: int, end_ms: int) -> tuple[ctypes.Array, dict]:
+    """Slice [start_ms, end_ms) directly from raw daily ZIPs.
+
+    Binance archives are ordered by exchange trade id. The same monotonic-id
+    filter used by the dataset builder is applied here so malformed/replayed
+    rows never reach the engine.
+    """
+    start = datetime.fromtimestamp(start_ms / 1000, tz=timezone.utc)
+    end = datetime.fromtimestamp(end_ms / 1000, tz=timezone.utc)
+    rows: list[tuple[int, int, float, float, int]] = []
+    previous_id = 0
+    skipped_nonmonotonic = 0
+    archives: list[str] = []
+
+    for day in iter_days(start, end):
+        stem = f"ETHUSDT-trades-{day.isoformat()}"
+        archive = data_root / "raw" / "trades" / f"{stem}.zip"
+        if not archive.is_file():
+            raise FileNotFoundError(f"missing raw trade archive: {archive}")
+        archives.append(str(archive))
+        with zipfile.ZipFile(archive) as zf:
+            members = [name for name in zf.namelist() if name.endswith(".csv")]
+            if len(members) != 1:
+                raise RuntimeError(f"expected one CSV in {archive}, got {members}")
+            with zf.open(members[0]) as raw:
+                reader = csv.DictReader(io.TextIOWrapper(raw, encoding="utf-8", newline=""))
+                for row in reader:
+                    timestamp = int(row["time"])
+                    if timestamp < start_ms:
+                        continue
+                    if timestamp >= end_ms:
+                        break
+                    sequence = int(row["id"])
+                    if sequence <= previous_id:
+                        skipped_nonmonotonic += 1
+                        continue
+                    previous_id = sequence
+                    rows.append((
+                        timestamp,
+                        sequence,
+                        float(row["price"]),
+                        float(row["qty"]),
+                    ))
+
+    ticks = (TradeTickC * len(rows))()
+    for i, (timestamp, sequence, price, quantity) in enumerate(rows):
+        ticks[i] = TradeTickC(timestamp, sequence, price, quantity)
+    return ticks, {
+        "count": len(rows),
+        "skipped_nonmonotonic": skipped_nonmonotonic,
+        "archives": archives,
+    }
+
+
+def load_case_config(case_dir: Path) -> dict:
+    path = case_dir / "inputs.json"
+    data = json.loads(path.read_text()) if path.is_file() else {}
+    params = {
+        key: value for key, value in data.items()
+        if not key.startswith("_") and key not in _VALIDATION_META_KEYS
+        and key not in {"validation_overrides", "expected_tier", "notes"}
+    }
+    return {
+        "params": params,
+        "strategy_overrides": data.get("strategy_overrides") or {},
+        "runtime": data.get("runtime_overrides") or {},
+        "chart_timezone": data.get("chart_timezone"),
+        "input_tf": str(data.get("input_tf") or "1"),
+        "script_tf": str(data.get("script_tf") or "1"),
+        "ohlcv_start_ms": data.get("ohlcv_start_ms"),
+    }
+
+
+def configure_state(strategy: Strategy, state, config: dict) -> None:
+    lib = strategy.lib
+    for key, value in config["strategy_overrides"].items():
+        lib.strategy_set_override(state, str(key).encode(), str(value).encode())
+    for key, value in config["params"].items():
+        lib.strategy_set_input(state, str(key).encode(), str(value).encode())
+    runtime = config["runtime"]
+    chart_tz = config["chart_timezone"]
+    if chart_tz and hasattr(lib, "strategy_set_chart_timezone"):
+        lib.strategy_set_chart_timezone(state, str(chart_tz).encode())
+    if runtime.get("timezone") and hasattr(lib, "strategy_set_syminfo_timezone"):
+        lib.strategy_set_syminfo_timezone(state, str(runtime["timezone"]).encode())
+    if runtime.get("session") and hasattr(lib, "strategy_set_syminfo_session"):
+        lib.strategy_set_syminfo_session(state, str(runtime["session"]).encode())
+    if runtime.get("mintick") is not None:
+        lib.strategy_set_syminfo_mintick(state, float(runtime["mintick"]))
+    if runtime.get("pointvalue") is not None:
+        lib.strategy_set_syminfo_pointvalue(state, float(runtime["pointvalue"]))
+    for key, value in (runtime.get("syminfo_metadata") or {}).items():
+        lib.strategy_set_syminfo_metadata(state, str(key).encode(), float(value))
+
+
+def engine_error(strategy: Strategy, state, operation: str) -> RuntimeError:
+    raw = strategy.lib.strategy_get_last_error(state)
+    message = raw.decode("utf-8", "replace") if raw else "unknown engine error"
+    return RuntimeError(f"{operation}: {message}")
+
+
+def run_batch(strategy: Strategy, bars, n: int, config: dict) -> dict:
+    lib = strategy.lib
+    state = lib.strategy_create(None)
+    report = ReportC()
+    try:
+        configure_state(strategy, state, config)
+        lib.run_backtest_full(
+            state, bars, n,
+            config["input_tf"].encode(), config["script_tf"].encode(),
+            0, 4, 3, ctypes.byref(report))
+        raw = lib.strategy_get_last_error(state)
+        if raw:
+            message = raw.decode("utf-8", "replace")
+            if message:
+                raise RuntimeError("batch run: " + message)
+        return _report_to_dict(report)
+    finally:
+        lib.report_free(ctypes.byref(report))
+        lib.strategy_free(state)
+
+
+def run_stream(strategy: Strategy, warmup, n_warmup: int, ticks,
+               end_ms: int, config: dict) -> dict:
+    lib = strategy.lib
+    if not hasattr(lib, "strategy_stream_begin"):
+        raise RuntimeError("strategy library does not export strategy_stream_begin")
+    state = lib.strategy_create(None)
+    report = ReportC()
+    try:
+        configure_state(strategy, state, config)
+        if lib.strategy_stream_begin(
+                state, warmup, n_warmup,
+                config["input_tf"].encode(), config["script_tf"].encode()) != 0:
+            raise engine_error(strategy, state, "stream begin")
+        if lib.strategy_stream_push_ticks(state, ticks, len(ticks)) != 0:
+            raise engine_error(strategy, state, "stream ticks")
+        if lib.strategy_stream_advance_time(state, end_ms) != 0:
+            raise engine_error(strategy, state, "stream advance")
+        if lib.strategy_stream_end(state, 0) != 0:
+            raise engine_error(strategy, state, "stream end")
+        if lib.strategy_stream_fill_report(state, ctypes.byref(report)) != 0:
+            raise engine_error(strategy, state, "stream report")
+        return _report_to_dict(report)
+    finally:
+        lib.report_free(ctypes.byref(report))
+        lib.strategy_free(state)
+
+
+def compare_reports(batch: dict, stream: dict) -> dict:
+    left = batch["trades"]
+    right = stream["trades"]
+    positional = 0
+    for a, b in zip(left, right):
+        same_minute = (
+            a["entry_time"] // 60_000 == b["entry_time"] // 60_000
+            and a["exit_time"] // 60_000 == b["exit_time"] // 60_000)
+        same_shape = (
+            a["is_long"] == b["is_long"]
+            and a["entry_bar_index"] == b["entry_bar_index"]
+            and a["exit_bar_index"] == b["exit_bar_index"]
+            and abs(a["qty"] - b["qty"]) <= 1e-8 * max(1.0, abs(a["qty"])))
+        if same_minute and same_shape:
+            positional += 1
+    denom = max(len(left), len(right))
+    def signature(trade: dict) -> tuple:
+        return (
+            trade["entry_time"] // 60_000,
+            trade["exit_time"] // 60_000,
+            trade["is_long"],
+            trade["entry_bar_index"],
+            trade["exit_bar_index"],
+            round(trade["qty"], 8),
+        )
+
+    matcher = difflib.SequenceMatcher(
+        None,
+        [signature(trade) for trade in left],
+        [signature(trade) for trade in right],
+        autojunk=False,
+    )
+    ordered_match = sum(block.size for block in matcher.get_matching_blocks())
+    batch_profit = float(batch["net_profit"])
+    stream_profit = float(stream["net_profit"])
+    return {
+        "batch_trades": len(left),
+        "stream_trades": len(right),
+        "trade_count_equal": len(left) == len(right),
+        "positional_trade_match": positional,
+        "positional_trade_match_pct": (
+            100.0 if denom == 0 else 100.0 * positional / denom),
+        "ordered_trade_match": ordered_match,
+        "ordered_trade_match_pct": (
+            100.0 if denom == 0 else 100.0 * ordered_match / denom),
+        "structural_equal": len(left) == len(right) and ordered_match == denom,
+        "batch_net_profit": batch_profit,
+        "stream_net_profit": stream_profit,
+        "net_profit_delta": stream_profit - batch_profit,
+        "input_bars_equal": batch["input_bars_processed"]
+            == stream["input_bars_processed"],
+        "script_bars_equal": batch["script_bars_processed"]
+            == stream["script_bars_processed"],
+    }
+
+
+def discover_cases(pattern: str, limit: int) -> list[Path]:
+    cases = sorted(path.parent for path in REPO_ROOT.glob(pattern)
+                   if path.name in {"strategy.so", "strategy.dylib"})
+    return cases[:limit] if limit > 0 else cases
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--data-root", type=Path, default=DEFAULT_DATA_ROOT)
+    parser.add_argument("--handoff", help="UTC ISO-8601; default is seeded random")
+    parser.add_argument("--seed", type=int, default=20250710)
+    parser.add_argument("--duration-minutes", type=int, default=10)
+    parser.add_argument("--cases", default="corpus/validation/*/strategy.dylib")
+    parser.add_argument("--limit", type=int, default=0)
+    parser.add_argument(
+        "--honor-case-start", action="store_true",
+        help="honor validation-only ohlcv_start_ms metadata (off by default)")
+    parser.add_argument("--output", type=Path,
+                        default=REPO_ROOT / "build" / "stream_corpus_report.json")
+    args = parser.parse_args()
+
+    handoff = parse_utc(args.handoff) if args.handoff else choose_handoff(
+        args.seed, args.duration_minutes)
+    if handoff.second or handoff.microsecond:
+        raise SystemExit("handoff must be aligned to a one-minute boundary")
+    if not (RANGE_START <= handoff < RANGE_END):
+        raise SystemExit(
+            f"handoff must be in [{RANGE_START.isoformat()}, {RANGE_END.isoformat()})")
+    end = handoff + timedelta(minutes=args.duration_minutes)
+    handoff_ms, end_ms = utc_ms(handoff), utc_ms(end)
+
+    ohlcv = args.data_root / DEFAULT_OHLCV
+    if not ohlcv.is_file():
+        raise SystemExit(f"canonical OHLCV not found: {ohlcv}")
+
+    started = time.perf_counter()
+    # These are independent slices from the canonical source on every run.
+    warmup, n_warmup = load_ohlcv_slice(ohlcv, end_ms=handoff_ms - 1)
+    full, n_full = load_ohlcv_slice(ohlcv, end_ms=end_ms - 1)
+    ticks, tick_meta = load_tick_slice(args.data_root, handoff_ms, end_ms)
+    if n_warmup == 0 or len(ticks) == 0:
+        raise SystemExit("experiment slice contains no warmup bars or no ticks")
+
+    cases = discover_cases(args.cases, args.limit)
+    if not cases:
+        raise SystemExit(f"no compiled corpus strategies match {args.cases!r}")
+
+    print(f"handoff={handoff.isoformat()} end={end.isoformat()} "
+          f"warmup_bars={n_warmup} full_bars={n_full} ticks={len(ticks)} "
+          f"cases={len(cases)}", flush=True)
+
+    results = []
+    for index, case in enumerate(cases, 1):
+        so = case / ("strategy.dylib" if (case / "strategy.dylib").is_file()
+                     else "strategy.so")
+        config = load_case_config(case)
+        if args.honor_case_start and config["ohlcv_start_ms"] is not None:
+            # Re-slice from the canonical source for this case; never trim an
+            # already-trimmed array in memory.
+            case_start = int(config["ohlcv_start_ms"])
+            case_warmup, case_n_warmup = load_ohlcv_slice(
+                ohlcv, start_ms=case_start, end_ms=handoff_ms - 1)
+            case_full, case_n_full = load_ohlcv_slice(
+                ohlcv, start_ms=case_start, end_ms=end_ms - 1)
+        else:
+            case_warmup, case_n_warmup = warmup, n_warmup
+            case_full, case_n_full = full, n_full
+        item = {"case": str(case.relative_to(REPO_ROOT)), "config": config}
+        try:
+            strategy = Strategy(so)
+            batch = run_batch(strategy, case_full, case_n_full, config)
+            stream = run_stream(
+                strategy, case_warmup, case_n_warmup, ticks, end_ms, config)
+            item["comparison"] = compare_reports(batch, stream)
+            item["status"] = "ok"
+        except Exception as exc:  # corpus sweep records failures and continues
+            item["status"] = "error"
+            item["error"] = str(exc)
+        results.append(item)
+        if index % 20 == 0 or index == len(cases):
+            print(f"[{index}/{len(cases)}]", flush=True)
+
+    ok = [item for item in results if item["status"] == "ok"]
+    summary = {
+        "cases": len(results),
+        "ok": len(ok),
+        "errors": len(results) - len(ok),
+        "trade_count_equal": sum(
+            bool(item["comparison"]["trade_count_equal"]) for item in ok),
+        "input_bars_equal": sum(
+            bool(item["comparison"]["input_bars_equal"]) for item in ok),
+        "script_bars_equal": sum(
+            bool(item["comparison"]["script_bars_equal"]) for item in ok),
+        "elapsed_seconds": time.perf_counter() - started,
+    }
+    payload = {
+        "experiment": {
+            "handoff": handoff.isoformat(),
+            "end": end.isoformat(),
+            "seed": args.seed,
+            "ohlcv": str(ohlcv),
+            "warmup_bars": n_warmup,
+            "full_bars": n_full,
+            "ticks": tick_meta,
+        },
+        "summary": summary,
+        "results": results,
+    }
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    print(json.dumps(summary, sort_keys=True))
+    print(f"report={args.output}")
+    return 0 if summary["errors"] == 0 else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/run_stream_corpus_mmap.py b/scripts/run_stream_corpus_mmap.py
new file mode 100644
index 0000000..7d3aab6
--- /dev/null
+++ b/scripts/run_stream_corpus_mmap.py
@@ -0,0 +1,344 @@
+#!/usr/bin/env python3
+"""Whole-session corpus sweep over one contiguous memory-mapped tick tape.
+
+Each probe receives exactly one strategy_stream_push_ticks() call spanning its
+handoff through the common end timestamp. The tape is a direct binary mirror of
+pf_trade_tick_t; mmap avoids constructing the full multi-gigabyte slice as
+Python objects without introducing feed chunks or intermediate session ends.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import ctypes
+import io
+import json
+import mmap
+import os
+import struct
+import sys
+import time
+import zipfile
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "scripts"))
+
+from run_strategy import ReportC, Strategy, TradeTickC, _report_to_dict  # noqa: E402
+from run_stream_corpus import (  # noqa: E402
+    DEFAULT_DATA_ROOT, DEFAULT_OHLCV, compare_reports, configure_state,
+    discover_cases, engine_error, load_case_config, load_ohlcv_slice,
+    parse_utc, run_batch, utc_ms,
+)
+
+TICK_STRUCT = struct.Struct("<qQdd")
+assert TICK_STRUCT.size == ctypes.sizeof(TradeTickC) == 32
+
+
+@dataclass
+class LiveRun:
+    case_label: str
+    session_index: int
+    strategy: Strategy
+    state: int
+    tick_pointer: ctypes.POINTER(TradeTickC)
+    tick_count: int
+    error: str | None = None
+
+
+def iter_days(start: datetime, end: datetime):
+    day = start.date()
+    while datetime.combine(day, datetime.min.time(), tzinfo=timezone.utc) < end:
+        yield day
+        day += timedelta(days=1)
+
+
+def expected_tick_count(data_root: Path, start_ms: int, end_ms: int) -> int:
+    source = data_root / "ETHUSDT.P_1m_OHLCV_from_trades_2025-01-01_2026-07-08.csv"
+    total = 0
+    with source.open(newline="", encoding="utf-8") as handle:
+        reader = csv.DictReader(handle)
+        timestamp_key = "open_time" if "open_time" in (reader.fieldnames or []) else "timestamp"
+        for row in reader:
+            timestamp = int(row[timestamp_key])
+            if timestamp < start_ms:
+                continue
+            if timestamp >= end_ms:
+                break
+            total += int(float(row["count"]))
+    return total
+
+
+def build_tick_tape(data_root: Path, tape: Path, start_ms: int, end_ms: int) -> dict:
+    metadata_path = tape.with_suffix(tape.suffix + ".json")
+    expected = expected_tick_count(data_root, start_ms, end_ms)
+    expected_size = expected * TICK_STRUCT.size
+    if tape.is_file() and metadata_path.is_file() and tape.stat().st_size == expected_size:
+        metadata = json.loads(metadata_path.read_text())
+        if (metadata.get("start_ms") == start_ms
+                and metadata.get("end_ms") == end_ms
+                and metadata.get("count") == expected):
+            return metadata
+
+    tape.parent.mkdir(parents=True, exist_ok=True)
+    with tape.open("w+b") as handle:
+        handle.truncate(expected_size)
+        view = mmap.mmap(handle.fileno(), expected_size, access=mmap.ACCESS_WRITE)
+        index = 0
+        previous_id = 0
+        skipped_nonmonotonic = 0
+        archives: list[str] = []
+        try:
+            start = datetime.fromtimestamp(start_ms / 1000, tz=timezone.utc)
+            end = datetime.fromtimestamp(end_ms / 1000, tz=timezone.utc)
+            for day in iter_days(start, end):
+                stem = f"ETHUSDT-trades-{day.isoformat()}"
+                archive = data_root / "raw" / "trades" / f"{stem}.zip"
+                if not archive.is_file():
+                    raise FileNotFoundError(f"missing raw trade archive: {archive}")
+                archives.append(str(archive))
+                with zipfile.ZipFile(archive) as zf:
+                    members = [name for name in zf.namelist() if name.endswith(".csv")]
+                    if len(members) != 1:
+                        raise RuntimeError(
+                            f"expected one CSV in {archive}, got {members}")
+                    with zf.open(members[0]) as raw:
+                        reader = csv.DictReader(io.TextIOWrapper(
+                            raw, encoding="utf-8", newline=""))
+                        for row in reader:
+                            timestamp = int(row["time"])
+                            if timestamp < start_ms:
+                                continue
+                            if timestamp >= end_ms:
+                                break
+                            sequence = int(row["id"])
+                            if sequence <= previous_id:
+                                skipped_nonmonotonic += 1
+                                continue
+                            previous_id = sequence
+                            if index >= expected:
+                                raise RuntimeError(
+                                    "raw trade count exceeds OHLCV count oracle")
+                            TICK_STRUCT.pack_into(
+                                view, index * TICK_STRUCT.size,
+                                timestamp, sequence, float(row["price"]),
+                                float(row["qty"]))
+                            index += 1
+                print(f"tape_archive={day.isoformat()} ticks={index}", flush=True)
+            if index != expected:
+                # The dataset builder intentionally drops non-monotonic ids;
+                # its per-minute count oracle can include those few source rows.
+                if index + skipped_nonmonotonic != expected:
+                    raise RuntimeError(
+                        f"tick tape count mismatch: wrote {index}, expected {expected}, "
+                        f"skipped {skipped_nonmonotonic}")
+                view.resize(index * TICK_STRUCT.size)
+            view.flush()
+        finally:
+            view.close()
+
+    metadata = {
+        "start_ms": start_ms,
+        "end_ms": end_ms,
+        "count": index,
+        "record_size": TICK_STRUCT.size,
+        "skipped_nonmonotonic": skipped_nonmonotonic,
+        "archives": archives,
+    }
+    metadata_path.write_text(json.dumps(metadata, indent=2, sort_keys=True) + "\n")
+    return metadata
+
+
+def tape_timestamp(view: mmap.mmap, index: int) -> int:
+    return struct.unpack_from("<q", view, index * TICK_STRUCT.size)[0]
+
+
+def lower_bound_timestamp(view: mmap.mmap, count: int, timestamp_ms: int) -> int:
+    lo, hi = 0, count
+    while lo < hi:
+        mid = (lo + hi) // 2
+        if tape_timestamp(view, mid) < timestamp_ms:
+            lo = mid + 1
+        else:
+            hi = mid
+    return lo
+
+
+def run_whole_session(live: LiveRun) -> str | None:
+    try:
+        lib = live.strategy.lib
+        if lib.strategy_stream_push_ticks(
+                live.state, live.tick_pointer, live.tick_count) != 0:
+            return str(engine_error(live.strategy, live.state, "stream ticks"))
+        return None
+    except Exception as exc:
+        return str(exc)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--data-root", type=Path, default=DEFAULT_DATA_ROOT)
+    parser.add_argument("--handoff", action="append", required=True,
+                        help="minute-aligned UTC ISO-8601; repeat three times")
+    parser.add_argument("--end", required=True, help="exclusive UTC session end")
+    parser.add_argument("--tape", type=Path, required=True)
+    parser.add_argument("--cases", default="corpus/validation/*/strategy.dylib")
+    parser.add_argument("--limit", type=int, default=0)
+    parser.add_argument("--workers", type=int, default=min(12, os.cpu_count() or 4))
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+
+    handoffs = [parse_utc(value) for value in args.handoff]
+    if len(handoffs) != 3 or len(set(handoffs)) != 3:
+        raise SystemExit("provide exactly three distinct --handoff values")
+    if any(value.second or value.microsecond for value in handoffs):
+        raise SystemExit("handoffs must be aligned to one-minute boundaries")
+    end = parse_utc(args.end)
+    if any(value >= end for value in handoffs):
+        raise SystemExit("every handoff must precede end")
+    handoffs.sort()
+    handoff_ms = [utc_ms(value) for value in handoffs]
+    end_ms = utc_ms(end)
+
+    tape_meta = build_tick_tape(args.data_root, args.tape, handoff_ms[0], end_ms)
+    ohlcv = args.data_root / DEFAULT_OHLCV
+    warmups = [load_ohlcv_slice(ohlcv, end_ms=value - 1) for value in handoff_ms]
+    full, n_full = load_ohlcv_slice(ohlcv, end_ms=end_ms - 1)
+    cases = discover_cases(args.cases, args.limit)
+    started = time.perf_counter()
+
+    tape_handle = args.tape.open("r+b")
+    tape_view = mmap.mmap(tape_handle.fileno(), 0, access=mmap.ACCESS_WRITE)
+    tape_count = tape_meta["count"]
+    starts = [lower_bound_timestamp(tape_view, tape_count, value)
+              for value in handoff_ms]
+
+    strategies: dict[str, Strategy] = {}
+    configs: dict[str, dict] = {}
+    live_runs: list[LiveRun] = []
+    results = [{"handoff": value.isoformat(), "results": []} for value in handoffs]
+    results_by_session = [dict() for _ in handoffs]
+
+    for case_index, case in enumerate(cases, 1):
+        label = str(case.relative_to(REPO_ROOT))
+        so = case / ("strategy.dylib" if (case / "strategy.dylib").is_file()
+                     else "strategy.so")
+        strategy = Strategy(so)
+        config = load_case_config(case)
+        strategies[label] = strategy
+        configs[label] = config
+        for session_index, (warmup, n_warmup) in enumerate(warmups):
+            item = {"case": label, "config": config}
+            results_by_session[session_index][label] = item
+            state = strategy.lib.strategy_create(None)
+            try:
+                configure_state(strategy, state, config)
+                if strategy.lib.strategy_stream_begin(
+                        state, warmup, n_warmup,
+                        config["input_tf"].encode(),
+                        config["script_tf"].encode()) != 0:
+                    raise engine_error(strategy, state, "stream begin")
+                byte_offset = starts[session_index] * TICK_STRUCT.size
+                pointer = ctypes.cast(
+                    ctypes.addressof(ctypes.c_char.from_buffer(tape_view, byte_offset)),
+                    ctypes.POINTER(TradeTickC))
+                live_runs.append(LiveRun(
+                    label, session_index, strategy, state, pointer,
+                    tape_count - starts[session_index]))
+            except Exception as exc:
+                item["status"] = "error"
+                item["error"] = str(exc)
+                strategy.lib.strategy_free(state)
+        if case_index % 20 == 0 or case_index == len(cases):
+            print(f"warmup=[{case_index}/{len(cases)}]", flush=True)
+
+    with ThreadPoolExecutor(max_workers=max(1, args.workers)) as pool:
+        futures = {pool.submit(run_whole_session, live): live for live in live_runs}
+        completed = 0
+        for future in as_completed(futures):
+            live = futures[future]
+            live.error = future.result()
+            completed += 1
+            if completed % 10 == 0 or completed == len(live_runs):
+                print(f"whole_session=[{completed}/{len(live_runs)}]", flush=True)
+
+    runs_by_case: dict[str, list[LiveRun]] = {}
+    for live in live_runs:
+        runs_by_case.setdefault(live.case_label, []).append(live)
+
+    for case_index, case in enumerate(cases, 1):
+        label = str(case.relative_to(REPO_ROOT))
+        strategy = strategies[label]
+        batch = run_batch(strategy, full, n_full, configs[label])
+        for live in runs_by_case.get(label, []):
+            item = results_by_session[live.session_index][label]
+            report = ReportC()
+            try:
+                if live.error:
+                    raise RuntimeError(live.error)
+                lib = strategy.lib
+                if lib.strategy_stream_advance_time(live.state, end_ms) != 0:
+                    raise engine_error(strategy, live.state, "stream advance")
+                if lib.strategy_stream_end(live.state, 0) != 0:
+                    raise engine_error(strategy, live.state, "stream end")
+                if lib.strategy_stream_fill_report(
+                        live.state, ctypes.byref(report)) != 0:
+                    raise engine_error(strategy, live.state, "stream report")
+                stream = _report_to_dict(report)
+                item["comparison"] = compare_reports(batch, stream)
+                item["status"] = "ok"
+            except Exception as exc:
+                item["status"] = "error"
+                item["error"] = str(exc)
+            finally:
+                strategy.lib.report_free(ctypes.byref(report))
+                strategy.lib.strategy_free(live.state)
+        if case_index % 20 == 0 or case_index == len(cases):
+            print(f"report=[{case_index}/{len(cases)}]", flush=True)
+
+    tape_view.close()
+    tape_handle.close()
+    session_payloads = []
+    for session_index, handoff in enumerate(handoffs):
+        session_results = [results_by_session[session_index][
+            str(case.relative_to(REPO_ROOT))] for case in cases]
+        ok = [item for item in session_results if item.get("status") == "ok"]
+        session_payloads.append({
+            "handoff": handoff.isoformat(),
+            "tick_offset": starts[session_index],
+            "tick_count": tape_count - starts[session_index],
+            "summary": {
+                "cases": len(session_results), "ok": len(ok),
+                "errors": len(session_results) - len(ok),
+                "input_bars_equal": sum(item["comparison"]["input_bars_equal"] for item in ok),
+                "script_bars_equal": sum(item["comparison"]["script_bars_equal"] for item in ok),
+                "trade_count_equal": sum(item["comparison"]["trade_count_equal"] for item in ok),
+                "structural_equal": sum(item["comparison"]["structural_equal"] for item in ok),
+            },
+            "results": session_results,
+        })
+
+    payload = {
+        "experiment": {
+            "end": end.isoformat(), "ohlcv": str(ohlcv),
+            "full_bars": n_full, "tape": str(args.tape),
+            "tape_metadata": tape_meta, "workers": args.workers,
+            "elapsed_seconds": time.perf_counter() - started,
+        },
+        "sessions": session_payloads,
+    }
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    print(json.dumps([session["summary"] for session in session_payloads],
+                     sort_keys=True), flush=True)
+    print(f"report={args.output}", flush=True)
+    return 0 if all(session["summary"]["errors"] == 0
+                    for session in session_payloads) else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/src/c_abi.cpp b/src/c_abi.cpp
index a494db2..317f156 100644
--- a/src/c_abi.cpp
+++ b/src/c_abi.cpp
@@ -8,8 +8,8 @@
  *     the C ABI has drifted from the internal representation — fix
  *     BEFORE shipping a .so that consumers depend on.
  *   - The runtime-library-side `extern "C"` symbols (the setters,
- *     strategy_get_last_error, pf_version_get/pf_version_string,
- *     pf_abi_version — the authoritative list is EXPECTED_RUNTIME in
+ *     strategy_get_last_error, the strategy_stream_* lifecycle,
+ *     pf_version_get/pf_version_string, pf_abi_version — the authoritative list is EXPECTED_RUNTIME in
  *     scripts/check_c_abi_runtime.py, enforced by CI). The other
  *     `extern "C"` symbols listed in pineforge.h (strategy_create,
  *     run_backtest, etc.) are emitted per-compiled-strategy by the
@@ -42,6 +42,23 @@ static_assert(offsetof(pf_bar_t, volume) == offsetof(pineforge::Bar, volume),
 static_assert(offsetof(pf_bar_t, timestamp) == offsetof(pineforge::Bar, timestamp),
               "pf_bar_t::timestamp offset mismatch");
 
+/* ── Normalized trade-tick layout parity ───────────────────────── */
+
+static_assert(sizeof(pf_trade_tick_t) == sizeof(pineforge::TradeTick),
+              "pf_trade_tick_t / pineforge::TradeTick size mismatch");
+static_assert(offsetof(pf_trade_tick_t, timestamp)
+                  == offsetof(pineforge::TradeTick, timestamp),
+              "pf_trade_tick_t::timestamp offset mismatch");
+static_assert(offsetof(pf_trade_tick_t, sequence)
+                  == offsetof(pineforge::TradeTick, sequence),
+              "pf_trade_tick_t::sequence offset mismatch");
+static_assert(offsetof(pf_trade_tick_t, price)
+                  == offsetof(pineforge::TradeTick, price),
+              "pf_trade_tick_t::price offset mismatch");
+static_assert(offsetof(pf_trade_tick_t, quantity)
+                  == offsetof(pineforge::TradeTick, quantity),
+              "pf_trade_tick_t::quantity offset mismatch");
+
 /* ── Trade layout parity ────────────────────────────────────────── */
 
 static_assert(sizeof(pf_trade_t) == sizeof(pineforge::TradeC),
@@ -159,6 +176,57 @@ PF_API void strategy_set_trade_start_time(pf_strategy_t s, int64_t timestamp_ms)
     static_cast<pineforge::BacktestEngine*>(s)->set_trade_start_time(timestamp_ms);
 }
 
+PF_API int strategy_stream_begin(pf_strategy_t s,
+                                 const pf_bar_t* warmup_bars,
+                                 int n_warmup,
+                                 const char* input_tf,
+                                 const char* script_tf) {
+    if (!s) return -1;
+    auto* engine = static_cast<pineforge::BacktestEngine*>(s);
+    const auto* bars = reinterpret_cast<const pineforge::Bar*>(warmup_bars);
+    return engine->stream_begin(
+        bars, n_warmup,
+        input_tf ? std::string(input_tf) : std::string(),
+        script_tf ? std::string(script_tf) : std::string()) ? 0 : -1;
+}
+
+PF_API int strategy_stream_push_tick(pf_strategy_t s,
+                                     const pf_trade_tick_t* tick) {
+    if (!s || !tick) return -1;
+    const auto* native = reinterpret_cast<const pineforge::TradeTick*>(tick);
+    return static_cast<pineforge::BacktestEngine*>(s)->stream_push_tick(*native)
+        ? 0
+        : -1;
+}
+
+PF_API int strategy_stream_push_ticks(pf_strategy_t s,
+                                      const pf_trade_tick_t* ticks,
+                                      int n) {
+    if (!s || n < 0 || (n > 0 && !ticks)) return -1;
+    auto* engine = static_cast<pineforge::BacktestEngine*>(s);
+    const auto* native = reinterpret_cast<const pineforge::TradeTick*>(ticks);
+    return engine->stream_push_ticks(native, n) ? 0 : -1;
+}
+
+PF_API int strategy_stream_advance_time(pf_strategy_t s, int64_t timestamp_ms) {
+    if (!s) return -1;
+    return static_cast<pineforge::BacktestEngine*>(s)
+        ->stream_advance_time(timestamp_ms) ? 0 : -1;
+}
+
+PF_API int strategy_stream_end(pf_strategy_t s, int finalize_partial_input_bar) {
+    if (!s) return -1;
+    return static_cast<pineforge::BacktestEngine*>(s)
+        ->stream_end(finalize_partial_input_bar != 0) ? 0 : -1;
+}
+
+PF_API int strategy_stream_fill_report(pf_strategy_t s, pf_report_t* out) {
+    if (!s || !out) return -1;
+    static_cast<pineforge::BacktestEngine*>(s)->fill_report(
+        reinterpret_cast<pineforge::ReportC*>(out));
+    return 0;
+}
+
 /* Override the chart TZ for ``hour``/``minute``/``dayofweek``/etc. See
  * pineforge.h docstring; NULL or empty are normalised to the legacy UTC
  * fast path. */
diff --git a/src/engine_run.cpp b/src/engine_run.cpp
index 878d4af..752df78 100644
--- a/src/engine_run.cpp
+++ b/src/engine_run.cpp
@@ -124,6 +124,24 @@ void BacktestEngine::reset_run_state() {
     session_isfirstbar_ = false;
     session_islastbar_ = false;
 
+    // A normal run starts a new lifecycle. stream_warmup_mode_ is deliberately
+    // preserved: stream_begin sets it before delegating to run() so historical
+    // bars are not mislabeled as the rightmost realtime bar.
+    stream_phase_ = StreamPhase::IDLE;
+    stream_input_tf_ms_ = 0;
+    stream_next_input_open_ms_ = 0;
+    stream_clock_ms_ = 0;
+    stream_last_tick_ms_ = 0;
+    stream_last_sequence_ = 0;
+    stream_seen_sequence_ = false;
+    stream_has_input_bar_ = false;
+    stream_input_bar_ = Bar{};
+    stream_last_price_ = 0.0;
+    stream_has_last_price_ = false;
+    stream_next_script_bar_index_ = 0;
+    stream_script_bar_had_tick_ = false;
+    stream_script_tick_seen_ = false;
+
     // Native source-series history (input.source(...) ring buffers). Must list
     // EVERY _src_*_ member declared in engine.hpp — a missing one leaks history
     // into a reused handle (see test_handle_reuse_reset all-series coverage).
@@ -190,7 +208,7 @@ void BacktestEngine::run(const Bar* bars, int n) {
         bar_index_ = i;
         is_first_tick_ = true;
         is_last_tick_ = true;
-        barstate_islast_ = (i == n - 1);
+        barstate_islast_ = !stream_warmup_mode_ && (i == n - 1);
         diag_script_bars_processed_++;
         // Reset per-bar pending-close accumulator. Each on_bar call
         // captures fresh ``strategy.close*`` qty for the same-bar
@@ -484,7 +502,7 @@ void BacktestEngine::run_simple_bar_loop(const Bar* input_bars, int n_input) {
         bar_index_ = i;
         is_first_tick_ = true;
         is_last_tick_ = true;
-        barstate_islast_ = (i == n_input - 1);
+        barstate_islast_ = !stream_warmup_mode_ && (i == n_input - 1);
         diag_script_bars_processed_++;
         // Reset per-bar pending-close accumulator. Each on_bar call captures
         // fresh ``strategy.close*`` qty for the same-bar close-then-entry
@@ -578,7 +596,8 @@ void BacktestEngine::run_aggregation_bar_loop(const Bar* input_bars, int n_input
             const int64_t script_bar_ts = ab.bar.timestamp;
             bar_index_ = script_bar_index++;
             emitted_script_bars++;
-            barstate_islast_ = (emitted_script_bars == expected_script_bars);
+            barstate_islast_ = !stream_warmup_mode_
+                && (emitted_script_bars == expected_script_bars);
             diag_script_bars_processed_++;
             // Reset per-bar pending-close accumulator. See run_simple_bar_loop
             // for the regression history; the aggregated path was missing the
diff --git a/src/engine_stream.cpp b/src/engine_stream.cpp
new file mode 100644
index 0000000..5e9bc09
--- /dev/null
+++ b/src/engine_stream.cpp
@@ -0,0 +1,379 @@
+/*
+ * engine_stream.cpp — continuous historical warmup -> realtime trade stream
+ */
+
+#include "engine_internal.hpp"
+
+#include <algorithm>
+#include <cmath>
+#include <limits>
+#include <stdexcept>
+#include <string>
+
+namespace pineforge {
+
+namespace {
+
+Bar price_point(double price, double volume, int64_t timestamp) {
+    return Bar{price, price, price, price, volume, timestamp};
+}
+
+}  // namespace
+
+bool BacktestEngine::stream_begin(const Bar* warmup_bars, int n_warmup,
+                                  const std::string& input_tf,
+                                  const std::string& script_tf) {
+    last_error_.clear();
+    try {
+        if (stream_phase_ == StreamPhase::REALTIME) {
+            throw std::runtime_error("stream is already realtime");
+        }
+        if (warmup_bars == nullptr || n_warmup <= 0) {
+            throw std::runtime_error(
+                "stream warmup requires at least one confirmed OHLCV bar");
+        }
+        const int input_seconds = tf_to_seconds(input_tf);
+        if (input_seconds <= 0) {
+            throw std::runtime_error(
+                "stream input timeframe must have a fixed positive duration: "
+                + input_tf);
+        }
+        for (int i = 1; i < n_warmup; ++i) {
+            if (warmup_bars[i].timestamp <= warmup_bars[i - 1].timestamp) {
+                throw std::runtime_error(
+                    "stream warmup timestamps must be strictly increasing");
+            }
+        }
+        if (!std::isfinite(warmup_bars[n_warmup - 1].close)
+            || warmup_bars[n_warmup - 1].close <= 0.0) {
+            throw std::runtime_error(
+                "stream warmup final close must be finite and positive");
+        }
+
+        // A stream's warmup is historical context, not the rightmost realtime
+        // bar. This keeps barstate.islast false until normalized trades take
+        // over.
+        stream_warmup_mode_ = true;
+        run(warmup_bars, n_warmup, input_tf, script_tf,
+            /*bar_magnifier=*/false, 4, MagnifierDistribution::ENDPOINTS);
+        stream_warmup_mode_ = false;
+        if (!last_error_.empty()) {
+            return false;
+        }
+
+        stream_input_tf_ms_ = static_cast<int64_t>(input_seconds) * 1000;
+        const int64_t last_open = warmup_bars[n_warmup - 1].timestamp;
+        if (last_open > std::numeric_limits<int64_t>::max() - stream_input_tf_ms_) {
+            throw std::runtime_error("stream warmup timestamp overflows next bar open");
+        }
+        stream_next_input_open_ms_ = last_open + stream_input_tf_ms_;
+        stream_clock_ms_ = stream_next_input_open_ms_;
+        stream_last_tick_ms_ = 0;
+        stream_last_sequence_ = 0;
+        stream_seen_sequence_ = false;
+        stream_has_input_bar_ = false;
+        stream_input_bar_ = Bar{};
+        stream_last_price_ = warmup_bars[n_warmup - 1].close;
+        stream_has_last_price_ = true;
+        stream_next_script_bar_index_ =
+            static_cast<int>(diag_script_bars_processed_);
+        stream_script_bar_had_tick_ = false;
+        stream_script_tick_seen_ = false;
+        stream_phase_ = StreamPhase::REALTIME;
+
+        // Exact normalized trades now drive the broker instead of inferred
+        // OHLC paths. Strategy code remains close-only unless codegen opts in
+        // to calc_on_every_tick; resting orders are nevertheless fillable on
+        // each normalized trade, as on TradingView's realtime broker emulator.
+        bar_magnifier_enabled_ = true;
+        bar_index_ = stream_next_script_bar_index_;
+        last_bar_index_ = bar_index_;
+        last_bar_time_ = stream_next_input_open_ms_;
+        barstate_islast_ = true;
+        return true;
+    } catch (const std::exception& e) {
+        stream_warmup_mode_ = false;
+        stream_phase_ = StreamPhase::IDLE;
+        last_error_ = e.what();
+        return false;
+    } catch (...) {
+        stream_warmup_mode_ = false;
+        stream_phase_ = StreamPhase::IDLE;
+        last_error_ = "unknown error during BacktestEngine::stream_begin";
+        return false;
+    }
+}
+
+bool BacktestEngine::stream_push_tick(const TradeTick& tick) {
+    last_error_.clear();
+    try {
+        if (stream_phase_ != StreamPhase::REALTIME) {
+            throw std::runtime_error("stream_push_tick requires a realtime stream");
+        }
+        if (!std::isfinite(tick.price) || tick.price <= 0.0) {
+            throw std::runtime_error("stream tick price must be finite and positive");
+        }
+        if (!std::isfinite(tick.quantity) || tick.quantity < 0.0) {
+            throw std::runtime_error("stream tick quantity must be finite and non-negative");
+        }
+        if (tick.timestamp < stream_clock_ms_) {
+            throw std::runtime_error("stream tick timestamp moved backwards");
+        }
+        if (tick.sequence != 0 && stream_seen_sequence_
+            && tick.sequence <= stream_last_sequence_) {
+            throw std::runtime_error("stream sequence must be strictly increasing");
+        }
+
+        if (!stream_finalize_until(tick.timestamp)) {
+            return false;
+        }
+
+        if (!stream_has_input_bar_) {
+            stream_input_bar_ = price_point(
+                tick.price, tick.quantity, stream_next_input_open_ms_);
+            stream_has_input_bar_ = true;
+        } else {
+            stream_input_bar_.high = std::max(stream_input_bar_.high, tick.price);
+            stream_input_bar_.low = std::min(stream_input_bar_.low, tick.price);
+            stream_input_bar_.close = tick.price;
+            stream_input_bar_.volume += tick.quantity;
+        }
+
+        stream_last_price_ = tick.price;
+        stream_has_last_price_ = true;
+        stream_last_tick_ms_ = tick.timestamp;
+        stream_clock_ms_ = tick.timestamp;
+        if (tick.sequence != 0) {
+            stream_last_sequence_ = tick.sequence;
+            stream_seen_sequence_ = true;
+        }
+
+        // Broker-only tick pass. Pine strategy code stays on its default
+        // close-only cadence, but orders created on the preceding close fill
+        // at the first actual source record and priced orders see the exact
+        // trade path rather than an inferred OHLC traversal.
+        current_bar_ = price_point(tick.price, tick.quantity, tick.timestamp);
+        bar_index_ = stream_next_script_bar_index_;
+        last_bar_index_ = bar_index_;
+        last_bar_time_ = tick.timestamp;
+        barstate_islast_ = true;
+        is_first_tick_ = !stream_script_tick_seen_;
+        is_last_tick_ = false;
+        // The overwhelming majority of source records arrive while many
+        // strategies are flat and have no order in the broker. Such a print
+        // still contributes to the forming OHLCV bar above, but there is no
+        // broker, excursion, or margin state it can possibly mutate. Avoiding
+        // the full order-sort/risk pass here is exact, not an approximation,
+        // and makes long shared-feed corpus replays tractable.
+        if (!pending_orders_.empty() || position_side_ != PositionSide::FLAT) {
+            if (!pending_orders_.empty()) {
+                process_pending_orders(current_bar_);
+            }
+            update_per_trade_extremes();
+            const std::size_t trades_before_mc = trades_.size();
+            process_margin_call(current_bar_);
+            if (trades_.size() != trades_before_mc) {
+                refresh_frozen_default_sizing_after_margin_call();
+            }
+        }
+        stream_script_tick_seen_ = true;
+        return true;
+    } catch (const std::exception& e) {
+        last_error_ = e.what();
+        return false;
+    } catch (...) {
+        last_error_ = "unknown error during BacktestEngine::stream_push_tick";
+        return false;
+    }
+}
+
+bool BacktestEngine::stream_push_ticks(const TradeTick* ticks, int n) {
+    last_error_.clear();
+    if (n < 0 || (n > 0 && ticks == nullptr)) {
+        last_error_ = "stream_push_ticks received an invalid tick array";
+        return false;
+    }
+    for (int i = 0; i < n; ++i) {
+        if (!stream_push_tick(ticks[i])) return false;
+    }
+    return true;
+}
+
+bool BacktestEngine::stream_advance_time(int64_t timestamp_ms) {
+    last_error_.clear();
+    try {
+        if (stream_phase_ != StreamPhase::REALTIME) {
+            throw std::runtime_error(
+                "stream_advance_time requires a realtime stream");
+        }
+        if (timestamp_ms < stream_clock_ms_) {
+            throw std::runtime_error("stream clock moved backwards");
+        }
+        if (!stream_finalize_until(timestamp_ms)) return false;
+        stream_clock_ms_ = timestamp_ms;
+        return true;
+    } catch (const std::exception& e) {
+        last_error_ = e.what();
+        return false;
+    } catch (...) {
+        last_error_ = "unknown error during BacktestEngine::stream_advance_time";
+        return false;
+    }
+}
+
+bool BacktestEngine::stream_end(bool finalize_partial_input_bar) {
+    last_error_.clear();
+    try {
+        if (stream_phase_ != StreamPhase::REALTIME) {
+            throw std::runtime_error("stream_end requires a realtime stream");
+        }
+        if (finalize_partial_input_bar && stream_has_input_bar_) {
+            stream_feed_input_bar(stream_input_bar_, /*had_tick=*/true);
+            stream_has_input_bar_ = false;
+            stream_next_input_open_ms_ += stream_input_tf_ms_;
+        }
+        stream_phase_ = StreamPhase::ENDED;
+        return true;
+    } catch (const std::exception& e) {
+        last_error_ = e.what();
+        return false;
+    } catch (...) {
+        last_error_ = "unknown error during BacktestEngine::stream_end";
+        return false;
+    }
+}
+
+bool BacktestEngine::stream_finalize_until(int64_t timestamp_ms) {
+    while (timestamp_ms >= stream_next_input_open_ms_ + stream_input_tf_ms_) {
+        const bool had_tick = stream_has_input_bar_;
+        const bool in_session = pine_session_ismarket(
+            syminfo_.session, syminfo_.timezone,
+            stream_next_input_open_ms_);
+
+        // A normalized provider may jump from one market session to the next.
+        // Do not turn the closed interval into synthetic tradable bars. A real
+        // source record is still honored even if the configured metadata is
+        // imperfect, so provider data remains authoritative.
+        if (!had_tick && !in_session) {
+            stream_input_bar_ = Bar{};
+            stream_next_input_open_ms_ += stream_input_tf_ms_;
+            continue;
+        }
+
+        Bar completed;
+        if (had_tick) {
+            completed = stream_input_bar_;
+        } else {
+            if (!stream_has_last_price_) {
+                last_error_ = "stream cannot synthesize a gap before any price";
+                return false;
+            }
+            completed = price_point(
+                stream_last_price_, 0.0, stream_next_input_open_ms_);
+        }
+
+        stream_feed_input_bar(completed, had_tick);
+        stream_has_input_bar_ = false;
+        stream_input_bar_ = Bar{};
+        stream_next_input_open_ms_ += stream_input_tf_ms_;
+    }
+    return true;
+}
+
+void BacktestEngine::stream_feed_input_bar(const Bar& bar, bool had_tick) {
+    ++diag_input_bars_processed_;
+    last_bar_time_ = bar.timestamp;
+
+    for (auto& state : security_eval_states_) {
+        feed_security_eval_state(state, bar);
+    }
+
+    if (!diag_needs_aggregation_) {
+        stream_dispatch_script_bar(bar, had_tick);
+        return;
+    }
+
+    AggregatedBar ab = script_tf_agg_.feed(bar);
+    const bool completed_on_boundary = ab.is_complete
+        && tf_change(ab.bar.timestamp, bar.timestamp, script_tf_);
+    if (completed_on_boundary) {
+        // The current input bar opened the next bucket; the aggregator emitted
+        // the preceding partial bucket before retaining this bar as its new
+        // current state.
+        stream_dispatch_script_bar(ab.bar, stream_script_bar_had_tick_);
+        stream_script_bar_had_tick_ = had_tick;
+    } else {
+        stream_script_bar_had_tick_ = stream_script_bar_had_tick_ || had_tick;
+        if (ab.is_complete) {
+            stream_dispatch_script_bar(ab.bar, stream_script_bar_had_tick_);
+            stream_script_bar_had_tick_ = false;
+        }
+    }
+}
+
+void BacktestEngine::stream_dispatch_script_bar(const Bar& bar, bool had_tick) {
+    const int this_bar_index = stream_next_script_bar_index_++;
+    bar_index_ = this_bar_index;
+    last_bar_index_ = this_bar_index;
+    last_bar_time_ = bar.timestamp;
+    barstate_islast_ = true;
+    is_first_tick_ = true;
+    is_last_tick_ = true;
+    ++diag_script_bars_processed_;
+    pending_close_qty_in_bar_ = 0.0;
+
+    // A synthesized zero-volume interval has no raw broker pass. Give resting
+    // market orders one carried-price point at its open so time advancement is
+    // deterministic even through quiet in-session intervals.
+    if (!had_tick) {
+        current_bar_ = price_point(bar.open, 0.0, bar.timestamp);
+        process_pending_orders(current_bar_);
+        update_per_trade_extremes();
+        const std::size_t trades_before_mc = trades_.size();
+        process_margin_call(current_bar_);
+        if (trades_.size() != trades_before_mc) {
+            refresh_frozen_default_sizing_after_margin_call();
+        }
+    }
+
+    current_bar_ = bar;
+    session_ismarket_ = pine_session_ismarket(
+        syminfo_.session, syminfo_.timezone, current_bar_.timestamp);
+    session_isfirstbar_ = session_ismarket_ && !prev_in_session_;
+    if (session_ismarket_ && script_tf_seconds_ > 0) {
+        const int64_t next_ts = current_bar_.timestamp
+            + static_cast<int64_t>(script_tf_seconds_) * 1000;
+        session_islastbar_ = !pine_session_ismarket(
+            syminfo_.session, syminfo_.timezone, next_ts);
+    } else {
+        session_islastbar_ = false;
+    }
+
+    _push_source_series();
+    on_bar(current_bar_);
+    if (process_orders_on_close_) {
+        flush_same_bar_close();
+        // New close-time orders only get the closing price point. Re-walking
+        // the full OHLC range would let a just-created order see prices that
+        // occurred before it existed.
+        const Bar completed_bar = current_bar_;
+        current_bar_ = price_point(
+            completed_bar.close, 0.0, completed_bar.timestamp);
+        process_pending_orders(current_bar_);
+        current_bar_ = completed_bar;
+    }
+
+    finalize_bar();
+    prev_in_session_ = session_ismarket_;
+    update_equity_extremes();
+    record_equity_point(bar.timestamp);
+    prev_bar_timestamp_ = bar.timestamp;
+
+    // Ticks belonging to the next script bar must compare pending-order
+    // created_bar values against the next index before that bar closes.
+    bar_index_ = stream_next_script_bar_index_;
+    last_bar_index_ = bar_index_;
+    stream_script_tick_seen_ = false;
+}
+
+}  // namespace pineforge
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 5eea755..20990f2 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -81,6 +81,7 @@ set(TEST_SOURCES
     test_metrics
     test_drawing
     test_margin_call
+    test_streaming
 )
 
 find_package(Threads REQUIRED)
diff --git a/tests/test_c_abi.c b/tests/test_c_abi.c
index 8813b8f..94d4f0b 100644
--- a/tests/test_c_abi.c
+++ b/tests/test_c_abi.c
@@ -65,6 +65,16 @@ int main(void) {
     CHECK(bar.open      == 100.0,           "bar.open roundtrip");
     CHECK(bar.close     == 103.0,           "bar.close roundtrip");
 
+    /* ── Realtime trade tick field access ───────────────────────── */
+    pf_trade_tick_t tick;
+    memset(&tick, 0, sizeof(tick));
+    tick.timestamp = 1700000000123LL;
+    tick.sequence = 987654321ULL;
+    tick.price = 103.25;
+    tick.quantity = 0.125;
+    CHECK(tick.sequence == 987654321ULL, "tick.sequence roundtrip");
+    CHECK(tick.quantity == 0.125,        "tick.quantity roundtrip");
+
     /* ── Trade ───────────────────────────────────────────────────── */
     pf_trade_t trade;
     memset(&trade, 0, sizeof(trade));
@@ -94,6 +104,8 @@ int main(void) {
      *     struct membership) ─────────────────────────────────────── */
     CHECK(sizeof(pf_bar_t)            >= 6 * sizeof(double),
           "pf_bar_t too small for OHLCV + timestamp");
+    CHECK(sizeof(pf_trade_tick_t)     == 32,
+          "pf_trade_tick_t must remain a four-field 32-byte POD");
     CHECK(sizeof(pf_trade_t)          >= 80,
           "pf_trade_t unexpectedly small");
     CHECK(sizeof(pf_report_t)         >= 80 + sizeof(pf_metrics_t)
diff --git a/tests/test_c_abi_setters.cpp b/tests/test_c_abi_setters.cpp
index d294d11..1f45964 100644
--- a/tests/test_c_abi_setters.cpp
+++ b/tests/test_c_abi_setters.cpp
@@ -97,6 +97,12 @@ int main() {
     strategy_set_syminfo_mintick(nullptr, 0.25);
     strategy_set_syminfo_pointvalue(nullptr, 50.0);
     strategy_set_syminfo_metadata(nullptr, "shares_outstanding_total", 1.0);
+    CHECK(strategy_stream_begin(nullptr, nullptr, 0, "1", "1") == -1);
+    CHECK(strategy_stream_push_tick(nullptr, nullptr) == -1);
+    CHECK(strategy_stream_push_ticks(nullptr, nullptr, 0) == -1);
+    CHECK(strategy_stream_advance_time(nullptr, 0) == -1);
+    CHECK(strategy_stream_end(nullptr, 0) == -1);
+    CHECK(strategy_stream_fill_report(nullptr, nullptr) == -1);
     CHECK(strategy_get_last_error(nullptr) == nullptr);
 
     // Secondary `|| !arg` guard arms (valid handle, NULL arg) — these
@@ -175,6 +181,32 @@ int main() {
     // A key that was never injected still reports na.
     CHECK(pineforge::is_na(eng.meta("never_injected")));
 
+    // ── Historical -> realtime lifecycle wrappers ─────────────────
+    ProbeEngine stream_eng;
+    pf_strategy_t sh = static_cast<pf_strategy_t>(&stream_eng);
+    pf_bar_t warmup{};
+    warmup.open = warmup.high = warmup.low = warmup.close = 100.0;
+    warmup.volume = 2.0;
+    warmup.timestamp = 0;
+    CHECK(strategy_stream_begin(sh, &warmup, 1, "1", "1") == 0);
+
+    pf_trade_tick_t tick{};
+    tick.timestamp = 60010;
+    tick.sequence = 7;
+    tick.price = 101.0;
+    tick.quantity = 0.5;
+    CHECK(strategy_stream_push_tick(sh, &tick) == 0);
+    CHECK(strategy_stream_push_ticks(sh, nullptr, 0) == 0);
+    CHECK(strategy_stream_advance_time(sh, 120000) == 0);
+
+    pf_report_t report{};
+    CHECK(strategy_stream_fill_report(sh, &report) == 0);
+    CHECK(report.input_bars_processed == 2);
+    CHECK(report.script_bars_processed == 2);
+    pineforge::BacktestEngine::free_report(
+        reinterpret_cast<pineforge::ReportC*>(&report));
+    CHECK(strategy_stream_end(sh, 0) == 0);
+
     if (g_fail == 0) {
         std::printf("test_c_abi_setters: OK (pineforge %s)\n", vs);
         return 0;
diff --git a/tests/test_streaming.cpp b/tests/test_streaming.cpp
new file mode 100644
index 0000000..2c0ffc7
--- /dev/null
+++ b/tests/test_streaming.cpp
@@ -0,0 +1,222 @@
+#include <pineforge/bar.hpp>
+#include <pineforge/engine.hpp>
+
+#include <cmath>
+#include <cstdio>
+#include <string>
+#include <vector>
+
+using namespace pineforge;
+
+namespace {
+
+int failures = 0;
+
+#define CHECK(cond)                                                            \
+    do {                                                                       \
+        if (!(cond)) {                                                         \
+            std::fprintf(stderr, "FAIL %s:%d  %s\n", __FILE__, __LINE__,     \
+                         #cond);                                               \
+            ++failures;                                                        \
+        }                                                                      \
+    } while (0)
+
+bool near(double a, double b, double eps = 1e-9) {
+    return std::fabs(a - b) <= eps;
+}
+
+Bar flat_bar(double price, int64_t timestamp, double volume = 1.0) {
+    return Bar{price, price, price, price, volume, timestamp};
+}
+
+TradeTick tick(int64_t timestamp, uint64_t id, double price,
+               double qty = 1.0) {
+    return TradeTick{timestamp, id, price, qty};
+}
+
+class ContinuityStrategy final : public BacktestEngine {
+public:
+    std::vector<bool> saw_islast;
+
+    void on_bar(const Bar&) override {
+        saw_islast.push_back(barstate_islast_);
+        if (bar_index_ == 0) strategy_entry("L", true);
+        if (bar_index_ == 1) strategy_close_all();
+    }
+
+    double position_size() const { return signed_position_size(); }
+    std::size_t pending_count() const { return pending_orders_.size(); }
+};
+
+class StopStrategy final : public BacktestEngine {
+public:
+    void on_bar(const Bar&) override {
+        if (bar_index_ == 1) {
+            strategy_entry("stop", true, na<double>(), 103.0);
+        }
+    }
+
+    double entry_price() const { return position_entry_price_; }
+    int64_t entry_time() const { return position_entry_time_; }
+    double position_size() const { return signed_position_size(); }
+};
+
+class CaptureStrategy final : public BacktestEngine {
+public:
+    std::vector<Bar> bars;
+    std::vector<int> indices;
+
+    void on_bar(const Bar& bar) override {
+        bars.push_back(bar);
+        indices.push_back(bar_index_);
+    }
+};
+
+void test_position_pending_order_and_equity_continue() {
+    ContinuityStrategy strategy;
+    const Bar warmup[] = {
+        flat_bar(100.0, 0),
+        flat_bar(101.0, 60'000),
+    };
+
+    CHECK(strategy.stream_begin(warmup, 2, "1", "1"));
+    CHECK(strategy.last_error().empty());
+    CHECK(strategy.stream_is_realtime());
+    CHECK(near(strategy.position_size(), 1.0));
+    CHECK(strategy.pending_count() == 1);
+    CHECK(strategy.trade_count() == 0);
+    CHECK(strategy.saw_islast.size() == 2);
+    CHECK(!strategy.saw_islast[0]);
+    CHECK(!strategy.saw_islast[1]);
+
+    // The close order created on the final historical bar fills at the first
+    // normalized source record. A second run() would have erased both the open
+    // lot and this pending order, so this is the core lifecycle regression test.
+    CHECK(strategy.stream_push_tick(tick(120'123, 1, 110.0, 0.25)));
+    CHECK(strategy.trade_count() == 1);
+    CHECK(near(strategy.position_size(), 0.0));
+    const Trade& trade = strategy.get_trade(0);
+    CHECK(near(trade.entry_price, 101.0));
+    CHECK(near(trade.exit_price, 110.0));
+    CHECK(trade.entry_time == 60'000);
+    CHECK(trade.exit_time == 120'123);
+    CHECK(trade.entry_bar_index == 1);
+    CHECK(trade.exit_bar_index == 2);
+    CHECK(near(trade.pnl, 9.0));
+
+    CHECK(strategy.stream_advance_time(180'000));
+    CHECK(strategy.saw_islast.size() == 3);
+    CHECK(strategy.saw_islast.back());
+
+    ReportC report{};
+    strategy.fill_report(&report);
+    CHECK(report.input_bars_processed == 3);
+    CHECK(report.script_bars_processed == 3);
+    CHECK(report.total_trades == 1);
+    CHECK(near(report.net_profit, 9.0));
+    BacktestEngine::free_report(&report);
+    CHECK(strategy.stream_end(false));
+}
+
+void test_raw_tick_gap_fill_uses_observed_price_and_time() {
+    StopStrategy strategy;
+    const Bar warmup[] = {
+        flat_bar(100.0, 0),
+        flat_bar(100.0, 60'000),
+    };
+    CHECK(strategy.stream_begin(warmup, 2, "1", "1"));
+    CHECK(strategy.stream_push_tick(tick(120'010, 10, 100.0)));
+    CHECK(near(strategy.position_size(), 0.0));
+
+    // No synthetic interpolation from 100 to 105: the first observed print
+    // beyond the 103 stop is 105, so a stop-market order gaps to 105.
+    CHECK(strategy.stream_push_tick(tick(120'250, 11, 105.0)));
+    CHECK(near(strategy.position_size(), 1.0));
+    CHECK(near(strategy.entry_price(), 105.0));
+    CHECK(strategy.entry_time() == 120'250);
+}
+
+void test_partial_mtf_aggregator_survives_handoff() {
+    CaptureStrategy strategy;
+    std::vector<Bar> warmup;
+    for (int i = 0; i < 7; ++i) {
+        warmup.push_back(flat_bar(static_cast<double>(i), i * 60'000LL));
+    }
+
+    CHECK(strategy.stream_begin(
+        warmup.data(), static_cast<int>(warmup.size()), "1", "5"));
+    CHECK(strategy.bars.size() == 1);
+    CHECK(strategy.indices.size() == 1 && strategy.indices[0] == 0);
+    CHECK(near(strategy.bars[0].open, 0.0));
+    CHECK(near(strategy.bars[0].close, 4.0));
+
+    CHECK(strategy.stream_push_tick(tick(420'000, 20, 7.0)));
+    CHECK(strategy.stream_push_tick(tick(480'000, 21, 8.0)));
+    CHECK(strategy.stream_push_tick(tick(540'000, 22, 9.0)));
+    CHECK(strategy.stream_advance_time(600'000));
+
+    CHECK(strategy.bars.size() == 2);
+    CHECK(strategy.indices[1] == 1);
+    // Minutes 5 and 6 came from historical OHLCV; 7, 8 and 9 came from raw
+    // ticks. One 5-minute candle must span both sources without a reset.
+    CHECK(strategy.bars[1].timestamp == 300'000);
+    CHECK(near(strategy.bars[1].open, 5.0));
+    CHECK(near(strategy.bars[1].close, 9.0));
+    CHECK(near(strategy.bars[1].volume, 5.0));
+}
+
+void test_clock_materializes_quiet_bars() {
+    CaptureStrategy strategy;
+    const Bar warmup[] = {flat_bar(42.0, 0, 3.0)};
+    CHECK(strategy.stream_begin(warmup, 1, "1", "1"));
+    CHECK(strategy.stream_advance_time(240'000));
+
+    CHECK(strategy.bars.size() == 4);
+    for (std::size_t i = 1; i < strategy.bars.size(); ++i) {
+        CHECK(near(strategy.bars[i].open, 42.0));
+        CHECK(near(strategy.bars[i].close, 42.0));
+        CHECK(near(strategy.bars[i].volume, 0.0));
+    }
+}
+
+void test_clock_skips_out_of_session_intervals() {
+    CaptureStrategy strategy;
+    strategy.set_syminfo_timezone("UTC");
+    strategy.set_syminfo_session("0000-0001");
+    const Bar warmup[] = {flat_bar(42.0, 0, 3.0)};
+    CHECK(strategy.stream_begin(warmup, 1, "1", "1"));
+    CHECK(strategy.stream_advance_time(240'000));
+
+    // Minute zero is the configured session. Minutes one through three are
+    // closed and must not become synthetic tradable bars.
+    CHECK(strategy.bars.size() == 1);
+}
+
+void test_rejects_replayed_or_out_of_order_ticks() {
+    CaptureStrategy strategy;
+    const Bar warmup[] = {flat_bar(100.0, 0)};
+    CHECK(strategy.stream_begin(warmup, 1, "1", "1"));
+    CHECK(strategy.stream_push_tick(tick(60'100, 100, 100.0)));
+    CHECK(!strategy.stream_push_tick(tick(60'200, 100, 101.0)));
+    CHECK(strategy.last_error().find("sequence") != std::string::npos);
+    CHECK(!strategy.stream_push_tick(tick(60'050, 101, 101.0)));
+    CHECK(strategy.last_error().find("backwards") != std::string::npos);
+}
+
+}  // namespace
+
+int main() {
+    test_position_pending_order_and_equity_continue();
+    test_raw_tick_gap_fill_uses_observed_price_and_time();
+    test_partial_mtf_aggregator_survives_handoff();
+    test_clock_materializes_quiet_bars();
+    test_clock_skips_out_of_session_intervals();
+    test_rejects_replayed_or_out_of_order_ticks();
+
+    if (failures == 0) {
+        std::puts("test_streaming: OK");
+        return 0;
+    }
+    std::fprintf(stderr, "test_streaming: %d failures\n", failures);
+    return 1;
+}
diff --git a/tutorial/CMakeLists.txt b/tutorial/CMakeLists.txt
index de34c23..6cdc5ae 100644
--- a/tutorial/CMakeLists.txt
+++ b/tutorial/CMakeLists.txt
@@ -67,3 +67,46 @@ pineforge_tutorial_strategy(strategy_tutorial_mtf_ltf
     strategy_ltf)
 message(STATUS "tutorial: configured strategy_tutorial_mtf_htf → tutorial/mtf/strategy_htf.so")
 message(STATUS "tutorial: configured strategy_tutorial_mtf_ltf → tutorial/mtf/strategy_ltf.so")
+
+# Keep the public streaming tutorial executable in the normal CTest lane so
+# its ctypes mirror and lifecycle calls cannot silently drift from the ABI.
+if(PINEFORGE_BUILD_TESTS)
+    find_package(Python3 COMPONENTS Interpreter QUIET)
+    if(Python3_Interpreter_FOUND)
+        add_test(NAME test_tutorial_stream
+            COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/run_stream.py)
+        set_tests_properties(test_tutorial_stream PROPERTIES
+            WORKING_DIRECTORY ${CMAKE_SOURCE_DIR})
+        if(PINEFORGE_ENABLE_SANITIZERS AND UNIX)
+            # The strategy DSO is ASan-instrumented, but the host Python
+            # interpreter is not. Preload the compiler's ASan runtime so its
+            # interceptors initialize before ctypes loads the strategy.
+            if(APPLE)
+                set(_pf_asan_runtime_name
+                    "libclang_rt.asan_osx_dynamic.dylib")
+                set(_pf_asan_preload_var "DYLD_INSERT_LIBRARIES")
+            else()
+                set(_pf_asan_runtime_name "libasan.so")
+                set(_pf_asan_preload_var "LD_PRELOAD")
+            endif()
+            execute_process(
+                COMMAND ${CMAKE_CXX_COMPILER}
+                        -print-file-name=${_pf_asan_runtime_name}
+                OUTPUT_VARIABLE _pf_asan_runtime
+                OUTPUT_STRIP_TRAILING_WHITESPACE)
+            if(IS_ABSOLUTE "${_pf_asan_runtime}"
+                    AND EXISTS "${_pf_asan_runtime}")
+                # Python keeps process-lifetime allocator state that LSan
+                # reports when its runtime is preloaded. Native C++ tests keep
+                # detect_leaks enabled; disable it only for this Python host.
+                set_tests_properties(test_tutorial_stream PROPERTIES
+                    ENVIRONMENT
+                        "${_pf_asan_preload_var}=${_pf_asan_runtime};ASAN_OPTIONS=detect_leaks=0:halt_on_error=1:abort_on_error=1")
+            else()
+                message(WARNING
+                    "ASan runtime not found for test_tutorial_stream: "
+                    "${_pf_asan_runtime}")
+            endif()
+        endif()
+    endif()
+endif()
diff --git a/tutorial/README.md b/tutorial/README.md
index 1bfc47a..e49e00b 100644
--- a/tutorial/README.md
+++ b/tutorial/README.md
@@ -16,6 +16,7 @@ tutorial/
 │   ├── btcusdt_15m_7d.csv  672 frozen bars (Binance)
 │   └── fetch_btcusdt.py    refresh from Binance public API
 ├── run.py                  ctypes harness + stats
+├── run_stream.py           OHLCV warmup → contiguous trade-stream replay
 ├── run_advanced.py         parameter sweep using ABI overrides
 ├── run_mtf.py              MTF demo — script_tf switch + lower_tf
 ├── run.sh                  one-shot: cmake build + run.py
@@ -44,6 +45,25 @@ MACD(12,26,9) on BTCUSDT 15m — 672 bars, 2026-04-29 18:15 → 2026-05-06 18:00
 
 Numbers depend on the OHLCV snapshot.
 
+## Historical warmup → realtime trades
+
+After `bash tutorial/run.sh` has built the MACD strategy, run:
+
+```bash
+python3 tutorial/run_stream.py
+```
+
+The script warms the strategy on the first 640 confirmed candles, converts the
+remaining 32 candles into 128 ordered trade ticks, and sends that entire tail
+through one `strategy_stream_push_ticks(...)` call. Positions, equity, pending
+orders, Pine/TA history, and partially aggregated timeframe state remain on the
+same strategy instance across the handoff.
+
+The frozen tutorial dataset contains OHLCV rather than raw trades, so this
+example expands each live candle into a deterministic OHLC trade path. Replace
+that expansion with your exchange feed in production. The public API and state
+lifecycle remain the same.
+
 ## Path B — Docker (no local toolchain)
 
 Mount the strategy + OHLCV into the published runtime image; get a
diff --git a/tutorial/run_stream.py b/tutorial/run_stream.py
new file mode 100644
index 0000000..f4908a6
--- /dev/null
+++ b/tutorial/run_stream.py
@@ -0,0 +1,148 @@
+#!/usr/bin/env python3
+"""Historical OHLCV warmup -> realtime trade-stream tutorial.
+
+The frozen tutorial dataset contains bars rather than normalized trades. To keep
+this example self-contained, the final LIVE_BARS candles are expanded into a
+deterministic open/high/low/close trade path. Production callers should
+normalize their provider's ordered trade feed into the same ABI.
+"""
+from __future__ import annotations
+
+import csv
+import ctypes
+import sys
+from datetime import datetime, timezone
+
+from run import BarC, OHLCV, ReportC, SO, check_abi
+
+LIVE_BARS = 32
+INPUT_TF_MS = 15 * 60 * 1000
+
+
+class TradeTickC(ctypes.Structure):
+    """ctypes mirror of pf_trade_tick_t."""
+
+    _fields_ = [
+        ("timestamp", ctypes.c_int64),
+        ("sequence", ctypes.c_uint64),
+        ("price", ctypes.c_double),
+        ("quantity", ctypes.c_double),
+    ]
+
+
+def error_message(lib: ctypes.CDLL, state: int) -> str:
+    raw = lib.strategy_get_last_error(state)
+    return raw.decode("utf-8", "replace") if raw else "unknown engine error"
+
+
+def check_call(lib: ctypes.CDLL, state: int, status: int, operation: str) -> None:
+    if status != 0:
+        raise RuntimeError(f"{operation}: {error_message(lib, state)}")
+
+
+def price_path(bar: BarC) -> tuple[float, float, float, float]:
+    """Choose the same nearest-extreme-first convention as an OHLC emulator."""
+    if abs(bar.open - bar.high) < abs(bar.open - bar.low):
+        return bar.open, bar.high, bar.low, bar.close
+    return bar.open, bar.low, bar.high, bar.close
+
+
+def main() -> int:
+    if not SO.exists():
+        sys.exit("strategy.so missing — run `bash tutorial/run.sh` first")
+
+    with OHLCV.open(newline="") as handle:
+        rows = list(csv.DictReader(handle))
+    if len(rows) <= LIVE_BARS:
+        sys.exit(f"need more than {LIVE_BARS} tutorial bars")
+
+    bars = (BarC * len(rows))()
+    for i, row in enumerate(rows):
+        bars[i] = BarC(
+            float(row["open"]), float(row["high"]), float(row["low"]),
+            float(row["close"]), float(row["volume"]), int(row["timestamp"]))
+
+    warmup_n = len(rows) - LIVE_BARS
+    tick_count = LIVE_BARS * 4
+    ticks = (TradeTickC * tick_count)()
+    offsets = (0, INPUT_TF_MS // 3, 2 * INPUT_TF_MS // 3,
+               INPUT_TF_MS - 1)
+    tick_index = 0
+    sequence = 1
+    for bar in bars[warmup_n:]:
+        for offset, price in zip(offsets, price_path(bar)):
+            ticks[tick_index] = TradeTickC(
+                bar.timestamp + offset, sequence, price,
+                bar.volume / 4.0)
+            tick_index += 1
+            sequence += 1
+
+    lib = ctypes.CDLL(str(SO))
+    check_abi(lib)
+    lib.strategy_create.argtypes = [ctypes.c_char_p]
+    lib.strategy_create.restype = ctypes.c_void_p
+    lib.strategy_free.argtypes = [ctypes.c_void_p]
+    lib.report_free.argtypes = [ctypes.POINTER(ReportC)]
+    lib.strategy_get_last_error.argtypes = [ctypes.c_void_p]
+    lib.strategy_get_last_error.restype = ctypes.c_char_p
+    lib.strategy_stream_begin.argtypes = [
+        ctypes.c_void_p, ctypes.POINTER(BarC), ctypes.c_int,
+        ctypes.c_char_p, ctypes.c_char_p]
+    lib.strategy_stream_begin.restype = ctypes.c_int
+    lib.strategy_stream_push_ticks.argtypes = [
+        ctypes.c_void_p, ctypes.POINTER(TradeTickC), ctypes.c_int]
+    lib.strategy_stream_push_ticks.restype = ctypes.c_int
+    lib.strategy_stream_advance_time.argtypes = [
+        ctypes.c_void_p, ctypes.c_int64]
+    lib.strategy_stream_advance_time.restype = ctypes.c_int
+    lib.strategy_stream_end.argtypes = [ctypes.c_void_p, ctypes.c_int]
+    lib.strategy_stream_end.restype = ctypes.c_int
+    lib.strategy_stream_fill_report.argtypes = [
+        ctypes.c_void_p, ctypes.POINTER(ReportC)]
+    lib.strategy_stream_fill_report.restype = ctypes.c_int
+
+    state = lib.strategy_create(b"{}")
+    if not state:
+        sys.exit("strategy_create failed")
+    report = ReportC()
+    try:
+        check_call(lib, state, lib.strategy_stream_begin(
+            state, bars, warmup_n, b"15", b"15"), "stream begin")
+
+        # One contiguous call covers the complete simulated live session.
+        check_call(lib, state, lib.strategy_stream_push_ticks(
+            state, ticks, tick_count), "stream ticks")
+
+        session_end = bars[-1].timestamp + INPUT_TF_MS
+        check_call(lib, state, lib.strategy_stream_advance_time(
+            state, session_end), "stream advance")
+        check_call(lib, state, lib.strategy_stream_end(
+            state, 0), "stream end")
+        check_call(lib, state, lib.strategy_stream_fill_report(
+            state, ctypes.byref(report)), "stream report")
+
+        if report.input_bars_processed != len(rows):
+            raise RuntimeError(
+                f"bar continuity failed: {report.input_bars_processed} != {len(rows)}")
+
+        fmt = lambda ms: datetime.fromtimestamp(
+            ms / 1000, tz=timezone.utc).strftime("%Y-%m-%d %H:%M")
+        print("MACD historical -> realtime stream")
+        print(f"  warmup:     {warmup_n} confirmed 15m bars")
+        print(f"  realtime:   {LIVE_BARS} bars from {tick_count} ordered trades")
+        print(f"  handoff:    {fmt(bars[warmup_n].timestamp)} UTC")
+        print(f"  processed:  {report.input_bars_processed} input / "
+              f"{report.script_bars_processed} script bars")
+        print(f"  trades:     {report.trades_len}")
+        print(f"  net pnl:    {report.net_profit:+.2f}")
+    except RuntimeError as exc:
+        print(f"stream error: {exc}", file=sys.stderr)
+        return 1
+    finally:
+        lib.report_free(ctypes.byref(report))
+        lib.strategy_free(state)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())