Commit Graph

3173 Commits (ffb57f0256e8333ab32cbc6ddb5820a538ada908)

Author SHA1 Message Date
Tyler Goodlet b3d1b1aa63 Port feed layer to use new `samplerd` APIs
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.
2023-01-13 13:21:15 -05:00
Tyler Goodlet 5ec1a72a3d Implement a `samplerd` singleton actor service
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.
2023-01-13 13:21:15 -05:00
Tyler Goodlet a342f7d2d4 Make `._daemon.Services` for use as singleton
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.
2023-01-13 13:21:15 -05:00
Tyler Goodlet 2c76cee928 Begin formalizing `Sampler` singleton API
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.
2023-01-13 13:21:15 -05:00
Tyler Goodlet b5f2ff854c Drop meaning the clearing rate, use per step count 2023-01-13 13:21:15 -05:00
Tyler Goodlet 3efb0b5884 Sync 1s (or less) sampler steps using rounded now-epoch 2023-01-13 13:21:15 -05:00
Tyler Goodlet 009bbe456e Always `.error()` log unknown queries for `marketstore` 2023-01-13 13:21:15 -05:00
Tyler Goodlet daf7b3f4a5 Only accept 6 tries for the same duplicate hist frame
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.
2023-01-13 13:21:15 -05:00
Tyler Goodlet b0a6dd46e4 Use recon set on stack closing during reconnect
Hopefully resolves https://github.com/pikers/piker/issues/434
2023-01-13 13:21:15 -05:00
Tyler Goodlet 1c5141f4c6 Fix f-str in duplicate frame msg print 2023-01-13 13:21:15 -05:00
Tyler Goodlet 4cdd2271b0 Drop `tractor` assert bug note 2023-01-13 13:21:15 -05:00
Tyler Goodlet 89095d4e9f Ensure FSPs last 2 times are synced with its source 2023-01-13 13:21:15 -05:00
Tyler Goodlet 04c0d77595 Frame ticks in helper routine
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
2023-01-13 13:21:15 -05:00
Tyler Goodlet d1b07c625f Copy timestamps from source to FSP dest buffer
Slice up to history's length worth of (latest) time stamps from source
series read at the start of the history init phase.
2023-01-13 13:21:15 -05:00
Tyler Goodlet a5bb33b0ff Avoid key error on already popped cancel 2023-01-13 13:21:15 -05:00
Tyler Goodlet 8e1ceca43d Add some data-flows jargon notes (re: #270) 2023-01-13 13:21:15 -05:00
Tyler Goodlet c85e7790de Rename `._flumes.py` -> `.flows.py` 2023-01-13 13:21:15 -05:00
Tyler Goodlet 2399c618b6 Expand sampler loop shm write lines 2023-01-13 13:21:15 -05:00
Tyler Goodlet 7ec88f8cac Make hist shm token optional to allow for FSPs 2023-01-13 13:21:15 -05:00
Tyler Goodlet eacd44dd65 Move `Flume` to a new `.data._flumes` module 2023-01-13 13:21:15 -05:00
Tyler Goodlet e5e70a6011 Extend `Flume` methods
Add some (untested) data slicing util methods for mapping time ranges to
source data indices:
- `.get_index()` which maps a single input epoch time to an equiv array
  (int) index.
- add `slice_from_time()` which returns a view of the shm data from an
  input epoch range presuming the underlying struct array contains
  a `'time'` field with epoch stamps.
- `.view_data()` which slices out the "in view" data according to the
  current state of the passed in `pg.PlotItem`'s view box.
2023-01-13 13:21:15 -05:00
Tyler Goodlet 7da5c2b238 Add epoch time index to fsp buffers 2023-01-13 13:21:15 -05:00
Tyler Goodlet 1ee49df31d Ensure a rt shm buffer without backfill has correct epoch timestamping 2023-01-13 13:21:15 -05:00
Tyler Goodlet f2df32a673 Use throttle period for wait-on-clearing-event timeout 2023-01-13 13:21:15 -05:00
Tyler Goodlet 125e31dbf3 Implement by-type tick-framing in throttler loop
This has been an outstanding idea for a while and changes the framing
format of tick events into a `dict[str, list[dict]]` wherein for each
tick "type" (eg. 'bid', 'ask', 'trade', 'asize'..etc) we create an FIFO
ordered `list` of events (data) and then pack this table into each
(throttled) send. This gives an additional implied downsample reduction
(in terms of iteration on the consumer side) from `N` tick-events to
a (max) `T` tick-types presuming the rx side only needs the latest tick
event.

Drop the `types: set` and adjust clearing event test to use the new
`ticks_by_type` map's keys.
2023-01-13 13:21:15 -05:00
Tyler Goodlet 715e693564 Improved clearing-tick-burst-oriented throttling
Instead of uniformly distributing the msg send rate for a given
aggregate subscription, choose to be more bursty around clearing ticks
so as to avoid saturating the consumer with L1 book updates and vs.
delivering real trade data as-fast-as-possible.

Presuming the consumer is in the "UI land of slow" (eg. modern display
frame rates) such an approach serves more useful for seeing "material
changes" in the market as-bursty-as-possible (i.e. more short lived fast
changes in last clearing price vs. many slower changes in the bid-ask
spread queues). Such an approach also lends better to multi-feed
overlays which in aggregate tend to scale linearly with the number of
feeds/overlays; centralization of bursty arrival rates allows for
a higher overall throttle rate if used cleverly with framing.
2023-01-13 13:21:15 -05:00
Tyler Goodlet 43717c92d9 Type annot-declare fsp-engine data `Feed` 2023-01-13 13:21:15 -05:00
Tyler Goodlet f370685c62 Init msg keys are always lower case 2023-01-13 13:21:15 -05:00
Tyler Goodlet 4300470786 Fix for empty tsdb query result case 2023-01-13 13:21:15 -05:00
Tyler Goodlet b89fd9652c `binance`: always request an extra 1min OHLC bar
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.
2023-01-13 13:14:35 -05:00
Tyler Goodlet 51f4afbd88 Don't raise on quote feed lags to dark clearing loop 2023-01-13 12:51:07 -05:00
Tyler Goodlet 7ef8111381 Provide `datetime`-sorted clears table iteration
Likely pertains to helping with stuff in issues #345 and #373 and just
generally is handy to have when processing ledgers / clearing event
tables.

