diff --git a/docs/trading/durable-paper.mdx b/docs/trading/durable-paper.mdx index 058f85f5..51e923a1 100644 --- a/docs/trading/durable-paper.mdx +++ b/docs/trading/durable-paper.mdx @@ -62,8 +62,56 @@ available through `inspect`. Queued model-1 jobs fail explicitly as unsupported; submit the original strategy, recording and assumptions again to run model 2. Old results are not silently recalculated or relabeled with the new model. +## Read-only experiment queries + +The experimental `neural.paper_query.PaperJournal` API lets consumers use saved +experiments without reading Neural's tables or starting a worker: + +```python +from neural.paper_query import PaperJournal + +journal = PaperJournal("/private/path/paper.sqlite3") +page = journal.history(offset=0, limit=25) # jobs, has_more; no result traces +if page["jobs"]: + job_id = page["jobs"][0]["id"] + saved = journal.inspect(job_id) # id, status, error, config, result + original_bytes = journal.recording(job_id) +``` + +Queries open an existing journal in SQLite read-only mode, without acquiring a +writer reservation, creating a queue, migrating its schema or running simulation. +Missing files raise `FileNotFoundError`; unknown jobs, invalid stored data and +unsupported journal versions raise `ValueError`. History orders by insertion, +newest first; offset is 0–100,000 and page size is 1–100. No timestamps are invented. +Reading preserves `sqlite3.OperationalError` for database-busy contention if SQLite +holds an exclusive lock, so callers can distinguish it from invalid stored data. +Only rollback-journal databases are supported. WAL-format journals are rejected +before SQLite opens them, preventing creation of WAL/SHM sidecars. The caller must +not change journal mode or replace the database during a read. + +`recording(job_id)` verifies the stored configuration/job identity and the original +recording's size and digest before returning its exact bytes. +Results are not decoded when recovering recording bytes; a damaged output does not +prevent recovery of an intact, verified input. Inspection preserves +historical results as recorded; it does not recalculate them or certify their +correctness. The caller selects the runtime archive and verifies its wheel and +environment identity. This API does not authenticate users or grant data rights. + +`compare(first_id, second_id)` returns two completed jobs with derived +`total_fees`, `fill_count` and `rejection_count` in a separate `summary` object on +each job. Saved `result` objects and their `result_id` values remain unchanged. Recording, +model, market/outcome, quantity, cash, fees and simulation limits must match; strategy +thresholds and risk limits may differ. Summaries use exact decimals and never +write back to the journal. Missing historical traces reject comparison explicitly. + +`compare_markets(first_id, second_id)` separately returns `ids`, supplied `markets` +and a sports `comparison`. It can inspect cross-venue contract metadata, but never +compares performance or execution costs. Missing sports metadata rejects the +request. Keep the existing `PaperJobs` write/worker API for submission and processing; +its methods and the journal format are unchanged. + ```sh -pytest tests/test_paper_worker.py -q +pytest tests/test_paper_worker.py tests/test_paper_query.py -q ``` Tests exercise CLI submission/inspection, concurrent processes and process death diff --git a/neural/paper_query.py b/neural/paper_query.py new file mode 100644 index 00000000..64b7e1e7 --- /dev/null +++ b/neural/paper_query.py @@ -0,0 +1,247 @@ +"""Read-only views of Neural's local paper journal; never run or migrate jobs.""" + +from __future__ import annotations + +import hashlib +import json +import math +import re +import sqlite3 +from collections.abc import Iterator +from contextlib import contextmanager +from decimal import Decimal, localcontext +from pathlib import Path +from typing import Any + +from neural.paper_worker import APPLICATION_ID, MAX_INPUT_BYTES +from neural.sports import SportsMarket, compare_sports_markets +from neural.strategy import StrategySpec, _decimal_text + + +class PaperJournal: + """Inspect an existing v1 journal without a writer lock or worker startup. + + Missing files raise FileNotFoundError; malformed/unsupported journals and + inputs raise ValueError. Historical model results remain opaque on reads. + The caller owns the journal path and selects the verified runtime archive. + """ + + def __init__(self, database: str | Path): + """Select an existing local journal path without opening or creating it.""" + if str(database) in ("", ":memory:"): + raise ValueError("paper journal requires an existing database file") + self.database = Path(database).resolve() + + @contextmanager + def _connection(self) -> Iterator[sqlite3.Connection]: + """Open a read snapshot and reject journals outside the supported schema.""" + if not self.database.exists(): + raise FileNotFoundError("paper job database not found") + with self.database.open("rb") as source: + header = source.read(20) + if header.startswith(b"SQLite format 3\x00") and header[18:20] != b"\x01\x01": + raise ValueError("WAL paper journals are unsupported; use a rollback-journal archive") + db = None + try: + db = sqlite3.connect(self.database.as_uri() + "?mode=ro", uri=True, timeout=5) + db.row_factory = sqlite3.Row + db.execute("PRAGMA query_only=ON") + db.execute("BEGIN") + if ( + db.execute("PRAGMA application_id").fetchone()[0] != APPLICATION_ID + or db.execute("PRAGMA user_version").fetchone()[0] != 1 + ): + raise ValueError("unsupported paper job database") + yield db + except sqlite3.Error as exc: + if isinstance(exc, sqlite3.OperationalError) and str(exc) in ( + "database is locked", + "database table is locked", + ): + raise + raise ValueError("cannot read paper job database") from exc + finally: + if db is not None: + db.close() + + @staticmethod + def _reject_constant(value: str) -> Any: + """Reject nonstandard JSON numbers that cannot be written by the worker.""" + raise ValueError("stored job JSON contains a non-finite constant") + + @staticmethod + def _finite_float(value: str) -> float: + """Decode standard JSON numbers without allowing floating-point overflow.""" + number = float(value) + if not math.isfinite(number): + raise ValueError("stored job JSON contains a non-finite number") + return number + + @staticmethod + def _view(row: sqlite3.Row) -> dict[str, Any]: + """Decode a saved row while checking configuration identity and state.""" + try: + if not isinstance(row["config"], str): + raise ValueError("stored job configuration is invalid") + config = json.loads( + row["config"], + parse_constant=PaperJournal._reject_constant, + parse_float=PaperJournal._finite_float, + ) + if ( + not isinstance(config, dict) + or hashlib.sha256(row["config"].encode()).hexdigest() != row["id"] + or row["status"] not in ("queued", "completed", "failed") + ): + raise ValueError("stored job input integrity mismatch") + view = {key: row[key] for key in ("id", "status", "error")} + if (row["error"] is not None and not isinstance(row["error"], str)) or ( + (row["status"] == "failed") != (row["error"] is not None) + ): + raise ValueError("stored job error is invalid") + view["config"] = config + if "has_result" in row.keys() and ( + (row["status"] == "completed") != bool(row["has_result"]) + ): + raise ValueError("stored job result is invalid") + if "result" in row.keys(): + if row["result"] is not None and not isinstance(row["result"], str): + raise ValueError("stored job result is invalid") + result = ( + json.loads( + row["result"], + parse_constant=PaperJournal._reject_constant, + parse_float=PaperJournal._finite_float, + ) + if row["result"] is not None + else None + ) + if (result is not None and not isinstance(result, dict)) or ( + (row["status"] == "completed") != (result is not None) + ): + raise ValueError("stored job result is invalid") + view["result"] = result + return view + except (TypeError, json.JSONDecodeError) as exc: + raise ValueError("stored job data is invalid") from exc + + @classmethod + def _job( + cls, db: sqlite3.Connection, identity: str, *, include_result: bool = True + ) -> dict[str, Any]: + """Read one saved job, optionally excluding its unrelated output.""" + if not isinstance(identity, str): + raise ValueError("job identity must be a string") + fields = "id,status,error,config,result" if include_result else "id,status,error,config" + row = db.execute(f"SELECT {fields} FROM jobs WHERE id=?", (identity,)).fetchone() + if row is None: + raise ValueError("unknown paper job") + return cls._view(row) + + def history(self, *, offset: int = 0, limit: int = 25) -> dict[str, Any]: + """Newest inserted jobs first, without result traces or invented timestamps.""" + if type(offset) is not int or not 0 <= offset <= 100000: + raise ValueError("offset must be an integer from 0 to 100000") + if type(limit) is not int or not 1 <= limit <= 100: + raise ValueError("limit must be an integer from 1 to 100") + with self._connection() as db: + rows = db.execute( + "SELECT id,status,error,config,result IS NOT NULL AS has_result " + "FROM jobs ORDER BY rowid DESC LIMIT ? OFFSET ?", + (limit + 1, offset), + ).fetchall() + return { + "jobs": [self._view(row) for row in rows[:limit]], + "has_more": len(rows) > limit, + } + + def inspect(self, identity: str) -> dict[str, Any]: + """Return saved configuration and result; do not execute or reinterpret it.""" + with self._connection() as db: + return self._job(db, identity) + + def recording(self, identity: str) -> bytes: + """Return original input bytes after checking size and the saved digest.""" + with self._connection() as db: + job = self._job(db, identity, include_result=False) + raw = db.execute( + "SELECT CASE WHEN length(recording) BETWEEN 1 AND ? THEN recording END " + "FROM jobs WHERE id=?", + (MAX_INPUT_BYTES, identity), + ).fetchone()[0] + if not isinstance(raw, bytes) or hashlib.sha256(raw).hexdigest() != job["config"].get( + "recording_sha256" + ): + raise ValueError("stored job recording integrity mismatch") + return raw + + def _pair(self, first: str, second: str) -> list[dict[str, Any]]: + """Read two distinct completed jobs from a single consistent snapshot.""" + if first == second: + raise ValueError("choose two distinct completed experiments") + with self._connection() as db: + jobs = [self._job(db, identity) for identity in (first, second)] + if any(job["status"] != "completed" for job in jobs): + raise ValueError("choose two distinct completed experiments") + return jobs + + @staticmethod + def _assumptions(job: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: + """Separate fixed replay inputs from tunable strategy thresholds and caps.""" + config = dict(job["config"]) + strategy = config.pop("strategy", None) + StrategySpec.from_dict(strategy) + return config, { + key: strategy[key] + for key in ("schema_version", "kind", "venue", "market_id", "outcome", "quantity") + } + + def compare(self, first: str, second: str) -> list[dict[str, Any]]: + """Compare strategy variants only when their recorded inputs/assumptions match.""" + jobs = self._pair(first, second) + if self._assumptions(jobs[0]) != self._assumptions(jobs[1]): + raise ValueError( + "comparison requires the same recording, model, market, outcome, " + "quantity, cash, fees and simulation limits" + ) + for job in jobs: + trace = job["result"].get("trace") + if not isinstance(trace, list) or any( + not isinstance(event, dict) or not isinstance(event.get("action"), str) + for event in trace + ): + raise ValueError("saved comparison trace unavailable or invalid") + with localcontext() as context: + context.prec = 100 + total_fees = sum((self._trace_fee(event) for event in trace), Decimal(0)) + job["summary"] = { + "total_fees": _decimal_text(total_fees), + "fill_count": sum(event["action"] == "fill" for event in trace), + "rejection_count": sum(event["action"] == "reject" for event in trace), + } + return jobs + + @staticmethod + def _trace_fee(event: dict[str, Any]) -> Decimal: + """Parse products of two 18-digit inputs without truncating their scale.""" + value = event.get("fees") if event["action"] == "fill" else event.get("fees", "0") + if ( + not isinstance(value, str) + or re.fullmatch(r"(0|[1-9][0-9]{0,35})(\.[0-9]{1,36})?", value) is None + ): + raise ValueError("saved trace fees must be a finite nonnegative decimal product") + fee = Decimal(value) + if event["action"] != "fill" and fee != 0: + raise ValueError("saved trace fees require a fill event") + return fee + + def compare_markets(self, first: str, second: str) -> dict[str, Any]: + """Inspect sports terms separately; this never compares execution costs/PnL.""" + jobs = self._pair(first, second) + markets = [job["result"].get("sports_market") for job in jobs] + if any(market is None for market in markets): + raise ValueError( + "market terms unavailable: both recordings need sports contract metadata" + ) + comparison = compare_sports_markets(*(SportsMarket.from_dict(market) for market in markets)) + return {"comparison": comparison.to_dict(), "ids": [first, second], "markets": markets} diff --git a/tests/test_paper_query.py b/tests/test_paper_query.py new file mode 100644 index 00000000..4ea3bf80 --- /dev/null +++ b/tests/test_paper_query.py @@ -0,0 +1,309 @@ +"""Read-only consumer behavior over current, historical and damaged journals.""" + +import hashlib +import json +import sqlite3 +from dataclasses import replace +from decimal import localcontext + +import pytest + +import neural.paper_worker as worker +from neural.paper_query import PaperJournal +from tests.test_recordings import inputs + + +def enqueue(jobs, *, venue="polymarket-us", entry="0.45", quantity=None, **assumptions): + spec, path = inputs(venue) + identity = jobs.submit( + replace(spec, entry_price=entry, quantity=quantity or spec.quantity), + path, + **{"initial_cash": "10", "fee_per_contract": "0.01", **assumptions}, + ) + return identity, path + + +def test_reopen_compare_and_source_bytes_across_synthetic_venues(tmp_path): + database = tmp_path / "journal.sqlite3" + jobs = worker.PaperJobs(database) + identities = [] + for venue in ("kalshi", "polymarket-us"): + for entry in ("0.45", "0.46"): + identity, source = enqueue(jobs, venue=venue, entry=entry) + identities.append(identity) + assert jobs.run_next()["status"] == "completed" + before = database.read_bytes() + journal = PaperJournal(database) + page = journal.history(limit=2) + assert [job["id"] for job in page["jobs"]] == list(reversed(identities[2:])) + assert page["has_more"] is True + assert "result" not in page["jobs"][0] + assert journal.history(offset=2, limit=2)["has_more"] is False + assert journal.history(offset=4) == {"jobs": [], "has_more": False} + first, second = identities[2:] + saved = journal.inspect(first) + assert saved["config"]["strategy"]["venue"] == "polymarket_us" + assert saved["result"] == jobs.inspect(first)["result"] + assert journal.recording(first) == source.read_bytes() + with localcontext() as context: + context.prec = 2 + compared = journal.compare(first, second) + assert [job["id"] for job in compared] == [first, second] + for job in compared: + assert job["result"]["cash"] == "10.52" + assert job["result"]["realized_pnl"] == "0.52" + assert job["summary"]["total_fees"] == "0.04" + assert job["summary"]["fill_count"] == 2 + assert job["summary"]["rejection_count"] == 0 + assert job["result"] == jobs.inspect(job["id"])["result"] + report = dict(job["result"]) + result_id = report.pop("result_id") + encoded = json.dumps(report, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + assert hashlib.sha256(encoded.encode()).hexdigest() == result_id + assert "total_fees" not in journal.inspect(first)["result"] + with pytest.raises(ValueError, match="same recording"): + journal.compare(identities[0], first) + terms = journal.compare_markets(identities[0], first) + assert terms["comparison"]["status"] == "compatible" # Supplied synthetic rules only. + assert terms["ids"] == [identities[0], first] + assert database.read_bytes() == before + + +def test_reads_do_not_acquire_writer_lock_or_initialize_journal(tmp_path): + database = tmp_path / "read only?#.sqlite3" + jobs = worker.PaperJobs(database) + identity, _ = enqueue(jobs) + journal = PaperJournal(database) + before = database.read_bytes() + with jobs._transaction(): + assert journal.inspect(identity)["status"] == "queued" + assert journal.history()["jobs"][0]["id"] == identity + assert journal.recording(identity) + assert database.read_bytes() == before + missing = tmp_path / "missing" / "never-created.sqlite3" + with pytest.raises(FileNotFoundError, match="not found"): + PaperJournal(missing).history() + assert not missing.parent.exists() + + +def test_historical_models_stay_opaque_and_read_only(tmp_path, monkeypatch): + database = tmp_path / "old.sqlite3" + jobs = worker.PaperJobs(database) + monkeypatch.setattr(worker, "PAPER_MODEL", "neural-paper/1") + first, source = enqueue(jobs) + second, _ = enqueue(jobs, entry="0.46") + historical = {"model": "neural-paper/1", "cash": "7.123456789012345678"} + with jobs._transaction() as db: + db.execute("UPDATE jobs SET status='completed',result=?", (json.dumps(historical),)) + before = database.read_bytes() + journal = PaperJournal(database) + assert journal.inspect(first)["result"] == historical + assert journal.recording(first) == source.read_bytes() + with pytest.raises(ValueError, match="trace unavailable"): + journal.compare(first, second) + with pytest.raises(ValueError, match="terms unavailable"): + journal.compare_markets(first, second) + assert database.read_bytes() == before + + +def test_derived_fees_preserve_product_scale(tmp_path): + database = tmp_path / "tiny-fees.sqlite3" + jobs = worker.PaperJobs(database) + spec, recording = inputs() + ids = [] + for entry in ("0.45", "0.46"): + ids.append( + jobs.submit( + replace(spec, entry_price=entry, quantity="0.01"), + recording, + initial_cash="10", + fee_per_contract="0.000000000000000001", + ) + ) + assert jobs.run_next()["status"] == "completed" + before = database.read_bytes() + for job in PaperJournal(jobs.database).compare(*ids): + assert job["summary"]["total_fees"] == "0.00000000000000000002" + assert job["summary"]["fill_count"] == 2 + assert database.read_bytes() == before + + +@pytest.mark.parametrize("fee", ["NaN", "Infinity", "-0.01", 0.01, "0." + "0" * 36 + "1", "0.01"]) +def test_invalid_saved_fees_reject_comparison(tmp_path, fee): + jobs = worker.PaperJobs(tmp_path / "invalid-fees.sqlite3") + first, _ = enqueue(jobs) + second, _ = enqueue(jobs, entry="0.46") + jobs.run_next() + jobs.run_next() + result = jobs.inspect(first)["result"] + result["trace"][0]["fees"] = fee + with sqlite3.connect(jobs.database) as db: + db.execute("UPDATE jobs SET result=? WHERE id=?", (json.dumps(result), first)) + with pytest.raises(ValueError, match="saved trace fees"): + PaperJournal(jobs.database).compare(first, second) + + +def test_exclusive_lock_stays_distinguishable_from_invalid_data(tmp_path, monkeypatch): + database = tmp_path / "busy.sqlite3" + worker.PaperJobs(database) + before = database.read_bytes() + connect = sqlite3.connect + monkeypatch.setattr( + sqlite3, "connect", lambda *args, **kwargs: connect(*args, **{**kwargs, "timeout": 0}) + ) + with connect(database) as writer: + writer.execute("BEGIN EXCLUSIVE") + with pytest.raises(sqlite3.OperationalError, match="database is locked"): + PaperJournal(database).history() + assert database.read_bytes() == before + + +def test_fill_without_fee_rejects_comparison(tmp_path): + database = tmp_path / "missing-fee.sqlite3" + jobs = worker.PaperJobs(database) + first, _ = enqueue(jobs) + second, _ = enqueue(jobs, entry="0.46") + jobs.run_next() + jobs.run_next() + result = jobs.inspect(first)["result"] + next(event for event in result["trace"] if event["action"] == "fill").pop("fees") + with sqlite3.connect(jobs.database) as db: + db.execute("UPDATE jobs SET result=? WHERE id=?", (json.dumps(result), first)) + before = database.read_bytes() + with pytest.raises(ValueError, match="saved trace fees"): + PaperJournal(jobs.database).compare(first, second) + assert database.read_bytes() == before + + +@pytest.mark.parametrize( + "change", + [{"initial_cash": "11"}, {"fee_per_contract": "0.02"}, {"max_events": 1000}, {"quantity": "1"}], +) +def test_incompatible_assumptions_and_unfinished_jobs_reject(tmp_path, change): + jobs = worker.PaperJobs(tmp_path / "jobs.sqlite3") + first, _ = enqueue(jobs) + second, _ = enqueue(jobs, **change) + journal = PaperJournal(jobs.database) + with pytest.raises(ValueError, match="completed"): + journal.compare(first, second) + jobs.run_next() + jobs.run_next() + with pytest.raises(ValueError, match="same recording"): + journal.compare(first, second) + with pytest.raises(ValueError, match="distinct"): + journal.compare(first, first) + with pytest.raises(ValueError, match="unknown"): + journal.inspect("missing") + + +@pytest.mark.parametrize("offset,limit", [(-1, 25), (True, 25), (100001, 25), (0, 0), (0, 101)]) +def test_pagination_is_bounded_before_opening_storage(tmp_path, offset, limit): + database = tmp_path / "absent.sqlite3" + with pytest.raises(ValueError): + PaperJournal(database).history(offset=offset, limit=limit) + assert not database.exists() + + +@pytest.mark.parametrize( + "result", + [ + '{"cash": NaN}', + '{"cash": Infinity}', + '{"cash": -Infinity}', + '{"cash": 1e999}', + '{"cash": -1e999}', + "{", + ], +) +def test_invalid_result_does_not_prevent_verified_recording_recovery(tmp_path, result): + database = tmp_path / "damaged-output.sqlite3" + jobs = worker.PaperJobs(database) + identity, source = enqueue(jobs) + jobs.run_next() + with sqlite3.connect(database) as db: + db.execute("UPDATE jobs SET result=? WHERE id=?", (result, identity)) + before = database.read_bytes() + journal = PaperJournal(database) + with pytest.raises(ValueError): + journal.inspect(identity) + assert journal.recording(identity) == source.read_bytes() + assert database.read_bytes() == before + + +@pytest.mark.parametrize("status,error", [("failed", b"bad"), ("failed", None), ("queued", "bad")]) +def test_invalid_error_state_rejects_history_and_inspection(tmp_path, status, error): + database = tmp_path / "bad-error.sqlite3" + jobs = worker.PaperJobs(database) + identity, _ = enqueue(jobs) + with sqlite3.connect(database) as db: + db.execute("PRAGMA ignore_check_constraints=ON") + db.execute("UPDATE jobs SET status=?,error=?", (status, error)) + before = database.read_bytes() + journal = PaperJournal(database) + with pytest.raises(ValueError, match="stored job error"): + journal.inspect(identity) + with pytest.raises(ValueError, match="stored job error"): + journal.history() + assert database.read_bytes() == before + + +def test_wal_archive_rejected_without_creating_sidecars(tmp_path): + database = tmp_path / "wal.sqlite3" + worker.PaperJobs(database) + db = sqlite3.connect(database) + db.execute("PRAGMA journal_mode=WAL") + db.close() + before = {path.name: path.read_bytes() for path in tmp_path.iterdir()} + assert set(before) == {database.name} + with pytest.raises(ValueError, match="WAL paper journals"): + PaperJournal(database).history() + assert {path.name: path.read_bytes() for path in tmp_path.iterdir()} == before + + +@pytest.mark.parametrize( + "status,result,error", + [("completed", None, None), ("queued", "{}", None), ("failed", "{}", "failure")], +) +def test_history_rejects_invalid_result_presence(tmp_path, status, result, error): + database = tmp_path / "bad-state.sqlite3" + jobs = worker.PaperJobs(database) + enqueue(jobs) + with sqlite3.connect(database) as db: + db.execute("PRAGMA ignore_check_constraints=ON") + db.execute("UPDATE jobs SET status=?,result=?,error=?", (status, result, error)) + before = database.read_bytes() + with pytest.raises(ValueError, match="stored job result"): + PaperJournal(database).history() + assert database.read_bytes() == before + + +@pytest.mark.parametrize( + "damage", ["config", "recording", "result", "result_bytes", "version", "not_sqlite"] +) +def test_damaged_journals_fail_without_repairing_them(tmp_path, damage): + database = tmp_path / "jobs.sqlite3" + jobs = worker.PaperJobs(database) + identity, _ = enqueue(jobs) + jobs.run_next() + if damage == "not_sqlite": + database.write_bytes(b"not a SQLite journal") + else: + with sqlite3.connect(database) as db: + if damage == "config": + db.execute("UPDATE jobs SET config=?", ('{"model":"changed"}',)) + elif damage == "recording": + db.execute("UPDATE jobs SET recording=?", (b"changed",)) + elif damage == "result": + db.execute("UPDATE jobs SET result=?", ("[]",)) + elif damage == "result_bytes": + db.execute("UPDATE jobs SET result=?", (b"\xff",)) + else: + db.execute("PRAGMA user_version=99") + before = database.read_bytes() + journal = PaperJournal(database) + with pytest.raises(ValueError): + if damage == "recording": + journal.recording(identity) + else: + journal.inspect(identity) + assert database.read_bytes() == before