diff --git a/background/finstitutions/trading/eod_capture.py b/background/finstitutions/trading/eod_capture.py index c58b230..c5a5391 100644 --- a/background/finstitutions/trading/eod_capture.py +++ b/background/finstitutions/trading/eod_capture.py @@ -433,7 +433,7 @@ if __name__ == "__main__": # Get the config. from the command-line: parser = argparse.ArgumentParser( description = ( - "To fetch and store EoD data for a given data from NSE's BhavCopy section. " + "To fetch and store EoD data for one or more dates from NSE's BhavCopy section. " "The data will be fetched for both, equities and derivatives." ) ) diff --git a/background/finstitutions/trading/tick_in_stateful.py b/background/finstitutions/trading/tick_in_stateful.py index bd2bda2..414160e 100644 --- a/background/finstitutions/trading/tick_in_stateful.py +++ b/background/finstitutions/trading/tick_in_stateful.py @@ -44,6 +44,7 @@ import os from utils_v2.string import json from utils_v2.string import regex from utils_v2.system import files +from utils_v2.date_time import date_time from utils_v2.database.async_mongo_v2 import AsyncMongo from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache from utils_v2.queue.kafka.controllers.kafka import ProducerKafka, ConsumerKafka @@ -158,15 +159,16 @@ def ticks_to_kafka(ticks: List[TradingTick]) -> int: # Push out the ticks to the queue: for tick in ticks: - # Flush the existing messages if needed: + # Flush the existing messages (if needed): TOTAL_TICK_COUNT += 1 TICKS_SINCE_FLUSH += 1 - if TICKS_SINCE_FLUSH > 5_000: - kafka_producer.flush() + if TICKS_SINCE_FLUSH > 50_000: + kafka_producer.flush(timeout = 0.0) TICKS_SINCE_FLUSH = 0 # Push this one tick to the queue: success = kafka_producer.produce(value = tick.summary) + kafka_producer.client.poll(0) if success: success_count += 1 else: failure_count += 1 @@ -211,7 +213,8 @@ def on_zerodha_kite_ticks(ws, ticks) -> None: # Model the raw input ticks: ticks = TradingTick.from_zerodha_kite( ticks = ticks, - instrument_lookup = ZERODHA_INSTRUMENT_LOOKUP + instrument_lookup = ZERODHA_INSTRUMENT_LOOKUP, + received_ts = date_time.get_current_utc_date_time(as_string = False) ) # Send the ticks to Kafka: @@ -428,7 +431,10 @@ def init( client_id = f"{SERVER_HOSTNAME}_{TOKEN_KEY}", acks = producer_creds["config"].get("acks", 1), retries = producer_creds["config"].get("retries", 1), - linger_ms = producer_creds["config"].get("lingerMs", 0) + linger_ms = producer_creds["config"].get("lingerMs", 0), + misc_json = { + "queue.buffering.max.messages": 2_00_000 + } ), topic = producer_creds["topic"], serializer = JSONSerializer(), diff --git a/models/finstitutions/trading/ticks.py b/models/finstitutions/trading/ticks.py index 8a1ff64..56dd884 100644 --- a/models/finstitutions/trading/ticks.py +++ b/models/finstitutions/trading/ticks.py @@ -291,29 +291,36 @@ class TradingTick(BaseModel): frozen = True ) - tradeTs: AwareDatetime | None = Field( + rcvdTs: AwareDatetime | None = Field( + description = "the time (utc) at which this update was received", + frozen = True, + default = None, + validate_default = True + ) + + tradeTs: AwareDatetime | None = Field( description = "the last trade time (utc) of this instrument", frozen = True ) - tradeTz: str | None = Field( + tradeTz: str | None = Field( description = "the timezone in which the last trade time should be interpreted; should be compatible with pytz", frozen = True, examples = ["UTC", "Asia/Kolkata"] ) - exchgTs: AwareDatetime | None = Field( + exchgTs: AwareDatetime | None = Field( description = "the time (utc) at which this update was received from the exchange", frozen = True ) - exchgTz: str | None = Field( + exchgTz: str | None = Field( description = "the timezone in which the exchange's time should be interpreted; should be compatible with pytz", frozen = True, examples = ["UTC", "Asia/Kolkata"] ) - depth: MarketDepth | None = Field( + depth: MarketDepth | None = Field( description = "the market depth data for this instrument at the time of this update" ) @@ -380,14 +387,18 @@ class TradingTick(BaseModel): "pChg": self.pChg, "vwap": self.vwap, "totVol": self.totVol, + "rcvdTs": self.rcvdTs.timestamp(), "tradeTs": self.tradeTs.timestamp(), - "tradeTz": self.tradeTz + "tradeTz": self.tradeTz, + "exchgTs": self.exchgTs.timestamp(), + "exchgTz": self.exchgTz } @staticmethod def from_zerodha_kite( ticks: dict | List[dict], - instrument_lookup: dict + instrument_lookup: dict, + received_ts: datetime.datetime = None ) -> list: # Ensure that we are working with a list: @@ -436,6 +447,7 @@ class TradingTick(BaseModel): oi = tick.get("oi"), oiDayHigh = tick.get("oi_day_high"), oiDayLow = tick.get("oi_day_low"), + rcvdTs = received_ts, tradeTs = tick.get("last_trade_time"), tradeTz = "Asia/Kolkata", exchgTs = tick.get("exchange_timestamp"), @@ -485,6 +497,11 @@ class TradingTick(BaseModel): # Done here: return value + @field_validator("rcvdTs", mode = "before") + def validate_received_timestamp(cls, value): + if value is None: value = date_time.get_current_utc_date_time(as_string = False) + return value + # ***************************************************************************************************************** # ***** **** diff --git a/wsio/finstitutions/trading/tick_out.py b/wsio/finstitutions/trading/tick_out.py index f76c1f1..a2e1d33 100644 --- a/wsio/finstitutions/trading/tick_out.py +++ b/wsio/finstitutions/trading/tick_out.py @@ -44,6 +44,7 @@ import os from utils_v2.string import json from utils_v2.string import regex from utils_v2.system import files +from utils_v2.date_time import date_time from utils_v2.database.async_mongo_v2 import AsyncMongo from utils_v2.queue.kafka.controllers.async_kafka import ConsumerKafka, get_ssl_context from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache @@ -82,7 +83,7 @@ from icecream import IceCreamDebugger # Debugging: printer = IceCreamDebugger(prefix = "Tick-Out | ", includeContext = True) -# printer.disable() +no_context_printer = IceCreamDebugger(prefix = "Tick-Out | ", includeContext = False) # To make API calls: http_client = httpx.AsyncClient( @@ -219,12 +220,14 @@ async def ticks_from_kafka( # Do the next part infinitely: while True: + # Note the time: + now_utc = date_time.get_current_utc_date_time().timestamp() + # Get messages form Kafka: ticks = await consumer.consume( count = fetch_count, timeout = fetch_timeout ) - printer(len(ticks)) # If there are no updates to give: if not ticks: continue @@ -235,6 +238,21 @@ async def ticks_from_kafka( tasks = [send_ticks(t.value if isinstance(t.value, list) else [t.value]) for t in ticks] results = await asyncio.gather(*tasks) + # Analyze the ticks: + # latency = [abs(now_utc - t.value.get("rcvdTs", t.value["tradeTs"])) for t in ticks] + latency = [abs(now_utc - t.ts.timestamp()) for t in ticks] + avg_latency = sum(latency) / len(latency) + total_ticks = len(ticks) + # late_cutoff_seconds = 3.0 + # + # late_ticks = 0 + # for tick in ticks: + # if now_utc - tick.value["tradeTs"] > late_cutoff_seconds: + # late_ticks += 1 + # ticks_str = f"COUNT: {total_ticks: >5,} | LATE: {late_ticks: >5,} ({(late_ticks/total_ticks)*100.0:.2f}%)" + ticks_str = f"COUNT: {total_ticks: >5,} | AVG. LATENCY: {avg_latency:.5f}" + no_context_printer(ticks_str) + # ---------------------------------------------------------------------------------------------------------------------