Initial support for real-time multi-symbol overlay charts using an
aggregate feed delivered by `Feed.open_multi_stream()`.
The setup steps for constructing the overlayed plot items is still very
very rough and will likely provide incentive for better refactoring high
level "charting APIs". For each fqsn passed into `display_symbol_data()`
we now synchronously,
- create a single call to `LinkedSplits.plot_ohlc_main() -> `ChartPlotWidget`
where we cache the chart in scope and for all other "sibling" fqsns
we,
- make a call to `ChartPlotWidget.overlay_plotitem()` -> `PlotItem`, hide its axes,
make another call with this plotitem input to
`ChartPlotWidget.draw_curve()`, set a sym-specific view box auto-yrange maxmin callback,
register the plotitem in a global `pis: dict[str, list[pgo.PlotItem, pgo.PlotItem]] = {}`
Once all plots have been created we then asynchronously for each symbol,
- maybe create a volume chart and register it in a similar task-global
table: `vlms: dict[str, ChartPlotWidget] = {}`
- start fsp displays for each symbol
Then common entrypoints are entered once for all symbols:
- a single `graphics_update_loop()` loop-task is started wherein
real-time graphics update components for each symbol are created,
* `L1Labels`
* y-axis last clearing price stickies
* `maxmin()` auto-ranger
* `DisplayState` (stored in a table `dss: dict[str, DisplayState] = {}`)
* an `increment_history_view()` task
and a single call to `Feed.open_multi_stream()` is used to create
a symbol-multiplexed quote stream which drives a single loop over all
symbols wherein for each quote the appropriate components are looked
up and passed to `graphics_update_cycle()`.
- a single call to `open_order_mode()` is made with the first symbol
provided as input, though eventually we want to support passing in the
entire list.
Further internal implementation details:
- special tweaks to the `pg.LinearRegionItem` setup wherein the region
is added with a zero opacity and *after* all plotitem overlays to
avoid and issue where overlays weren't being shown within the region
area in the history chart.
- all symbol-specific graphics oriented update calls are adjusted to
pass in the fqsn:
* `update_fsp_chart()`
* `ChartView._set_yrange()`
* ChartPlotWidget.update_graphics_from_flow()`
- avoid a double increment on sample step updates by not calling the
increment on any vlm chart since it seems the vlm-ohlc chart linking
already takes care of this now?
- use global counters for the last epoch time step to avoid incrementing
all views more then once per new time step given underlying shm array
buffers may be on different array-index values from one another.
Main "public" API change is to make `GodWidget.get/set_chart_symbol()`
accept and cache-on fqsn tuples to allow handling overlayed chart groups
and adjust method names to be plural to match.
Wrt `LinkedSplits`,
- create all chart widget axes with a `None` plotitem argument and set
the `.pi` field after axis creation (since apparently we have another
object reference causality dilemma..)
- set a monkeyed `PlotItem.chart_widget` for use in axes that still need
the widget reference.
- drop feed pause/resume for now since it's leaking feed tasks on the
`brokerd` side and we probably don't really need it any more, and if
we still do it should be done on the feed not the flume.
Wrt `ChartPlotItem`,
- drop `._add_sticky()` and use the `Axis` method instead and add some
overlay + axis sanity checks.
- refactor `.draw_ohlc()` to be a lighter wrapper around a call to
`.add_plot()`.
We have this method on our `ChartPlotWidget` but it makes more sense to
directly associate axis-labels with, well, the label's parent axis XD.
We add `._stickies: dict[str, YAxisLabel]` to replace
`ChartPlotWidget._ysticks` and pass in the `pg.PlotItem` to each axis
instance, stored as `Axis.pi` instead of handing around linked split
references (which are way out of scope for a single axis).
More work needs to be done to remove dependence on `.chart:
ChartPlotWidget` references in the date axis type as per comments.
More or less a revamp (and possibly first draft for something similar in
`tractor` core) which ensures all actor trees attempt to discover the
`pikerd` registry actor.
Implementation improvements include:
- new `Registry` singleton which houses the `pikerd` discovery
socket-address `Registry.addr` + a `open_registry()` manager which
provides bootstrapped actor-local access.
- refine `open_piker_runtime()` to do the work of opening a root actor
and call the new `open_registry()` depending on whether a runtime has
yet been bootstrapped.
- rejig `[maybe_]open_pikerd()` in terms of the above.
Allows running simultaneous data feed services on the same (linux) host
by avoiding file-name collisions instead keying shm buffer sets by the
given `brokerd` instance. This allows, for example, either multiple dev
versions of the data layer to run side-by-side or for the test suite to
be seamlessly run alongside a production instance.
Previously we were relying on implicit actor termination in
`maybe_spawn_daemon()` but really on `pikerd` teardown we should be sure
to tear down not only all service tasks in each actor but also the actor
runtimes. This adjusts `Services.cancel_service()` to only cancel the
service task scope and wait on the `complete` event and reworks the
`open_context_in_task()` inner closure body to,
- always cancel the service actor at exit.
- not call `.cancel_service()` (potentially causing recursion issues on
cancellation).
- allocate a `complete: trio.Event` to signal full task + actor termination.
- pop the service task from the `.service_tasks` registry.
Further, add a `maybe_set_global_registry_sockaddr()` helper-cm to do
the work of checking whether a registry socket needs-to/has-been set
and use it for discovery calls to the `pikerd` service tree.
Seems that by default their history indexing rounds down/back to the
previous time step, so make sure we add a minute inside `Client.bars()`
when the `end_dt=None`, indicating "get the latest bar". Add
a breakpoint block that should trigger whenever the latest bar vs. the
latest epoch time is mismatched; we'll remove this after some testing
verifying the history bars issue is resolved.
Further this drops the legacy `backfill_bars()` endpoint which has been
deprecated and unused for a while.
Always use `open_sample_stream()` to register fast and slow quote feed
buffers and get a sampler stream which we use to trigger
`Sampler.broadcast_all()` calls on the service side after backfill
events.
Now spawned under the `pikerd` tree as a singleton-daemon-actor we offer
a slew of new routines in support of this micro-service:
- `maybe_open_samplerd()` and `spawn_samplerd()` which provide the
`._daemon.Services` integration to conduct service spawning.
- `open_sample_stream()` which is a client-side endpoint which does all
the work of (lazily) starting the `samplerd` service (if dne) and
registers shm buffers for update as well as connect a sample-index
stream for iterator by the caller.
- `register_with_sampler()` which is the `samplerd`-side service task
endpoint implementing all the shm buffer and index-stream registry
details as well as logic to ensure a lone service task runs
`Services.increment_ohlc_buffer()`; it increments at the shortest period
registered which, for now, is the default 1s duration.
Further impl notes:
- fixes to `Services.broadcast()` to ensure broken streams get discarded
gracefully.
- we use a `pikerd` side singleton mutex `trio.Lock()` to ensure
one-and-only-one `samplerd` is ever spawned per `pikerd` actor tree.
Drop the `_services` module level ref and adjust all client code to
match. Drop struct inheritance and convert all methods to class level.
Move `Brokerd.locks` -> `Services.locks` and add sampling mod to pikerd
enabled set.
We're moving toward a single actor managing sampler work and distributed
independently of `brokerd` services such that a user can run samplers on
different hosts then real-time data feed infra. Most of the
implementation details include aggregating `.data._sampling` routines
into a new `Sampler` singleton type.
Move the following methods to class methods:
- `.increment_ohlc_buffer()` to allow a single task to increment all
registered shm buffers.
- `.broadcast()` for IPC relay to all registered clients/shms.
Further add a new `maybe_open_global_sampler()` which allocates
a service nursery and assigns it to the `Sampler.service_nursery`; this
is prep for putting the step incrementer in a singleton service task
higher up the data-layer actor tree.
When we see multiple history frames that are duplicate to the request
set, bail re-trying after a number of tries (6 just cuz) and return
early from the tsdb backfill loop; presume that this many duplicates
means we've hit the beginning of history. Use a `collections.Counter`
for the duplicate counts. Make sure and warn log in such cases.
Wow, turns out tick framing was totally borked since we weren't framing
on "greater then throttle period long waits" XD
This moves all the framing logic into a common func and calls it in
every case:
- every (normal) "pre throttle period expires" quote receive
- each "no new quote before throttle period expires" (slow case)
- each "no clearing tick yet received" / only burst on clears case
Tyler Goodlet