# piker: trading gear for hackers # Copyright (C) 2018-present Tyler Goodlet (in stewardship of pikers) # 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 . ''' Startup-only schedule reservations and explicit synthetic provenance. No published rows are moved. Synthetic rows never enter storage here. ''' from collections.abc import Awaitable, Callable import math import numpy as np from piker.calc import SECONDS_PER_DAY from piker.log import get_logger log = get_logger(__name__) # Actor-local metadata, scoped by market and sampling period. # Each range is half-open UTC and identifies synthetic SHM samples. type SyntheticRanges = list[tuple[float, float]] _reservations: dict[tuple[str, float], SyntheticRanges] = {} def reservation_ranges( fqme: str, timeframe: float, ) -> list[tuple[float, float]]: ''' Return a copy for RPC readers; do not expose mutable ownership. ''' return list(_reservations.get((fqme, timeframe), [])) async def reserve_history_gaps( frame: np.ndarray, timeframe: float, query_schedule: Callable[..., Awaitable[dict]], max_rows: int, newest_gaps: int = 32, ) -> tuple[np.ndarray, list[tuple[float, float]]]: ''' Reserve missing scheduled samples before publishing history. Keep venue closures compressed. Fully covered schedule replies permit flat OHLC placeholders at the expected sampling cadence. Unknown coverage and oversized reservations remain time gaps. Inspect at most `newest_gaps` gaps in retained history, newest first; zero disables reservations. Return the expanded frame and half-open UTC synthetic ranges separately. `_history.tsdb_backfill()` calls this before prepending stored history. It publishes the returned ranges in `_reservations`. Local readers use `reservation_ranges()`; remote readers use `piker.brokers.ib.feed.get_history_reservations()` on datad.ib. Future segment repair will replace these slots by timestamp; chart/FSP consumers still need synthetic-aware handling. No synthetic rows are written to parquet here. TODO: coordinated insertion when coverage or capacity cannot support a reservation; first pass never rewrites published rows. ''' if newest_gaps < 0: raise ValueError('newest_gaps must be nonnegative') # A gap needs two observed margins and room for both. A zero # request budget disables work without slicing `gaps[-0:]`. if ( max_rows < 2 or len(frame) < 2 or newest_gaps == 0 ): return frame, [] frame = frame[-max_rows:] times: np.ndarray = frame['time'] # A nonpositive cadence cannot define sample positions. Duplicate # or decreasing timestamps make the left/right margins ambiguous. # Leave the retained data unchanged; this helper does not sort or # deduplicate provider observations to guess a valid sample grid. if ( timeframe <= 0 or np.any(np.diff(times) <= 0) ): return frame, [] gaps: np.ndarray = np.flatnonzero(np.diff(times) > timeframe) inserts: dict[int, np.ndarray] = {} ranges: list[tuple[float, float]] = [] budget: int = max_rows # Match IB Client.history_schedule()'s first-pass 31-day bound. # TODO: advertise query limits through backend history config. max_schedule_span_s: int = 31 * SECONDS_PER_DAY for offset in reversed(gaps[-newest_gaps:]): left: float = float(times[offset]) right: float = float(times[offset + 1]) if right - left > max_schedule_span_s: continue try: # Backend-owned callback supplied in history config. # IB's open_history_client() routes it through the proxy # to Client.history_schedule(), returning UTC intervals. schedule: dict = await query_schedule(left, right) except Exception as exc: log.warning(f'Schedule unavailable for {left}:{right}: ' f'{exc!r}') continue if ( schedule['start'] > left or schedule['end'] < right or schedule['use_rth'] or not schedule['sessions'] ): continue stamps: list[np.ndarray] = [] count: int = 0 for start, end in schedule['sessions']: first: float = max(left + timeframe, start) # Preserve the existing sample grid across closures. first = left + math.ceil( (first - left) / timeframe ) * timeframe stop: float = min(right, end) size: int = max(0, math.ceil((stop - first) / timeframe)) count += size if count > budget: break if size: stamps.append(first + np.arange(size) * timeframe) if not count: continue if count > budget: # Recoverable: leave this entire gap packed/unresolved. # No rows or provenance have been published for it, so # later gaps can still use the remaining reservation # budget. Do not partially reserve or reindex this gap. log.warning(f'Reservation exceeds SHM capacity: ' f'{left}:{right}') continue missing: np.ndarray = np.concatenate(stamps) reserved: np.ndarray = np.zeros(count, dtype=frame.dtype) reserved['time'] = missing # COPY CLOSE / SYNTHESIZE TIMESTAMPS for scheduled slots. # Return explicit provenance; zero volume is NOT a flag. reserved[['open', 'high', 'low', 'close']] = ( frame['close'][offset] ) inserts[int(offset)] = reserved budget -= count # `np.diff` locates breaks in the synthetic sample cadence. # Add 1 to each break's left index to select the first row # of the next run. `np.split` returns one array per run; # encode each as [first timestamp, last + timeframe). # E.g. [101, 102, 108] -> [101, 103), [108, 109) at 1s. # Separate runs keep venue closures OUT of synthetic ranges. for group in np.split( missing, np.flatnonzero(np.diff(missing) != timeframe) + 1, ): ranges.append(( float(group[0]), float(group[-1] + timeframe), )) if not inserts: return frame, [] pieces: list[np.ndarray] = [] previous: int = 0 for offset, reserved in sorted(inserts.items()): pieces.extend([frame[previous:offset + 1], reserved]) previous = offset + 1 pieces.append(frame[previous:]) result: np.ndarray = np.concatenate(pieces)[-max_rows:] first_t: float = float(result['time'][0]) ranges = [(max(a, first_t), b) for a, b in ranges if b > first_t] return result, sorted(ranges)