ditus/uniswap_auto_clp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add asymmetric compensation & exact liquidity support to hedger

Browse Source
This commit is contained in:
DiTus
2025-12-26 23:01:00 +01:00
parent 4f7bb429b7
commit c04eb3f377
2 changed files with 79 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
clp_hedger.py
View File

@ -222,7 +222,8 @@ def update_position_stats(token_id: int, stats_data: Dict):
class HyperliquidStrategy:
def __init__(self, entry_amount0: Decimal, entry_amount1: Decimal, target_value: Decimal,
entry_price: Decimal, low_range: Decimal, high_range: Decimal, start_price: Decimal):
entry_price: Decimal, low_range: Decimal, high_range: Decimal, start_price: Decimal,
liquidity: int = 0):
self.entry_amount0 = entry_amount0
self.entry_amount1 = entry_amount1
self.target_value = target_value
@ -235,40 +236,53 @@ class HyperliquidStrategy:
self.recovery_target = entry_price + (Decimal("2") * self.gap)
self.L = Decimal("0.0")
try:
sqrt_P = entry_price.sqrt()
sqrt_Pa = low_range.sqrt()
sqrt_Pb = high_range.sqrt()
# Priority: Use exact Liquidity from Contract if available
if liquidity > 0:
# Scale raw liquidity (uint128) to human liquidity
# Formula: L_human = L_raw * 10^(-(d0+d1)/2)
# For ETH(18) / USDC(6) -> 10^(-12)
scale = Decimal("1e-12")
self.L = Decimal(liquidity) * scale
# Method 1: Amount0 (WETH)
if entry_amount0 > 0:
# Assuming amount0 is already in standard units (ETH) from JSON
denom0 = (Decimal("1") / sqrt_P) - (Decimal("1") / sqrt_Pb)
if denom0 > Decimal("1e-10"):
self.L = entry_amount0 / denom0
logger.info(f"Calculated L from Amount0: {self.L:.4f}")
# Method 2: Amount1 (USDC)
if self.L == 0 and entry_amount1 > 0:
denom1 = sqrt_P - sqrt_Pa
if denom1 > Decimal("1e-10"):
self.L = entry_amount1 / denom1
logger.info(f"Calculated L from Amount1: {self.L:.4f}")
# Method 3: Target Value Heuristic
if self.L == 0:
logger.warning("Amounts missing. Using Target Value Heuristic.")
max_eth = target_value / low_range
denom_h = (Decimal("1") / sqrt_Pa) - (Decimal("1") / sqrt_Pb)
if denom_h > 0:
self.L = max_eth / denom_h
logger.info(f"Calculated L from Target Value: {self.L:.4f}")
else:
logger.error("Critical: Invalid Range for L calculation")
# Calculate implied delta at entry for verification
implied_delta = self.get_pool_delta(entry_price)
logger.info(f"Using Exact Liquidity: {self.L:.4f} (Raw: {liquidity}) -> Implied Delta: {implied_delta:.4f} ETH")
else:
try:
sqrt_P = entry_price.sqrt()
sqrt_Pa = low_range.sqrt()
sqrt_Pb = high_range.sqrt()
# Method 1: Amount0 (WETH)
if entry_amount0 > 0:
# Assuming amount0 is already in standard units (ETH) from JSON
denom0 = (Decimal("1") / sqrt_P) - (Decimal("1") / sqrt_Pb)
if denom0 > Decimal("1e-10"):
self.L = entry_amount0 / denom0
logger.info(f"Calculated L from Amount0: {self.L:.4f}")
# Method 2: Amount1 (USDC)
if self.L == 0 and entry_amount1 > 0:
denom1 = sqrt_P - sqrt_Pa
if denom1 > Decimal("1e-10"):
self.L = entry_amount1 / denom1
logger.info(f"Calculated L from Amount1: {self.L:.4f}")
# Method 3: Target Value Heuristic
if self.L == 0:
logger.warning("Amounts missing. Using Target Value Heuristic.")
max_eth = target_value / low_range
denom_h = (Decimal("1") / sqrt_Pa) - (Decimal("1") / sqrt_Pb)
if denom_h > 0:
self.L = max_eth / denom_h
logger.info(f"Calculated L from Target Value: {self.L:.4f}")
else:
logger.error("Critical: Invalid Range for L calculation")
except Exception as e:
logger.error(f"Error calculating liquidity: {e}")
sys.exit(1)
except Exception as e:
logger.error(f"Error calculating liquidity: {e}")
sys.exit(1)
def get_pool_delta(self, current_price: Decimal) -> Decimal:
if current_price >= self.high_range:
@ -286,21 +300,28 @@ class HyperliquidStrategy:
def calculate_rebalance(self, current_price: Decimal, current_short_size: Decimal) -> Dict:
pool_delta = self.get_pool_delta(current_price)
# Over-Hedge Logic
overhedge_pct = Decimal("0.0")
# --- ASYMMETRIC COMPENSATION (0.35% Leakage Fix) ---
# Over-hedge on drops, Under-hedge on rises to offset execution friction.
# Max adjustment at edges: 7.5%
adj_pct = Decimal("0.0")
range_width = self.high_range - self.low_range
if range_width > 0:
price_pct = (current_price - self.low_range) / range_width
# Distance from entry price
dist = current_price - self.entry_price
# Normalize to range half-width (approx)
half_width = range_width / Decimal("2")
norm_dist = dist / half_width
# Adjustment: -7.5% at +1.0 (High), +7.5% at -1.0 (Low)
max_boost = Decimal("0.075")
adj_pct = -norm_dist * max_boost
# Safety Cap
adj_pct = max(-max_boost, min(max_boost, adj_pct))
# If below 80% of range
if price_pct < Decimal("0.8"):
# Formula: 0.75% boost for every 0.1 drop below 0.8
diff_factor = (Decimal("0.8") - max(Decimal("0.0"), price_pct)) / Decimal("0.1")
overhedge_pct = diff_factor * Decimal("0.0075")
raw_target_short = pool_delta
adjusted_target_short = raw_target_short * (Decimal("1.0") + overhedge_pct)
adjusted_target_short = raw_target_short * (Decimal("1.0") + adj_pct)
diff = adjusted_target_short - abs(current_short_size)
@ -311,7 +332,7 @@ class HyperliquidStrategy:
"current_short": abs(current_short_size),
"diff": diff,
"action": "SELL" if diff > 0 else "BUY",
"overhedge_pct": overhedge_pct
"adj_pct": adj_pct
}
# --- MAIN HEDGER CLASS ---
@ -434,6 +455,8 @@ class ScalperHedger:
lower = to_decimal(position_data['range_lower'])
upper = to_decimal(position_data['range_upper'])
liquidity_val = int(position_data.get('liquidity', 0))
start_price = self.get_market_price(COIN_SYMBOL)
if start_price is None:
logger.warning("Waiting for initial price to start strategy...")
@ -446,7 +469,8 @@ class ScalperHedger:
entry_price=entry_price,
low_range=lower,
high_range=upper,
start_price=start_price
start_price=start_price,
liquidity=liquidity_val
)
# Reset State
@ -1017,7 +1041,7 @@ class ScalperHedger:
create_shadow = False
urgency = "URGENT" if bypass_cooldown else "NORMAL"
logger.info(f"[TRIG] Rebalance ({urgency}): {calc['action']} {diff_abs:.4f} > {rebalance_threshold:.4f} | Book: {levels['bid']}/{levels['ask']} | Vol: {vol_pct*100:.3f}% x{vol_multiplier:.1f} | Thresh: {final_threshold_pct*100:.1f}%")
logger.info(f"[TRIG] Rebalance ({urgency}): {calc['action']} {diff_abs:.4f} > {rebalance_threshold:.4f} | Adj: {calc['adj_pct']*100:+.1f}% | Book: {levels['bid']}/{levels['ask']} | Vol: {vol_pct*100:.3f}% x{vol_multiplier:.1f} | Thresh: {final_threshold_pct*100:.1f}%")
oid = self.place_limit_order(COIN_SYMBOL, is_buy, diff_abs, exec_price, order_type)
if oid:
@ -1074,7 +1098,7 @@ class ScalperHedger:
p_sell = self.strategy.low_range + safety_margin
net_pnl = self.accumulated_pnl - self.accumulated_fees
logger.info(f"[IDLE] Px: {price:.2f} | M: {p_mid:.1f} | B: {p_buy:.1f} / S: {p_sell:.1f} (Vol: {vol_pct*100:.3f}% x{vol_multiplier:.1f} | Thresh: {final_threshold_pct*100:.1f}%) | TotPnL: {net_pnl:.2f}")
logger.info(f"[IDLE] Px: {price:.2f} | M: {p_mid:.1f} | B: {p_buy:.1f} / S: {p_sell:.1f} | Adj: {calc['adj_pct']*100:+.1f}% (Vol: {vol_pct*100:.3f}% x{vol_multiplier:.1f} | Thresh: {final_threshold_pct*100:.1f}%) | TotPnL: {net_pnl:.2f}")
self.last_idle_log_time = time.time()
# --- FISHING ORDER LOGIC (SAFE ZONE) ---

