Compare commits

..

No commits in common. "9ebb977731f34b8f5a53204d2ed2e3e03dd7fe62" and "176090b234446f5b67cc040f93032d14f20de664" have entirely different histories.

4 changed files with 29 additions and 58 deletions

View File

@ -102,12 +102,12 @@ class AggTrade(Struct, frozen=True):
a: int # Aggregate trade ID
p: float # Price
q: float # Quantity with all the market trades
nq: float # Normal quantity without the trades involving RPI orders
f: int # First trade ID
l: int # noqa Last trade ID
T: int # Trade time
m: bool # Is the buyer the market maker?
M: bool|None = None # Ignore
nq: float|None = None # Normal quantity without the trades involving RPI orders
M: bool | None = None # Ignore
async def stream_messages(

View File

@ -1187,7 +1187,7 @@ async def load_aio_clients(
# the API TCP in `ib_insync` connection can be flaky af so instead
# retry a few times to get the client going..
connect_retries: int = 3,
connect_timeout: float = 30, # in case a remote-host
connect_timeout: float = 10,
disconnect_on_exit: bool = True,
) -> dict[str, Client]:

View File

@ -242,7 +242,6 @@ def anal(
trio.run(main)
# TODO, move to `.tsp._annotate`
async def markup_gaps(
fqme: str,
timeframe: float,
@ -289,38 +288,18 @@ async def markup_gaps(
)
# XXX: probably a gap in the (newly sorted or de-duplicated)
# dt-df, so we might need to re-index first..
dt: pl.Series = row['dt']
dt_prev: pl.Series = row['dt_prev']
if prev_r.is_empty():
# XXX, filter out any special ignore cases,
# - UNIX-epoch stamped datums
# - first row
if (
dt_prev.dt.epoch()[0] == 0
or
dt.dt.epoch()[0] == 0
):
log.warning('Skipping row with UNIX epoch timestamp ??')
continue
if wdts[0]['index'][0] == iend: # first row
log.warning('Skipping first-row (has no previous obvi) !!')
continue
# XXX, if the previous-row by shm-index is missing,
# meaning there is a missing sample (set), get the prior
# row by df index and attempt to use it?
i_wdts: pl.DataFrame = wdts.with_row_index(name='i')
i_row: int = i_wdts.filter(pl.col('index') == iend)['i'][0]
prev_row_by_i = wdts[i_row]
prev_r: pl.DataFrame = prev_row_by_i
# debug any missing pre-row
if tractor._state.is_debug_mode():
await tractor.pause()
istart: int = prev_r['index'][0]
# dt_start_t: float = dt_prev.timestamp()
# start_t: float = prev_r['time']
# assert (
# dt_start_t
# ==
# start_t
# )
# TODO: implement px-col width measure
# and ensure at least as many px-cols
@ -379,7 +358,6 @@ def ldshm(
fqme: str,
write_parquet: bool = True,
reload_parquet_to_shm: bool = True,
pdb: bool = False, # --pdb passed?
) -> None:
'''
@ -399,7 +377,7 @@ def ldshm(
open_piker_runtime(
'polars_boi',
enable_modules=['piker.data._sharedmem'],
debug_mode=pdb,
debug_mode=True,
),
open_storage_client() as (
mod,
@ -419,9 +397,6 @@ def ldshm(
times: np.ndarray = shm.array['time']
d1: float = float(times[-1] - times[-2])
d2: float = 0
# XXX, take a median sample rate if sufficient data
if times.size > 2:
d2: float = float(times[-2] - times[-3])
med: float = np.median(np.diff(times))
if (
@ -432,6 +407,7 @@ def ldshm(
raise ValueError(
f'Something is wrong with time period for {shm}:\n{times}'
)
period_s: float = float(max(d1, d2, med))
null_segs: tuple = tsp.get_null_segs(
@ -441,8 +417,6 @@ def ldshm(
# TODO: call null-seg fixer somehow?
if null_segs:
if tractor._state.is_debug_mode():
await tractor.pause()
# async with (
# trio.open_nursery() as tn,
@ -524,11 +498,8 @@ def ldshm(
if (
not venue_gaps.is_empty()
or (
not step_gaps.is_empty()
# XXX, i presume i put this bc i was guarding
# for ib venue gaps?
# and
# period_s < 60
period_s < 60
and not step_gaps.is_empty()
)
):
# write repaired ts to parquet-file?

View File

@ -237,8 +237,8 @@ class LevelLabel(YAxisLabel):
class L1Label(LevelLabel):
text_flags = (
QtCore.Qt.TextFlag.TextDontClip
| QtCore.Qt.AlignmentFlag.AlignLeft
QtCore.Qt.TextDontClip
| QtCore.Qt.AlignLeft
)
def set_label_str(