diff --git a/bittensor_cli/cli.py b/bittensor_cli/cli.py index 16e68c254..5b4591813 100755 --- a/bittensor_cli/cli.py +++ b/bittensor_cli/cli.py @@ -29,7 +29,7 @@ is_valid_ss58_address as btwallet_is_valid_ss58_address, ) from rich import box -from rich.prompt import FloatPrompt, Prompt, IntPrompt +from rich.prompt import Confirm, FloatPrompt, Prompt, IntPrompt from rich.table import Column, Table from rich.tree import Tree from typing_extensions import Annotated @@ -8424,6 +8424,18 @@ def liquidity_add( "--liquidity_price_high", help="High price for the adding liquidity position.", ), + amount_tao: Optional[float] = typer.Option( + None, + "--amount-tao", + "--amount_tao", + help="Amount of TAO to provide (when only TAO is needed).", + ), + amount_alpha: Optional[float] = typer.Option( + None, + "--amount-alpha", + "--amount_alpha", + help="Amount of Alpha to provide (when only Alpha is needed).", + ), prompt: bool = Options.prompt, decline: bool = Options.decline, quiet: bool = Options.quiet, @@ -8433,60 +8445,285 @@ def liquidity_add( """Add liquidity to the swap (as a combination of TAO + Alpha).""" self.verbosity_handler(quiet, verbose, json_output, prompt, decline) proxy = self.is_valid_proxy_name_or_ss58(proxy, announce_only) + + # Step 1: Ask for netuid if not netuid: - netuid = Prompt.ask( - f"Enter the [{COLORS.G.SUBHEAD_MAIN}]netuid[/{COLORS.G.SUBHEAD_MAIN}] to use", - default=None, - show_default=False, + netuid = IntPrompt.ask( + f"Enter the [{COLORS.G.SUBHEAD_MAIN}]netuid[/{COLORS.G.SUBHEAD_MAIN}] to use" ) - wallet, hotkey = self.wallet_ask( - wallet_name=wallet_name, - wallet_path=wallet_path, - wallet_hotkey=wallet_hotkey, - ask_for=[WO.NAME, WO.HOTKEY, WO.PATH], - validate=WV.WALLET, - return_wallet_and_hotkey=True, - ) - # Determine the liquidity amount. - if liquidity_: - liquidity_ = Balance.from_tao(liquidity_) - else: - liquidity_ = prompt_liquidity("Enter the amount of liquidity") + # Step 2: Check if the subnet exists (early validation) + subtensor = self.initialize_chain(network) + + async def check_subnet(): + if not await subtensor.subnet_exists(netuid=netuid): + return ( + False, + f"Subnet with netuid: {netuid} does not exist in {subtensor}.", + ) + return True, None + + success, err_msg = self._run_command(check_subnet()) + if not success: + print_error(err_msg) + return False - # Determine price range + # Step 3: Ask user to enter low and high position prices if price_low: price_low = Balance.from_tao(price_low) else: - price_low = prompt_liquidity("Enter liquidity position low price") + price_low = prompt_liquidity( + "Enter liquidity position [blue]low price[/blue]" + ) if price_high: price_high = Balance.from_tao(price_high) else: price_high = prompt_liquidity( - "Enter liquidity position high price (must be greater than low price)" + "Enter liquidity position [blue]high price[/blue] (must be greater than low price)" ) if price_low >= price_high: print_error("The low price must be lower than the high price.") return False + + # Step 4: Fetch current subnet price + async def get_current_price(): + return await subtensor.get_subnet_price(netuid=netuid) + + current_price = self._run_command(get_current_price()) + console.print( + f"\n[cyan]Current subnet price:[/cyan] [green]{current_price.tao:.6f}[/green] τ/α\n" + ) + + # Step 5: Conditional logic based on price comparison + wallet = None + hotkey = None + liquidity_to_provide = None + + if price_low.tao >= current_price.tao: + # Only Alpha is needed + console.print( + "[yellow]Based on your price range, only [bold]Alpha[/bold] tokens are needed.[/yellow]\n" + ) + + # Ask for hotkey (optional but needed for taking stake) + if wallet_hotkey: + wallet, hotkey = self.wallet_ask( + wallet_name=wallet_name, + wallet_path=wallet_path, + wallet_hotkey=wallet_hotkey, + ask_for=[WO.NAME, WO.HOTKEY, WO.PATH], + validate=WV.WALLET, + return_wallet_and_hotkey=True, + ) + else: + # Prompt for optional hotkey + use_hotkey = Confirm.ask( + "Do you want to specify a hotkey to take Alpha stake from?", + default=False, + ) + if use_hotkey: + wallet, hotkey = self.wallet_ask( + wallet_name=wallet_name, + wallet_path=wallet_path, + wallet_hotkey=None, + ask_for=[WO.NAME, WO.HOTKEY, WO.PATH], + validate=WV.WALLET, + return_wallet_and_hotkey=True, + ) + else: + wallet = self.wallet_ask( + wallet_name=wallet_name, + wallet_path=wallet_path, + wallet_hotkey=wallet_hotkey, + ask_for=[WO.NAME, WO.PATH], + validate=WV.WALLET, + ) + hotkey = wallet.hotkey.ss58_address + + # Ask for Alpha amount + if amount_alpha: + amount_alpha_balance = Balance.from_tao(amount_alpha).set_unit(netuid) + else: + amount_alpha_balance = prompt_liquidity( + f"Enter the amount of [blue]Alpha[/blue] to provide" + ).set_unit(netuid) + + # Calculate liquidity from Alpha amount + from bittensor_cli.src.commands.liquidity.utils import ( + calculate_max_liquidity_from_amounts, + ) + + liquidity_to_provide = calculate_max_liquidity_from_amounts( + amount_tao=Balance.from_tao(0), + amount_alpha=amount_alpha_balance, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + elif price_high.tao <= current_price.tao: + # Only TAO is needed + console.print( + "[yellow]Based on your price range, only [bold]TAO[/bold] tokens are needed.[/yellow]\n" + ) + + # Ask for hotkey (optional, has no effect for TAO-only) + if wallet_hotkey: + wallet, hotkey = self.wallet_ask( + wallet_name=wallet_name, + wallet_path=wallet_path, + wallet_hotkey=wallet_hotkey, + ask_for=[WO.NAME, WO.HOTKEY, WO.PATH], + validate=WV.WALLET, + return_wallet_and_hotkey=True, + ) + else: + # Prompt for optional hotkey + use_hotkey = Confirm.ask( + "Do you want to specify a hotkey? (Note: hotkey has no effect for TAO-only liquidity)", + default=False, + ) + if use_hotkey: + wallet, hotkey = self.wallet_ask( + wallet_name=wallet_name, + wallet_path=wallet_path, + wallet_hotkey=None, + ask_for=[WO.NAME, WO.HOTKEY, WO.PATH], + validate=WV.WALLET, + return_wallet_and_hotkey=True, + ) + else: + wallet = self.wallet_ask( + wallet_name=wallet_name, + wallet_path=wallet_path, + wallet_hotkey=wallet_hotkey, + ask_for=[WO.NAME, WO.PATH], + validate=WV.WALLET, + ) + hotkey = wallet.hotkey.ss58_address + + # Ask for TAO amount + if amount_tao: + amount_tao_balance = Balance.from_tao(amount_tao) + else: + amount_tao_balance = prompt_liquidity( + f"Enter the amount of [blue]TAO[/blue] to provide" + ) + + # Calculate liquidity from TAO amount + from bittensor_cli.src.commands.liquidity.utils import ( + calculate_max_liquidity_from_amounts, + ) + + liquidity_to_provide = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao_balance, + amount_alpha=Balance.from_tao(0).set_unit(netuid), + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + else: + # Both TAO and Alpha are needed + console.print( + "[yellow]Based on your price range, both [bold]TAO and Alpha[/bold] tokens are needed.[/yellow]\n" + ) + + # Ask for wallet and hotkey + wallet, hotkey = self.wallet_ask( + wallet_name=wallet_name, + wallet_path=wallet_path, + wallet_hotkey=wallet_hotkey, + ask_for=[WO.NAME, WO.HOTKEY, WO.PATH], + validate=WV.WALLET, + return_wallet_and_hotkey=True, + ) + + # Fetch TAO and Alpha balances + async def fetch_balances(): + tao_balance, alpha_balance = await asyncio.gather( + subtensor.get_balance(wallet.coldkeypub.ss58_address), + subtensor.get_stake_for_coldkey_and_hotkey( + hotkey_ss58=hotkey, + coldkey_ss58=wallet.coldkeypub.ss58_address, + netuid=netuid, + ), + ) + return tao_balance, alpha_balance + + tao_balance, alpha_balance = self._run_command(fetch_balances()) + + # Calculate maximum liquidity that can be provided + from bittensor_cli.src.commands.liquidity.utils import ( + calculate_max_liquidity_from_amounts, + calculate_token_amounts_from_liquidity, + ) + + max_liquidity = calculate_max_liquidity_from_amounts( + amount_tao=tao_balance, + amount_alpha=alpha_balance, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Calculate the token amounts for max liquidity + max_alpha, max_tao = calculate_token_amounts_from_liquidity( + liquidity=max_liquidity, + current_price=current_price, + price_low=price_low, + price_high=price_high, + netuid=netuid, + ) + + # Display max amounts + console.print( + f"[cyan]Your available balances:[/cyan]\n" + f" TAO: [green]{tao_balance.tao:.6f}[/green] τ\n" + f" Alpha: [green]{alpha_balance.tao:.6f}[/green] α\n" + ) + console.print( + f"[cyan]Maximum liquidity that can be provided:[/cyan]\n" + f" Liquidity: [green]{max_liquidity.tao:.6f}[/green]\n" + f" Requires TAO: [green]{max_tao.tao:.6f}[/green] τ\n" + f" Requires Alpha: [green]{max_alpha.tao:.6f}[/green] α\n" + ) + + # Ask user to enter amount + if liquidity_: + liquidity_to_provide = Balance.from_tao(liquidity_) + else: + liquidity_to_provide = prompt_liquidity( + "Enter the amount of [blue]liquidity[/blue] to provide" + ) + + # Validate that user doesn't exceed max + if liquidity_to_provide.tao > max_liquidity.tao: + print_error( + f"Requested liquidity ({liquidity_to_provide.tao:.6f}) exceeds maximum available ({max_liquidity.tao:.6f})." + ) + return False + + # Step 6: Execute the extrinsic logger.debug( f"args:\n" f"hotkey: {hotkey}\n" f"netuid: {netuid}\n" - f"liquidity: {liquidity_}\n" + f"liquidity: {liquidity_to_provide}\n" f"price_low: {price_low}\n" f"price_high: {price_high}\n" f"proxy: {proxy}\n" ) return self._run_command( liquidity.add_liquidity( - subtensor=self.initialize_chain(network), + subtensor=subtensor, wallet=wallet, hotkey_ss58=hotkey, netuid=netuid, proxy=proxy, - liquidity=liquidity_, + liquidity=liquidity_to_provide, price_low=price_low, price_high=price_high, prompt=prompt, diff --git a/bittensor_cli/src/commands/liquidity/utils.py b/bittensor_cli/src/commands/liquidity/utils.py index f364a64e4..197d3f3ea 100644 --- a/bittensor_cli/src/commands/liquidity/utils.py +++ b/bittensor_cli/src/commands/liquidity/utils.py @@ -200,3 +200,100 @@ def prompt_position_id() -> int: console.print("[red]Please enter a valid number[/red].") # will never return this, but fixes the type checker return 0 + + +def calculate_max_liquidity_from_amounts( + amount_tao: Balance, + amount_alpha: Balance, + current_price: Balance, + price_low: Balance, + price_high: Balance, +) -> Balance: + """Calculate the maximum liquidity that can be provided given token amounts. + + Arguments: + amount_tao: Available TAO balance. + amount_alpha: Available Alpha balance. + current_price: Current subnet price. + price_low: Lower bound of the price range. + price_high: Upper bound of the price range. + + Returns: + Maximum liquidity that can be provided as Balance. + + This function reverses the liquidity-to-amounts formula from Uniswap V3: + - If current price < low price: Only Alpha is needed + - If current price > high price: Only TAO is needed + - Otherwise: Both TAO and Alpha are needed, limited by whichever runs out first + """ + sqrt_price_low = math.sqrt(price_low.tao) + sqrt_price_high = math.sqrt(price_high.tao) + sqrt_current_price = math.sqrt(current_price.tao) + + if sqrt_current_price < sqrt_price_low: + # Only Alpha is needed + denom = 1 / sqrt_price_low - 1 / sqrt_price_high + if denom == 0: + return Balance.from_tao(0) + liquidity = amount_alpha.rao / denom + elif sqrt_current_price > sqrt_price_high: + # Only TAO is needed + denom = sqrt_price_high - sqrt_price_low + if denom == 0: + return Balance.from_tao(0) + liquidity = amount_tao.rao / denom + else: + # Both TAO and Alpha are needed + # Calculate liquidity from each and take the minimum + alpha_denom = 1 / sqrt_current_price - 1 / sqrt_price_high + tao_denom = sqrt_current_price - sqrt_price_low + + if alpha_denom == 0 or tao_denom == 0: + return Balance.from_tao(0) + + liquidity_from_alpha = amount_alpha.rao / alpha_denom + liquidity_from_tao = amount_tao.rao / tao_denom + liquidity = min(liquidity_from_alpha, liquidity_from_tao) + + return Balance.from_rao(int(liquidity)) + + +def calculate_token_amounts_from_liquidity( + liquidity: Balance, + current_price: Balance, + price_low: Balance, + price_high: Balance, + netuid: int, +) -> tuple[Balance, Balance]: + """Calculate token amounts needed for a given liquidity. + + Arguments: + liquidity: The amount of liquidity to provide. + current_price: Current subnet price. + price_low: Lower bound of the price range. + price_high: Upper bound of the price range. + netuid: The subnet ID (for setting Alpha unit). + + Returns: + tuple[Balance, Balance]: (amount_alpha, amount_tao) needed for the liquidity. + """ + sqrt_price_low = math.sqrt(price_low.tao) + sqrt_price_high = math.sqrt(price_high.tao) + sqrt_current_price = math.sqrt(current_price.tao) + + if sqrt_current_price < sqrt_price_low: + # Only Alpha is needed + amount_alpha = liquidity.rao * (1 / sqrt_price_low - 1 / sqrt_price_high) + amount_tao = 0 + elif sqrt_current_price > sqrt_price_high: + # Only TAO is needed + amount_alpha = 0 + amount_tao = liquidity.rao * (sqrt_price_high - sqrt_price_low) + else: + # Both TAO and Alpha are needed + amount_alpha = liquidity.rao * (1 / sqrt_current_price - 1 / sqrt_price_high) + amount_tao = liquidity.rao * (sqrt_current_price - sqrt_price_low) + + return Balance.from_rao(int(amount_alpha)).set_unit(netuid), Balance.from_rao( + int(amount_tao) + ) diff --git a/tests/unit_tests/test_liquidity_utils.py b/tests/unit_tests/test_liquidity_utils.py new file mode 100644 index 000000000..3b3f12d64 --- /dev/null +++ b/tests/unit_tests/test_liquidity_utils.py @@ -0,0 +1,488 @@ +"""Unit tests for liquidity utility functions.""" + +import pytest + +from bittensor_cli.src.bittensor.balances import Balance +from bittensor_cli.src.commands.liquidity.utils import ( + calculate_max_liquidity_from_amounts, + calculate_token_amounts_from_liquidity, + price_to_tick, + tick_to_price, +) + + +class TestCalculateMaxLiquidityFromAmounts: + """Tests for calculate_max_liquidity_from_amounts function.""" + + def test_only_alpha_needed_below_range(self): + """Test when current price is below price range (only Alpha needed).""" + amount_tao = Balance.from_tao(10.0) + amount_alpha = Balance.from_tao(100.0).set_unit(1) + current_price = Balance.from_tao(0.5) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + + result = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Should only use Alpha + assert result.rao > 0 + # TAO balance should not affect the result + result_with_less_tao = calculate_max_liquidity_from_amounts( + amount_tao=Balance.from_tao(1.0), + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + assert result.rao == result_with_less_tao.rao + + def test_only_tao_needed_above_range(self): + """Test when current price is above price range (only TAO needed).""" + amount_tao = Balance.from_tao(10.0) + amount_alpha = Balance.from_tao(100.0).set_unit(1) + current_price = Balance.from_tao(3.0) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + + result = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Should only use TAO + assert result.rao > 0 + # Alpha balance should not affect the result + result_with_less_alpha = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=Balance.from_tao(1.0).set_unit(1), + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + assert result.rao == result_with_less_alpha.rao + + def test_both_tokens_needed_in_range(self): + """Test when current price is within range (both tokens needed).""" + amount_tao = Balance.from_tao(10.0) + amount_alpha = Balance.from_tao(100.0).set_unit(1) + current_price = Balance.from_tao(1.5) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + + result = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Should use both tokens + assert result.rao > 0 + + # Reducing either balance should reduce liquidity + result_with_less_tao = calculate_max_liquidity_from_amounts( + amount_tao=Balance.from_tao(5.0), + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + result_with_less_alpha = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=Balance.from_tao(50.0).set_unit(1), + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # At least one should be smaller + assert ( + result_with_less_tao.rao < result.rao + or result_with_less_alpha.rao < result.rao + ) + + def test_zero_balances(self): + """Test with zero balances.""" + amount_tao = Balance.from_tao(0.0) + amount_alpha = Balance.from_tao(0.0).set_unit(1) + current_price = Balance.from_tao(1.5) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + + result = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + assert result.rao == 0 + + def test_equal_price_range(self): + """Test when price_low equals price_high (edge case).""" + amount_tao = Balance.from_tao(10.0) + amount_alpha = Balance.from_tao(100.0).set_unit(1) + current_price = Balance.from_tao(1.5) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(1.0) + + result = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Should return 0 due to invalid range + assert result.rao == 0 + + +class TestCalculateTokenAmountsFromLiquidity: + """Tests for calculate_token_amounts_from_liquidity function.""" + + def test_only_alpha_below_range(self): + """Test token amounts when current price is below range.""" + liquidity = Balance.from_tao(1000.0) + current_price = Balance.from_tao(0.5) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + netuid = 1 + + amount_alpha, amount_tao = calculate_token_amounts_from_liquidity( + liquidity=liquidity, + current_price=current_price, + price_low=price_low, + price_high=price_high, + netuid=netuid, + ) + + # Should only need Alpha + assert amount_alpha.rao > 0 + assert amount_tao.rao == 0 + # Alpha unit is set to the subnet symbol + assert amount_alpha.unit is not None + + def test_only_tao_above_range(self): + """Test token amounts when current price is above range.""" + liquidity = Balance.from_tao(1000.0) + current_price = Balance.from_tao(3.0) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + netuid = 1 + + amount_alpha, amount_tao = calculate_token_amounts_from_liquidity( + liquidity=liquidity, + current_price=current_price, + price_low=price_low, + price_high=price_high, + netuid=netuid, + ) + + # Should only need TAO + assert amount_alpha.rao == 0 + assert amount_tao.rao > 0 + + def test_both_tokens_in_range(self): + """Test token amounts when current price is within range.""" + liquidity = Balance.from_tao(1000.0) + current_price = Balance.from_tao(1.5) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + netuid = 1 + + amount_alpha, amount_tao = calculate_token_amounts_from_liquidity( + liquidity=liquidity, + current_price=current_price, + price_low=price_low, + price_high=price_high, + netuid=netuid, + ) + + # Should need both tokens + assert amount_alpha.rao > 0 + assert amount_tao.rao > 0 + # Alpha unit is set to the subnet symbol + assert amount_alpha.unit is not None + + def test_zero_liquidity(self): + """Test with zero liquidity.""" + liquidity = Balance.from_tao(0.0) + current_price = Balance.from_tao(1.5) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + netuid = 1 + + amount_alpha, amount_tao = calculate_token_amounts_from_liquidity( + liquidity=liquidity, + current_price=current_price, + price_low=price_low, + price_high=price_high, + netuid=netuid, + ) + + assert amount_alpha.rao == 0 + assert amount_tao.rao == 0 + + def test_at_price_boundaries(self): + """Test when current price equals price_low or price_high.""" + liquidity = Balance.from_tao(1000.0) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + netuid = 1 + + # Current price at lower bound + amount_alpha_low, amount_tao_low = calculate_token_amounts_from_liquidity( + liquidity=liquidity, + current_price=price_low, + price_low=price_low, + price_high=price_high, + netuid=netuid, + ) + + # Current price at upper bound + amount_alpha_high, amount_tao_high = calculate_token_amounts_from_liquidity( + liquidity=liquidity, + current_price=price_high, + price_low=price_low, + price_high=price_high, + netuid=netuid, + ) + + # At lower bound, should have more Alpha, less TAO + # At upper bound, should have less Alpha, more TAO + assert amount_alpha_low.rao >= amount_alpha_high.rao + assert amount_tao_low.rao <= amount_tao_high.rao + + +class TestRoundTripConsistency: + """Test that calculations are consistent when going back and forth.""" + + def test_roundtrip_alpha_only(self): + """Test roundtrip: amounts -> liquidity -> amounts (Alpha only).""" + amount_tao = Balance.from_tao(10.0) + amount_alpha = Balance.from_tao(100.0).set_unit(1) + current_price = Balance.from_tao(0.5) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + netuid = 1 + + # Calculate liquidity from amounts + liquidity = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Calculate amounts from liquidity + calc_alpha, calc_tao = calculate_token_amounts_from_liquidity( + liquidity=liquidity, + current_price=current_price, + price_low=price_low, + price_high=price_high, + netuid=netuid, + ) + + # Should match (allowing for rounding errors) + assert abs(calc_alpha.rao - amount_alpha.rao) < 2 # Within 1 rao tolerance + assert calc_tao.rao == 0 + + def test_roundtrip_tao_only(self): + """Test roundtrip: amounts -> liquidity -> amounts (TAO only).""" + amount_tao = Balance.from_tao(10.0) + amount_alpha = Balance.from_tao(100.0).set_unit(1) + current_price = Balance.from_tao(3.0) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + netuid = 1 + + # Calculate liquidity from amounts + liquidity = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Calculate amounts from liquidity + calc_alpha, calc_tao = calculate_token_amounts_from_liquidity( + liquidity=liquidity, + current_price=current_price, + price_low=price_low, + price_high=price_high, + netuid=netuid, + ) + + # Should match (allowing for rounding errors) + assert calc_alpha.rao == 0 + assert abs(calc_tao.rao - amount_tao.rao) < 2 # Within 1 rao tolerance + + def test_roundtrip_both_tokens(self): + """Test roundtrip: amounts -> liquidity -> amounts (both tokens).""" + amount_tao = Balance.from_tao(10.0) + amount_alpha = Balance.from_tao(100.0).set_unit(1) + current_price = Balance.from_tao(1.5) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + netuid = 1 + + # Calculate liquidity from amounts + liquidity = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Calculate amounts from liquidity + calc_alpha, calc_tao = calculate_token_amounts_from_liquidity( + liquidity=liquidity, + current_price=current_price, + price_low=price_low, + price_high=price_high, + netuid=netuid, + ) + + # Both amounts should be less than or equal to available + # (liquidity is limited by the constraining token) + assert calc_alpha.rao <= amount_alpha.rao + assert calc_tao.rao <= amount_tao.rao + + # At least one should be close to fully utilized (within 1%) + alpha_utilization = ( + calc_alpha.rao / amount_alpha.rao if amount_alpha.rao > 0 else 0 + ) + tao_utilization = calc_tao.rao / amount_tao.rao if amount_tao.rao > 0 else 0 + + assert alpha_utilization >= 0.99 or tao_utilization >= 0.99 + + +class TestPriceTickConversions: + """Test that price-tick conversions still work correctly.""" + + def test_price_to_tick_basic(self): + """Test basic price to tick conversion.""" + price = 1.0001**100 + tick = price_to_tick(price) + # Allow for rounding due to floating point precision + assert abs(tick - 100) <= 1 + + def test_tick_to_price_basic(self): + """Test basic tick to price conversion.""" + tick = 100 + price = tick_to_price(tick) + expected_price = 1.0001**100 + assert abs(price - expected_price) < 1e-10 + + def test_roundtrip_price_tick(self): + """Test roundtrip price -> tick -> price.""" + original_price = 1.5 + tick = price_to_tick(original_price) + converted_price = tick_to_price(tick) + # Should be close (tick is discrete) + assert ( + abs(converted_price - original_price) / original_price < 0.001 + ) # Within 0.1% + + def test_price_to_tick_invalid(self): + """Test that invalid prices raise errors.""" + with pytest.raises(ValueError): + price_to_tick(0) + + with pytest.raises(ValueError): + price_to_tick(-1.0) + + +class TestEdgeCases: + """Test edge cases and boundary conditions.""" + + def test_very_small_amounts(self): + """Test with very small token amounts.""" + amount_tao = Balance.from_rao(1000) # 0.000001 TAO + amount_alpha = Balance.from_rao(1000) # 0.000001 Alpha + current_price = Balance.from_tao(1.5) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + + result = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Should handle small amounts without errors + assert result.rao >= 0 + + def test_very_large_amounts(self): + """Test with very large token amounts.""" + amount_tao = Balance.from_tao(1_000_000.0) + amount_alpha = Balance.from_tao(1_000_000.0).set_unit(1) + current_price = Balance.from_tao(1.5) + price_low = Balance.from_tao(1.0) + price_high = Balance.from_tao(2.0) + + result = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Should handle large amounts without overflow + assert result.rao > 0 + + def test_narrow_price_range(self): + """Test with very narrow price range.""" + amount_tao = Balance.from_tao(10.0) + amount_alpha = Balance.from_tao(100.0).set_unit(1) + current_price = Balance.from_tao(1.5) + price_low = Balance.from_tao(1.49) + price_high = Balance.from_tao(1.51) + + result = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Should handle narrow range + assert result.rao > 0 + + def test_wide_price_range(self): + """Test with very wide price range.""" + amount_tao = Balance.from_tao(10.0) + amount_alpha = Balance.from_tao(100.0).set_unit(1) + current_price = Balance.from_tao(5.0) + price_low = Balance.from_tao(0.1) + price_high = Balance.from_tao(100.0) + + result = calculate_max_liquidity_from_amounts( + amount_tao=amount_tao, + amount_alpha=amount_alpha, + current_price=current_price, + price_low=price_low, + price_high=price_high, + ) + + # Should handle wide range + assert result.rao > 0