From 3b8d248d8259e7f5e0cb2dce680a1ac7d2e966ee Mon Sep 17 00:00:00 2001 From: khushal Date: Mon, 27 Jan 2025 17:17:58 +0530 Subject: [PATCH] (20250127) Strategy check ready for live test. --- .../trading/strategy_bhandari_0.py | 294 ++++++++++++++++-- 1 file changed, 262 insertions(+), 32 deletions(-) diff --git a/background/finstitutions/trading/strategy_bhandari_0.py b/background/finstitutions/trading/strategy_bhandari_0.py index 3b42024..9111c9a 100644 --- a/background/finstitutions/trading/strategy_bhandari_0.py +++ b/background/finstitutions/trading/strategy_bhandari_0.py @@ -127,10 +127,10 @@ SERVER_HOSTNAME = str(socket.gethostname()) # ***** **** # ***************************************************************************************************************** -# Redis: +# For cache: redis_cache: AsyncRedisCache | None = None -# For database: +# For database(s): sql_writer: AsyncMySQL | None = None sql_reader: AsyncMySQL | None = None data_mongo: AsyncMongo | None = None @@ -141,13 +141,10 @@ kafka_consumer: ConsumerKafka | None = None # Session-awareness and maintenance of this script's state: SCRIPT_DATA = {} STRATEGY_REFERENCE = {} +ACTIVE_CALLS = {} strategy_reference_lock = asyncio.Semaphore(1) ticks_since_telegram = 0 -# For Zerodha-Kite: -ZERODHA_INSTRUMENT_TOKENS = [] -ZERODHA_INSTRUMENT_LOOKUP = {} - # ***************************************************************************************************************** # ***** **** @@ -156,13 +153,209 @@ ZERODHA_INSTRUMENT_LOOKUP = {} # ***************************************************************************************************************** +async def mark_call_as_active( + tick_ref: dict, + tick: dict +) -> bool: + + global ACTIVE_CALLS + + # get the keys: + tick_key = tick_ref["tickKey"] + redis_key = tick_ref["redisKey"] + + # Mark locally: + print(f"CALL ({redis_key}):", json.to_string(tick_ref)) + ACTIVE_CALLS[redis_key] = tick_ref + + # Mark in cache: + await redis_cache.set( + key = redis_key, + value = tick_ref + ) + + # Mark in SQL: + proc_name = "portfolio_trade_add" + proc_args = ( + tick_ref["userId"], # ........................................................ p_user_id + tick_ref["billingAccountId"], # .............................................. p_billing_account_id + 0, # ......................................................................... p_entity_integration_id + date_time.get_current_ist_date_time().strftime("%Y-%m-%d"), # ................ p_date + tick_ref["exchange"], # ...................................................... p_exchange + tick_ref["segment"], # ....................................................... p_segment + tick_ref["symbol"], # ........................................................ p_symbol + tick_ref["expiry"], # ........................................................ p_expiry + tick_ref["right"], # ......................................................... p_right + tick_ref["strike"], # ........................................................ p_strike + 0, # ......................................................................... p_qty + 0, # ......................................................................... p_price + tick["ltp"], # ............................................................... p_refPrice + tick_ref["eqQty"] if tick_ref["segment"] == "EQ" else tick_ref["foQty"], # ... p_refQty + tick["exchangeToken"] # ...................................................... p_exchange_code + ) + db_json, db_exception = await sql_reader.call_procedure_and_get_json( + procedure_name = proc_name, + procedure_args = proc_args, + return_exception = True, + retry_count = 3, + backoff_seconds = 0.5, + backoff_multiplier = 1.1 + ) + + # Alert if the portfolio adding fails: + if db_json["status"] != 1 or db_exception: + await send_telegram( + message = ( + "*Strategies (0) SQL Procedure Failure*\n\n" + f"Procedure Name: `{proc_name}`\n\n" + f"Procedure Args: `{proc_args}`\n\n" + f"Message: `{db_json['message']}`\n\n" + f"Exception: `{db_exception}`\n\n" + ), + message_type = "error" + ) + + # Done here: + return True + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def call_is_active( + tick_ref: dict, + tick: dict +) -> bool: + + global ACTIVE_CALLS + + # get the keys: + tick_key = tick_ref["tickKey"] + redis_key = tick_ref["redisKey"] + + if ACTIVE_CALLS.get(redis_key): return True + if await redis_cache.get(redis_key): + ACTIVE_CALLS[redis_key] = tick_ref + return True + return False + + +# --------------------------------------------------------------------------------------------------------------------- + + +def key_from_tick(tick: dict) -> str: + + if tick["broker"] == "zerodhaKite": + + is_eq = True if tick["type"] == "EQ" else False + + exchange = { + "NFO": "NSE", + }.get(tick["exchange"], tick["exchange"]) + + segment = { + "NFO-FUT": "FUT", + "BFO-FUT": "FUT", + "MCX-FUT": "FUT", + }.get(tick["segment"], tick["segment"]) + + symbol = tick["symbol"] + name = tick["name"] + + expiry = tick["expiry"] + + if is_eq: key_items = [exchange, tick["type"], symbol] + else: key_items = [exchange, segment, name, expiry] + return "_".join(key_items) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def get_reference_for_tick(tick_key: str, tick: dict) -> dict | None: + + tick_reference = None + + async with strategy_reference_lock: + for ref in STRATEGY_REFERENCE: + if ref["tickKey"] == tick_key: + tick_reference = ref + break + + return tick_reference + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def test_one_tick(tick: dict) -> None: + + # Get the reference for this ticker: + tick_key = key_from_tick(tick) + tick_reference = await get_reference_for_tick(tick_key, tick) + if not tick_reference: return + redis_key = tick_reference["redisKey"] + + # The call must NOT be active already: + # print(f"FOUND REF ({tick_key} | {tick['ltp']})", json.to_string(tick_reference, default = str)) + if await call_is_active( + tick_ref = tick_reference, + tick = tick + ): return + + # Check if the price is favourable: + if ( + ( + tick_reference["entry"] == "BUY" and + tick["ltp"] <= tick_reference["rate"] + ) or + ( + tick_reference["entry"] == "SELL" and + tick["ltp"] >= tick_reference["rate"] + ) + ): + mark_success = await mark_call_as_active( + tick_ref = tick_reference, + tick = tick + ) + if not mark_success: await send_telegram( + message = ( + "*Strategies (0) Alert*\n\n" + "Message: `Failed to mark active call.`\n\n" + f"Tick Key; `{tick_key}`\n" + f"Redis Key; `{redis_key}`\n" + f"Entry: `{tick_reference['entry']}`\n\n" + f"Rate: `{tick_reference['rate']}`\n\n" + f"LTP: `{tick['ltp']}`\n\n" + ), + message_type = "error" + ) + else: await send_telegram( + message = ( + "*Strategies (0) Alert*\n\n" + "Message: `Marked active call.`\n\n" + f"Tick Key; `{tick_key}`\n" + f"Redis Key; `{redis_key}`\n" + f"Entry: `{tick_reference['entry']}`\n\n" + f"Rate: `{tick_reference['rate']}`\n\n" + f"LTP: `{tick['ltp']}`\n\n" + ), + message_type = "info" + ) + + +# --------------------------------------------------------------------------------------------------------------------- + + async def test_strategies(ticks: List[dict]): # declare the needed global variables: global ticks_since_telegram # Test needed strategies here: - pass + # print("TICK 0:", json.to_string(ticks[0])) + tasks = [test_one_tick(t) for t in ticks] + results = await asyncio.gather(*tasks) # Send out the alert if needed: ticks_since_telegram += len(ticks) @@ -190,7 +383,7 @@ async def ticks_from_kafka( :return: None """ - printer("Starting Kafka consumer (ticks).") + no_context_printer("Starting Kafka consumer (ticks).") # Do the next part infinitely: while True: @@ -235,13 +428,14 @@ async def init( # Declare the required global variables: global SCRIPT_DATA + global redis_cache global sql_writer global sql_reader global kafka_consumer # Basic stuff: if debug: printer.enable() - printer("Initializing.") + no_context_printer("Initializing.") # ┏┓ ┓ ┓ ┳┓ # ┃ ┏┓┏┓┏┫ ┏┓┏┓┏┫ ┃┃┏┓╋┏┓ @@ -268,7 +462,7 @@ async def init( SCRIPT_DATA = response.json().get("data") # Done with this step: - printer("Cred and Data loaded.") + no_context_printer("Cred and Data loaded.") # ┳┳┓ • ┳┓┳┓ # ┃┃┃┏┓┏┓┓┏┓┃┃┣┫ @@ -297,7 +491,7 @@ async def init( print("FATAL: MARIA-DB READER CONNECTION FAILED!") return False - printer("MariaDB connected.") + no_context_printer("MariaDB connected.") # ┓┏┓ ┏┓ ┏┓┓• # ┃┫ ┏┓╋┃┏┏┓ ┃ ┃┓┏┓┏┓╋┏ @@ -306,8 +500,9 @@ async def init( # Create the consumer that will listen to changes in watchlist: consumer_creds = script_cred["kafka"]["consumer"] kafka_consumer = ConsumerKafka( - topic = consumer_creds["topic"], - group_id = "tick_save", + # topic = consumer_creds["topic"], + topic = "tickers", + group_id = "tick_strategy", bootstrap_servers = consumer_creds["config"]["bootstrapServers"], security_protocol = consumer_creds["config"].get("securityProtocol", "PLAINTEXT"), ssl_context = get_ssl_context( @@ -321,14 +516,29 @@ async def init( if not await kafka_consumer.connect(): print("FATAL: KAFKA CONSUMER NOT CREATED!") return False - printer("Kafka consumer ready.") + no_context_printer("Kafka consumer ready.") + + # ┳┓ ┓• ┏┓ ┓ + # ┣┫┏┓┏┫┓┏ ━━ ┃ ┏┓┏┣┓┏┓ + # ┛┗┗ ┗┻┗┛ ┗┛┗┻┗┛┗┗ + + # Caching connections: + redis_cache = AsyncRedisCache( + connection_string = script_cred["redisCache"]["general"]["connectionString"], + debug = False, + debug_prefix = "Cache | " + ) + if not await redis_cache.connect(): + print("FATAL: REDIS CACHE NOT CREATED!") + return False + no_context_printer("Redis cache ready.") # ┳┓ # ┃┃┏┓┏┓┏┓ # ┻┛┗┛┛┗┗ # If everything went well, we return with success: - printer("Initialization done.") + no_context_printer("Initialization done.") return True @@ -410,25 +620,42 @@ async def refresh_strategy_reference() -> bool: return False # Construct the reference structure: - formatted_reference = {} + formatted_reference = [] for i in db_json["data"]["rs0"]: key_items = [i["exchange"], i["segment"], i["exchange_code"]] if i["expiry_date"]: key_items.append(str(i["expiry_date"])) if i["right"]: key_items.append(str(i["right"])) key_items.append(str(i["strike"])) - formatted_reference["_".join(key_items)] = { - "stratName": i["name"], # ................ As inserted by the user. - "watchlistId": i["watch_list_id"], # ..... A unique id in case 2 users come up with the same name. - "entry": i["entry"].upper().strip(), # ... BUY/SELL. - "rate": i["entry_rate"], # ............... The price at which to enter the trade. - "t": i["target1"], # ..................... The 1st target. - "sl": i["stop_loss"] # ................... The stop loss value. - } + tick_key = "_".join(key_items) + strategy_name = i["name"] + strategy_entry = i["entry"].upper().strip() + formatted_reference.append({ + "userId": i["user_id"], + "billingAccountId": i["billing_account_id"], + "eqQty": i["eq_qty"], + "foQty": i["fno_qty"], + "tickKey": tick_key, + "redisKey": f"{tick_key}_{strategy_name}_{strategy_entry}", + "stratId": i["strategy_id"], # .......... As inserted by the user. + "stratName": strategy_name, # .......... As inserted by the user. + "watchlistId": i["watch_list_id"], # ... A unique id in case 2 users come up with the same name. + "entry": strategy_entry, # ............. BUY/SELL. + "rate": i["entry_rate"], # ............. The price at which to enter the trade. + "t": i["target1"], # ................... The 1st target. + "sl": i["stop_loss"], # ................. The stop loss value. + "exchange": i["exchange"], + "segment": i["segment"], + "expiry": i["expiry_date"], + "strike": i["strike"], + "right": i["right"], + "symbol": i["symbol"] + }) # Save the new reference in the global variable: async with strategy_reference_lock: STRATEGY_REFERENCE = formatted_reference + print(json.to_string(list(STRATEGY_REFERENCE))) # Done here: return True @@ -461,8 +688,11 @@ async def main( debug: bool = False ): + # Get the reference before the scheduler kicks in: + await refresh_strategy_reference() + # Start configuring the scheduler: - printer("Configuring the schedule-manager.") + no_context_printer("Configuring the schedule-manager.") schedule_manager = Scheduler() # Create all the timestamps at which the job must be done: @@ -476,15 +706,15 @@ async def main( # Add the jobs: for ts in all_job_ts: schedule_manager.daily(ts, refresh_strategy_reference) - printer(len(all_job_ts)) + no_context_printer(len(all_job_ts)) # Run the heartbeat task and the infinite tick-reading loop: tasks = [ - ticks_from_kafka( - consumer = kafka_consumer, - fetch_count = 500, - fetch_timeout = 2.5 - ), + # ticks_from_kafka( + # consumer = kafka_consumer, + # fetch_count = 500, + # fetch_timeout = 2.5 + # ), heartbeat(interval_seconds = heartbeat_interval) ] await asyncio.gather(*tasks) @@ -499,7 +729,7 @@ async def main( if __name__ == "__main__": - printer("Main.") + no_context_printer("Main.") # To get args from the terminal: import argparse