280 lines
10 KiB
Python
280 lines
10 KiB
Python
# piker: trading gear for hackers
|
|
# Copyright (C) 2018-present Tyler Goodlet (in stewardship of pikers)
|
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
|
|
# This program is free software: you can redistribute it and/or
|
|
# modify it under the terms of the GNU Affero General Public License
|
|
# as published by the Free Software Foundation, either version 3
|
|
# of the License, or (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU Affero General Public
|
|
# License along with this program. If not, see
|
|
# <https://www.gnu.org/licenses/>.
|
|
|
|
'''
|
|
Exercise native Trend Calculus publication through real Tractor SHM.
|
|
|
|
'''
|
|
from collections.abc import AsyncIterator, Callable
|
|
from functools import partial
|
|
from types import SimpleNamespace
|
|
from typing import Any, cast
|
|
from uuid import uuid4
|
|
|
|
import numpy as np
|
|
import tractor
|
|
from tractor.ipc._shm import (
|
|
NDToken,
|
|
ShmArray,
|
|
attach_shm_ndarray,
|
|
open_shm_ndarray,
|
|
)
|
|
import trio
|
|
from trio.testing import wait_all_tasks_blocked
|
|
|
|
from piker.fsp._engine import (
|
|
Cascade,
|
|
QuoteFrame,
|
|
_next_step_output,
|
|
_sample_wakeups,
|
|
connect_streams,
|
|
)
|
|
from piker.fsp._trend import TrendCalculator
|
|
from piker.fsp._trend_fsp import _history_fields, trend_calculus
|
|
|
|
|
|
async def _read_in_child(
|
|
token: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
'''
|
|
Attach from an independent actor using the normal token codec.
|
|
|
|
'''
|
|
shm: ShmArray = attach_shm_ndarray(NDToken.from_msg(token))
|
|
rows: np.ndarray = shm.array.copy()
|
|
fields: dict[str, list] = {}
|
|
name: str
|
|
for name in rows.dtype.names:
|
|
value: int|float # noqa: F842
|
|
fields[name] = [
|
|
value if np.isfinite(value) else None
|
|
for value in rows[name].tolist()
|
|
]
|
|
return {
|
|
'first': shm._first.value,
|
|
'last': shm._last.value,
|
|
'readonly': not shm.array.flags.writeable,
|
|
'fields': fields,
|
|
}
|
|
|
|
|
|
def test_real_shm_bootstrap_step_and_in_place_replay() -> None:
|
|
'''
|
|
Validate actual SHM bounds, field layout and replay visibility.
|
|
|
|
Bootstrap the real engine against a nonzero source origin and
|
|
read its destination through a separate readonly attachment.
|
|
Advance a quiet bar through the sampler publication protocol,
|
|
then repair an interior close without changing source bounds.
|
|
``Cascade.resync()`` must rewrite every affected output and
|
|
notify its
|
|
consumer. A child actor then attaches using the transported token
|
|
and verifies the repaired data independently. All actors and
|
|
memory names belong to this test; no broker, model API or
|
|
existing Piker daemon is required. The child reads after replay
|
|
completes; this proves visibility, not concurrent atomicity.
|
|
|
|
'''
|
|
async def main() -> None:
|
|
'''
|
|
Own actor lifetimes and SHM names for this replay proof.
|
|
|
|
'''
|
|
suffix: str = uuid4().hex
|
|
key: str = f'test_tc_{suffix}'
|
|
async with tractor.open_root_actor(
|
|
name=key,
|
|
tpt_bind_addrs=[('127.0.0.1', 0)],
|
|
):
|
|
source: ShmArray = open_shm_ndarray(
|
|
size=128,
|
|
key=f'{key}_source',
|
|
append_start_index=37,
|
|
dtype=np.dtype([
|
|
('index', '<i8'),
|
|
('time', '<f8'),
|
|
('close', '<f8'),
|
|
]),
|
|
)
|
|
name: str # noqa: F842
|
|
destination: ShmArray = open_shm_ndarray(
|
|
size=128,
|
|
key=f'{key}_output',
|
|
append_start_index=0,
|
|
dtype=np.dtype([
|
|
('index', '<i8'),
|
|
('time', '<f8'),
|
|
] + [
|
|
(name, '<f8')
|
|
for name in trend_calculus.outputs
|
|
]),
|
|
)
|
|
seed: np.ndarray = np.zeros(6, dtype=source.array.dtype)
|
|
seed['time'] = np.arange(1000., 1006.)
|
|
seed['close'] = [1., 2., 4., 3., 2., 5.]
|
|
source.push(seed)
|
|
reader: ShmArray = attach_shm_ndarray(destination._token)
|
|
notices: list[dict] = []
|
|
|
|
class Sink:
|
|
'''
|
|
Record resync notices without involving a live UI
|
|
stream.
|
|
|
|
'''
|
|
async def send(self, message: dict) -> None:
|
|
'''
|
|
Retain each engine notice for deterministic
|
|
assertions.
|
|
|
|
'''
|
|
notices.append(message)
|
|
|
|
def check_publication() -> None:
|
|
'''
|
|
Compare the readonly attachment with causal source
|
|
replay.
|
|
|
|
'''
|
|
expected: dict[str, np.ndarray] = _history_fields(
|
|
TrendCalculator().extend(
|
|
source.array['close'], source.array['time'],
|
|
source.array['index'],
|
|
),
|
|
)
|
|
assert reader._first.value == source._first.value
|
|
assert reader._last.value == source._last.value
|
|
assert not reader.array.flags.writeable
|
|
field: str
|
|
for field in ('index', 'time'):
|
|
np.testing.assert_array_equal(
|
|
reader.array[field], source.array[field],
|
|
)
|
|
field: str
|
|
values: np.ndarray
|
|
for field, values in expected.items():
|
|
np.testing.assert_array_equal(
|
|
reader.array[field], values,
|
|
)
|
|
|
|
send: trio.MemorySendChannel[QuoteFrame]
|
|
quotes: trio.MemoryReceiveChannel[QuoteFrame]
|
|
send, quotes = trio.open_memory_channel[QuoteFrame](0)
|
|
market: SimpleNamespace = SimpleNamespace(
|
|
fqme='synthetic.test',
|
|
)
|
|
wakeups: AsyncIterator[QuoteFrame]
|
|
notify: Callable[[], None]|None
|
|
async with send, quotes, _sample_wakeups(
|
|
quotes, market.fqme, True,
|
|
) as (wakeups, notify):
|
|
nursery: trio.Nursery
|
|
async with trio.open_nursery() as nursery:
|
|
casc: Cascade = Cascade(
|
|
src=SimpleNamespace(
|
|
rt_shm=source,
|
|
mkt=market,
|
|
),
|
|
dst=SimpleNamespace(rt_shm=destination),
|
|
tn=nursery,
|
|
fsp=trend_calculus,
|
|
client_stream=cast(
|
|
tractor.MsgStream, Sink(),
|
|
),
|
|
)
|
|
compute: Callable = partial(
|
|
connect_streams,
|
|
casc=casc,
|
|
mkt=market,
|
|
quote_stream=wakeups,
|
|
src=casc.src,
|
|
dst=casc.dst,
|
|
edge_func=trend_calculus.func,
|
|
)
|
|
async with casc.open_edge(compute):
|
|
await wait_all_tasks_blocked()
|
|
check_publication()
|
|
assert reader.array['top_1'][-1] == 4.
|
|
assert reader.array['bottom_1'][-1] == 2.
|
|
|
|
step: np.ndarray = source.array[-1:].copy()
|
|
step['time'] += 1
|
|
source.push(step)
|
|
out: np.ndarray = _next_step_output(
|
|
destination.array[-1:],
|
|
trend_calculus,
|
|
False,
|
|
)
|
|
out['time'] = step['time']
|
|
destination.push(out)
|
|
if casc.sample_ready is not None:
|
|
casc.sample_ready.set()
|
|
casc.sample_ready = trio.Event()
|
|
assert notify is not None
|
|
notify()
|
|
await wait_all_tasks_blocked()
|
|
check_publication()
|
|
|
|
source.array['close'][2] = 100.
|
|
await casc.resync()
|
|
await wait_all_tasks_blocked()
|
|
check_publication()
|
|
assert len(notices) == 1
|
|
info: dict = notices[0]['fsp_update']
|
|
assert info['first'] == 37
|
|
assert info['last'] == 44
|
|
|
|
actors: tractor.ActorNursery
|
|
async with tractor.open_nursery() as actors:
|
|
portal: tractor.Portal = (
|
|
await actors.start_actor(
|
|
'trend-shm-reader',
|
|
enable_modules=[__name__],
|
|
)
|
|
)
|
|
try:
|
|
token: dict[str, Any] = (
|
|
destination._token.as_msg()
|
|
)
|
|
remote: dict[str, Any] = (
|
|
await portal.run(
|
|
_read_in_child,
|
|
token=token,
|
|
)
|
|
)
|
|
assert remote['first'] == 37
|
|
assert remote['last'] == 44
|
|
assert remote['readonly']
|
|
fields: tuple[str, ...] = (
|
|
reader.array.dtype.names
|
|
)
|
|
field: str
|
|
for field in fields:
|
|
np.testing.assert_array_equal(
|
|
np.asarray(
|
|
remote['fields'][field],
|
|
dtype=float,
|
|
),
|
|
reader.array[field],
|
|
)
|
|
finally:
|
|
await actors.cancel()
|
|
assert casc.complete.is_set()
|
|
|
|
trio.run(main)
|