import os import time import logging import sys import math import json import threading from dotenv import load_dotenv from web3 import Web3 # --- FIX: Add project root to sys.path to import local modules --- current_dir = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.dirname(current_dir) sys.path.append(project_root) # Now we can import from root from logging_utils import setup_logging from eth_account import Account from hyperliquid.exchange import Exchange from hyperliquid.info import Info from hyperliquid.utils import constants # Load environment variables from .env in current directory dotenv_path = os.path.join(current_dir, '.env') if os.path.exists(dotenv_path): load_dotenv(dotenv_path) else: # Fallback to default search load_dotenv() setup_logging("normal", "SCALPER_HEDGER") # --- CONFIGURATION --- COIN_SYMBOL = "ETH" CHECK_INTERVAL = 4 # Optimized for speed (was 5) LEVERAGE = 5 # 3x Leverage STATUS_FILE = "hedge_status.json" RPC_URL = os.environ.get("MAINNET_RPC_URL") # Required for Uniswap Monitor # Uniswap V3 Pool (Arbitrum WETH/USDC 0.05%) UNISWAP_POOL_ADDRESS = "0xC31E54c7a869B9FcBEcc14363CF510d1c41fa443" UNISWAP_POOL_ABI = json.loads('[{"inputs":[],"name":"slot0","outputs":[{"internalType":"uint160","name":"sqrtPriceX96","type":"uint160"},{"internalType":"int24","name":"tick","type":"int24"},{"internalType":"uint16","name":"observationIndex","type":"uint16"},{"internalType":"uint16","name":"observationCardinality","type":"uint16"},{"internalType":"uint16","name":"observationCardinalityNext","type":"uint16"},{"internalType":"uint8","name":"feeProtocol","type":"uint8"},{"internalType":"bool","name":"unlocked","type":"bool"}],"stateMutability":"view","type":"function"}]') # --- STRATEGY ZONES (Percent of Range Width) --- # Bottom Hedge Zone: Covers entire range (0.0 to 1.5) -> Always Active ZONE_BOTTOM_HEDGE_LIMIT = 1 # Close Zone: Disabled (Set > 1.0) ZONE_CLOSE_START = 10.0 ZONE_CLOSE_END = 11.0 # Top Hedge Zone: Disabled/Redundant ZONE_TOP_HEDGE_START = 10.0 # --- ORDER SETTINGS --- PRICE_BUFFER_PCT = 0.0001 # 0.2% price move triggers order update (Relaxed for cost) MIN_THRESHOLD_ETH = 0.0025 # Minimum trade size in ETH (~$60, Reduced frequency) MIN_ORDER_VALUE_USD = 10.0 # Minimum order value for API safety class UniswapPriceMonitor: def __init__(self, rpc_url, pool_address): self.w3 = Web3(Web3.HTTPProvider(rpc_url)) self.pool_contract = self.w3.eth.contract(address=pool_address, abi=UNISWAP_POOL_ABI) self.latest_price = None self.running = True self.thread = threading.Thread(target=self._loop, daemon=True) self.thread.start() def _loop(self): logging.info("Uniswap Monitor Started.") while self.running: try: slot0 = self.pool_contract.functions.slot0().call() sqrt_price_x96 = slot0[0] # Price = (sqrtPriceX96 / 2^96)^2 * 10^(18-6) (WETH/USDC) # But typically WETH is token1? Let's verify standard Arbitrum Pool. # 0xC31E... Token0=WETH, Token1=USDC. # Price = (sqrt / 2^96)^2 * (10^12) -> This gives USDC per ETH? No, Token1/Token0. # Wait, usually Token0 is WETH (18) and Token1 is USDC (6). # P = (1.0001^tick) * 10^(decimals0 - decimals1)? No. # Standard conversion: Price = (sqrtRatioX96 / Q96) ** 2 # Adjusted for decimals: Price = Price_raw / (10**(Dec0 - Dec1)) ? No. # Price (Quote/Base) = (sqrt / Q96)^2 * 10^(BaseDec - QuoteDec) # Let's rely on standard logic: Price = (sqrt / 2^96)^2 * 10^(12) for ETH(18)/USDC(6) raw_price = (sqrt_price_x96 / (2**96)) ** 2 price = raw_price * (10**(18-6)) # 10^12 # If Token0 is WETH, price is USDC per WETH. # Note: If the pool is inverted (USDC/WETH), we invert. # On Arb, WETH is usually Token0? # 0x82aF... < 0xaf88... (WETH < USDC). So WETH is Token0. # Price is Token1 per Token0. self.latest_price = 1 / price if price < 1 else price # Sanity check, ETH should be > 2000 except Exception as e: # logging.error(f"Uniswap Monitor Error: {e}") pass time.sleep(5) def get_price(self): return self.latest_price def get_active_automatic_position(): if not os.path.exists(STATUS_FILE): return None try: with open(STATUS_FILE, 'r') as f: data = json.load(f) for entry in data: if entry.get('type') == 'AUTOMATIC' and entry.get('status') == 'OPEN': return entry except Exception as e: logging.error(f"ERROR reading status file: {e}") return None def update_position_zones_in_json(token_id, zones_data): """Updates the active position in JSON with calculated zone prices and formats the entry.""" if not os.path.exists(STATUS_FILE): return try: with open(STATUS_FILE, 'r') as f: data = json.load(f) updated = False for i, entry in enumerate(data): if entry.get('type') == 'AUTOMATIC' and entry.get('status') == 'OPEN' and entry.get('token_id') == token_id: # Merge Zones for k, v in zones_data.items(): entry[k] = v # Format & Reorder open_ts = entry.get('timestamp_open', int(time.time())) opened_str = time.strftime('%H:%M %d/%m/%y', time.localtime(open_ts)) # Reconstruct Dict in Order new_entry = { "type": entry.get('type'), "token_id": entry.get('token_id'), "opened": opened_str, "status": entry.get('status'), "entry_price": round(entry.get('entry_price', 0), 2), "target_value": round(entry.get('target_value', 0), 2), # Amounts might be string or float or int. Ensure float. "amount0_initial": round(float(entry.get('amount0_initial', 0)), 4), "amount1_initial": round(float(entry.get('amount1_initial', 0)), 2), "range_upper": round(entry.get('range_upper', 0), 2), "zone_top_start_price": entry.get('zone_top_start_price'), "zone_close_top_price": entry.get('zone_close_top_price'), "zone_close_bottom_price": entry.get('zone_close_bottom_price'), "zone_bottom_limit_price": entry.get('zone_bottom_limit_price'), "range_lower": round(entry.get('range_lower', 0), 2), "static_long": entry.get('static_long', 0.0), "timestamp_open": open_ts, "timestamp_close": entry.get('timestamp_close') } data[i] = new_entry updated = True break if updated: with open(STATUS_FILE, 'w') as f: json.dump(data, f, indent=2) logging.info(f"Updated JSON with Formatted Zone Prices for Position {token_id}") except Exception as e: logging.error(f"Error updating JSON zones: {e}") def round_to_sig_figs(x, sig_figs=5): if x == 0: return 0.0 return round(x, sig_figs - int(math.floor(math.log10(abs(x)))) - 1) def round_to_sz_decimals(amount, sz_decimals=4): return round(abs(amount), sz_decimals) class HyperliquidStrategy: def __init__(self, entry_amount0, entry_amount1, target_value, entry_price, low_range, high_range, start_price, static_long=0.0): self.entry_amount0 = entry_amount0 self.entry_amount1 = entry_amount1 self.target_value = target_value self.entry_price = entry_price self.low_range = low_range self.high_range = high_range self.static_long = static_long self.start_price = start_price self.gap = max(0.0, entry_price - start_price) self.recovery_target = entry_price + (2 * self.gap) self.current_mode = "NORMAL" self.last_switch_time = 0 logging.info(f"Strategy Init. Start Px: {start_price:.2f} | Gap: {self.gap:.2f} | Recovery Tgt: {self.recovery_target:.2f}") try: sqrt_P = math.sqrt(entry_price) sqrt_Pa = math.sqrt(low_range) sqrt_Pb = math.sqrt(high_range) self.L = 0.0 # Method 1: Use Amount0 (WETH) if entry_amount0 > 0: # If amount is huge (Wei), scale it. If small (ETH), use as is. if entry_amount0 > 1000: amount0_eth = entry_amount0 / 10**18 else: amount0_eth = entry_amount0 denom0 = (1/sqrt_P) - (1/sqrt_Pb) if denom0 > 0.00000001: self.L = amount0_eth / denom0 logging.info(f"Calculated L from Amount0: {self.L:.4f}") # Method 2: Use Amount1 (USDC) if self.L == 0.0 and entry_amount1 > 0: if entry_amount1 > 100000: amount1_usdc = entry_amount1 / 10**6 else: amount1_usdc = entry_amount1 denom1 = sqrt_P - sqrt_Pa if denom1 > 0.00000001: self.L = amount1_usdc / denom1 logging.info(f"Calculated L from Amount1: {self.L:.4f}") # Method 3: Fallback Heuristic if self.L == 0.0: logging.warning("Amounts missing or 0. Using Target Value Heuristic.") max_eth_heuristic = target_value / low_range denom_h = (1/sqrt_Pa) - (1/sqrt_Pb) if denom_h > 0: self.L = max_eth_heuristic / denom_h logging.info(f"Calculated L from Target Value: {self.L:.4f}") else: logging.error("Critical: Denominator 0 in Heuristic. Invalid Range?") self.L = 0.0 except Exception as e: logging.error(f"Error calculating liquidity: {e}") sys.exit(1) def get_pool_delta(self, current_price): if current_price >= self.high_range: return 0.0 if current_price <= self.low_range: sqrt_Pa = math.sqrt(self.low_range) sqrt_Pb = math.sqrt(self.high_range) return self.L * ((1/sqrt_Pa) - (1/sqrt_Pb)) sqrt_P = math.sqrt(current_price) sqrt_Pb = math.sqrt(self.high_range) return self.L * ((1/sqrt_P) - (1/sqrt_Pb)) def calculate_rebalance(self, current_price, current_short_position_size): pool_delta = self.get_pool_delta(current_price) # --- Over-Hedge Logic --- overhedge_pct = 0.0 range_width = self.high_range - self.low_range if range_width > 0: price_pct = (current_price - self.low_range) / range_width # If below 0.8 (80%) of range if price_pct < 0.8: # Formula: 0.75% boost for every 0.1 drop below 0.8 # Example: At 0.6 (60%), diff is 0.2. (0.2/0.1)*0.0075 = 0.015 (1.5%) overhedge_pct = ((0.8 - max(0.0, price_pct)) / 0.1) * 0.0075 raw_target_short = pool_delta + self.static_long # Apply Boost adjusted_target_short = raw_target_short * (1.0 + overhedge_pct) target_short_size = adjusted_target_short diff = target_short_size - abs(current_short_position_size) return { "current_price": current_price, "pool_delta": pool_delta, "target_short": target_short_size, "current_short": abs(current_short_position_size), "diff": diff, "action": "SELL" if diff > 0 else "BUY", "mode": "OVERHEDGE" if overhedge_pct > 0 else "NORMAL", "overhedge_pct": overhedge_pct } class ScalperHedger: def __init__(self): self.private_key = os.environ.get("SCALPER_AGENT_PK") self.vault_address = os.environ.get("MAIN_WALLET_ADDRESS") if not self.private_key: logging.error("No SCALPER_AGENT_PK found in .env") sys.exit(1) self.account = Account.from_key(self.private_key) self.info = Info(constants.MAINNET_API_URL, skip_ws=True) self.exchange = Exchange(self.account, constants.MAINNET_API_URL, account_address=self.vault_address) try: logging.info(f"Setting leverage to {LEVERAGE}x (Cross)...") self.exchange.update_leverage(LEVERAGE, COIN_SYMBOL, is_cross=True) except Exception as e: logging.error(f"Failed to update leverage: {e}") self.strategy = None self.sz_decimals = self._get_sz_decimals(COIN_SYMBOL) self.active_position_id = None self.active_order = None # --- Start Uniswap Monitor --- self.uni_monitor = UniswapPriceMonitor(RPC_URL, UNISWAP_POOL_ADDRESS) logging.info(f"Scalper Hedger initialized. Agent: {self.account.address}") def _init_strategy(self, position_data): try: entry_amount0 = position_data.get('amount0_initial', 0) entry_amount1 = position_data.get('amount1_initial', 0) target_value = position_data.get('target_value', 50.0) entry_price = position_data['entry_price'] lower = position_data['range_lower'] upper = position_data['range_upper'] static_long = position_data.get('static_long', 0.0) start_price = self.get_market_price(COIN_SYMBOL) if start_price is None: logging.warning("Waiting for initial price to start strategy...") return self.strategy = HyperliquidStrategy( entry_amount0=entry_amount0, entry_amount1=entry_amount1, target_value=target_value, entry_price=entry_price, low_range=lower, high_range=upper, start_price=start_price, static_long=static_long ) logging.info(f"Strategy Initialized for Position {position_data['token_id']}.") self.active_position_id = position_data['token_id'] except Exception as e: logging.error(f"Failed to init strategy: {e}") self.strategy = None def _get_sz_decimals(self, coin): try: meta = self.info.meta() for asset in meta["universe"]: if asset["name"] == coin: return asset["szDecimals"] return 4 except: return 4 def get_order_book_levels(self, coin): try: l2_snapshot = self.info.l2_snapshot(coin) if l2_snapshot and 'levels' in l2_snapshot: bids = l2_snapshot['levels'][0] asks = l2_snapshot['levels'][1] if bids and asks: best_bid = float(bids[0]['px']) best_ask = float(asks[0]['px']) mid = (best_bid + best_ask) / 2 return {'bid': best_bid, 'ask': best_ask, 'mid': mid} # Fallback px = self.get_market_price(coin) return {'bid': px, 'ask': px, 'mid': px} except: px = self.get_market_price(coin) return {'bid': px, 'ask': px, 'mid': px} def get_market_price(self, coin): try: mids = self.info.all_mids() if coin in mids: return float(mids[coin]) except: pass return None def get_order_book_mid(self, coin): try: l2_snapshot = self.info.l2_snapshot(coin) if l2_snapshot and 'levels' in l2_snapshot: bids = l2_snapshot['levels'][0] asks = l2_snapshot['levels'][1] if bids and asks: best_bid = float(bids[0]['px']) best_ask = float(asks[0]['px']) return (best_bid + best_ask) / 2 return self.get_market_price(coin) except: return self.get_market_price(coin) def get_funding_rate(self, coin): try: meta, asset_ctxs = self.info.meta_and_asset_ctxs() for i, asset in enumerate(meta["universe"]): if asset["name"] == coin: return float(asset_ctxs[i]["funding"]) return 0.0 except: return 0.0 def get_current_position(self, coin): try: user_state = self.info.user_state(self.vault_address or self.account.address) for pos in user_state["assetPositions"]: if pos["position"]["coin"] == coin: return { 'size': float(pos["position"]["szi"]), 'pnl': float(pos["position"]["unrealizedPnl"]) } return {'size': 0.0, 'pnl': 0.0} except: return {'size': 0.0, 'pnl': 0.0} def get_open_orders(self): try: return self.info.open_orders(self.vault_address or self.account.address) except: return [] def cancel_order(self, coin, oid): logging.info(f"Cancelling order {oid}...") try: return self.exchange.cancel(coin, oid) except Exception as e: logging.error(f"Error cancelling order: {e}") def place_limit_order(self, coin, is_buy, size, price): logging.info(f"🕒 PLACING LIMIT: {coin} {'BUY' if is_buy else 'SELL'} {size} @ {price:.2f}") reduce_only = is_buy try: # Gtc order (Maker) -> Changed to Alo to force Maker limit_px = round_to_sig_figs(price, 5) # Use 'Alo' (Add Liquidity Only) to ensure Maker rebate. # If price crosses spread, order is rejected (safe cost-wise). order_result = self.exchange.order(coin, is_buy, size, limit_px, {"limit": {"tif": "Alo"}}, reduce_only=reduce_only) status = order_result["status"] if status == "ok": response_data = order_result["response"]["data"] if "statuses" in response_data: status_obj = response_data["statuses"][0] if "error" in status_obj: logging.error(f"Order API Error: {status_obj['error']}") return None # Parse OID from nested structure oid = None if "resting" in status_obj: oid = status_obj["resting"]["oid"] elif "filled" in status_obj: oid = status_obj["filled"]["oid"] logging.info("Order filled immediately.") if oid: logging.info(f"✅ Limit Order Placed: OID {oid}") return oid else: logging.warning(f"Order placed but OID not found in: {status_obj}") return None else: logging.error(f"Order Failed: {order_result}") return None except Exception as e: logging.error(f"Exception during trade: {e}") return None def manage_orders(self): """ Checks open orders. Returns: True if an order exists and is valid (don't trade), False if no order (can trade). """ open_orders = self.get_open_orders() my_orders = [o for o in open_orders if o['coin'] == COIN_SYMBOL] if not my_orders: self.active_order = None return False if len(my_orders) > 1: logging.warning("Multiple open orders found. Cancelling all for safety.") for o in my_orders: self.cancel_order(COIN_SYMBOL, o['oid']) self.active_order = None return False order = my_orders[0] oid = order['oid'] order_price = float(order['limitPx']) current_mid = self.get_order_book_mid(COIN_SYMBOL) pct_diff = abs(current_mid - order_price) / order_price if pct_diff > PRICE_BUFFER_PCT: logging.info(f"Price moved {pct_diff*100:.3f}% > {PRICE_BUFFER_PCT*100}%. Cancelling/Replacing order {oid}.") self.cancel_order(COIN_SYMBOL, oid) self.active_order = None return False else: logging.info(f"Pending Order {oid} @ {order_price:.2f} is within range ({pct_diff*100:.3f}%). Waiting.") return True def close_all_positions(self): logging.info("Closing all positions (Market Order)...") try: # Cancel open orders first open_orders = self.get_open_orders() for o in open_orders: if o['coin'] == COIN_SYMBOL: self.cancel_order(COIN_SYMBOL, o['oid']) price = self.get_market_price(COIN_SYMBOL) pos_data = self.get_current_position(COIN_SYMBOL) current_pos = pos_data['size'] if current_pos == 0: return is_buy = current_pos < 0 final_size = round_to_sz_decimals(abs(current_pos), self.sz_decimals) if final_size == 0: return price = self.get_market_price(COIN_SYMBOL) # Get mid price for safety fallback pos_data = self.get_current_position(COIN_SYMBOL) current_pos = pos_data['size'] if current_pos == 0: return is_buy_to_close = current_pos < 0 final_size = round_to_sz_decimals(abs(current_pos), self.sz_decimals) if final_size == 0: return # --- ATTEMPT MAKER CLOSE (Alo) --- try: book_levels = self.get_order_book_levels(COIN_SYMBOL) TICK_SIZE = 0.1 if is_buy_to_close: # We are short, need to buy to close maker_price = book_levels['bid'] - TICK_SIZE else: # We are long, need to sell to close maker_price = book_levels['ask'] + TICK_SIZE logging.info(f"Attempting MAKER CLOSE (Alo): {COIN_SYMBOL} {'BUY' if is_buy_to_close else 'SELL'} {final_size} @ {maker_price:.2f}") order_result = self.exchange.order(COIN_SYMBOL, is_buy_to_close, final_size, round_to_sig_figs(maker_price, 5), {"limit": {"tif": "Alo"}}, reduce_only=True) status = order_result["status"] if status == "ok": response_data = order_result["response"]["data"] if "statuses" in response_data and "resting" in response_data["statuses"][0]: logging.info(f"✅ MAKER CLOSE Order Placed (Alo). OID: {response_data['statuses'][0]['resting']['oid']}") return elif "statuses" in response_data and "filled" in response_data["statuses"][0]: logging.info(f"✅ MAKER CLOSE Order Filled (Alo). OID: {response_data['statuses'][0]['filled']['oid']}") return else: # Fallback if Alo didn't rest or fill immediately in an expected way logging.warning(f"Alo order result unclear: {order_result}. Falling back to Market Close.") elif status == "error": if "Post only order would have immediately matched" in order_result["response"]["data"]["statuses"][0].get("error", ""): logging.warning("Alo order would have immediately matched. Falling back to Market Close for guaranteed fill.") else: logging.error(f"Alo order failed with unknown error: {order_result}. Falling back to Market Close.") else: logging.warning(f"Alo order failed with status {status}. Falling back to Market Close.") except Exception as e: logging.error(f"Exception during Alo close attempt: {e}. Falling back to Market Close.", exc_info=True) # --- FALLBACK TO MARKET CLOSE (Ioc) for guaranteed fill --- logging.info(f"Falling back to MARKET CLOSE (Ioc): {COIN_SYMBOL} {'BUY' if is_buy_to_close else 'SELL'} {final_size} @ {price:.2f} (guaranteed)") self.exchange.order(COIN_SYMBOL, is_buy_to_close, final_size, round_to_sig_figs(price * (1.05 if is_buy_to_close else 0.95), 5), {"limit": {"tif": "Ioc"}}, reduce_only=True) self.active_position_id = None logging.info("✅ MARKET CLOSE Order Placed (Ioc).") except Exception as e: logging.error(f"Error closing positions: {e}", exc_info=True) def run(self): logging.info(f"Starting Scalper Monitor Loop. Interval: {CHECK_INTERVAL}s") while True: try: active_pos = get_active_automatic_position() # Check Global Enable Switch if not active_pos or not active_pos.get('hedge_enabled', True): if self.strategy is not None: logging.info("Hedge Disabled or Position Closed. Closing remaining positions.") self.close_all_positions() self.strategy = None else: pass time.sleep(CHECK_INTERVAL) continue if self.strategy is None or self.active_position_id != active_pos['token_id']: logging.info(f"New position {active_pos['token_id']} detected or strategy not initialized. Initializing strategy.") self._init_strategy(active_pos) if self.strategy is None: time.sleep(CHECK_INTERVAL) continue if self.strategy is None: continue # --- ORDER MANAGEMENT --- if self.manage_orders(): time.sleep(CHECK_INTERVAL) continue # 2. Market Data book_levels = self.get_order_book_levels(COIN_SYMBOL) price = book_levels['mid'] if price is None: time.sleep(5) continue funding_rate = self.get_funding_rate(COIN_SYMBOL) pos_data = self.get_current_position(COIN_SYMBOL) current_pos_size = pos_data['size'] current_pnl = pos_data['pnl'] # --- SPREAD MONITOR LOG --- uni_price = self.uni_monitor.get_price() spread_text = "" if uni_price: diff = price - uni_price pct = (diff / uni_price) * 100 spread_text = f" | Sprd: {pct:+.2f}% (H:{price:.0f}/U:{uni_price:.0f})" # 3. Calculate Logic calc = self.strategy.calculate_rebalance(price, current_pos_size) diff_abs = abs(calc['diff']) # --- LOGGING OVERHEDGE --- oh_text = "" if calc.get('overhedge_pct', 0) > 0: oh_text = f" | 🔥 OH: +{calc['overhedge_pct']*100:.2f}%" # 4. Dynamic Threshold Calculation sqrt_Pa = math.sqrt(self.strategy.low_range) sqrt_Pb = math.sqrt(self.strategy.high_range) max_potential_eth = self.strategy.L * ((1/sqrt_Pa) - (1/sqrt_Pb)) # Use MIN_THRESHOLD_ETH from config rebalance_threshold = max(MIN_THRESHOLD_ETH, max_potential_eth * 0.05) # 5. Determine Hedge Zone clp_low_range = self.strategy.low_range clp_high_range = self.strategy.high_range range_width = clp_high_range - clp_low_range # Calculate Prices for Zones # If config > 9, set to None (Disabled Zone) zone_bottom_limit_price = (clp_low_range + (range_width * ZONE_BOTTOM_HEDGE_LIMIT)) if ZONE_BOTTOM_HEDGE_LIMIT <= 9 else None zone_close_bottom_price = (clp_low_range + (range_width * ZONE_CLOSE_START)) if ZONE_CLOSE_START <= 9 else None zone_close_top_price = (clp_low_range + (range_width * ZONE_CLOSE_END)) if ZONE_CLOSE_END <= 9 else None zone_top_start_price = (clp_low_range + (range_width * ZONE_TOP_HEDGE_START)) if ZONE_TOP_HEDGE_START <= 9 else None # Update JSON with zone prices if they are None (initially set by uniswap_manager.py) if active_pos.get('zone_bottom_limit_price') is None: update_position_zones_in_json(active_pos['token_id'], { 'zone_top_start_price': round(zone_top_start_price, 2) if zone_top_start_price else None, 'zone_close_top_price': round(zone_close_top_price, 2) if zone_close_top_price else None, 'zone_close_bottom_price': round(zone_close_bottom_price, 2) if zone_close_bottom_price else None, 'zone_bottom_limit_price': round(zone_bottom_limit_price, 2) if zone_bottom_limit_price else None }) # Check Zones (Handle None) # If zone price is None, condition fails safe (False) in_close_zone = False if zone_close_bottom_price is not None and zone_close_top_price is not None: in_close_zone = (price >= zone_close_bottom_price and price <= zone_close_top_price) in_hedge_zone = False if zone_bottom_limit_price is not None and price <= zone_bottom_limit_price: in_hedge_zone = True if zone_top_start_price is not None and price >= zone_top_start_price: in_hedge_zone = True # --- Execute Logic --- if in_close_zone: logging.info(f"ZONE: CLOSE ({price:.2f} in {zone_close_bottom_price:.2f}-{zone_close_top_price:.2f}). PNL: ${current_pnl:.2f}. Closing all hedge positions.") self.close_all_positions() time.sleep(CHECK_INTERVAL) continue elif in_hedge_zone: # HEDGE NORMALLY if diff_abs > rebalance_threshold: trade_size = round_to_sz_decimals(diff_abs, self.sz_decimals) min_trade_size = MIN_ORDER_VALUE_USD / price if trade_size < min_trade_size: logging.info(f"Idle. Trade size {trade_size} < Min Order Size {min_trade_size:.4f} (${MIN_ORDER_VALUE_USD:.2f}). PNL: ${current_pnl:.2f}{spread_text}{oh_text}") elif trade_size > 0: logging.info(f"⚡ THRESHOLD TRIGGERED ({diff_abs:.4f} >= {rebalance_threshold:.4f}). In Hedge Zone. PNL: ${current_pnl:.2f}{spread_text}{oh_text}") # Execute Passively for Alo # Force 1 tick offset (0.1) away from BBO to ensure rounding doesn't cause cross # Sell at Ask + 0.1, Buy at Bid - 0.1 TICK_SIZE = 0.1 is_buy = (calc['action'] == "BUY") if is_buy: exec_price = book_levels['bid'] - TICK_SIZE else: exec_price = book_levels['ask'] + TICK_SIZE self.place_limit_order(COIN_SYMBOL, is_buy, trade_size, exec_price) else: logging.info(f"Trade size rounds to 0. Skipping. PNL: ${current_pnl:.2f}{spread_text}{oh_text}") else: logging.info(f"Idle. Diff {diff_abs:.4f} < Threshold {rebalance_threshold:.4f}. In Hedge Zone. PNL: ${current_pnl:.2f}{spread_text}{oh_text}") else: # MIDDLE ZONE (IDLE) pct_position = (price - clp_low_range) / range_width logging.info(f"Idle. In Middle Zone ({pct_position*100:.1f}%). PNL: ${current_pnl:.2f}{spread_text}{oh_text}. No Actions.") time.sleep(CHECK_INTERVAL) except KeyboardInterrupt: logging.info("Stopping Hedger...") self.close_all_positions() break except Exception as e: logging.error(f"Loop Error: {e}", exc_info=True) time.sleep(10) if __name__ == "__main__": hedger = ScalperHedger() hedger.run()