uniswap_auto_clp/doc/LOW_LATENCY_OPTIMIZATION_PLAN.md

Low Latency Optimization Plan: Memory Sharing Integration

Overview

Currently, the system consists of two separate processes (uniswap_manager_refactored.py and clp_hedger.py) communicating via a file (hedge_status.json). This introduces inevitable latency due to:

1. Polling Intervals: The hedger must "sleep" and "wake up" to check the file.
2. File I/O: Reading/writing to disk is thousands of times slower than memory operations.
3. Synchronization: Potential race conditions if both try to access the file simultaneously.

Goal

Eliminate file reliance to achieve sub-millisecond reaction times between "Uniswap Position Out of Range" detection and "Hedge Close" execution.

Proposed Architecture: Unified Multi-Threaded Bot

Instead of two independent scripts, we will merge them into a single Python application running two concurrent threads that share a common data object in memory.

Key Components

1. SharedState Class (The Brain)

   • A thread-safe data structure (using threading.Lock) that holds the current position status, price, and range.
   • Events: Uses threading.Event (e.g., close_signal) to allow the Manager to instantly wake up the Hedger without waiting for a sleep cycle to finish.

2. UniswapManager Thread

   • Role: Monitors on-chain data (RPC).
   • Action: When it detects "Out of Range", it updates SharedState and sets close_signal.set().

3. ClpHedger Thread

   • Role: Manages the Hyperliquid hedge.
   • Action: Instead of time.sleep(1), it waits on close_signal.wait(timeout=1).
   • Reaction: If close_signal is triggered, it executes the close logic immediately (0 latency).

4. main_bot.py (The Entry Point)

   • Initializes SharedState.
   • Starts UniswapManager and ClpHedger as threads.
   • Handles centralized logging and clean shutdown.

Implementation Steps

Step 1: Create SharedState

Define a class that replaces the JSON file structure.

class SharedState:
    def __init__(self):
        self.lock = threading.Lock()
        self.close_event = threading.Event()
        self.position_data = {} # Stores the dict formerly in JSON
    
    def update_position(self, data):
        with self.lock:
            self.position_data.update(data)
            
    def get_position(self):
        with self.lock:
            return self.position_data.copy()
            
    def trigger_emergency_close(self):
        self.close_event.set()

Step 2: Refactor uniswap_manager_refactored.py

• Convert the script into a class UniswapManager.
• Replace all load_status_data() and save_status_data() calls with self.shared_state.update_position(...).
• When "Out of Range" is detected:

  # Old
  update_position_status(token_id, "CLOSING")
  
  # New
  self.shared_state.update_position({'status': 'CLOSING'})
  self.shared_state.trigger_emergency_close() # Wakes up Hedger instantly
  

Step 3: Refactor clp_hedger.py

• Convert the script into a class ClpHedger.
• Replace file reading logic with self.shared_state.get_position().
• Update the main loop to handle the event:

  # Old
  time.sleep(CHECK_INTERVAL)
  
  # New
  # Wait for 1 second OR immediate signal
  if self.shared_state.close_event.wait(timeout=1.0):
      self.close_all_positions()
      self.shared_state.close_event.clear()
  

Step 4: Create main_bot.py

if __name__ == "__main__":
    state = SharedState()
    
    manager = UniswapManager(state)
    hedger = ClpHedger(state)
    
    t1 = threading.Thread(target=manager.run)
    t2 = threading.Thread(target=hedger.run)
    
    t1.start()
    t2.start()
    
    t1.join()
    t2.join()

Benefits

1. Zero Latency: The moment the Manager sets the event, the Hedger reacts. No polling delay.
2. Reliability: No file corruption risks (like the JSON error experienced earlier).
3. Efficiency: Reduces disk I/O, extending SD card/drive life and reducing CPU usage.