diff --git a/ai/prompt-io/opencode/20260903T192742Z_aadfd1cf_prompt_io.md b/ai/prompt-io/opencode/20260903T192742Z_aadfd1cf_prompt_io.md new file mode 100644 index 00000000..ac934ad2 --- /dev/null +++ b/ai/prompt-io/opencode/20260903T192742Z_aadfd1cf_prompt_io.md @@ -0,0 +1,51 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: 8737d0a9-98da-4fa1-ba35-ca8a3a42bbc9 +timestamp: 2026-09-03T19:27:42Z +git_ref: wkt/replay_provider_e2e +scope: code +substantive: true +raw_file: 20260903T192742Z_aadfd1cf_prompt_io.raw.md +--- + +## Prompt + +Continue Phase 2 of the human-facing E2E plan by implementing a +protocol-faithful offline replay provider. Use normal backend discovery, +datad, history, quote, search, and typed Tractor control contracts with +versioned fixtures, deterministic ordering, no network, no positive +sleeps, and no log parsing. + +During review, explain the provider's purpose and placement, evaluate a +general package-entry-point backend mechanism, then park replay when the +human prioritizes real-backend chart E2E instead. + +## Response summary + +Adds a versioned offline market-data provider implementing normal Piker +backend endpoint shapes. Typed control commands advance, pause, resume, +snapshot, and inject declared failures while preserving deterministic +transcripts. Focused tests validate fixture symbology and finite, +repeatable OHLCV history behavior. + +## Files changed + +- `piker/brokers/replay.py` - implement replay scenario, provider, and + control protocol machinery. +- `tests/_inputs/replay/basic-v1.json` - provide deterministic history + and quote input. +- `tests/_inputs/replay/failure-v1.json` - provide an acknowledged fault + point. +- `tests/replay/test_contract.py` - validate scenario and history + contracts. + +## Human edits + +The human challenged the original replay-first premise, questioned why +test machinery lived beside production backends, proposed dynamic +registration, and selected standard package entry points as the future +direction for external backend support. The human directly added a +source note questioning replay placement, then explicitly placed replay +on hold and requested that the completed experiment be committed only +as parked work with plugin migration recorded for possible resumption. diff --git a/ai/prompt-io/opencode/20260903T192742Z_aadfd1cf_prompt_io.raw.md b/ai/prompt-io/opencode/20260903T192742Z_aadfd1cf_prompt_io.raw.md new file mode 100644 index 00000000..4760d996 --- /dev/null +++ b/ai/prompt-io/opencode/20260903T192742Z_aadfd1cf_prompt_io.raw.md @@ -0,0 +1,41 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-09-03T19:27:42Z +git_ref: wkt/replay_provider_e2e +diff_cmd: git diff HEAD~1..HEAD +--- + +## Generated changes + +> `git diff HEAD~1..HEAD -- piker/brokers/replay.py` + +Implements a fixture-driven market-data backend with typed scenario, +quote, failure, command, acknowledgement, snapshot, and transcript +structures. It provides the normal market-info, bounded history, +quote-stream, search, and Tractor control endpoints without external +network access. + +> `git diff HEAD~1..HEAD -- tests/_inputs/replay/ tests/replay/test_contract.py` + +Adds versioned basic and failure scenarios plus focused validation and +history-boundary tests. The scenarios carry normalized market metadata, +one-second and one-minute OHLCV arrays, sequenced quotes, and a named +failure point. + +## Verification + +- Python compilation, Ruff, JSON parsing, and staged whitespace checks + pass. +- Scenario normalization and bounded repeatable history tests pass. +- The production history contract exposed a need for Pendulum datetime + values; the implementation was corrected before final verification. + +## Design discussion + +The user questioned placing test infrastructure beside production +backends and proposed dynamic test registration. The resulting review +identified standard Python package entry points as the preferred future +mechanism for external and testing backends. The user then deprioritized +replay entirely in favor of real-backend chart E2E and requested that +the completed experiment be parked with that follow-up recorded. diff --git a/piker/brokers/replay.py b/piker/brokers/replay.py new file mode 100644 index 00000000..bdb6bebf --- /dev/null +++ b/piker/brokers/replay.py @@ -0,0 +1,715 @@ +''' +Deterministic offline market-data replay backend. + +''' +from __future__ import annotations + +from contextlib import ( + asynccontextmanager as acm, +) +from dataclasses import dataclass +from datetime import datetime +import os +from pathlib import Path +from typing import ( + Any, + Literal, +) + +import msgspec +import numpy as np +import pendulum +import tractor +import trio +from trio_typing import TaskStatus + +from piker.accounting import MktPair +from piker.brokers import ( + DataUnavailable, + SymbolNotFound, +) +from piker.data._source import def_iohlcv_fields +from piker.data.validate import FeedInit +from piker.types import Struct + + +name: str = 'replay' +_scenario_env: str = 'PIKER_REPLAY_SCENARIO' + + +class ReplayBar(Struct, frozen=True): + ''' + One normalized OHLCV fixture row. + + ''' + index: int + time: int + open: float + high: float + low: float + close: float + volume: float + + +class ReplayTick(Struct, frozen=True): + ''' + One normalized tick in a replay quote. + + ''' + type: str + price: float + size: float + + +class ReplayQuote(Struct, frozen=True): + ''' + One sequenced provider event from a replay scenario. + + ''' + sequence: int + fqme: str + broker_ts: int + last: float + ticks: list[ReplayTick] + + +class ReplayFailure(Struct, frozen=True): + ''' + A failure which can be armed at one event sequence. + + ''' + sequence: int + code: str + message: str + + +class ReplayScenario(Struct, frozen=True): + ''' + Versioned input for one deterministic replay generation. + + ''' + version: int + scenario_id: str + markets: list[MktPair] + history_1s: list[ReplayBar] + history_1m: list[ReplayBar] + quotes: list[ReplayQuote] + failures: list[ReplayFailure] = [] + + +ReplayState = Literal[ + 'ready', + 'paused', + 'failed', + 'exhausted', +] + + +class Advance(Struct, frozen=True, tag=True): + ''' + Publish one exact replay event. + + ''' + command_id: int + event_sequence: int + + +class Pause(Struct, frozen=True, tag=True): + ''' + Close the replay producer gate. + + ''' + command_id: int + + +class Resume(Struct, frozen=True, tag=True): + ''' + Open the replay producer gate. + + ''' + command_id: int + + +class FailAt(Struct, frozen=True, tag=True): + ''' + Arm one scenario-declared failure injection. + + ''' + command_id: int + event_sequence: int + + +class AwaitState(Struct, frozen=True, tag=True): + ''' + Wait for one provider state without scheduler polling. + + ''' + command_id: int + state: ReplayState + + +class Snapshot(Struct, frozen=True, tag=True): + ''' + Request the current deterministic replay state. + + ''' + command_id: int + + +ReplayCommand = ( + Advance + | Pause + | Resume + | FailAt + | AwaitState + | Snapshot +) + + +class ReplayRecord(Struct, frozen=True): + ''' + One deterministic control transcript entry. + + ''' + command_id: int + command: str + state: ReplayState + event_sequence: int + ok: bool + error: str = '' + + +class ReplaySnapshot(Struct, frozen=True, tag=True): + ''' + Provider state returned through the control protocol. + + ''' + scenario_id: str + state: ReplayState + command_id: int + event_sequence: int + subscriber_active: bool + armed_failure: int | None + transcript: list[ReplayRecord] + + +class ReplayAck(Struct, frozen=True, tag=True): + ''' + Correlated result for one replay control command. + + ''' + command_id: int + ok: bool + snapshot: ReplaySnapshot + error: str = '' + + +ReplayPayload = ( + ReplayCommand + | ReplayAck + | ReplaySnapshot +) + + +# hmm maybe put this in a new piker._testing.brokers ? +# or is there some reason to put it alongside the production +# backends? +def load_scenario( + path: Path | str, +) -> ReplayScenario: + ''' + Decode and validate one versioned replay scenario. + + ''' + scenario_path: Path = Path(path) + scenario: ReplayScenario = msgspec.json.decode( + scenario_path.read_bytes(), + type=ReplayScenario, + ) + if scenario.version != 1: + raise ValueError( + f'Unsupported replay scenario version: ' + f'{scenario.version}' + ) + if not scenario.markets: + raise ValueError('Replay scenario has no markets') + if not scenario.quotes: + raise ValueError('Replay scenario has no quotes') + + quote_sequences: list[int] = [ + quote.sequence + for quote in scenario.quotes + ] + expected: list[int] = list(range( + quote_sequences[0], + quote_sequences[-1] + 1, + )) + if quote_sequences != expected: + raise ValueError( + 'Replay quote sequences must be contiguous' + ) + + market_fqmes: set[str] = { + mkt.fqme + for mkt in scenario.markets + } + unknown_fqmes: set[str] = { + quote.fqme + for quote in scenario.quotes + } - market_fqmes + if unknown_fqmes: + raise ValueError( + f'Replay quotes reference unknown markets: ' + f'{sorted(unknown_fqmes)!r}' + ) + return scenario + + +def _scenario_path() -> Path: + path: str | None = os.environ.get(_scenario_env) + if not path: + raise RuntimeError( + f'Set `{_scenario_env}` to a replay scenario path' + ) + return Path(path) + + +def _quote_to_msg( + quote: ReplayQuote, +) -> dict[str, Any]: + return { + 'symbol': quote.fqme, + 'last': quote.last, + 'broker_ts': quote.broker_ts, + 'brokerd_ts': quote.broker_ts, + 'replay_seq': quote.sequence, + 'ticks': [ + { + 'type': tick.type, + 'price': tick.price, + 'size': tick.size, + } + for tick in quote.ticks + ], + } + + +def _bars_to_array( + bars: list[ReplayBar], +) -> np.ndarray: + return np.asarray( + [ + ( + bar.index, + bar.time, + bar.open, + bar.high, + bar.low, + bar.close, + bar.volume, + ) + for bar in bars + ], + dtype=def_iohlcv_fields, + ) + + +def _find_market( + scenario: ReplayScenario, + fqme: str, +) -> MktPair: + normalized: str = fqme.lower() + for mkt in scenario.markets: + aliases: set[str] = { + mkt.fqme.lower(), + mkt.bs_fqme.lower(), + mkt.bs_mktid.lower(), + } + if normalized in aliases: + return mkt + raise SymbolNotFound(fqme) + + +async def get_mkt_info( + fqme: str, +) -> tuple[MktPair, MktPair]: + ''' + Return normalized and backend market records from the fixture. + + ''' + scenario: ReplayScenario = load_scenario(_scenario_path()) + mkt: MktPair = _find_market(scenario, fqme) + return mkt, mkt + + +@acm +async def open_history_client( + mkt: MktPair, +): + ''' + Open a bounded in-memory OHLCV history client. + + ''' + scenario: ReplayScenario = load_scenario(_scenario_path()) + _find_market(scenario, mkt.fqme) + frames: dict[int, np.ndarray] = { + 1: _bars_to_array(scenario.history_1s), + 60: _bars_to_array(scenario.history_1m), + } + + async def get_ohlc( + timeframe: float, + end_dt: datetime | None = None, + start_dt: datetime | None = None, + + ) -> tuple[np.ndarray, datetime, datetime]: + period: int = int(timeframe) + try: + frame: np.ndarray = frames[period] + except KeyError: + raise DataUnavailable( + f'Unsupported replay timeframe: {timeframe}' + ) from None + + selected: np.ndarray = frame + if start_dt is not None: + start_ts: float = start_dt.timestamp() + selected = selected[selected['time'] >= start_ts] + if end_dt is not None: + end_ts: float = end_dt.timestamp() + selected = selected[selected['time'] < end_ts] + if not selected.size: + raise DataUnavailable( + f'No replay history before {end_dt!r}' + ) + + result: np.ndarray = selected.copy() + start: datetime = pendulum.from_timestamp( + int(result['time'][0]), + ) + end: datetime = pendulum.from_timestamp( + int(result['time'][-1]), + ) + return result, start, end + + yield get_ohlc, { + 'erlangs': 1, + 'rate': 1, + } + + +@dataclass +class _RuntimeRequest: + command: ReplayCommand + reply: trio.MemorySendChannel[ReplayAck] + + +class _ReplayRuntime: + ''' + Actor-local owner of replay progression and transcript state. + + ''' + def __init__( + self, + scenario: ReplayScenario, + ) -> None: + self.scenario = scenario + self.command_tx: ( + trio.MemorySendChannel[_RuntimeRequest] + | None + ) = None + self.command_id: int = 0 + self.event_sequence: int = scenario.quotes[0].sequence + self.producer_paused: bool = False + self.subscriber_active: bool = True + self.armed_failure: int | None = None + self.failed: bool = False + self.transcript: list[ReplayRecord] = [] + self._state_changed: trio.Event = trio.Event() + + @property + def state(self) -> ReplayState: + if self.failed: + return 'failed' + if ( + self.event_sequence + == self.scenario.quotes[-1].sequence + ): + return 'exhausted' + if ( + self.producer_paused + or not self.subscriber_active + ): + return 'paused' + return 'ready' + + def snapshot(self) -> ReplaySnapshot: + return ReplaySnapshot( + scenario_id=self.scenario.scenario_id, + state=self.state, + command_id=self.command_id, + event_sequence=self.event_sequence, + subscriber_active=self.subscriber_active, + armed_failure=self.armed_failure, + transcript=list(self.transcript), + ) + + def signal_state_change(self) -> None: + changed: trio.Event = self._state_changed + self._state_changed = trio.Event() + changed.set() + + async def await_state( + self, + state: ReplayState, + ) -> None: + while self.state != state: + changed: trio.Event = self._state_changed + await changed.wait() + + def record( + self, + command: ReplayCommand, + ok: bool, + error: str = '', + ) -> ReplayAck: + self.transcript.append(ReplayRecord( + command_id=command.command_id, + command=type(command).__name__, + state=self.state, + event_sequence=self.event_sequence, + ok=ok, + error=error, + )) + return ReplayAck( + command_id=command.command_id, + ok=ok, + error=error, + snapshot=self.snapshot(), + ) + + +_runtime: _ReplayRuntime | None = None +_runtime_path: Path | None = None + + +def _get_runtime() -> _ReplayRuntime: + global _runtime, _runtime_path + path: Path = _scenario_path().resolve() + if ( + _runtime is None + or _runtime_path != path + ): + _runtime = _ReplayRuntime(load_scenario(path)) + _runtime_path = path + return _runtime + + +def on_feed_subscription_change( + bs_fqme: str, + active: bool, +) -> None: + ''' + Synchronize the provider gate with feed-bus subscriptions. + + ''' + runtime: _ReplayRuntime = _get_runtime() + _find_market(runtime.scenario, bs_fqme) + if runtime.subscriber_active != active: + runtime.subscriber_active = active + runtime.signal_state_change() + + +async def _execute_command( + runtime: _ReplayRuntime, + command: ReplayCommand, + send_chan: trio.abc.SendChannel, +) -> ReplayAck: + expected_id: int = runtime.command_id + 1 + if command.command_id != expected_id: + error: str = ( + f'Expected command_id={expected_id}, got ' + f'{command.command_id}' + ) + return runtime.record(command, False, error) + + runtime.command_id = command.command_id + match command: + case Pause(): + runtime.producer_paused = True + runtime.signal_state_change() + + case Resume(): + runtime.producer_paused = False + runtime.signal_state_change() + + case AwaitState(state=state): + await runtime.await_state(state) + + case FailAt(event_sequence=sequence): + failures: dict[int, ReplayFailure] = { + failure.sequence: failure + for failure in runtime.scenario.failures + } + if sequence not in failures: + error = ( + f'No scenario failure at event_sequence=' + f'{sequence}' + ) + return runtime.record(command, False, error) + runtime.armed_failure = sequence + + case Advance(event_sequence=sequence): + if runtime.state != 'ready': + error = ( + f'Cannot advance replay while state=' + f'{runtime.state!r}' + ) + return runtime.record(command, False, error) + + expected_sequence: int = runtime.event_sequence + 1 + if sequence != expected_sequence: + error = ( + f'Expected event_sequence={expected_sequence}, ' + f'got {sequence}' + ) + return runtime.record(command, False, error) + + if runtime.armed_failure == sequence: + failure: ReplayFailure = next( + failure + for failure in runtime.scenario.failures + if failure.sequence == sequence + ) + runtime.failed = True + runtime.signal_state_change() + error = f'{failure.code}: {failure.message}' + return runtime.record(command, False, error) + + quote: ReplayQuote = runtime.scenario.quotes[ + sequence + - runtime.scenario.quotes[0].sequence + ] + mkt: MktPair = _find_market( + runtime.scenario, + quote.fqme, + ) + await send_chan.send({ + mkt.bs_fqme: _quote_to_msg(quote), + }) + runtime.event_sequence = sequence + runtime.signal_state_change() + + case Snapshot(): + pass + + return runtime.record(command, True) + + +async def stream_quotes( + send_chan: trio.abc.SendChannel, + symbols: list[str], + feed_is_live: trio.Event, + loglevel: str | None = None, + task_status: TaskStatus[ + tuple[list[FeedInit], dict[str, Any]] + ] = trio.TASK_STATUS_IGNORED, +) -> None: + ''' + Publish fixture quotes only after acknowledged control commands. + + ''' + if len(symbols) != 1: + raise ValueError( + 'The replay backend currently supports one symbol' + ) + + runtime: _ReplayRuntime = _get_runtime() + mkt: MktPair = _find_market(runtime.scenario, symbols[0]) + first: ReplayQuote = runtime.scenario.quotes[0] + if first.fqme != mkt.fqme: + raise ValueError( + f'First replay quote does not target {mkt.fqme!r}' + ) + + command_tx: trio.MemorySendChannel[_RuntimeRequest] + command_rx: trio.MemoryReceiveChannel[_RuntimeRequest] + command_tx, command_rx = trio.open_memory_channel(0) + runtime.command_tx = command_tx + try: + async with ( + send_chan, + command_rx, + ): + task_status.started(( + [FeedInit(mkt_info=mkt)], + _quote_to_msg(first), + )) + feed_is_live.set() + + async for request in command_rx: + ack: ReplayAck = await _execute_command( + runtime, + request.command, + send_chan, + ) + async with request.reply: + await request.reply.send(ack) + finally: + runtime.command_tx = None + + +@tractor.context(pld_spec=ReplayPayload) +async def open_replay_control( + ctx: tractor.Context, +) -> None: + ''' + Open the typed replay command and acknowledgement stream. + + ''' + runtime: _ReplayRuntime = _get_runtime() + if runtime.command_tx is None: + raise RuntimeError('Replay quote stream is not running') + + await ctx.started(runtime.snapshot()) + async with ctx.open_stream() as stream: + with ctx.pld_rx.limit_plds(spec=ReplayCommand): + command: ReplayCommand + async for command in stream: + reply_tx: trio.MemorySendChannel[ReplayAck] + reply_rx: trio.MemoryReceiveChannel[ReplayAck] + reply_tx, reply_rx = trio.open_memory_channel(1) + request = _RuntimeRequest( + command=command, + reply=reply_tx, + ) + await runtime.command_tx.send(request) + async with reply_rx: + ack: ReplayAck = await reply_rx.receive() + await stream.send(ack) + + +@tractor.context +async def open_symbol_search( + ctx: tractor.Context, +) -> None: + ''' + Search replay fixture markets by normalized FQME substring. + + ''' + scenario: ReplayScenario = load_scenario(_scenario_path()) + await ctx.started() + async with ctx.open_stream() as stream: + pattern: str + async for pattern in stream: + lowered: str = pattern.lower() + matches: dict[str, dict[str, Any]] = { + mkt.fqme: mkt.to_dict() + for mkt in scenario.markets + if lowered in mkt.fqme.lower() + } + await stream.send(matches) + + +_datad_mods: list[str] = [] +__enable_modules__: list[str] = [] diff --git a/tests/_inputs/replay/basic-v1.json b/tests/_inputs/replay/basic-v1.json new file mode 100644 index 00000000..0c2e20e6 --- /dev/null +++ b/tests/_inputs/replay/basic-v1.json @@ -0,0 +1,45 @@ +{ + "version": 1, + "scenario_id": "basic-v1", + "markets": [ + { + "dst": { + "name": "btc", + "atype": "crypto", + "tx_tick": "0.00000001" + }, + "src": { + "name": "usd", + "atype": "fiat", + "tx_tick": "0.01" + }, + "price_tick": "0.01", + "size_tick": "0.0001", + "bs_mktid": "BTCUSD", + "broker": "replay", + "venue": "test" + } + ], + "history_1s": [ + {"index": 0, "time": 1700000035, "open": 100.0, "high": 100.2, "low": 99.9, "close": 100.1, "volume": 1.0}, + {"index": 1, "time": 1700000036, "open": 100.1, "high": 100.3, "low": 100.0, "close": 100.2, "volume": 2.0}, + {"index": 2, "time": 1700000037, "open": 100.2, "high": 100.4, "low": 100.1, "close": 100.3, "volume": 3.0}, + {"index": 3, "time": 1700000038, "open": 100.3, "high": 100.5, "low": 100.2, "close": 100.4, "volume": 4.0}, + {"index": 4, "time": 1700000039, "open": 100.4, "high": 100.6, "low": 100.3, "close": 100.5, "volume": 5.0}, + {"index": 5, "time": 1700000040, "open": 100.5, "high": 100.7, "low": 100.4, "close": 100.6, "volume": 6.0} + ], + "history_1m": [ + {"index": 0, "time": 1699999740, "open": 99.0, "high": 99.4, "low": 98.8, "close": 99.2, "volume": 10.0}, + {"index": 1, "time": 1699999800, "open": 99.2, "high": 99.6, "low": 99.0, "close": 99.4, "volume": 11.0}, + {"index": 2, "time": 1699999860, "open": 99.4, "high": 99.8, "low": 99.2, "close": 99.6, "volume": 12.0}, + {"index": 3, "time": 1699999920, "open": 99.6, "high": 100.0, "low": 99.4, "close": 99.8, "volume": 13.0}, + {"index": 4, "time": 1699999980, "open": 99.8, "high": 100.4, "low": 99.6, "close": 100.2, "volume": 14.0}, + {"index": 5, "time": 1700000040, "open": 100.2, "high": 100.8, "low": 100.0, "close": 100.6, "volume": 15.0} + ], + "quotes": [ + {"sequence": 1, "fqme": "btcusd.test.replay", "broker_ts": 1700000041, "last": 100.7, "ticks": [{"type": "trade", "price": 100.7, "size": 0.1}]}, + {"sequence": 2, "fqme": "btcusd.test.replay", "broker_ts": 1700000042, "last": 100.8, "ticks": [{"type": "trade", "price": 100.8, "size": 0.2}]}, + {"sequence": 3, "fqme": "btcusd.test.replay", "broker_ts": 1700000043, "last": 100.9, "ticks": [{"type": "trade", "price": 100.9, "size": 0.3}]}, + {"sequence": 4, "fqme": "btcusd.test.replay", "broker_ts": 1700000044, "last": 101.0, "ticks": [{"type": "trade", "price": 101.0, "size": 0.4}]} + ] +} diff --git a/tests/_inputs/replay/failure-v1.json b/tests/_inputs/replay/failure-v1.json new file mode 100644 index 00000000..6069acf0 --- /dev/null +++ b/tests/_inputs/replay/failure-v1.json @@ -0,0 +1,45 @@ +{ + "version": 1, + "scenario_id": "failure-v1", + "markets": [ + { + "dst": { + "name": "btc", + "atype": "crypto", + "tx_tick": "0.00000001" + }, + "src": { + "name": "usd", + "atype": "fiat", + "tx_tick": "0.01" + }, + "price_tick": "0.01", + "size_tick": "0.0001", + "bs_mktid": "BTCUSD", + "broker": "replay", + "venue": "test" + } + ], + "history_1s": [ + {"index": 0, "time": 1700000038, "open": 100.0, "high": 100.2, "low": 99.9, "close": 100.1, "volume": 1.0}, + {"index": 1, "time": 1700000039, "open": 100.1, "high": 100.3, "low": 100.0, "close": 100.2, "volume": 2.0}, + {"index": 2, "time": 1700000040, "open": 100.2, "high": 100.4, "low": 100.1, "close": 100.3, "volume": 3.0} + ], + "history_1m": [ + {"index": 0, "time": 1699999920, "open": 99.0, "high": 99.5, "low": 98.8, "close": 99.3, "volume": 10.0}, + {"index": 1, "time": 1699999980, "open": 99.3, "high": 99.9, "low": 99.1, "close": 99.7, "volume": 11.0}, + {"index": 2, "time": 1700000040, "open": 99.7, "high": 100.5, "low": 99.5, "close": 100.3, "volume": 12.0} + ], + "quotes": [ + {"sequence": 1, "fqme": "btcusd.test.replay", "broker_ts": 1700000041, "last": 100.4, "ticks": [{"type": "trade", "price": 100.4, "size": 0.1}]}, + {"sequence": 2, "fqme": "btcusd.test.replay", "broker_ts": 1700000042, "last": 100.5, "ticks": [{"type": "trade", "price": 100.5, "size": 0.2}]}, + {"sequence": 3, "fqme": "btcusd.test.replay", "broker_ts": 1700000043, "last": 100.6, "ticks": [{"type": "trade", "price": 100.6, "size": 0.3}]} + ], + "failures": [ + { + "sequence": 3, + "code": "fixture_disconnect", + "message": "offline provider disconnected" + } + ] +} diff --git a/tests/replay/test_contract.py b/tests/replay/test_contract.py new file mode 100644 index 00000000..cac1aee9 --- /dev/null +++ b/tests/replay/test_contract.py @@ -0,0 +1,100 @@ +''' +Offline replay provider contract tests. + +''' +from datetime import ( + UTC, + datetime, +) +from pathlib import Path + +import numpy as np +import pytest +import trio + +from piker.brokers import DataUnavailable +from piker.brokers import replay + + +INPUTS: Path = ( + Path(__file__).parents[1] + / '_inputs' + / 'replay' +) + + +def test_versioned_scenario_normalizes_market_data() -> None: + ''' + Reject fixture drift before a datad actor obscures its cause. + + Replay scenarios are durable test inputs rather than loose mock + dictionaries. Decode the versioned basic scenario through the + production loader and prove its market identity, Decimal fields, + contiguous event IDs, and normalized tick records survive typed + decoding. These assertions catch schema or symbology changes at + the provider boundary without starting services or using network + resources. + + ''' + scenario: replay.ReplayScenario = replay.load_scenario( + INPUTS / 'basic-v1.json' + ) + + assert scenario.version == 1 + assert scenario.scenario_id == 'basic-v1' + assert scenario.markets[0].fqme == 'btcusd.test.replay' + assert str(scenario.markets[0].price_tick) == '0.01' + assert [quote.sequence for quote in scenario.quotes] == [ + 1, + 2, + 3, + 4, + ] + assert scenario.quotes[1].ticks[0].type == 'trade' + + +def test_history_queries_are_bounded_and_repeatable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Keep history replay finite, offline, and repeatable. + + A history fixture which mutates a cursor per request can make the + two concurrent 1-second and 1-minute backfill tasks race, + while an unbounded latest frame can make datad backfill + forever. Select the same 1-minute frame twice, then request + data ending at its first timestamp. Equal arrays prove calls + are immutable and the explicit `DataUnavailable` proves + reverse backfill terminates at the fixture boundary without + a clock delay or external request. + + ''' + scenario_path: Path = INPUTS / 'basic-v1.json' + monkeypatch.setenv( + 'PIKER_REPLAY_SCENARIO', + str(scenario_path), + ) + scenario: replay.ReplayScenario = replay.load_scenario( + scenario_path + ) + mkt = scenario.markets[0] + + async def main() -> None: + async with replay.open_history_client(mkt) as ( + get_hist, + config, + ): + first, start, end = await get_hist(60) + second, second_start, second_end = await get_hist(60) + np.testing.assert_array_equal(first, second) + assert (start, end) == (second_start, second_end) + assert config == {'erlangs': 1, 'rate': 1} + + boundary: datetime = datetime.fromtimestamp( + int(first['time'][0]), + tz=UTC, + ) + with pytest.raises(DataUnavailable): + await get_hist(60, end_dt=boundary) + + trio.run(main)