pikers
/
piker
4
1
Fork 0
Code Issues 20 Pull Requests 12 Packages Projects Releases Wiki Activity

Add deterministic `replay` data backend 

Deats,
- decode versioned market, OHLCV, quote and failure fixtures
- expose normal symbology, history, search and quote-feed eps
- sequence typed control msgs with snapshots and transcripts
- prove fixture validation and bounded repeatable history

Prompt-IO: ai/prompt-io/opencode/20260903T192742Z_aadfd1cf_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
wkt/replay_provider_e2e
Gud Boi 2026-09-03 15:31:04 -04:00
parent aadfd1cfcc
commit dcca60debc
6 changed files with 997 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
ai/prompt-io/opencode/20260903T192742Z_aadfd1cf_prompt_io.md 100644
View File

@ -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.

41
ai/prompt-io/opencode/20260903T192742Z_aadfd1cf_prompt_io.raw.md 100644
View File

@ -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.

715
piker/brokers/replay.py 100644
View File

@ -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] = []

45
tests/_inputs/replay/basic-v1.json 100644
View File

@ -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}]}
]
}

45
tests/_inputs/replay/failure-v1.json 100644
View File

@ -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"
}
]
}

100
tests/replay/test_contract.py 100644
View File

@ -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)