hyper/clp_auto_hedger/velocity_sqrt_fix.py

#!/usr/bin/env python3
"""
Fix for velocity calculation sqrt domain error in CLP Scalper Hedger

The error occurs in the liquidity velocity calculation when trying to compute:
sqrt((new_increased_liquidity ** 2) - (4 * net_cash_proceeds))

This happens when (new_increased_liquidity ** 2) < (4 * net_cash_proceeds), 
making the discriminant negative.

This fix provides defensive programming patterns to handle such cases.
"""

import math
import logging

def safe_sqrt_with_fallback(value: float, fallback_value: float = 0.0, context: str = "sqrt calculation") -> float:
    """
    Safely compute square root with fallback for negative values
    
    Args:
        value: The value to compute square root of
        fallback_value: Value to return if input is negative
        context: Context description for logging
    
    Returns:
        Square root of value if positive, fallback_value if negative
    """
    if value >= 0:
        return math.sqrt(value)
    else:
        logging.warning(
            f"Negative value in {context}: {value:.6f}. "
            f"Using fallback value: {fallback_value:.6f}"
        )
        return fallback_value

def calculate_liquidity_velocity_safe(
    current_liquidity: float, 
    new_increased_liquidity: float, 
    net_cash_proceeds: float,
    current_tick: int,
    lower_tick: int,
    upper_tick: int
) -> tuple[float, float]:
    """
    Safe calculation of liquidity velocity with proper error handling
    
    Args:
        current_liquidity: Current liquidity amount
        new_increased_liquidity: New increased liquidity amount
        net_cash_proceeds: Net cash proceeds from liquidity change
        current_tick: Current price tick
        lower_tick: Lower tick boundary
        upper_tick: Upper tick boundary
    
    Returns:
        Tuple of (velocity, price_impact)
    """
    try:
        # Basic velocity calculation
        velocity = new_increased_liquidity - current_liquidity
        price_impact = 0.0
        
        # Inside position range - use square root formula
        if lower_tick <= current_tick <= upper_tick:
            if net_cash_proceeds >= 0:
                # Validate discriminant to prevent sqrt of negative number
                discriminant = (new_increased_liquidity ** 2) - (4 * net_cash_proceeds)
                
                if discriminant >= 0:
                    # Safe calculation
                    sqrt_term = math.sqrt(discriminant)
                    denominator = 2 * max(current_liquidity, 1e-10)  # Prevent division by zero
                    price_impact = (new_increased_liquidity - sqrt_term) / denominator
                else:
                    # Edge case: negative discriminant
                    # This can happen due to:
                    # 1. Floating point precision errors
                    # 2. Extreme market conditions 
                    # 3. Invalid input parameters
                    
                    logging.warning(
                        f"Negative discriminant in liquidity velocity: {discriminant:.6f}. "
                        f"Liquidity: {current_liquidity:.6f} -> {new_increased_liquidity:.6f}, "
                        f"Cash: {net_cash_proceeds:.6f}. Using zero price impact."
                    )
                    
                    # Use approximation methods
                    price_impact = 0.0
                    
                    # Alternative: Use small positive approximation
                    # discriminant = max(discriminant, 0)
                    # sqrt_term = math.sqrt(discriminant)
                    # price_impact = (new_increased_liquidity - sqrt_term) / (2 * current_liquidity)
                    
            else:
                # Negative cash flow means additional capital required
                # No price impact calculation needed
                price_impact = 0.0
        
        return velocity, price_impact
        
    except Exception as e:
        logging.error(f"Error in liquidity velocity calculation: {e}")
        # Return safe defaults
        return 0.0, 0.0

def validate_liquidity_inputs(
    current_liquidity: float,
    new_increased_liquidity: float, 
    net_cash_proceeds: float
) -> bool:
    """
    Validate inputs for liquidity velocity calculation
    
    Args:
        current_liquidity: Current liquidity amount
        new_increased_liquidity: New increased liquidity amount
        net_cash_proceeds: Net cash proceeds from liquidity change
    
    Returns:
        True if inputs are valid, False otherwise
    """
    # Check for NaN or infinite values
    if any(math.isnan(x) or math.isinf(x) for x in [current_liquidity, new_increased_liquidity, net_cash_proceeds]):
        logging.error("Invalid inputs: NaN or infinite values detected")
        return False
    
    # Check for negative liquidity (should be non-negative)
    if current_liquidity < 0 or new_increased_liquidity < 0:
        logging.error(f"Invalid liquidity values: current={current_liquidity}, new={new_increased_liquidity}")
        return False
    
    # Check for reasonable ranges (adjust based on your specific needs)
    max_liquidity = 1e20  # Very large number for safety
    if current_liquidity > max_liquidity or new_increased_liquidity > max_liquidity:
        logging.error(f"Liquidity values too large: current={current_liquidity}, new={new_increased_liquidity}")
        return False
    
    return True

# Example usage and test cases
def test_liquidity_velocity_calculation():
    """Test the safe liquidity velocity calculation with various scenarios"""
    
    test_cases = [
        # Normal case
        {
            "name": "Normal case",
            "current_liquidity": 1000.0,
            "new_increased_liquidity": 1200.0,
            "net_cash_proceeds": 100.0,
            "current_tick": 200000,
            "lower_tick": 195000,
            "upper_tick": 205000
        },
        
        # Edge case: negative discriminant
        {
            "name": "Negative discriminant",
            "current_liquidity": 100.0,
            "new_increased_liquidity": 100.0,
            "net_cash_proceeds": 3000.0,  # This will cause negative discriminant
            "current_tick": 200000,
            "lower_tick": 195000,
            "upper_tick": 205000
        },
        
        # Edge case: very small liquidity
        {
            "name": "Small liquidity",
            "current_liquidity": 1e-10,
            "new_increased_liquidity": 2e-10,
            "net_cash_proceeds": 0.0,
            "current_tick": 200000,
            "lower_tick": 195000,
            "upper_tick": 205000
        }
    ]
    
    print("Testing Liquidity Velocity Calculation")
    print("=" * 50)
    
    for case in test_cases:
        print(f"\nTest: {case['name']}")
        print(f"Inputs: {case}")
        
        # Validate inputs
        if validate_liquidity_inputs(
            case["current_liquidity"], 
            case["new_increased_liquidity"],
            case["net_cash_proceeds"]
        ):
            # Calculate safely
            velocity, price_impact = calculate_liquidity_velocity_safe(
                case["current_liquidity"],
                case["new_increased_liquidity"], 
                case["net_cash_proceeds"],
                case["current_tick"],
                case["lower_tick"],
                case["upper_tick"]
            )
            
            print(f"Results: velocity={velocity:.6f}, price_impact={price_impact:.6f}")
        else:
            print("Results: Invalid inputs - calculation skipped")

if __name__ == "__main__":
    # Set up logging
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s'
    )
    
    # Run tests
    test_liquidity_velocity_calculation()
    
    print("\n" + "=" * 50)
    print("Integration Instructions:")
    print("1. Replace the problematic sqrt calculation with calculate_liquidity_velocity_safe()")
    print("2. Add input validation using validate_liquidity_inputs()")
    print("3. Use safe_sqrt_with_fallback() for any other sqrt operations")
    print("4. Add proper logging to track edge cases and errors")