diff --git a/models/finstitutions/trading/ticks.py b/models/finstitutions/trading/ticks.py index ffa006e..5e7d71f 100644 --- a/models/finstitutions/trading/ticks.py +++ b/models/finstitutions/trading/ticks.py @@ -208,6 +208,11 @@ class TradingTick(BaseModel): examples = ["UTC", "Asia/Kolkata"] ) + prevClose: float = Field( + description = "the closing price of this instrument on the previous day", + frozen = True + ) + ltp: float = Field( description = "the last price of this instrument at the time of requesting the symbol list", frozen = True @@ -233,17 +238,17 @@ class TradingTick(BaseModel): ) h: float = Field( - description = "this session's high price", + description = "this session's highest price", frozen = True ) l: float = Field( - description = "this session's low price", + description = "this session's lowest price", frozen = True ) c: float = Field( - description = "this session's close price", + description = "this session's close price; typically the same as the ltp", frozen = True ) @@ -364,7 +369,7 @@ class TradingTick(BaseModel): "ltp": self.ltp, "chg": self.chg, "pChg": self.pChg, - "totVol": self.totVol, + "totVol": self.totVol } @staticmethod @@ -383,8 +388,10 @@ class TradingTick(BaseModel): # Stash frequently needed vars: broker_token = tick["instrument_token"] tick_lookup = instrument_lookup[broker_token] - change = tick["change"] last_price = tick["last_price"] + prev_close = tick["ohlc"]["close"] + change = last_price - prev_close + p_change = tick["change"] # Model the currently picked tick: modelled_ticks.append( @@ -400,14 +407,15 @@ class TradingTick(BaseModel): strike = tick_lookup.get("strike"), expiryTs = tick_lookup.get("expiryTs"), expiryTz = tick_lookup.get("expiryTz"), + prevClose = prev_close, ltp = last_price, qty = tick.get("last_traded_quantity"), chg = change, - pChg = change / (last_price - change), + pChg = p_change, o = tick["ohlc"]["open"], h = tick["ohlc"]["high"], l = tick["ohlc"]["low"], - c = tick["ohlc"]["close"], + c = last_price, totVol = tick.get("volume_traded"), vwap = tick.get("average_traded_price"), totBuyQty = tick.get("total_buy_quantity"), diff --git a/playground/socketio/to_kafka.py b/playground/socketio/to_kafka.py index 650bdd9..e74cecd 100644 --- a/playground/socketio/to_kafka.py +++ b/playground/socketio/to_kafka.py @@ -88,13 +88,14 @@ kafka_producer = ProducerKafka( topic = "tickers", config = create_config( bootstrap_servers = "del.ditscentre.in:9092", + buffer_size = 1, security_protocol = "SSL", - # ca_file = r"../../creds/kafka/cert_authority.pem", - # cert_file = r"../../creds/kafka/fullchain.pem", - # key_file = r"../../creds/kafka/privkey.pem" - ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"), - cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"), - key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem") + ca_file = r"../../creds/kafka/cert_authority.pem", + cert_file = r"../../creds/kafka/fullchain.pem", + key_file = r"../../creds/kafka/privkey.pem" + # ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"), + # cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"), + # key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem") ) ) @@ -110,6 +111,7 @@ def to_kafka(tick: TradingTick) -> bool: success = False summary = tick.summary + summary["messageType"] = "ticks" success = kafka_producer.produce(value = summary) return success @@ -133,7 +135,10 @@ def on_connect(ws, response): def on_ticks(ws, ticks): + # print("TICK SAMPLE:", json.to_string(ticks, default=str)) ticks = TradingTick.from_zerodha_kite(ticks = ticks, instrument_lookup = INSTRUMENT_LOOKUP) + # print("TICK SAMPLE:", json.to_string(ticks[0].model_dump(), default=str)) + # print("TICK SAMPLE:", json.to_string(ticks[0].summary, default=str)) results = [to_kafka(tick) for tick in ticks] success = sum(results) print(f"TICKS: {len(ticks): <4} | PRODUCED: {success: <4}{' | FAILURE(S)!' if success < len(results)else ''}") @@ -149,8 +154,8 @@ def main(): global INSTRUMENT_LOOKUP # Load Zerodha credentials: - # creds = json.from_file(r"../../creds/zerodha/api.json") - creds = json.from_file(os.path.join(parent_dir, "creds", "zerodha", "api.json")) + creds = json.from_file(r"../../creds/zerodha/api.json") + # creds = json.from_file(os.path.join(parent_dir, "creds", "zerodha", "api.json")) api_key = creds["apiKey"] access_token = creds["accessToken"] @@ -158,19 +163,18 @@ def main(): kite = KiteConnect(api_key = api_key) kite.set_access_token(access_token) - # Get a list of instruments to work with: + # Get the entire list of instruments: instruments = kite.instruments(exchange = "NSE") - instruments = instruments[:100] - instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments] + instruments += kite.instruments(exchange = "NFO") + instruments += kite.instruments(exchange = "BSE") + instruments += kite.instruments(exchange = "BFO") + instruments += kite.instruments(exchange = "MCX") + instruments += kite.instruments(exchange = "CDS") + instruments += kite.instruments(exchange = "BCD") - instruments_csv = [] - instruments_csv += kite.instruments(exchange = "NSE") - instruments_csv += kite.instruments(exchange = "NFO") - # instruments_csv += kite.instruments(exchange = "BSE") - # instruments_csv += kite.instruments(exchange = "BFO") - # instruments_csv += kite.instruments(exchange = "MCX") - # instruments_csv += kite.instruments(exchange = "CDS") - # instruments_csv += kite.instruments(exchange = "BCD") + # Pick the instruments of interest: + instruments = instruments[:1000] + instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments] # Create the lookup: for i in instruments: diff --git a/utils_v2/queue/kafka.py b/utils_v2/queue/kafka.py index b36015b..12a5230 100644 --- a/utils_v2/queue/kafka.py +++ b/utils_v2/queue/kafka.py @@ -88,7 +88,8 @@ def create_config( ca_file: str | None = None, cert_file: str | None = None, key_file: str | None = None, - client_id: str | None = None + client_id: str | None = None, + buffer_size: int = 32 ) -> dict: """ @@ -100,6 +101,7 @@ def create_config( :param cert_file: Needed for 'SSL' security protocol. :param key_file: Needed for 'SSL' security protocol. :param client_id: An identifier for one producer/consumer. Useful for debugging later. + :param buffer_size: The size of the local message buffer in MBs. :return: The dictionary that needs to be passed as the 'conf' param when creating the producer/consumer. """ @@ -107,7 +109,8 @@ def create_config( if not isinstance(bootstrap_servers, list): bootstrap_servers = [bootstrap_servers] config = { "bootstrap.servers": ",".join(bootstrap_servers), - "security.protocol": security_protocol + "security.protocol": security_protocol, + # "buffer.memory": buffer_size } # Add the SSL security details: @@ -194,6 +197,16 @@ class ProducerKafka: if not self.__connected: self.connect() return self.__connected + def flush(self): + + """ + Flushes the buffer entirely. + :return: None + """ + + self.__producer.flush() + self.__printer("Producer flushed.") + def close(self): """