# 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
# .
'''
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', ' 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)