diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 638ae3f..0ce5e19 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -11,4 +11,8 @@ jobs:
           cache: pip
       - run: pip install -e ".[dev]"
       - run: python -m unittest discover -s tests -v
-      - run: ruff check engine tests run.py run_cross_sectional.py
+      - run: ruff check engine tests run.py run_cross_sectional.py run_offline.py run_experiments.py benchmark.py
+      - run: python run_offline.py > /tmp/offline.txt
+      - run: diff -u fixtures/synthetic/expected.txt /tmp/offline.txt
+      - run: python run_experiments.py --db /tmp/experiments.sqlite demo
+      - run: python benchmark.py --symbols 3 --bars 90 --repeats 1
diff --git a/.gitignore b/.gitignore
index b661980..36752c3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,3 +11,7 @@ data/*.png
 data/**/*.csv
 data/**/*.png
 !data/.gitkeep
+
+.venv-runtime/
+*.sqlite
+*.sqlite-*
diff --git a/README.md b/README.md
index 0319aa0..62d207a 100644
--- a/README.md
+++ b/README.md
@@ -30,6 +30,40 @@ pip install -e ".[dev]"
 python -m unittest discover -s tests -v
 ```
 
+## Offline accounting and stored replay
+
+After installation, run from this repository's root:
+
+```bash
+python run_offline.py
+python run_experiments.py --db experiments.sqlite demo
+```
+
+The first command prints a versioned synthetic OHLC fixture's expected ledgers
+and timeline. The second stores four cost scenarios in SQLite and verifies exact
+replay from stored configuration and prices. No credentials or network are used.
+
+```text
+Jan 1 close: signal A at 10.00; B rejected for insufficient cash
+Jan 2 open:  buy 8 A at 12.012; commission 1.00; cash 2.904
+Jan 2 close: 8 A marked at 13.00; equity 106.904
+Jan 3 open:  exit A; final cash and equity 113.792
+```
+
+Today's close signal cannot receive today's price. The invented example is an
+accounting demonstration, not a historical trading result. See the
+[fixture and regeneration command](fixtures/synthetic/README.md),
+[storage, architecture and data policies](docs/EXPERIMENTS.md), and
+[measured benchmark](docs/PERFORMANCE.md).
+
+`compare RUN_ID_A RUN_ID_B`, `costs RUN_ID`, `replay RUN_ID` and `plan RUN_ID`
+are subcommands of `run_experiments.py`. Comparison rejects mismatched data,
+timing or costs; the separate cost report varies only slippage. The original
+`Backtest.run()` ledger API is preserved. Strict calendar validation is available;
+the default retains the documented common-calendar policy.
+
+## Historical examples
+
 Run the SPY example (this step downloads market data):
 
 ```bash
diff --git a/benchmark.py b/benchmark.py
new file mode 100644
index 0000000..d4e4845
--- /dev/null
+++ b/benchmark.py
@@ -0,0 +1,166 @@
+"""Seeded offline benchmark; correctness checks run before any measurements."""
+
+import argparse
+import cProfile
+import json
+import os
+import platform
+import resource
+import statistics
+import subprocess
+import tempfile
+import time
+import tracemalloc
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+from engine.backtest import Backtest
+from engine.reproducibility import digest, ledger_records, versions
+from engine.storage import SQLiteStore, snapshot
+from engine.strategy import MovingAverageCross
+
+
+def generate_data(symbols, bars, seed):
+    if symbols < 1 or bars < 31:
+        raise ValueError("benchmark requires at least one symbol and 31 bars")
+    rng = np.random.default_rng(seed)
+    dates = pd.bdate_range("2000-01-03", periods=bars)
+    closes = 100 * np.exp(np.cumsum(rng.normal(0, 0.01, (bars, symbols)), axis=0))
+    opens = np.vstack([closes[0], closes[:-1]]) * np.exp(rng.normal(0, 0.002, (bars, symbols)))
+    return {
+        f"S{i:03}": pd.DataFrame({"Open": opens[:, i], "Close": closes[:, i]}, index=dates)
+        for i in range(symbols)
+    }
+
+
+def make_backtest(data, seed):
+    return Backtest(
+        list(data),
+        price_data=data,
+        strategy_cls=MovingAverageCross,
+        strategy_kwargs={"short": 5, "long": 30},
+        initial_capital=1_000_000,
+        calendar_policy="strict",
+        provenance={"kind": "synthetic", "seed": seed, "source": "benchmark.generate_data v1"},
+    )
+
+
+def check_accounting(result, initial):
+    fills = result["fills"]
+    if not fills.empty:
+        if not (fills.fill_dt > fills.submitted_dt).all():
+            raise AssertionError("same-bar fill")
+        flows = fills.groupby("fill_dt").cash_flow.sum()
+    else:
+        flows = pd.Series(dtype=float)
+    cash = result["cash"].set_index("dt")
+    expected_cash = initial + flows.reindex(cash.index, fill_value=0).cumsum()
+    np.testing.assert_allclose(cash.cash, expected_cash, rtol=1e-12, atol=1e-7)
+    marked = result["positions"].groupby("dt").market_value.sum()
+    np.testing.assert_allclose(cash.equity, cash.cash + marked, rtol=1e-12, atol=1e-7)
+    if cash.cash.min() < -1e-8 or result["positions"].quantity.min() < 0:
+        raise AssertionError("negative cash or short position")
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--symbols", type=int, default=20)
+    parser.add_argument("--bars", type=int, default=2520)
+    parser.add_argument("--seed", type=int, default=17)
+    parser.add_argument("--repeats", type=int, default=3)
+    parser.add_argument("--profile", type=Path)
+    args = parser.parse_args()
+    if args.repeats < 1:
+        parser.error("repeats must be positive")
+    data = generate_data(args.symbols, args.bars, args.seed)
+
+    # Untimed correctness run precedes throughput, allocation and profiling runs.
+    baseline = make_backtest(data, args.seed)
+    result = baseline.run()
+    check_accounting(result, 1_000_000)
+    expected = digest(ledger_records(result))
+    event_count = len(baseline.event_records)
+    ledger_rows = sum(len(frame) for frame in result.values())
+    del baseline, result
+    seconds = []
+    for _ in range(args.repeats):
+        started = time.perf_counter()
+        bt = make_backtest(data, args.seed)
+        result = bt.run()
+        seconds.append(time.perf_counter() - started)
+        if digest(ledger_records(result)) != expected:
+            raise AssertionError("benchmark replay mismatch")
+        del bt, result
+
+    # Separate allocation measurement so tracemalloc does not distort throughput.
+    tracemalloc.start()
+    bt = make_backtest(data, args.seed)
+    result = bt.run()
+    _, peak = tracemalloc.get_traced_memory()
+    tracemalloc.stop()
+    started = time.perf_counter()
+    experiment = snapshot(bt, result)
+    snapshot_seconds = time.perf_counter() - started
+    with tempfile.TemporaryDirectory() as directory:
+        path = Path(directory) / "benchmark.sqlite"
+        with SQLiteStore(path) as store:
+            started = time.perf_counter()
+            store.save(experiment)
+            save_seconds = time.perf_counter() - started
+            started = time.perf_counter()
+            store.load(bt.run_id)
+            load_seconds = time.perf_counter() - started
+            plan = store.query_plan(bt.run_id)
+        db_bytes = path.stat().st_size
+    if args.profile:
+        args.profile.parent.mkdir(parents=True, exist_ok=True)
+        profiler = cProfile.Profile()
+        profiler.enable()
+        profiled = make_backtest(data, args.seed)
+        profiled.run()
+        profiler.disable()
+        profiler.dump_stats(args.profile)
+    cpu = platform.processor() or platform.machine()
+    if platform.system() == "Darwin":
+        try:
+            cpu = subprocess.check_output(
+                ["sysctl", "-n", "machdep.cpu.brand_string"], text=True, stderr=subprocess.DEVNULL
+            ).strip()
+        except (OSError, subprocess.CalledProcessError):
+            cpu += " (CPU model unavailable in sandbox)"
+    rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+    rss_bytes = rss if platform.system() == "Darwin" else rss * 1024
+    report = {
+        "kind": "synthetic",
+        "seed": args.seed,
+        "symbols": args.symbols,
+        "bars_per_symbol": args.bars,
+        "input_rows": args.symbols * args.bars,
+        "consumed_columns": ["Open", "Close"],
+        "events": event_count,
+        "ledger_rows": ledger_rows,
+        "repeats": args.repeats,
+        "seconds": seconds,
+        "median_seconds": statistics.median(seconds),
+        "events_per_second": event_count / statistics.median(seconds),
+        "tracemalloc_peak_bytes": peak,
+        "process_peak_rss_bytes": rss_bytes,
+        "snapshot_seconds": snapshot_seconds,
+        "sqlite_save_seconds": save_seconds,
+        "sqlite_load_seconds": load_seconds,
+        "database_bytes": db_bytes,
+        "ledger_hash": expected,
+        "versions": versions(),
+        "hardware": {"os": platform.platform(), "cpu": cpu, "logical_cpus": os.cpu_count()},
+        "query_plan": plan,
+        "measurement_scope": "Throughput includes Backtest construction, input hashing and run; excludes "
+        "data generation, validation, persistence, profiling and allocation tracing. Allocation peak "
+        "is a separate constructor/run; RSS is process lifetime and includes other passes.",
+    }
+    print(json.dumps(report, indent=2))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/docs/DESIGN.md b/docs/DESIGN.md
index 3367b04..3608664 100644
--- a/docs/DESIGN.md
+++ b/docs/DESIGN.md
@@ -19,3 +19,10 @@ An order submitted on the final bar is cancelled with `end_of_data`; it is never
 ## Intentionally excluded
 
 Version 1 does not add more strategies. Its goal is a reliable engine boundary, auditable ledgers, and honest evaluation—not a strategy catalogue.
