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

clp hedge zones

Browse Source
This commit is contained in:
DiTus
2025-12-14 19:03:50 +01:00
parent 84242f3654
commit e31079cdbb
3 changed files with 742 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

287
clp_hedger/clp_scalper_hedger.py
View File

@ -30,13 +30,28 @@ setup_logging("normal", "SCALPER_HEDGER")
# --- CONFIGURATION ---
COIN_SYMBOL = "ETH"
CHECK_INTERVAL = 10 # Faster check for scalper
CHECK_INTERVAL = 1 # Faster check for scalper
LEVERAGE = 5
STATUS_FILE = "hedge_status.json"
# Gap Recovery Configuration
PRICE_BUFFER_PCT = 0.002 # 0.25% buffer
TIME_BUFFER_SECONDS = 120 # 2 minutes wait
# --- STRATEGY ZONES (Percent of Range Width) ---
# Bottom Hedge Zone: 0% to 15% -> Active Hedging
ZONE_BOTTOM_HEDGE_LIMIT = 0.1
# Close Zone: 15% to 20% -> Close All Hedges (Flatten)
ZONE_CLOSE_START = 0.18
ZONE_CLOSE_END = 0.20
# Middle Zone: 20% to 85% -> Idle (No new orders, keep existing)
# Implied by gaps between other zones.
# Top Hedge Zone: 85% to 100% -> Active Hedging
ZONE_TOP_HEDGE_START = 0.8
# --- ORDER SETTINGS ---
PRICE_BUFFER_PCT = 0.0005 # 0.05% price move triggers order update
MIN_THRESHOLD_ETH = 0.005 # Minimum trade size in ETH
MIN_ORDER_VALUE_USD = 10.0 # Minimum order value for API safety
def get_active_automatic_position():
if not os.path.exists(STATUS_FILE):
@ -51,6 +66,29 @@ def get_active_automatic_position():
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."""
if not os.path.exists(STATUS_FILE): return
try:
with open(STATUS_FILE, 'r') as f:
data = json.load(f)
updated = False
for entry in data:
if entry.get('type') == 'AUTOMATIC' and entry.get('status') == 'OPEN' and entry.get('token_id') == token_id:
# Update keys
for k, v in zones_data.items():
entry[k] = v
updated = True
break
if updated:
with open(STATUS_FILE, 'w') as f:
json.dump(data, f, indent=2)
logging.info(f"Updated JSON with 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)
@ -85,7 +123,6 @@ class HyperliquidStrategy:
self.L = 0.0
# Method 1: Use Amount0 (WETH)
# Formula: L = amount0 / (1/sqrtP - 1/sqrtPb)
if entry_amount0 > 0:
amount0_eth = entry_amount0 / 10**18
denom0 = (1/sqrt_P) - (1/sqrt_Pb)
@ -93,20 +130,15 @@ class HyperliquidStrategy:
self.L = amount0_eth / denom0
logging.info(f"Calculated L from Amount0: {self.L:.4f}")
# Method 2: Use Amount1 (USDC) if Method 1 failed or yielded 0
# Formula: L = amount1 / (sqrtP - sqrtPa)
# Note: Price in formula is Token1/Token0? No, sqrtPrice is sqrt(Token1/Token0).
# Yes. Amount1 = L * (sqrtP - sqrtPa)
# Method 2: Use Amount1 (USDC)
if self.L == 0.0 and entry_amount1 > 0:
amount1_usdc = entry_amount1 / 10**6 # USDC is 6 decimals
amount1_usdc = entry_amount1 / 10**6
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 using Target Value
# Max ETH = Value / LowerPrice.
# L = MaxETH / (1/sqrtPa - 1/sqrtPb)
# 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
@ -137,33 +169,7 @@ class HyperliquidStrategy:
pool_delta = self.get_pool_delta(current_price)
raw_target_short = pool_delta + self.static_long
entry_upper = self.entry_price * (1 + PRICE_BUFFER_PCT)
entry_lower = self.entry_price * (1 - PRICE_BUFFER_PCT)
desired_mode = self.current_mode
if self.current_mode == "NORMAL":
if current_price > entry_upper and current_price < self.recovery_target:
desired_mode = "RECOVERY"
elif self.current_mode == "RECOVERY":
if current_price < entry_lower or current_price >= self.recovery_target:
desired_mode = "NORMAL"
now = time.time()
if desired_mode != self.current_mode:
if (now - self.last_switch_time) >= TIME_BUFFER_SECONDS:
logging.info(f"🔄 MODE SWITCH: {self.current_mode} -> {desired_mode} (Px: {current_price:.2f})")
self.current_mode = desired_mode
self.last_switch_time = now
else:
logging.info(f"⏳ Mode Switch Delayed (Time Buffer). Pending: {desired_mode}")
if self.current_mode == "RECOVERY":
target_short_size = 0.0
logging.info(f"🩹 RECOVERY MODE ACTIVE (0% Hedge). Target: {self.recovery_target:.2f}")
else:
target_short_size = raw_target_short
target_short_size = raw_target_short
diff = target_short_size - abs(current_short_position_size)
return {
@ -173,7 +179,7 @@ class HyperliquidStrategy:
"current_short": abs(current_short_position_size),
"diff": diff,
"action": "SELL" if diff > 0 else "BUY",
"mode": self.current_mode
"mode": "NORMAL"
}
class ScalperHedger:
@ -198,6 +204,8 @@ class ScalperHedger:
self.strategy = None
self.sz_decimals = self._get_sz_decimals(COIN_SYMBOL)
self.active_position_id = None
self.active_order = None
logging.info(f"Scalper Hedger initialized. Agent: {self.account.address}")
def _init_strategy(self, position_data):
@ -249,6 +257,20 @@ class ScalperHedger:
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()
@ -267,29 +289,101 @@ class ScalperHedger:
return 0.0
except: return 0.0
def execute_trade(self, coin, is_buy, size, price):
logging.info(f"🚀 EXECUTING: {coin} {'BUY' if is_buy else 'SELL'} {size} @ ~{price}")
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:
raw_limit_px = price * (1.05 if is_buy else 0.95)
limit_px = round_to_sig_figs(raw_limit_px, 5)
# Gtc order (Maker)
limit_px = round_to_sig_figs(price, 5)
order_result = self.exchange.order(coin, is_buy, size, limit_px, {"limit": {"tif": "Ioc"}}, reduce_only=reduce_only)
order_result = self.exchange.order(coin, is_buy, size, limit_px, {"limit": {"tif": "Gtc"}}, reduce_only=reduce_only)
status = order_result["status"]
if status == "ok":
response_data = order_result["response"]["data"]
if "statuses" in response_data and "error" in response_data["statuses"][0]:
logging.error(f"Order API Error: {response_data['statuses'][0]['error']}")
else:
logging.info(f"✅ Trade Success")
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 (Safety/Closed State)...")
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)
current_pos = self.get_current_position(COIN_SYMBOL)
if current_pos == 0: return
@ -298,7 +392,8 @@ class ScalperHedger:
final_size = round_to_sz_decimals(abs(current_pos), self.sz_decimals)
if final_size == 0: return
self.execute_trade(COIN_SYMBOL, is_buy, final_size, price)
# Market order for closing
self.exchange.order(COIN_SYMBOL, is_buy, final_size, round_to_sig_figs(price * (1.05 if is_buy else 0.95), 5), {"limit": {"tif": "Ioc"}}, reduce_only=True)
self.active_position_id = None
except Exception as e:
logging.error(f"Error closing: {e}")
@ -321,7 +416,6 @@ class ScalperHedger:
time.sleep(CHECK_INTERVAL)
continue
# Initialize Strategy if needed
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)
@ -329,12 +423,15 @@ class ScalperHedger:
time.sleep(CHECK_INTERVAL)
continue
# Double Check Strategy validity
if self.strategy is None:
if self.strategy is None: continue
# --- ORDER MANAGEMENT ---
if self.manage_orders():
time.sleep(CHECK_INTERVAL)
continue
# 2. Market Data
price = self.get_market_price(COIN_SYMBOL)
price = self.get_order_book_mid(COIN_SYMBOL)
if price is None:
time.sleep(5)
continue
@ -342,7 +439,7 @@ class ScalperHedger:
funding_rate = self.get_funding_rate(COIN_SYMBOL)
current_pos_size = self.get_current_position(COIN_SYMBOL)
# 3. Calculate
# 3. Calculate Logic
calc = self.strategy.calculate_rebalance(price, current_pos_size)
diff_abs = abs(calc['diff'])
@ -351,26 +448,68 @@ class ScalperHedger:
sqrt_Pb = math.sqrt(self.strategy.high_range)
max_potential_eth = self.strategy.L * ((1/sqrt_Pa) - (1/sqrt_Pb))
min_threshold = 0.001
rebalance_threshold = max(min_threshold, max_potential_eth * 0.05)
# Use MIN_THRESHOLD_ETH from config
rebalance_threshold = max(MIN_THRESHOLD_ETH, max_potential_eth * 0.05)
# 5. Execute with Min Order Value check
if diff_abs > rebalance_threshold:
trade_size = round_to_sz_decimals(diff_abs, self.sz_decimals)
min_order_value_usd = 10.0
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})")
elif trade_size > 0:
logging.info(f"⚡ THRESHOLD TRIGGERED ({diff_abs:.4f} >= {rebalance_threshold:.4f})")
is_buy = (calc['action'] == "BUY")
self.execute_trade(COIN_SYMBOL, is_buy, trade_size, price)
# 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
zone_bottom_limit_price = clp_low_range + (range_width * ZONE_BOTTOM_HEDGE_LIMIT)
zone_close_start_price = clp_low_range + (range_width * ZONE_CLOSE_START)
zone_close_end_price = clp_low_range + (range_width * ZONE_CLOSE_END)
zone_top_start_price = clp_low_range + (range_width * ZONE_TOP_HEDGE_START)
# Update JSON with zone prices if missing
if 'zone_bottom_limit_price' not in active_pos:
update_position_zones_in_json(active_pos['token_id'], {
'zone_bottom_limit_price': zone_bottom_limit_price,
'zone_close_start_price': zone_close_start_price,
'zone_close_end_price': zone_close_end_price,
'zone_top_start_price': zone_top_start_price
})
# Check Zones
in_close_zone = (price >= zone_close_start_price and price <= zone_close_end_price)
in_hedge_zone = (price <= zone_bottom_limit_price) or (price >= zone_top_start_price)
# --- Execute Logic ---
if in_close_zone:
logging.info(f"ZONE: CLOSE ({price:.2f} in {zone_close_start_price:.2f}-{zone_close_end_price:.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)
# --- SOFT START LOGIC (Bottom Zone Only) ---
# If in Bottom Zone, opening a NEW Short (SELL), and current position is 0 -> Cut size by 50%
if (price <= zone_bottom_limit_price) and (current_pos_size == 0) and (calc['action'] == "SELL"):
logging.info(f"🔰 SOFT START: Reducing initial hedge size by 50% in Bottom Zone.")
trade_size = round_to_sz_decimals(trade_size * 0.5, 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})")
elif trade_size > 0:
logging.info(f"⚡ THRESHOLD TRIGGERED ({diff_abs:.4f} >= {rebalance_threshold:.4f}). In Hedge Zone.")
is_buy = (calc['action'] == "BUY")
self.place_limit_order(COIN_SYMBOL, is_buy, trade_size, price)
else:
logging.info("Trade size rounds to 0. Skipping.")
else:
logging.info("Trade size rounds to 0. Skipping.")
logging.info(f"Idle. Diff {diff_abs:.4f} < Threshold {rebalance_threshold:.4f}. In Hedge Zone.")
else:
logging.info(f"Idle. Diff {diff_abs:.4f} < Threshold {rebalance_threshold:.4f}")
# MIDDLE ZONE (IDLE)
pct_position = (price - clp_low_range) / range_width
logging.info(f"Idle. In Middle Zone ({pct_position*100:.1f}%). No Actions.")
time.sleep(CHECK_INTERVAL)