13
uniswap_manager.py
View File

@ -128,8 +128,8 @@ CLOSE_POSITION_ENABLED = True
OPEN_POSITION_ENABLED = True
REBALANCE_ON_CLOSE_BELOW_RANGE = True
TARGET_INVESTMENT_VALUE_USDC = 2000
INITIAL_HEDGE_CAPITAL_USDC = 2000 # Your starting Hyperliquid balance for Benchmark calc
RANGE_WIDTH_PCT = Decimal("0.01") # +/- 1% (2% total width)
INITIAL_HEDGE_CAPITAL_USDC = 1000 # Your starting Hyperliquid balance for Benchmark calc
RANGE_WIDTH_PCT = Decimal("0.05") # +/- 5% (10% total width)
SLIPPAGE_TOLERANCE = Decimal("0.02") # do not change, or at least remember it ( 0.02 = 2.0% slippage tolerance)
TRANSACTION_TIMEOUT_SECONDS = 30
@ -179,9 +179,10 @@ def send_transaction_robust(
"""
try:
# 1. Prepare Params
# Use 'pending' to ensure we get the correct nonce if a tx was just sent/mined
tx_params = {
'from': account.address,
'nonce': w3.eth.get_transaction_count(account.address),
'nonce': w3.eth.get_transaction_count(account.address, 'pending'),
'value': value,
'chainId': w3.eth.chain_id,
}
@ -541,7 +542,7 @@ def mint_new_position(w3: Web3, npm_contract, account: LocalAccount, token0: str
# IncreaseLiquidity Event (Topic0)
increase_liq_topic = Web3.keccak(text="IncreaseLiquidity(uint256,uint128,uint256,uint256)").hex()
minted_data = {'token_id': None, 'amount0': 0, 'amount1': 0}
minted_data = {'token_id': None, 'liquidity': 0, 'amount0': 0, 'amount1': 0}
for log in receipt.logs:
topics = [t.hex() for t in log['topics']]
@ -560,6 +561,7 @@ def mint_new_position(w3: Web3, npm_contract, account: LocalAccount, token0: str
data = data[2:]
# liquidity is first 32 bytes (padded), amt0 next 32, amt1 next 32
minted_data['liquidity'] = int(data[0:64], 16)
minted_data['amount0'] = int(data[64:128], 16)
minted_data['amount1'] = int(data[128:192], 16)
@ -872,6 +874,7 @@ def main():
"entry_price": round(entry_price, 2),
"amount0_initial": round(fmt_amt0, 4),
"amount1_initial": round(fmt_amt1, 2),
"liquidity": str(minted['liquidity']),
"range_upper": round(float(price_from_tick(tick_upper, d0, d1)), 2),
"range_lower": round(float(price_from_tick(tick_lower, d0, d1)), 2),
"timestamp_open": int(time.time())
@ -891,4 +894,4 @@ def main():
time.sleep(MONITOR_INTERVAL_SECONDS)
if __name__ == "__main__":
main()
main()