+
+## Experiment persistence
+
+The completed run can now be saved behind a save/load interface, with immutable
+audit events, content identities and exact configuration replay. See
+[replayable experiments](EXPERIMENTS.md) for storage architecture, SQL reports,
+compatibility details and data policies.
diff --git a/docs/EXPERIMENTS.md b/docs/EXPERIMENTS.md
new file mode 100644
index 0000000..8f5c28b
--- /dev/null
+++ b/docs/EXPERIMENTS.md
@@ -0,0 +1,132 @@
+# Replayable experiments
+
+## Architecture and public API
+
+`Backtest(...).run()` still returns the original seven pandas ledgers. Existing
+strategy classes and injected DataFrames continue to work. `event_records`,
+`run_id`, `config`, `input_snapshot` and `metadata` are additional attributes.
+Instances are single-use; create a fresh instance for each run.
+
+The queue retains MarketEvent, SignalEvent, OrderEvent and FillEvent. The audit
+stream uses immutable records for signals, accepted orders, rejected quantities,
+booked fills and valuations. Accepted records contain the order and total cash
+reservation; broker fill proposals become booked fill events only after actual
+cash/position checks. This distinction prevents replay from booking a rejected
+or reduced quantity as if it filled. Sequence numbers start at zero and order
+all records within a run; timestamps alone cannot order events on the same day.
+A final cancellation releases reservations and emits a terminal valuation at the
+last timestamp, without adding another return observation.
+
+The portfolio remains the sole accounting implementation. The store receives a
+completed experiment through `ExperimentStore.save` and retrieves it through
+`load`; neither implementation computes trades. MemoryStore provides isolated
+copies for tests. SQLiteStore implements the same boundary with explicit SQL.
+There is no service, ORM, broker integration or production database connection.
+
+## Identity and reproducibility
+
+SHA-256 run IDs address canonical JSON containing the configuration, normalized
+input hash, engine source hash, strategy source hash, dependency versions, timing,
+provenance and excluded dates. Ordered symbols are part of configuration because
+competing orders consume cash in that order. Configuration and data also have
+separate hashes. Identical inputs in the identical environment reproduce the
+same ID. Different costs, versions, source or provenance produce different IDs;
+an ID is a reproducible experiment identity, not a timestamped execution attempt.
+
+The database retains every normalized Open/Close observation before calendar
+intersection, plus the excluded-date list and provenance. Replay uses these
+stored observations, never a fresh download or the current fixture file. High,
+Low and unused input columns are not part of the consumed dataset. Preserve an
+external raw-data artifact if those columns are needed for separate research.
+
+Use `provenance={"kind": "synthetic", "source": "...", "seed": 17}` for synthetic
+inputs and identify source, adjustment conventions and dates for historical
+inputs. An unspecified injected frame is labeled `injected`, not claimed to be
+historical. The downloader uses adjusted historical prices. No live feed is
+implemented. Callers are responsible for truthful source labels.
+
+`replay` requires matching engine source, strategy source and Python/pandas/NumPy/
+SQLite versions, then checks every event and ledger against the saved manifest.
+The CLI permits only named built-in strategies; the Python API can accept an
+explicit registry for custom strategies. Database text is never imported or
+executed. Interactive classes without inspectable source may run but cannot be
+exactly replayed. Configuration and provenance must be JSON-serializable.
+Custom strategy determinism remains the caller's responsibility: source hashing
+does not capture arbitrary external files, global state or helper dependencies.
+
+## Storage and queries
+
+`engine/sql/001_initial.sql` creates runs, events and ledgers, keyed by run ID.
+Foreign keys, event-kind checks, JSON validity, nonnegative accounting fields,
+unique sequence/row keys and projection-to-JSON checks constrain storage. SQL
+report columns are checked against their complete JSON records to avoid divergent
+copies. Indexes cover run comparison groups, slippage groups, event kinds and
+ledger timelines. `PRAGMA user_version` tracks the migration. Initialization and
+whole-run saves use explicit transactions; failure rolls back the entire run.
+Newer unknown schema versions fail. A repeated identical save is idempotent;
+different output for the same ID fails. There is no partial-run resume support.
+
+`summary.sql` computes drawdown with a running MAX window and sums actual fill
+commissions and slippage. Tests reconcile these with `engine.metrics.performance`
+and pandas fill totals, including a declining equity curve. Queries bind run IDs
+and comparison hashes as parameters. `plan` runs SQLite's real EXPLAIN QUERY PLAN;
+see [measured performance](PERFORMANCE.md) for an observed plan.
+
+```bash
+python run_experiments.py --db experiments.sqlite demo
+python run_experiments.py --db experiments.sqlite replay RUN_ID
+python run_experiments.py --db experiments.sqlite compare RUN_ID_A RUN_ID_B
+python run_experiments.py --db experiments.sqlite costs RUN_ID
+python run_experiments.py --db experiments.sqlite plan RUN_ID
+```
+
+Copy IDs printed by `demo`. It stores and verifies replay for 0, 1, 5 and 10 bps.
+`compare` requires identical inputs, timing, initial capital, costs, source labels,
+calendar policy and engine/dependency versions; only strategy/configuration may
+vary. `costs` holds strategy/configuration constant and permits only slippage to
+vary. It reports separate reruns, not a post-hoc fee subtraction from one ledger.
+Comparisons are accounting reports, not selection or untouched evaluation claims.
+
+## Data policy and limits
+
+- `calendar_policy="common"` preserves the original intersection policy. A next
+  bar means the next date shared by every symbol, potentially several days later.
+  Excluded dates are retained in metadata. This can truncate a delisted symbol's
+  tail for the whole portfolio and is unsuitable for delisting research.
+- `calendar_policy="strict"` requires identical dates and is used by the stored
+  demo and benchmark. Missing internal bars or unequal tails fail before trading.
+  Neither policy forward-fills or invents an open. There is no exchange calendar:
+  a date missing from every symbol cannot be detected automatically. Daily
+  timestamps are session labels; timezone information is stripped for legacy
+  compatibility. Callers must align session labels across markets before injection.
+- Duplicate/invalid timestamps, nonfinite/missing prices and nonpositive prices
+  fail. Explicit nonzero `Stale`, `Delisted`, `Dividends` or `Stock Splits` columns
+  fail as unsupported. Repeated prices alone cannot establish staleness. Supply
+  quality flags or validate freshness upstream. Unflagged corporate actions
+  cannot be inferred reliably from price changes. The engine does not process
+  action cash flows, share adjustments or delisting recoveries.
+- A last-bar order is cancelled; positions already held remain marked at the
+  last observed close. There is no invented terminal liquidation.
+- Buy affordability uses commission on actual filled quantity. A reduced fill
+  still pays the configured minimum. A sale whose proceeds plus cash cannot
+  cover commission is rejected and the position remains held. These fix edge
+  cases where the earlier implementation prorated away minimum fees or allowed
+  negative cash on a collapsed-price sale.
+- Accounting uses floating-point currency and integer shares; tests use explicit
+  tolerances. No shorting, financing, borrow, margin, partial liquidity or taxes.
+- Completed runs, events, snapshots and ledgers remain in memory. This is a daily
+  research engine, not an unbounded streaming storage system. Database manifests
+  detect accidental corruption; they are not signatures against malicious edits.
+
+## Validation and references
+
+The original offline tests remain in CI, alongside the golden fixture, event
+reduction, replay, constraints, rollback, missing/stale-bar, fee and comparison
+tests. Benchmarking follows these correctness tests and has no machine-specific
+speed threshold. Installation requires dependency downloads once; all demos,
+replay, tests and benchmarks then work without network access or credentials.
+
+Implementation references: [Python 3.12 sqlite3 transactions and parameters](https://docs.python.org/3.12/library/sqlite3.html),
+[SQLite window functions](https://www.sqlite.org/windowfunctions.html),
+[pandas 3.0 DataFrame records](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.to_dict.html),
+and [NumPy Generator](https://numpy.org/doc/stable/reference/random/generator.html).
diff --git a/docs/PERFORMANCE.md b/docs/PERFORMANCE.md
new file mode 100644
index 0000000..0d08458
--- /dev/null
+++ b/docs/PERFORMANCE.md
@@ -0,0 +1,83 @@
+# Seeded performance baseline
+
+This is an observed local measurement, not a speed guarantee. No performance
+optimization was applied. Correctness checks and deterministic result-hash
+comparison precede profiling. CI runs a small correctness/performance smoke
+workload without a hardware-specific speed threshold.
+
+## Reproduce
+
+After installing the project dependencies, from the repository root:
+
+```bash
+python -m unittest discover -s tests -v
+python benchmark.py --symbols 20 --bars 2520 --seed 17 --repeats 3 \
+  --profile /tmp/backtest-profile.pstats > /tmp/backtest-benchmark.json
+python -c 'import pstats; pstats.Stats("/tmp/backtest-profile.pstats").strip_dirs().sort_stats("cumulative").print_stats(12)'
+```
+
+The seeded workload is synthetic lognormal prices with overnight gaps, 20
+symbols, 2,520 business-date bars per symbol, 50,400 input rows and two consumed
+columns (Open/Close). It uses the existing 5/30 moving-average strategy, one
+million starting capital, default 1 bp slippage and default commission. It is
+not a model calibrated to historical markets. The date grid is not an exchange
+calendar. Runs retain complete event and ledger histories.
+
+## Observed run
+
+Recorded September 22, 2026 UTC on macOS 26.5.1, Apple M5, 10 logical CPUs and
+16 GiB RAM. Hardware was independently read with
+`sysctl -n machdep.cpu.brand_string hw.memsize hw.logicalcpu`; the benchmark's
+sandbox could report architecture and CPU count but not the CPU model.
+Python 3.12.14, pandas 3.0.6, NumPy 2.5.3 and SQLite 3.53.1 were used.
+Full installed package versions are in [environment.txt](measurements/environment.txt).
+The accounting source at measurement was local commit `8923562`; the benchmark
+script accompanies this report.
+
+| Measurement | Observed value |
+|---|---:|
+| Audit events | 9,838 |
+| Total rows across all seven ledgers | 62,756 |
+| Three constructor/run timings | 4.791, 4.820, 6.405 seconds |
+| Median constructor/run time | 4.820 seconds |
+| Audit-event throughput | 2,041 events/second |
+| Separate traced allocation peak | 50,021,681 bytes (47.7 MiB) |
+| Whole-process peak RSS | 382,484,480 bytes (364.8 MiB) |
+| Canonical snapshot conversion | 0.870 seconds |
+| Atomic SQLite save | 1.343 seconds |
+| Validated SQLite load | 0.596 seconds |
+| SQLite file size | 39,616,512 bytes (37.8 MiB) |
+
+[Raw measurements and query plan](measurements/benchmark.json) include the exact
+ledger hash. Throughput includes construction, normalization, input hashing,
+strategy execution and ledger construction. It excludes data generation,
+validation, persistence, profiling and allocation tracing. Tracemalloc runs
+separately and excludes the pre-existing input frames; it may not capture all
+native allocations. RSS is the process lifetime maximum, including validation,
+serialization, persistence and profiling passes. Disk timings are a single local
+sample and are sensitive to caches and filesystem behavior. Other machine load,
+thermal state and scheduler effects were not controlled. These results do not
+establish linear scaling or intraday capacity.
+
+## Profile and SQL plan
+
+The separate [cProfile pass](measurements/profile.txt) took 16.836 seconds under
+instrumentation. It reported 202,419 `get_latest_close` calls taking 8.838 seconds
+cumulatively, with pandas positional indexing prominent. Instrumented time is
+not the throughput timing, and cumulative rows overlap. Caching price arrays
+could be investigated later with the same correctness suite and input hash;
+this baseline does not claim that optimization has been performed.
+
+The observed summary query plan searches the composite primary-key index:
+
+```text
+SEARCH ledgers USING INDEX sqlite_autoindex_ledgers_1 (run_id=? AND ledger=?)
+MATERIALIZE curve
+USE TEMP B-TREE FOR ORDER BY
+```
+
+The index selects one run and ledger, then orders its rows for the running peak
+window. SQLite materializes that curve because both final equity and drawdown
+consume it; the final descending lookup uses a temporary sort. This is expected
+for this small report and is not described as an index-only query. Inspect a
+stored experiment directly using `python run_experiments.py plan RUN_ID`.
diff --git a/docs/measurements/benchmark.json b/docs/measurements/benchmark.json
new file mode 100644
index 0000000..e411af7
--- /dev/null
+++ b/docs/measurements/benchmark.json
@@ -0,0 +1,114 @@
+{
+  "kind": "synthetic",
+  "seed": 17,
+  "symbols": 20,
+  "bars_per_symbol": 2520,
+  "input_rows": 50400,
+  "consumed_columns": [
+    "Open",
+    "Close"
+  ],
+  "events": 9838,
+  "ledger_rows": 62756,
+  "repeats": 3,
+  "seconds": [
+    4.79093991604168,
+    4.819878124981187,
+    6.404577417008113
+  ],
+  "median_seconds": 4.819878124981187,
+  "events_per_second": 2041.1304487161487,
+  "tracemalloc_peak_bytes": 50021681,
+  "process_peak_rss_bytes": 382484480,
+  "snapshot_seconds": 0.8704346669837832,
+  "sqlite_save_seconds": 1.3427781659993343,
+  "sqlite_load_seconds": 0.5959005000186153,
+  "database_bytes": 39616512,
+  "ledger_hash": "2273f0c763f0240580ea209af4f8607728cbb131faa164e4cb8b669f27cc2838",
+  "versions": {
+    "python": "3.12.14",
+    "pandas": "3.0.6",
+    "numpy": "2.5.3",
+    "sqlite": "3.53.1"
+  },
+  "hardware": {
+    "os": "macOS-26.5.1-arm64-arm-64bit",
+    "cpu": "arm (CPU model unavailable in sandbox)",
+    "logical_cpus": 10
+  },
+  "query_plan": [
+    [
+      2,
+      0,
+      0,
+      "CO-ROUTINE costs"
+    ],
+    [
+      6,
+      2,
+      61,
+      "SEARCH ledgers USING INDEX sqlite_autoindex_ledgers_1 (run_id=? AND ledger=?)"
+    ],
+    [
+      34,
+      0,
+      16,
+      "SCAN costs"
+    ],
+    [
+      39,
+      0,
+      0,
+      "SCALAR SUBQUERY 3"
+    ],
+    [
+      43,
+      39,
+      0,
+      "MATERIALIZE curve"
+    ],
+    [
+      46,
+      43,
+      0,
+      "CO-ROUTINE (subquery-6)"
+    ],
+    [
+      50,
+      46,
+      61,
+      "SEARCH ledgers USING INDEX sqlite_autoindex_ledgers_1 (run_id=? AND ledger=?)"
+    ],
+    [
+      73,
+      43,
+      47,
+      "SCAN (subquery-6)"
+    ],
+    [
+      123,
+      39,
+      47,
+      "SCAN curve"
+    ],
+    [
+      137,
+      39,
+      0,
+      "USE TEMP B-TREE FOR ORDER BY"
+    ],
+    [
+      145,
+      0,
+      0,
+      "SCALAR SUBQUERY 4"
+    ],
+    [
+      151,
+      145,
+      216,
+      "SEARCH curve"
+    ]
+  ],
+  "measurement_scope": "Throughput includes Backtest construction, input hashing and run; excludes data generation, validation, persistence, profiling and allocation tracing. Allocation peak is a separate constructor/run; RSS is process lifetime and includes other passes."
+}
diff --git a/docs/measurements/environment.txt b/docs/measurements/environment.txt
new file mode 100644
index 0000000..b9a1d09
--- /dev/null
+++ b/docs/measurements/environment.txt
@@ -0,0 +1,36 @@
+beautifulsoup4==4.15.0
+certifi==2026.7.22
+cffi==2.1.1
+charset-normalizer==3.5.1
+contourpy==1.4.0
+curl_cffi==0.16.3
+cycler==0.12.1
+fonttools==4.65.0
+idna==3.20
+iniconfig==2.3.0
+kiwisolver==1.5.1
+lxml==6.1.3
+matplotlib==3.11.2
+multitasking==0.0.13
+numpy==2.5.3
+packaging==26.3
+pandas==3.0.6
+peewee==4.5.1
+pillow==12.3.0
+platformdirs==4.11.12
+pluggy==1.6.0
+protobuf==7.36.2
+pycparser==3.0
+Pygments==2.21.0
+pyparsing==3.3.3
+pytest==9.1.1
+python-dateutil==2.9.0.post0
+pytz==2026.3.post1
+requests==2.34.2
+ruff==0.16.8
+six==1.17.0
+soupsieve==2.9.2
+typing_extensions==4.16.0
+urllib3==2.8.0
+websockets==17.1
+yfinance==1.7.0
diff --git a/docs/measurements/profile.txt b/docs/measurements/profile.txt
new file mode 100644
index 0000000..41a206f
--- /dev/null
+++ b/docs/measurements/profile.txt
@@ -0,0 +1,20 @@
+Tue Sep 22 02:04:28 2026    /tmp/backtest-profile.pstats
+
+         39894603 function calls (39436450 primitive calls) in 16.836 seconds
+
+   Ordered by: cumulative time
+   List reduced from 643 to 12 due to restriction <12>
+
+   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
+        1    0.024    0.024   14.282   14.282 backtest.py:93(run)
+   255297    0.310    0.000    9.793    0.000 indexing.py:1192(__getitem__)
+   255277    0.403    0.000    9.035    0.000 indexing.py:1740(_getitem_axis)
+   202419    0.278    0.000    8.838    0.000 data.py:115(get_latest_close)
+     2520    0.059    0.000    6.668    0.003 portfolio.py:178(mark_to_market)
+   255277    0.337    0.000    6.157    0.000 frame.py:4292(_ixs)
+     2520    0.072    0.000    4.604    0.002 strategy.py:32(calculate_signals)
+     6184    0.019    0.000    4.525    0.001 {built-in method builtins.sum}
+     4959    0.005    0.000    4.475    0.001 portfolio.py:35(_equity)
+   104139    0.049    0.000    4.452    0.000 portfolio.py:36(<genexpr>)
+    50400    0.138    0.000    3.984    0.000 data.py:118(get_latest_closes)
+        1    0.000    0.000    2.573    2.573 benchmark.py:38(make_backtest)
diff --git a/engine/backtest.py b/engine/backtest.py
index 5317d88..de1a502 100644
--- a/engine/backtest.py
+++ b/engine/backtest.py
@@ -2,13 +2,17 @@
 
 from __future__ import annotations
 
+import math
 import queue
+from dataclasses import replace
 
 import pandas as pd
 
 from .data import DataHandler
+from .events import EventRecord
 from .execution import SimulatedExecutionHandler
 from .portfolio import Portfolio
+from .reproducibility import TIMING, digest, encode_frames, engine_hash, strategy_hash, versions
 
 
 class Backtest:
@@ -24,16 +28,23 @@ def __init__(
         minimum_commission=1.0,
         slippage_bps=1.0,
         strategy_kwargs=None,
+        calendar_policy="common",
+        provenance=None,
         **legacy_strategy_kwargs,
     ):
         if strategy_cls is None:
             raise ValueError("strategy_cls is required")
+        if not math.isfinite(initial_capital) or initial_capital <= 0:
+            raise ValueError("initial_capital must be finite and positive")
+        self.event_records = []
         self.events = queue.Queue()
-        self.data = DataHandler(self.events, symbols, start, end, price_data=price_data)
+        self.data = DataHandler(
+            self.events, symbols, start, end, price_data=price_data, calendar_policy=calendar_policy
+        )
         kwargs = dict(strategy_kwargs or {})
         kwargs.update(legacy_strategy_kwargs)
         self.strategy = strategy_cls(self.data, self.events, **kwargs)
-        self.portfolio = Portfolio(self.data, self.events, initial_capital)
+
         self.execution = SimulatedExecutionHandler(
             self.data,
             self.events,
@@ -42,7 +53,47 @@ def __init__(
             slippage_bps=slippage_bps,
         )
 
+        self.portfolio = Portfolio(
+            self.data, self.events, initial_capital, record=self._record, commission=self.execution.commission
+        )
+        self.config = {
+            "symbols": self.data.symbols,
+            "initial_capital": float(initial_capital),
+            "strategy": f"{strategy_cls.__module__}:{strategy_cls.__qualname__}",
+            "strategy_kwargs": kwargs,
+            "commission_per_share": float(commission_per_share),
+            "minimum_commission": float(minimum_commission),
+            "slippage_bps": float(slippage_bps),
+            "calendar_policy": calendar_policy,
+        }
+        self.input_snapshot = encode_frames(self.data.source_frames)
+        self.provenance = provenance or {
+            "kind": "injected" if price_data is not None else "historical",
+            "source": "caller DataFrames" if price_data is not None else "yfinance",
+            "requested_start": start,
+            "requested_end": end,
+        }
+        self.metadata = {
+            "config": self.config,
+            "data_hash": digest(self.input_snapshot),
+            "config_hash": digest(self.config),
+            "timing": TIMING,
+            "engine_hash": engine_hash(),
+            "strategy_hash": strategy_hash(strategy_cls),
+            "versions": versions(),
+            "provenance": self.provenance,
+            "excluded_dates": self.data.excluded_dates,
+        }
+        self.run_id = digest(self.metadata)
+        self._has_run = False
+
+    def _record(self, event):
+        self.event_records.append(EventRecord(self.run_id, len(self.event_records), event))
+
     def run(self):
+        if self._has_run:
+            raise RuntimeError("Backtest instances are single-use; construct a fresh run for replay")
+        self._has_run = True
         while self.data.continue_backtest:
             self.data.update_bars()
             if not self.data.continue_backtest:
@@ -58,6 +109,7 @@ def run(self):
                     self.execution.on_market(event)
                     self.strategy.calculate_signals(event)
                 elif event.type == "SIGNAL":
+                    self._record(event)
                     self.portfolio.update_signal(event)
                 elif event.type == "ORDER":
                     self.execution.execute_order(event)
@@ -66,8 +118,14 @@ def run(self):
             if market_event is not None:
                 self.portfolio.mark_to_market(market_event)
 
-        for order in self.execution.cancel_all():
+        final_valuation = self.event_records[-1].event
+        cancelled = self.execution.cancel_all()
+        for order in cancelled:
             self.portfolio.cancel_order(order)
+        if cancelled:
+            # End-of-data cancellation changes reservations, not marked equity.
+            self.portfolio.cash_ledger[-1]["reserved_cash"] = self.portfolio.reserved_cash
+            self._record(replace(final_valuation, reserved_cash=self.portfolio.reserved_cash))
         return self.results()
 
     def results(self):
diff --git a/engine/data.py b/engine/data.py
index 26609df..05d9363 100644
--- a/engine/data.py
+++ b/engine/data.py
@@ -42,26 +42,35 @@ def __init__(
         start=None,
         end=None,
         price_data: Mapping[str, pd.DataFrame] | None = None,
+        calendar_policy="common",
     ):
+        if calendar_policy not in ("common", "strict"):
+            raise ValueError("calendar_policy must be common or strict")
+        self.calendar_policy = calendar_policy
         self.events = events
         self.symbols = list(symbols)
+        if len(set(self.symbols)) != len(self.symbols):
+            raise ValueError("symbols must be unique")
         if not self.symbols:
             raise ValueError("at least one symbol is required")
 
-        source = (
-            dict(price_data)
-            if price_data is not None
-            else download_price_data(self.symbols, start, end)
-        )
+        source = dict(price_data) if price_data is not None else download_price_data(self.symbols, start, end)
         missing = set(self.symbols).difference(source)
         if missing:
             raise ValueError(f"missing price data for: {sorted(missing)}")
 
         self.frames = {symbol: self._normalize(source[symbol], symbol) for symbol in self.symbols}
+        self.source_frames = self.frames.copy()
         common = self.frames[self.symbols[0]].index
         for symbol in self.symbols[1:]:
             common = common.intersection(self.frames[symbol].index)
         common = common.sort_values()
+        self.excluded_dates = {
+            symbol: [dt.isoformat() for dt in frame.index.difference(common)]
+            for symbol, frame in self.frames.items()
+        }
+        if calendar_policy == "strict" and any(self.excluded_dates.values()):
+            raise ValueError("missing bars or delisting: strict calendar requires identical dates")
         if len(common) < 2:
             raise ValueError("symbols need at least two common bars")
 
@@ -75,12 +84,19 @@ def __init__(
     def _normalize(frame: pd.DataFrame, symbol: str) -> pd.DataFrame:
         if not isinstance(frame, pd.DataFrame):
             raise TypeError(f"price_data[{symbol!r}] must be a DataFrame")
-        renamed = frame.rename(columns={str(c).lower(): str(c).title() for c in frame.columns})
+        renamed = frame.rename(columns={c: str(c).title() for c in frame.columns})
+        for column in ("Dividends", "Stock Splits", "Delisted", "Stale"):
+            if column in renamed and (renamed[column].fillna(0) != 0).any():
+                raise ValueError(f"{symbol}: unsupported data flag {column}")
         if not {"Open", "Close"}.issubset(renamed.columns):
             raise ValueError(f"{symbol} requires Open and Close columns")
         out = renamed[["Open", "Close"]].copy()
         out.index = pd.DatetimeIndex(pd.to_datetime(out.index)).tz_localize(None)
-        out = out[~out.index.duplicated(keep="last")].sort_index().astype(float).dropna()
+        if out.index.has_duplicates or out.index.isna().any():
+            raise ValueError(f"{symbol}: duplicate or invalid timestamps")
+        out = out.sort_index().astype(float)
+        if not np.isfinite(out.to_numpy()).all():
+            raise ValueError(f"{symbol}: prices must be finite; missing/stale bars cannot be filled")
         if (out <= 0).any().any():
             raise ValueError(f"{symbol} contains non-positive prices")
         return out
@@ -105,4 +121,3 @@ def get_latest_closes(self, symbol: str, n: int) -> np.ndarray:
 
     def frame(self, symbol: str) -> pd.DataFrame:
         return self.frames[symbol].copy()
-
diff --git a/engine/demo.py b/engine/demo.py
new file mode 100644
index 0000000..f5a636c
--- /dev/null
+++ b/engine/demo.py
@@ -0,0 +1,68 @@
+"""Tiny synthetic accounting example; never downloads market data."""
+
+from pathlib import Path
+
+import pandas as pd
+
+from .backtest import Backtest
+from .events import SignalEvent
+from .strategy import Strategy
+
+
+class DemoStrategy(Strategy):
+    def __init__(self, data, events):
+        self.data, self.events = data, events
+
+    def calculate_signals(self, event):
+        if self.data.i == 0:
+            # A reserves almost all cash; B competes for the remainder.
+            for symbol in self.data.symbols:
+                self.events.put(SignalEvent(symbol, event.dt, 1.0))
+        elif self.data.i == 1:
+            for symbol in self.data.symbols:
+                self.events.put(SignalEvent(symbol, event.dt, 0.0))
+
+
+def demo_data():
+    path = Path(__file__).resolve().parents[1] / "fixtures/synthetic/ohlc.csv"
+    frame = pd.read_csv(path, parse_dates=["date"])
+    return {
+        symbol: group.set_index("date").drop(columns="symbol")
+        for symbol, group in frame.groupby("symbol", sort=False)
+    }
+
+
+def demo_backtest():
+    return Backtest(
+        ["A", "B"],
+        price_data=demo_data(),
+        initial_capital=100,
+        strategy_cls=DemoStrategy,
+        commission_per_share=0,
+        minimum_commission=1,
+        slippage_bps=10,
+        provenance={"kind": "synthetic", "source": "fixtures/synthetic/ohlc.csv v1"},
+    )
+
+
+def render_demo(result):
+    lines = [
+        "SYNTHETIC OHLC: 2 symbols x 4 daily bars; initial cash 100.00",
+        "Timing: close[t] signal -> next common open fill -> close mark.",
+        "A's Jan 1 close is 10.00; its Jan 2 open is 12.00.",
+    ]
+    for name in ("orders", "rejections", "fills", "cash", "positions", "equity"):
+        lines.extend([f"\n{name.upper()}", result[name].to_csv(index=False, float_format="%.6f").rstrip()])
+    lines.append("\nTIMELINE")
+    for row in result["cash"].itertuples():
+        day = str(row.dt.date())
+        fills = result["fills"]
+        for fill in fills[fills.fill_dt == row.dt].itertuples():
+            lines.append(
+                f"{day} OPEN {fill.direction} {fill.quantity} {fill.symbol} at "
+                f"{fill.fill_price:.3f}: cash flow {fill.cash_flow:.6f}"
+            )
+        lines.append(
+            f"{day} CLOSE cash={row.cash:.6f} reserved={row.reserved_cash:.6f} equity={row.equity:.6f}"
+        )
+    return "\n".join(lines) + "\n"
diff --git a/engine/events.py b/engine/events.py
index fc9c809..e16c62f 100644
--- a/engine/events.py
+++ b/engine/events.py
@@ -45,3 +45,47 @@ class FillEvent:
     slippage_cost: float
     type: str = field(default="FILL", init=False)
 
+
+@dataclass(frozen=True)
+class OrderAcceptedEvent:
+    order: OrderEvent
+    reserved_cash: float
+    type: str = field(default="ORDER_ACCEPTED", init=False)
+
+
+@dataclass(frozen=True)
+class OrderRejectedEvent:
+    dt: datetime
+    order_id: str | None
+    symbol: str
+    quantity: int
+    reason: str
+    type: str = field(default="ORDER_REJECTED", init=False)
+
+
+@dataclass(frozen=True)
+class AccountingFillEvent:
+    """Actual booked fill, after affordability checks (not a broker proposal)."""
+
+    fill: FillEvent
+    cash_flow: float
+    cash: float
+    position: int
+    type: str = field(default="FILL_BOOKED", init=False)
+
+
+@dataclass(frozen=True)
+class ValuationEvent:
+    dt: datetime
+    cash: float
+    reserved_cash: float
+    equity: float
+    positions: tuple[tuple[str, int, float], ...]
+    type: str = field(default="VALUATION", init=False)
+
+
+@dataclass(frozen=True)
+class EventRecord:
+    run_id: str
+    sequence: int
+    event: SignalEvent | OrderAcceptedEvent | OrderRejectedEvent | AccountingFillEvent | ValuationEvent
diff --git a/engine/execution.py b/engine/execution.py
index 6db2b39..de9c275 100644
--- a/engine/execution.py
+++ b/engine/execution.py
@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import math
+
 from .events import FillEvent
 
 
@@ -19,18 +21,30 @@ def __init__(
         self.commission_per_share = float(commission_per_share)
         self.minimum_commission = float(minimum_commission)
         self.slippage_bps = float(slippage_bps)
+        if any(
+            not math.isfinite(v) or v < 0
+            for v in (self.commission_per_share, self.minimum_commission, self.slippage_bps)
+        ):
+            raise ValueError("costs must be finite and nonnegative")
+        if self.slippage_bps >= 10_000:
+            raise ValueError("slippage_bps must be below 10000")
         self.pending_orders = []
 
+    def commission(self, quantity):
+        return max(self.minimum_commission, quantity * self.commission_per_share)
+
     def execute_order(self, event) -> None:
         self.pending_orders.append(event)
 
     def on_market(self, event) -> None:
         pending, self.pending_orders = self.pending_orders, []
         for order in pending:
+            if event.dt <= order.submitted_dt:
+                raise ValueError("fill must be strictly after signal timestamp")
             open_price = self.data.get_latest_open(order.symbol)
             sign = 1.0 if order.direction == "BUY" else -1.0
             fill_price = open_price * (1.0 + sign * self.slippage_bps / 10_000.0)
-            commission = max(self.minimum_commission, order.quantity * self.commission_per_share)
+            commission = self.commission(order.quantity)
             self.events.put(
                 FillEvent(
                     order_id=order.order_id,
diff --git a/engine/experiments.py b/engine/experiments.py
new file mode 100644
index 0000000..a7f773c
--- /dev/null
+++ b/engine/experiments.py
@@ -0,0 +1,49 @@
+"""Store and reproduce runs without importing executable names from a database."""
+
+from .backtest import Backtest
+from .demo import DemoStrategy
+from .reproducibility import decode_frames, digest, engine_hash, strategy_hash, versions
+from .storage import ExperimentStore, snapshot, validate
+from .strategy import CrossSectionalMomentum, EqualWeightBuyAndHold, MovingAverageCross
+
+STRATEGIES = {
+    f"{cls.__module__}:{cls.__qualname__}": cls
+    for cls in (DemoStrategy, CrossSectionalMomentum, EqualWeightBuyAndHold, MovingAverageCross)
+}
+
+
+def record_run(backtest: Backtest, store: ExperimentStore):
+    result = backtest.run()
+    store.save(snapshot(backtest, result))
+    return result
+
+
+def replay(store: ExperimentStore, run_id: str, registry=None):
+    saved = store.load(run_id)
+    validate(saved)
+    metadata = saved["metadata"]
+    config = dict(metadata["config"])
+    name = config.pop("strategy")
+    strategies = STRATEGIES if registry is None else registry
+    if name not in strategies:
+        raise ValueError(f"strategy must be explicitly registered for replay: {name}")
+    cls = strategies[name]
+    if (
+        metadata["engine_hash"] != engine_hash()
+        or metadata["versions"] != versions()
+        or metadata["strategy_hash"] is None
+        or metadata["strategy_hash"] != strategy_hash(cls)
+    ):
+        raise ValueError(
+            "exact replay requires matching engine source, strategy source and dependency versions"
+        )
+    bt = Backtest(
+        **config,
+        strategy_cls=cls,
+        price_data=decode_frames(saved["inputs"]),
+        provenance=metadata["provenance"],
+    )
+    result = bt.run()
+    if digest(snapshot(bt, result)) != digest(saved):
+        raise ValueError("deterministic replay mismatch")
+    return result
diff --git a/engine/portfolio.py b/engine/portfolio.py
index 888dc85..1fd26f6 100644
--- a/engine/portfolio.py
+++ b/engine/portfolio.py
@@ -2,11 +2,17 @@
 
 from __future__ import annotations
 
-from .events import OrderEvent
+from dataclasses import replace
+
+from .events import AccountingFillEvent, OrderAcceptedEvent, OrderEvent, OrderRejectedEvent, ValuationEvent
 
 
 class Portfolio:
-    def __init__(self, data, events, initial_capital=100_000.0, reserve_buffer=0.02):
+    def __init__(
+        self, data, events, initial_capital=100_000.0, reserve_buffer=0.02, record=None, commission=None
+    ):
+        self.record = record or (lambda event: None)
+        self.commission = commission
         self.data = data
         self.events = events
         self.symbols = data.symbols
@@ -99,6 +105,7 @@ def update_signal(self, event) -> None:
                 "status": "pending",
             }
         )
+        self.record(OrderAcceptedEvent(order, self.reserved_cash))
         self.events.put(order)
 
     def update_fill(self, event) -> None:
@@ -107,14 +114,23 @@ def update_fill(self, event) -> None:
         quantity = event.quantity
 
         if event.direction == "BUY":
-            affordable = int(max(0.0, self.cash - event.commission) // event.fill_price)
+            # Monotone search charges the configured fee on the actual quantity.
+            low, high = 0, min(quantity, int(self.cash // event.fill_price))
+            while low < high:
+                mid = (low + high + 1) // 2
+                fee = self.commission(mid) if self.commission else event.commission
+                if mid * event.fill_price + fee <= self.cash:
+                    low = mid
+                else:
+                    high = mid - 1
+            affordable = low
             if affordable < quantity:
                 self._reject(event.order_id, event.symbol, quantity - affordable, "gap_exceeded_cash")
                 quantity = affordable
             if quantity <= 0:
                 self._set_order_status(event.order_id, "rejected")
                 return
-            commission = event.commission * quantity / event.quantity
+            commission = self.commission(quantity) if self.commission else event.commission
             self.positions[event.symbol] += quantity
             self.cash -= quantity * event.fill_price + commission
             cash_flow = -(quantity * event.fill_price + commission)
@@ -124,7 +140,11 @@ def update_fill(self, event) -> None:
                 self._reject(event.order_id, event.symbol, event.quantity, "no_position_to_sell")
                 self._set_order_status(event.order_id, "rejected")
                 return
-            commission = event.commission * quantity / event.quantity
+            commission = self.commission(quantity) if self.commission else event.commission
+            if quantity * event.fill_price + self.cash < commission:
+                self._reject(event.order_id, event.symbol, quantity, "sale_cannot_cover_commission")
+                self._set_order_status(event.order_id, "rejected")
+                return
             self.positions[event.symbol] -= quantity
             self.cash += quantity * event.fill_price - commission
             cash_flow = quantity * event.fill_price - commission
@@ -143,12 +163,32 @@ def update_fill(self, event) -> None:
             "slippage_cost": slippage_cost,
             "cash_flow": cash_flow,
         }
+        self.record(
+            AccountingFillEvent(
+                replace(event, quantity=quantity, commission=commission, slippage_cost=slippage_cost),
+                cash_flow,
+                self.cash,
+                self.positions[event.symbol],
+            )
+        )
         self.fills.append(record)
         self.trades.append(record.copy())
         self._set_order_status(event.order_id, "filled" if quantity == event.quantity else "partially_filled")
 
     def mark_to_market(self, event) -> None:
         equity = self._equity()
+        self.record(
+            ValuationEvent(
+                event.dt,
+                self.cash,
+                self.reserved_cash,
+                equity,
+                tuple(
+                    (symbol, self.positions[symbol], self.data.get_latest_close(symbol))
+                    for symbol in self.symbols
+                ),
+            )
+        )
         self.equity_curve.append({"dt": event.dt, "equity": equity})
         self.cash_ledger.append(
             {"dt": event.dt, "cash": self.cash, "reserved_cash": self.reserved_cash, "equity": equity}
@@ -178,6 +218,7 @@ def _set_order_status(self, order_id, status) -> None:
                 return
 
     def _reject(self, order_id, symbol, quantity, reason) -> None:
+        self.record(OrderRejectedEvent(self.data.current_dt, order_id, symbol, quantity, reason))
         self.rejections.append(
             {
                 "dt": self.data.current_dt,
diff --git a/engine/reproducibility.py b/engine/reproducibility.py
new file mode 100644
index 0000000..8c818fe
--- /dev/null
+++ b/engine/reproducibility.py
@@ -0,0 +1,75 @@
+"""Canonical, lossless JSON and content identities for reproducible experiments."""
+
+import hashlib
+import inspect
+import json
+import platform
+import sqlite3
+from dataclasses import asdict, is_dataclass
+from datetime import datetime
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+TIMING = "close-signal/next-common-open/close-mark-v1"
+
+
+def canonical(value):
+    def convert(item):
+        if is_dataclass(item):
+            return asdict(item)
+        if isinstance(item, (datetime, pd.Timestamp)):
+            return item.isoformat()
+        if isinstance(item, np.generic):
+            return item.item()
+        raise TypeError(f"not serializable: {type(item).__name__}")
+
+    return json.dumps(value, default=convert, sort_keys=True, separators=(",", ":"), allow_nan=False)
+
+
+def digest(value):
+    return hashlib.sha256(canonical(value).encode()).hexdigest()
+
+
+def engine_hash():
+    return digest({p.name: p.read_text() for p in sorted(Path(__file__).parent.glob("*.py"))})
+
+
+def strategy_hash(cls):
+    try:
+        return digest(inspect.getsource(cls))
+    except (OSError, TypeError):
+        return None  # Interactive classes still run; stored replay must register source explicitly.
+
+
+def versions():
+    return {
+        "python": platform.python_version(),
+        "pandas": pd.__version__,
+        "numpy": np.__version__,
+        "sqlite": sqlite3.sqlite_version,
+    }
+
+
+def encode_frames(frames):
+    return {
+        symbol: [
+            {"dt": dt.isoformat(), **{c: float(v) for c, v in row.items()}} for dt, row in frame.iterrows()
+        ]
+        for symbol, frame in frames.items()
+    }
+
+
+def decode_frames(frames):
+    return {
+        symbol: pd.DataFrame(rows).assign(dt=lambda f: pd.to_datetime(f.dt)).set_index("dt")
+        for symbol, rows in frames.items()
+    }
+
+
+def ledger_records(results):
+    return {
+        name: frame.astype(object).where(pd.notna(frame), None).to_dict("records")
+        for name, frame in results.items()
+    }
diff --git a/engine/sql/001_initial.sql b/engine/sql/001_initial.sql
new file mode 100644
index 0000000..59c5228
--- /dev/null
+++ b/engine/sql/001_initial.sql
@@ -0,0 +1,43 @@
+CREATE TABLE runs (
+    run_id TEXT PRIMARY KEY CHECK(length(run_id) = 64),
+    data_hash TEXT NOT NULL CHECK(length(data_hash) = 64),
+    config_hash TEXT NOT NULL CHECK(length(config_hash) = 64),
+    comparison_key TEXT NOT NULL,
+    sensitivity_key TEXT NOT NULL,
+    slippage_bps REAL NOT NULL CHECK(slippage_bps >= 0),
+    metadata TEXT NOT NULL CHECK(json_valid(metadata)),
+    inputs TEXT NOT NULL CHECK(json_valid(inputs)),
+    manifest_hash TEXT NOT NULL CHECK(length(manifest_hash) = 64)
+);
+CREATE INDEX runs_comparison ON runs(comparison_key, run_id);
+CREATE INDEX runs_sensitivity ON runs(sensitivity_key, slippage_bps);
+CREATE TABLE events (
+    run_id TEXT NOT NULL REFERENCES runs(run_id) ON DELETE CASCADE,
+    sequence INTEGER NOT NULL CHECK(sequence >= 0),
+    kind TEXT NOT NULL CHECK(kind IN ('SIGNAL','ORDER_ACCEPTED','ORDER_REJECTED','FILL_BOOKED','VALUATION')),
+    payload TEXT NOT NULL CHECK(json_valid(payload)),
+    PRIMARY KEY(run_id, sequence),
+    CHECK(kind = json_extract(payload, '$.type'))
+);
+CREATE INDEX events_kind ON events(run_id, kind, sequence);
+CREATE TABLE ledgers (
+    run_id TEXT NOT NULL REFERENCES runs(run_id) ON DELETE CASCADE,
+    ledger TEXT NOT NULL CHECK(ledger IN ('orders','fills','trades','positions','cash','equity','rejections')),
+    row_number INTEGER NOT NULL CHECK(row_number >= 0),
+    dt TEXT NOT NULL,
+    equity REAL,
+    cash REAL CHECK(cash >= -0.00000001),
+    commission REAL CHECK(commission >= 0),
+    slippage_cost REAL CHECK(slippage_cost >= 0),
+    payload TEXT NOT NULL CHECK(json_valid(payload)),
+    PRIMARY KEY(run_id, ledger, row_number),
+    CHECK(ledger != 'cash' OR (equity IS NOT NULL AND cash IS NOT NULL)),
+    CHECK(ledger != 'fills' OR (commission IS NOT NULL AND slippage_cost IS NOT NULL)),
+    CHECK(equity IS json_extract(payload, '$.equity')),
+    CHECK(cash IS json_extract(payload, '$.cash')),
+    CHECK(commission IS json_extract(payload, '$.commission')),
+    CHECK(slippage_cost IS json_extract(payload, '$.slippage_cost')),
+    CHECK(dt = COALESCE(json_extract(payload, '$.dt'), json_extract(payload, '$.fill_dt'),
+                        json_extract(payload, '$.submitted_dt')))
+);
+CREATE INDEX ledgers_timeline ON ledgers(run_id, ledger, dt);
diff --git a/engine/sql/summary.sql b/engine/sql/summary.sql
new file mode 100644
index 0000000..1ef14e1
--- /dev/null
+++ b/engine/sql/summary.sql
@@ -0,0 +1,14 @@
+WITH curve AS (
+    SELECT row_number, equity,
+           MAX(equity) OVER (ORDER BY row_number ROWS UNBOUNDED PRECEDING) AS peak
+    FROM ledgers WHERE run_id = :run_id AND ledger = 'cash'
+), costs AS (
+    SELECT COALESCE(SUM(commission), 0) AS commission,
+           COALESCE(SUM(slippage_cost), 0) AS slippage_cost,
+           COUNT(*) AS fills
+    FROM ledgers WHERE run_id = :run_id AND ledger = 'fills'
+)
+SELECT (SELECT equity FROM curve ORDER BY row_number DESC LIMIT 1) AS final_equity,
+       (SELECT MIN(equity / peak - 1.0) FROM curve) AS max_drawdown,
+       commission, slippage_cost, fills
+FROM costs;
diff --git a/engine/storage.py b/engine/storage.py
new file mode 100644
index 0000000..6edccb2
--- /dev/null
+++ b/engine/storage.py
@@ -0,0 +1,241 @@
+"""Atomic experiment storage behind a two-method persistence protocol."""
+
+import json
+import sqlite3
+from copy import deepcopy
+from pathlib import Path
+from typing import Protocol
+
+from .reproducibility import canonical, digest, ledger_records
+
+SQL = Path(__file__).with_name("sql")
+LEDGERS = ("equity", "orders", "fills", "trades", "positions", "cash", "rejections")
+
+
+class ExperimentStore(Protocol):
+    def save(self, experiment: dict) -> None: ...
+    def load(self, run_id: str) -> dict: ...
+
+
+def snapshot(backtest, results):
+    return json.loads(
+        canonical(
+            {
+                "run_id": backtest.run_id,
+                "metadata": backtest.metadata,
+                "inputs": backtest.input_snapshot,
+                "events": backtest.event_records,
+                "ledgers": ledger_records(results),
+            }
+        )
+    )
+
+
+def validate(experiment):
+    metadata = experiment["metadata"]
+    if digest(metadata) != experiment["run_id"]:
+        raise ValueError("run identity mismatch")
+    if digest(metadata["config"]) != metadata["config_hash"]:
+        raise ValueError("configuration hash mismatch")
+    if digest(experiment["inputs"]) != metadata["data_hash"]:
+        raise ValueError("input hash mismatch")
+    if set(experiment["ledgers"]) != set(LEDGERS):
+        raise ValueError("incomplete ledger set")
+    for i, event in enumerate(experiment["events"]):
+        if event["sequence"] != i or event["run_id"] != experiment["run_id"]:
+            raise ValueError("event sequence or run identity mismatch")
+
+
+def comparison_key(metadata, sensitivity=False):
+    config = deepcopy(metadata["config"])
+    if sensitivity:
+        config.pop("slippage_bps")
+    else:
+        config.pop("strategy")
+        config.pop("strategy_kwargs")
+    return digest(
+        {
+            "config": config,
+            "data_hash": metadata["data_hash"],
+            "timing": metadata["timing"],
+            "engine_hash": metadata["engine_hash"],
+            "versions": metadata["versions"],
+            "provenance": metadata["provenance"],
+            "strategy_hash": metadata["strategy_hash"] if sensitivity else None,
+        }
+    )
+
+
+class MemoryStore:
+    def __init__(self):
+        self._runs = {}
+
+    def save(self, experiment):
+        validate(experiment)
+        run_id = experiment["run_id"]
+        if run_id in self._runs and self._runs[run_id] != experiment:
+            raise ValueError("non-deterministic result for existing run identity")
+        self._runs[run_id] = deepcopy(experiment)
+
+    def load(self, run_id):
+        return deepcopy(self._runs[run_id])
+
+
+class SQLiteStore:
+    def __init__(self, path):
+        self.connection = sqlite3.connect(path, isolation_level=None)
+        self.connection.row_factory = sqlite3.Row
+        self.connection.execute("PRAGMA foreign_keys = ON")
+        self._migrate()
+
+    def _migrate(self):
+        # Lock before inspecting the version so concurrent openers cannot race.
+        self.connection.execute("BEGIN IMMEDIATE")
+        try:
+            version = self.connection.execute("PRAGMA user_version").fetchone()[0]
+            if version > 1:
+                raise ValueError("database schema is newer than this engine")
+            if version == 0:
+                statement = ""
+                for line in (SQL / "001_initial.sql").read_text().splitlines(True):
+                    statement += line
+                    if sqlite3.complete_statement(statement):
+                        self.connection.execute(statement)
+                        statement = ""
+                self.connection.execute("PRAGMA user_version = 1")
+            self.connection.commit()
+        except BaseException:
+            self.connection.rollback()
+            raise
+
+    def close(self):
+        self.connection.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.close()
+
+    def save(self, experiment):
+        validate(experiment)
+        run_id, metadata = experiment["run_id"], experiment["metadata"]
+        manifest = digest(experiment)
+        self.connection.execute("BEGIN IMMEDIATE")
+        try:
+            existing = self.connection.execute(
+                "SELECT manifest_hash FROM runs WHERE run_id = ?", (run_id,)
+            ).fetchone()
+            if existing:
+                if existing[0] != manifest:
+                    raise ValueError("non-deterministic result for existing run identity")
+                self.connection.commit()
+                return
+            self.connection.execute(
+                "INSERT INTO runs VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    run_id,
+                    metadata["data_hash"],
+                    metadata["config_hash"],
+                    comparison_key(metadata),
+                    comparison_key(metadata, True),
+                    metadata["config"]["slippage_bps"],
+                    canonical(metadata),
+                    canonical(experiment["inputs"]),
+                    manifest,
+                ),
+            )
+            self.connection.executemany(
+                "INSERT INTO events VALUES (?, ?, ?, ?)",
+                (
+                    (run_id, e["sequence"], e["event"]["type"], canonical(e["event"]))
+                    for e in experiment["events"]
+                ),
+            )
+            self.connection.executemany(
+                "INSERT INTO ledgers VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    (
+                        run_id,
+                        name,
+                        i,
+                        row.get("dt", row.get("fill_dt", row.get("submitted_dt"))),
+                        row.get("equity"),
+                        row.get("cash"),
+                        row.get("commission"),
+                        row.get("slippage_cost"),
+                        canonical(row),
+                    )
+                    for name, rows in experiment["ledgers"].items()
+                    for i, row in enumerate(rows)
+                ),
+            )
+            self.connection.commit()
+        except BaseException:
+            self.connection.rollback()
+            raise
+
+    def load(self, run_id):
+        row = self.connection.execute("SELECT * FROM runs WHERE run_id = ?", (run_id,)).fetchone()
+        if row is None:
+            raise KeyError(run_id)
+        result = {
+            "run_id": run_id,
+            "metadata": json.loads(row["metadata"]),
+            "inputs": json.loads(row["inputs"]),
+            "events": [],
+            "ledgers": {name: [] for name in LEDGERS},
+        }
+        for event in self.connection.execute(
+            "SELECT sequence, payload FROM events WHERE run_id = ? ORDER BY sequence", (run_id,)
+        ):
+            result["events"].append({"run_id": run_id, "sequence": event[0], "event": json.loads(event[1])})
+        for ledger in self.connection.execute(
+            "SELECT ledger, payload, row_number FROM ledgers WHERE run_id = ? ORDER BY ledger, row_number",
+            (run_id,),
+        ):
+            if ledger[2] != len(result["ledgers"][ledger[0]]):
+                raise ValueError("stored ledger ordering mismatch")
+            result["ledgers"][ledger[0]].append(json.loads(ledger[1]))
+        validate(result)
+        if digest(result) != row["manifest_hash"]:
+            raise ValueError("stored experiment integrity check failed")
+        return result
+
+    def summary(self, run_id):
+        self.load(run_id)  # Validate stored content before using the report projections.
+        return dict(self.connection.execute((SQL / "summary.sql").read_text(), {"run_id": run_id}).fetchone())
+
+    def compare(self, run_ids, sensitivity=False):
+        if not run_ids:
+            raise ValueError("at least one run is required")
+        runs = [self.load(run_id) for run_id in run_ids]
+        if len({comparison_key(run["metadata"], sensitivity) for run in runs}) != 1:
+            raise ValueError("incompatible data, timing, capital, costs, provenance or engine versions")
+        return [
+            dict(
+                run_id=r["run_id"],
+                slippage_bps=r["metadata"]["config"]["slippage_bps"],
+                **self.summary(r["run_id"]),
+            )
+            for r in runs
+        ]
+
+    def cost_sensitivity(self, run_id):
+        run = self.load(run_id)
+        ids = [
+            row[0]
+            for row in self.connection.execute(
+                "SELECT run_id FROM runs WHERE sensitivity_key = ? ORDER BY slippage_bps, run_id",
+                (comparison_key(run["metadata"], True),),
+            )
+        ]
+        return self.compare(ids, sensitivity=True)
+
+    def query_plan(self, run_id):
+        return [
+            tuple(row)
+            for row in self.connection.execute(
+                "EXPLAIN QUERY PLAN " + (SQL / "summary.sql").read_text(), {"run_id": run_id}
+            )
+        ]
diff --git a/fixtures/synthetic/README.md b/fixtures/synthetic/README.md
new file mode 100644
index 0000000..acfc4a6
--- /dev/null
+++ b/fixtures/synthetic/README.md
@@ -0,0 +1,25 @@
+# Synthetic fixture v1
+
+`ohlc.csv` contains invented prices for two symbols over four daily bars. It is
+not historical or live market data. Open and Close drive the engine; High and
+Low make the fixture a readable OHLC source but are not execution inputs.
+
+From the repository root after installing dependencies:
+
+```bash
+python run_offline.py
+python run_offline.py > fixtures/synthetic/expected.txt
+python -m unittest discover -s tests -v
+```
+
+The second command intentionally regenerates the committed golden output.
+Review changes to it; do not regenerate merely to silence a failing test.
+Float output is formatted to six decimals for readability. Replay compares the
+full serialized float values, events and all seven ledgers.
+
+A closes at 10 on January 1. Its buy fills at the January 2 open of 12, plus
+10 bps adverse slippage: 8 shares at 12.012 with a 1.00 commission. Cash becomes
+2.904; marking 8 shares at the close of 13 produces equity of 106.904.
+B's competing buy is rejected for insufficient cash. A exits at the January 3
+open with adverse sell slippage and a second commission; final cash and equity
+are 113.792. These invented prices demonstrate accounting, not strategy returns.
diff --git a/fixtures/synthetic/expected.txt b/fixtures/synthetic/expected.txt
new file mode 100644
index 0000000..612f1fd
--- /dev/null
+++ b/fixtures/synthetic/expected.txt
@@ -0,0 +1,52 @@
+SYNTHETIC OHLC: 2 symbols x 4 daily bars; initial cash 100.00
+Timing: close[t] signal -> next common open fill -> close mark.
+A's Jan 1 close is 10.00; its Jan 2 open is 12.00.
+
+ORDERS
+order_id,submitted_dt,symbol,direction,quantity,reference_price,status
+O00001,2024-01-01,A,BUY,9,10.000000,partially_filled
+O00002,2024-01-02,A,SELL,8,13.000000,filled
+
+REJECTIONS
+dt,order_id,symbol,quantity,reason
+2024-01-01,,A,1,partially_sized_for_cash
+2024-01-01,,B,5,insufficient_cash
+2024-01-02,O00001,A,1,gap_exceeded_cash
+
+FILLS
+order_id,submitted_dt,fill_dt,symbol,direction,quantity,reference_price,fill_price,commission,slippage_cost,cash_flow
+O00001,2024-01-01,2024-01-02,A,BUY,8,12.000000,12.012000,1.000000,0.096000,-97.096000
+O00002,2024-01-02,2024-01-03,A,SELL,8,14.000000,13.986000,1.000000,0.112000,110.888000
+
+CASH
+dt,cash,reserved_cash,equity
+2024-01-01,100.000000,92.800000,100.000000
+2024-01-02,2.904000,0.000000,106.904000
+2024-01-03,113.792000,0.000000,113.792000
+2024-01-04,113.792000,0.000000,113.792000
+
+POSITIONS
+dt,symbol,quantity,close,market_value
+2024-01-01,A,0,10.000000,0.000000
+2024-01-01,B,0,20.000000,0.000000
+2024-01-02,A,8,13.000000,104.000000
+2024-01-02,B,0,22.000000,0.000000
+2024-01-03,A,0,12.000000,0.000000
+2024-01-03,B,0,24.000000,0.000000
+2024-01-04,A,0,12.000000,0.000000
+2024-01-04,B,0,26.000000,0.000000
+
+EQUITY
+dt,equity
+2024-01-01,100.000000
+2024-01-02,106.904000
+2024-01-03,113.792000
+2024-01-04,113.792000
+
+TIMELINE
+2024-01-01 CLOSE cash=100.000000 reserved=92.800000 equity=100.000000
+2024-01-02 OPEN BUY 8 A at 12.012: cash flow -97.096000
+2024-01-02 CLOSE cash=2.904000 reserved=0.000000 equity=106.904000
+2024-01-03 OPEN SELL 8 A at 13.986: cash flow 110.888000
+2024-01-03 CLOSE cash=113.792000 reserved=0.000000 equity=113.792000
+2024-01-04 CLOSE cash=113.792000 reserved=0.000000 equity=113.792000
diff --git a/fixtures/synthetic/ohlc.csv b/fixtures/synthetic/ohlc.csv
new file mode 100644
index 0000000..f109517
--- /dev/null
+++ b/fixtures/synthetic/ohlc.csv
@@ -0,0 +1,9 @@
+date,symbol,Open,High,Low,Close
+2024-01-01,A,9,10,9,10
+2024-01-01,B,19,20,19,20
+2024-01-02,A,12,13,11,13
+2024-01-02,B,21,22,20,22
+2024-01-03,A,14,14,12,12
+2024-01-03,B,23,24,22,24
+2024-01-04,A,11,12,10,12
+2024-01-04,B,25,26,24,26
diff --git a/pyproject.toml b/pyproject.toml
index 2c1f310..dcf4959 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -17,3 +17,6 @@ include = ["engine*"]
 
 [tool.ruff]
 line-length = 110
+
+[tool.setuptools.package-data]
+engine = ["sql/*.sql"]
diff --git a/run_cross_sectional.py b/run_cross_sectional.py
index 803eb1c..6e22c4d 100644
--- a/run_cross_sectional.py
+++ b/run_cross_sectional.py
@@ -10,7 +10,6 @@
 from engine.metrics import performance
 from engine.strategy import CrossSectionalMomentum, EqualWeightBuyAndHold
 
-
 SYMBOLS = ["XLB", "XLE", "XLF", "XLI", "XLK", "XLP", "XLU", "XLV", "XLY"]
 CAPITAL = 100_000.0
 
diff --git a/run_experiments.py b/run_experiments.py
new file mode 100644
index 0000000..b53ecaa
--- /dev/null
+++ b/run_experiments.py
@@ -0,0 +1,73 @@
+"""Offline experiment storage, replay and comparable-run reporting."""
+
+import argparse
+import json
+
+from engine.backtest import Backtest
+from engine.demo import DemoStrategy, demo_data, render_demo
+from engine.experiments import record_run, replay
+from engine.storage import SQLiteStore
+from engine.strategy import EqualWeightBuyAndHold
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--db", default="experiments.sqlite")
+    sub = parser.add_subparsers(dest="command", required=True)
+    sub.add_parser("demo")
+    replayer = sub.add_parser("replay")
+    replayer.add_argument("run_id")
+    compare = sub.add_parser("compare")
+    compare.add_argument("run_ids", nargs="+")
+    costs = sub.add_parser("costs")
+    costs.add_argument("run_id")
+    plan = sub.add_parser("plan")
+    plan.add_argument("run_id")
+    args = parser.parse_args()
+    with SQLiteStore(args.db) as store:
+        if args.command == "demo":
+            for bps in (0, 1, 5, 10):
+                bt = Backtest(
+                    ["A", "B"],
+                    price_data=demo_data(),
+                    initial_capital=100,
+                    strategy_cls=DemoStrategy,
+                    commission_per_share=0,
+                    minimum_commission=1,
+                    slippage_bps=bps,
+                    calendar_policy="strict",
+                    provenance={"kind": "synthetic", "source": "fixtures/synthetic/ohlc.csv v1"},
+                )
+                result = record_run(bt, store)
+                replay(store, bt.run_id)
+                print(f"{bps} bps: {bt.run_id} (stored and exact replay verified)")
+            print(render_demo(result))
+            print(json.dumps(store.cost_sensitivity(bt.run_id), indent=2))
+            baseline = Backtest(
+                ["A", "B"],
+                price_data=demo_data(),
+                initial_capital=100,
+                strategy_cls=EqualWeightBuyAndHold,
+                commission_per_share=0,
+                minimum_commission=1,
+                slippage_bps=10,
+                calendar_policy="strict",
+                provenance={"kind": "synthetic", "source": "fixtures/synthetic/ohlc.csv v1"},
+            )
+            record_run(baseline, store)
+            replay(store, baseline.run_id)
+            print("SAME-DATA, SAME-COST COMPARISON: scripted accounting example vs equal-weight hold")
+            print(json.dumps(store.compare([bt.run_id, baseline.run_id]), indent=2))
+        elif args.command == "replay":
+            replay(store, args.run_id)
+            print(f"Exact replay verified: {args.run_id}")
+        elif args.command == "compare":
+            print(json.dumps(store.compare(args.run_ids), indent=2))
+        elif args.command == "costs":
+            print(json.dumps(store.cost_sensitivity(args.run_id), indent=2))
+        else:
+            print(json.dumps(store.query_plan(args.run_id), indent=2))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/run_offline.py b/run_offline.py
new file mode 100644
index 0000000..e5f0177
--- /dev/null
+++ b/run_offline.py
@@ -0,0 +1,6 @@
+"""Run the fully offline, versioned synthetic demonstration."""
+
+from engine.demo import demo_backtest, render_demo
+
+if __name__ == "__main__":
+    print(render_demo(demo_backtest().run()), end="")
diff --git a/tests/test_benchmark.py b/tests/test_benchmark.py
new file mode 100644
index 0000000..f99936c
--- /dev/null
+++ b/tests/test_benchmark.py
@@ -0,0 +1,14 @@
+import unittest
+
+from benchmark import check_accounting, generate_data, make_backtest
+from engine.reproducibility import digest, ledger_records
+
+
+class BenchmarkTests(unittest.TestCase):
+    def test_seeded_workload_is_reproducible_and_conserves_cash(self):
+        data = generate_data(3, 90, 17)
+        first = make_backtest(data, 17).run()
+        second = make_backtest(generate_data(3, 90, 17), 17).run()
+        check_accounting(first, 1_000_000)
+        self.assertEqual(digest(ledger_records(first)), digest(ledger_records(second)))
+        self.assertGreater(len(first["fills"]), 0)
diff --git a/tests/test_demo.py b/tests/test_demo.py
new file mode 100644
index 0000000..1355afe
--- /dev/null
+++ b/tests/test_demo.py
@@ -0,0 +1,19 @@
+import unittest
+from pathlib import Path
+
+from engine.demo import demo_backtest, render_demo
+
+
+class DemoTests(unittest.TestCase):
+    def test_golden_output_and_conservation(self):
+        result = demo_backtest().run()
+        expected = Path(__file__).resolve().parents[1] / "fixtures/synthetic/expected.txt"
+        self.assertEqual(render_demo(result), expected.read_text())
+        self.assertIn("insufficient_cash", set(result["rejections"].reason))
+        self.assertEqual(list(result["fills"].direction), ["BUY", "SELL"])
+        cash = 100.0
+        for row in result["cash"].itertuples():
+            cash += result["fills"].loc[result["fills"].fill_dt == row.dt, "cash_flow"].sum()
+            self.assertAlmostEqual(row.cash, cash)
+            marked = result["positions"].loc[result["positions"].dt == row.dt, "market_value"].sum()
+            self.assertAlmostEqual(row.equity, cash + marked)
diff --git a/tests/test_experiments.py b/tests/test_experiments.py
new file mode 100644
index 0000000..844024c
--- /dev/null
+++ b/tests/test_experiments.py
@@ -0,0 +1,321 @@
+import queue
+import sqlite3
+import tempfile
+import unittest
+from copy import deepcopy
+from pathlib import Path
+
+import pandas as pd
+
+from engine.backtest import Backtest
+from engine.data import DataHandler
+from engine.demo import DemoStrategy, demo_backtest, demo_data
+from engine.experiments import record_run, replay
+from engine.metrics import performance
+from engine.reproducibility import canonical
+from engine.storage import MemoryStore, SQLiteStore, snapshot
+from engine.strategy import EqualWeightBuyAndHold
+
+
+class ExperimentTests(unittest.TestCase):
+    def test_memory_and_sqlite_roundtrip_and_exact_replay(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "runs.sqlite"
+            bt = demo_backtest()
+            memory = MemoryStore()
+            results = record_run(bt, memory)
+            experiment = memory.load(bt.run_id)
+            with SQLiteStore(path) as sql:
+                sql.save(experiment)
+                sql.save(experiment)  # Idempotent; never duplicates rows.
+                self.assertEqual(sql.load(bt.run_id), experiment)
+                stats = sql.summary(bt.run_id)
+                metrics, _ = performance(results["equity"], 100)
+                for key in ("final_equity", "max_drawdown"):
+                    self.assertAlmostEqual(stats[key], metrics[key])
+                self.assertAlmostEqual(stats["commission"], results["fills"].commission.sum())
+                self.assertAlmostEqual(stats["slippage_cost"], results["fills"].slippage_cost.sum())
+                self.assertEqual(stats["fills"], len(results["fills"]))
+                self.assertTrue(any("INDEX" in str(row) for row in sql.query_plan(bt.run_id)))
+            with SQLiteStore(path) as reopened:
+                for name, frame in replay(reopened, bt.run_id).items():
+                    pd.testing.assert_frame_equal(frame, results[name])
+            self.assertEqual(canonical(snapshot(bt, replay(memory, bt.run_id))), canonical(experiment))
+
+    def test_hashes_and_events_are_deterministic_and_actual(self):
+        first, second = demo_backtest(), demo_backtest()
+        first.run()
+        second.run()
+        self.assertEqual(first.run_id, second.run_id)
+        self.assertEqual(first.event_records, second.event_records)
+        self.assertEqual([r.sequence for r in first.event_records], list(range(len(first.event_records))))
+        fills = [r.event for r in first.event_records if r.event.type == "FILL_BOOKED"]
+        self.assertEqual(fills[0].fill.quantity, 8)
+        self.assertEqual(fills[0].fill.commission, 1)
+        self.assertEqual(first.portfolio.positions, {"A": 0, "B": 0})
+        with self.assertRaisesRegex(RuntimeError, "single-use"):
+            first.run()
+
+    def test_transaction_rolls_back_on_invalid_ledger(self):
+        bt = demo_backtest()
+        experiment = snapshot(bt, bt.run())
+        experiment["ledgers"]["cash"][0]["cash"] = -100
+        with SQLiteStore(":memory:") as sql:
+            with self.assertRaises(sqlite3.IntegrityError):
+                sql.save(experiment)
+            for table in ("runs", "events", "ledgers"):
+                self.assertEqual(sql.connection.execute(f"SELECT count(*) FROM {table}").fetchone()[0], 0)
+            self.assertEqual(sql.connection.execute("PRAGMA foreign_key_check").fetchall(), [])
+
+    def test_integrity_and_identity_collision(self):
+        bt = demo_backtest()
+        experiment = snapshot(bt, bt.run())
+        for store in (MemoryStore(), SQLiteStore(":memory:")):
+            try:
+                store.save(experiment)
+                changed = deepcopy(experiment)
+                changed["ledgers"]["cash"][0]["cash"] += 1
+                with self.assertRaisesRegex(ValueError, "non-deterministic"):
+                    store.save(changed)
+                changed = deepcopy(experiment)
+                changed["inputs"]["A"][0]["Open"] += 1
+                with self.assertRaisesRegex(ValueError, "input hash"):
+                    store.save(changed)
+            finally:
+                if isinstance(store, SQLiteStore):
+                    store.close()
+        with SQLiteStore(":memory:") as sql:
+            sql.save(experiment)
+            sql.connection.execute("DELETE FROM events WHERE run_id = ? AND sequence = 0", (bt.run_id,))
+            with self.assertRaisesRegex(ValueError, "sequence"):
+                sql.load(bt.run_id)
+
+    def test_replay_rejects_unregistered_or_changed_strategy(self):
+        bt = demo_backtest()
+        store = MemoryStore()
+        record_run(bt, store)
+        with self.assertRaisesRegex(ValueError, "registered"):
+            replay(store, bt.run_id, {})
+        with self.assertRaisesRegex(ValueError, "matching"):
+            replay(store, bt.run_id, {bt.config["strategy"]: EqualWeightBuyAndHold})
+
+    def test_comparisons_and_cost_sensitivity(self):
+        with SQLiteStore(":memory:") as sql:
+            ids = []
+            for bps in (0, 10):
+                bt = Backtest(
+                    ["A", "B"],
+                    price_data=demo_data(),
+                    strategy_cls=DemoStrategy,
+                    initial_capital=100,
+                    slippage_bps=bps,
+                )
+                record_run(bt, sql)
+                ids.append(bt.run_id)
+            with self.assertRaisesRegex(ValueError, "incompatible"):
+                sql.compare(ids)
+            sensitivity = sql.cost_sensitivity(ids[0])
+            self.assertEqual([row["slippage_bps"] for row in sensitivity], [0, 10])
+            self.assertGreater(sensitivity[0]["final_equity"], sensitivity[1]["final_equity"])
+            hold = Backtest(
+                ["A", "B"],
+                price_data=demo_data(),
+                strategy_cls=EqualWeightBuyAndHold,
+                initial_capital=100,
+                slippage_bps=0,
+            )
+            record_run(hold, sql)
+            self.assertEqual(len(sql.compare([ids[0], hold.run_id])), 2)
+            changed = demo_data()
+            changed["A"].iloc[0, 0] += 1
+            other = Backtest(
+                ["A", "B"], price_data=changed, strategy_cls=DemoStrategy, initial_capital=100, slippage_bps=0
+            )
+            record_run(other, sql)
+            with self.assertRaisesRegex(ValueError, "incompatible"):
+                sql.compare([ids[0], other.run_id])
+            with self.assertRaises(KeyError):
+                sql.load("' OR 1=1 --")
+
+    def test_drawdown_reconciles_on_falling_prices(self):
+        data = demo_data()
+        data["A"]["Close"] = [10, 8, 6, 4]
+        bt = Backtest(["A"], price_data=data, initial_capital=100, strategy_cls=EqualWeightBuyAndHold)
+        with SQLiteStore(":memory:") as sql:
+            results = record_run(bt, sql)
+            expected, _ = performance(results["equity"], 100)
+            self.assertLess(expected["max_drawdown"], 0)
+            self.assertAlmostEqual(sql.summary(bt.run_id)["max_drawdown"], expected["max_drawdown"])
+
+    def test_common_calendar_replay_retains_exclusions(self):
+        data = demo_data()
+        data["B"] = data["B"].iloc[1:]
+        bt = Backtest(["A", "B"], price_data=data, strategy_cls=DemoStrategy)
+        store = MemoryStore()
+        record_run(bt, store)
+        self.assertEqual(bt.metadata["excluded_dates"]["A"], ["2024-01-01T00:00:00"])
+        replay(store, bt.run_id)
+
+
+class DataPolicyTests(unittest.TestCase):
+    def test_missing_internal_or_delisted_tail_strict_fails(self):
+        for index in (1, 3):
+            data = demo_data()
+            data["B"] = data["B"].drop(data["B"].index[index])
+            with self.assertRaisesRegex(ValueError, "missing bars or delisting"):
+                DataHandler(queue.Queue(), ["A", "B"], price_data=data, calendar_policy="strict")
+
+    def test_nonfinite_duplicate_stale_and_actions_fail(self):
+        for value in (float("nan"), float("inf"), 0, -1):
+            data = demo_data()
+            data["A"] = data["A"].astype(float)
+            data["A"].iloc[1, 0] = value
+            with self.assertRaises(ValueError):
+                DataHandler(queue.Queue(), ["A"], price_data=data)
+        for flag in ("Stale", "Delisted", "Dividends", "Stock Splits"):
+            data = demo_data()
+            data["A"][flag] = [0, 1, 0, 0]
+            with self.assertRaisesRegex(ValueError, "unsupported data flag"):
+                DataHandler(queue.Queue(), ["A"], price_data=data)
+        data = demo_data()
+        data["A"] = pd.concat([data["A"], data["A"].iloc[:1]])
+        with self.assertRaisesRegex(ValueError, "duplicate"):
+            DataHandler(queue.Queue(), ["A"], price_data=data)
+
+    def test_repeated_valid_prices_are_not_inferred_to_be_stale(self):
+        data = demo_data()
+        data["A"][["Open", "Close"]] = 10
+        handler = DataHandler(queue.Queue(), ["A"], price_data=data, calendar_policy="strict")
+        self.assertEqual(len(handler.dates), 4)
+
+    def test_invalid_costs_and_capital(self):
+        for config in (
+            {"initial_capital": 0},
+            {"initial_capital": float("inf")},
+            {"slippage_bps": -1},
+            {"slippage_bps": 10000},
+            {"minimum_commission": float("nan")},
+        ):
+            with self.assertRaises(ValueError):
+                Backtest(["A"], price_data=demo_data(), strategy_cls=DemoStrategy, **config)
+
+
+class AccountingBoundaryTests(unittest.TestCase):
+    def test_final_cancellation_releases_reserved_cash(self):
+        from test_engine import BuyThenExit
+
+        bt = Backtest(["A"], price_data=demo_data(), strategy_cls=BuyThenExit, strategy_kwargs={"buy_bar": 3})
+        result = bt.run()
+        self.assertTrue(result["fills"].empty)
+        self.assertEqual(result["cash"].iloc[-1].reserved_cash, 0)
+        self.assertEqual(bt.event_records[-1].event.reserved_cash, 0)
+        self.assertEqual(result["orders"].iloc[-1].status, "cancelled")
+
+    def test_partial_fill_uses_actual_per_share_fee(self):
+        from test_engine import BuyThenExit, bars
+
+        bt = Backtest(
+            ["A"],
+            price_data={"A": bars([1, 10], [1, 10])},
+            strategy_cls=BuyThenExit,
+            initial_capital=100,
+            commission_per_share=1,
+            minimum_commission=1,
+            slippage_bps=0,
+        )
+        result = bt.run()
+        fill = result["fills"].iloc[0]
+        self.assertEqual(fill.quantity, 9)
+        self.assertEqual(fill.commission, 9)
+        self.assertEqual(result["cash"].iloc[-1].cash, 1)
+
+    def test_sale_fee_cannot_create_negative_cash(self):
+        from test_engine import BuyThenExit, bars
+
+        bt = Backtest(
+            ["A"],
+            price_data={"A": bars([1, 1, 0.001], [1, 1, 0.001])},
+            strategy_cls=BuyThenExit,
+            strategy_kwargs={"exit_bar": 1, "weight": 1},
+            initial_capital=100,
+            commission_per_share=0,
+            minimum_commission=10,
+            slippage_bps=0,
+        )
+        result = bt.run()
+        self.assertGreaterEqual(result["cash"].cash.min(), 0)
+        self.assertIn("sale_cannot_cover_commission", result["rejections"].reason.values)
+        self.assertGreater(result["positions"].iloc[-1].quantity, 0)
+
+    def test_event_fills_reduce_to_ledger_cash_and_positions(self):
+        bt = demo_backtest()
+        bt.run()
+        cash, positions = 100, {"A": 0, "B": 0}
+        for record in bt.event_records:
+            event = record.event
+            if event.type == "FILL_BOOKED":
+                fill = event.fill
+                sign = 1 if fill.direction == "BUY" else -1
+                positions[fill.symbol] += sign * fill.quantity
+                cash -= sign * fill.quantity * fill.fill_price + fill.commission
+                self.assertAlmostEqual(event.cash, cash)
+                self.assertEqual(event.position, positions[fill.symbol])
+                self.assertGreaterEqual(cash, 0)
+            elif event.type == "VALUATION":
+                marked = sum(positions[symbol] * close for symbol, quantity, close in event.positions)
+                self.assertAlmostEqual(event.equity, cash + marked)
+
+
+class PersistenceConstraintTests(unittest.TestCase):
+    def test_foreign_keys_uniqueness_and_projection_constraints(self):
+        bt = demo_backtest()
+        experiment = snapshot(bt, bt.run())
+        with SQLiteStore(":memory:") as sql:
+            sql.save(experiment)
+            with self.assertRaises(sqlite3.IntegrityError):
+                sql.connection.execute(
+                    "INSERT INTO events VALUES (?, ?, ?, ?)",
+                    ("missing-run", 0, "SIGNAL", '{"type":"SIGNAL"}'),
+                )
+            with self.assertRaises(sqlite3.IntegrityError):
+                sql.connection.execute(
+                    "INSERT INTO events SELECT * FROM events WHERE run_id = ? AND sequence = 0",
+                    (bt.run_id,),
+                )
+            with self.assertRaises(sqlite3.IntegrityError):
+                sql.connection.execute(
+                    "UPDATE ledgers SET equity = equity + 100 WHERE run_id = ? AND ledger = 'cash'",
+                    (bt.run_id,),
+                )
+            self.assertEqual(sql.load(bt.run_id), experiment)
+
+    def test_no_trade_run_replays_with_zero_costs(self):
+        from engine.strategy import MovingAverageCross
+
+        bt = Backtest(["A"], price_data=demo_data(), strategy_cls=MovingAverageCross)
+        with SQLiteStore(":memory:") as sql:
+            record_run(bt, sql)
+            result = replay(sql, bt.run_id)
+            self.assertTrue(result["fills"].empty)
+            summary = sql.summary(bt.run_id)
+            self.assertEqual(summary["commission"], 0)
+            self.assertEqual(summary["fills"], 0)
+            self.assertEqual(summary["max_drawdown"], 0)
+
+    def test_common_calendar_never_fills_an_absent_date(self):
+        data = demo_data()
+        data["B"] = data["B"].drop(data["B"].index[1])
+        bt = Backtest(["A", "B"], price_data=data, strategy_cls=DemoStrategy, initial_capital=100)
+        result = bt.run()
+        self.assertEqual(result["fills"].iloc[0].fill_dt, pd.Timestamp("2024-01-03"))
+        self.assertEqual(result["fills"].iloc[0].reference_price, 14)
+        self.assertEqual(bt.metadata["excluded_dates"]["A"], ["2024-01-02T00:00:00"])
+
+    def test_offline_demo_and_stored_replay_do_not_open_network(self):
+        from unittest.mock import patch
+
+        with patch("socket.socket.connect", side_effect=AssertionError("network forbidden")):
+            bt = demo_backtest()
+            with SQLiteStore(":memory:") as sql:
+                record_run(bt, sql)
+                replay(sql, bt.run_id)