Adds the following helper methods:
- `iter_by_dt()` to iter-sort an arbitrary `Transaction`-like table of
  clear entries.
- `Position.iter_clears()` as a convenience wrapper for the above.
2023-01-13 12:51:01 -05:00
Tyler Goodlet 35b097469b Round spread (slap) offset to min tick digits 2023-01-13 12:51:01 -05:00
Tyler Goodlet 94290c7d8b `kraken`: ignore mismatched zero-ed pps (for now)
See more details in the GH comment:
https://github.com/pikers/piker/issues/373#issuecomment-1380988581

More or less we need to pull and include the transfer fees for
withdrawals in our ledger tracking but this serves as a sloppy
workaround for the moment.
2023-01-13 12:48:18 -05:00
Tyler Goodlet 73379d3627 Run CI on all PRs 2023-01-13 12:39:17 -05:00
Tyler Goodlet 23835f2c08 `deribit`: drop old `backfill_bars()` ep 2023-01-13 12:39:17 -05:00
Tyler Goodlet d2aee00a56 `kraken`: only do unsub if connected
Trying to send a message in the `NoBsWs.fixture()` exit when the ws is
not currently disconnected causes a double `._stack.close()` call which
will corrupt `trio`'s coro stack. Instead only do the unsub if we detect
the ws is still up.

Also drops the legacy `backfill_bars()` module endpoint.

Fixes #437
2023-01-13 12:39:17 -05:00
Tyler Goodlet cf6e44cb9c Add `NoBsWs.connected()` predicate 2023-01-13 12:39:17 -05:00
Tyler Goodlet a146ad9e69 Never restart `ib-gw` containers on boot 2023-01-13 12:37:49 -05:00
Tyler Goodlet 70ad1a1860 `kraken`: don't presume src fiat symbol size in pos predicate 2023-01-13 12:37:49 -05:00
Tyler Goodlet f3ef73ef41 `kraken`: drop symbol token size =6 check 2023-01-13 12:37:49 -05:00
Tyler Goodlet a9832dc0cb `ib`: fix position log msg 2023-01-13 12:37:49 -05:00
Tyler Goodlet 9be245e955 `ib`: Add treasury yield futs to adhoc fqsn set 2023-01-13 12:37:49 -05:00
Tyler Goodlet 800773e585 ib: ignore throttles on `.get_head_time()` 2023-01-13 12:37:49 -05:00
goodboy 8d1eb81f16
Merge pull request #414 from pikers/agg_feedz
Agg feedz
2023-01-13 12:20:47 -05:00
Tyler Goodlet 963e5bdd62 Go back to `Feed.pause/resume()`, new flume APIs coming later 2023-01-10 11:09:19 -05:00
Tyler Goodlet 55de9abc41 Adjust cli mod imports of daemon sockaddr vars 2023-01-10 11:09:19 -05:00
Tyler Goodlet 593db0ed0d Only run `kraken` feed tests in CI, use `open_test_pikerd()` 2023-01-10 11:09:19 -05:00
Tyler Goodlet 06622105cd Add a `open_test_pikerd()` acm fixture for easy booting of the service stack 2023-01-10 11:09:19 -05:00
Tyler Goodlet 008ae47e14 Reset `._registry_addr` to any passed in value from caller 2023-01-10 11:09:19 -05:00