110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Test script for hedge execution functionality
|
||
"""
|
||
|
||
import json
|
||
import sys
|
||
import os
|
||
from datetime import datetime
|
||
|
||
# Add current directory to path for imports
|
||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.append(current_dir)
|
||
|
||
from uniswap_manager import execute_hedge_sync, get_token_symbol, get_token_decimals
|
||
from web3 import Web3
|
||
from eth_account import Account
|
||
from dotenv import load_dotenv
|
||
|
||
def test_hedge_execution():
|
||
"""Test hedge execution with data from hedge_status.json"""
|
||
print("🧪 Testing Hedge Execution Functionality")
|
||
print("=" * 50)
|
||
|
||
# Load environment
|
||
load_dotenv(override=True)
|
||
|
||
# Check required environment variables
|
||
rpc_url = os.environ.get("MAINNET_RPC_URL")
|
||
private_key = os.environ.get("MAIN_WALLET_PRIVATE_KEY") or os.environ.get("PRIVATE_KEY")
|
||
|
||
if not rpc_url or not private_key:
|
||
print("❌ Missing RPC URL or Private Key in environment")
|
||
return False
|
||
|
||
# Load hedge status
|
||
try:
|
||
with open("hedge_status.json", 'r') as f:
|
||
hedge_data = json.load(f)
|
||
except Exception as e:
|
||
print(f"❌ Error loading hedge_status.json: {e}")
|
||
return False
|
||
|
||
# Find positions requiring hedges
|
||
hedge_positions = []
|
||
for position in hedge_data:
|
||
if position.get("hedge_required", False) and position.get("hedge_amount", 0) > 0:
|
||
hedge_positions.append(position)
|
||
|
||
if not hedge_positions:
|
||
print("ℹ️ No positions requiring hedges found")
|
||
return True
|
||
|
||
print(f"📊 Found {len(hedge_positions)} positions requiring hedges:")
|
||
for i, pos in enumerate(hedge_positions, 1):
|
||
print(f" {i}. Token: {pos.get('token', 'Unknown')}")
|
||
print(f" Amount: {pos.get('hedge_amount', 0):.6f}")
|
||
print(f" Reason: {pos.get('hedge_reason', 'Unknown')}")
|
||
print(f" Confidence: {pos.get('hedge_confidence', 0):.2f}")
|
||
|
||
# Initialize Web3
|
||
try:
|
||
w3 = Web3(Web3.HTTPProvider(rpc_url))
|
||
if not w3.is_connected():
|
||
print("❌ Failed to connect to RPC")
|
||
return False
|
||
|
||
account = Account.from_key(private_key)
|
||
print(f"✅ Connected to Chain ID: {w3.eth.chain_id}")
|
||
print(f"✅ Wallet: {account.address}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ Web3 initialization error: {e}")
|
||
return False
|
||
|
||
# Test with first position (dry run)
|
||
if hedge_positions:
|
||
test_pos = hedge_positions[0]
|
||
token_address = test_pos.get("token_address", "0x82aF49447D8a07e3bd95BD0d56f35241523fBab1") # Default to WETH
|
||
hedge_amount = test_pos.get("hedge_amount", 0.01)
|
||
|
||
print(f"\n🎯 Testing hedge execution for:")
|
||
print(f" Token Address: {token_address}")
|
||
print(f" Amount: {hedge_amount:.6f}")
|
||
|
||
# Test token info functions
|
||
try:
|
||
symbol = get_token_symbol(w3, token_address)
|
||
decimals = get_token_decimals(w3, token_address)
|
||
print(f" Token Symbol: {symbol}")
|
||
print(f" Token Decimals: {decimals}")
|
||
except Exception as e:
|
||
print(f"⚠️ Error getting token info: {e}")
|
||
|
||
# For dry run, we won't actually execute the hedge
|
||
print("\n🔍 DRY RUN MODE - Not executing actual hedge")
|
||
print(" To execute real hedge, set DRY_RUN = False")
|
||
|
||
# Uncomment the following lines to execute real hedge:
|
||
# DRY_RUN = False
|
||
# if not DRY_RUN:
|
||
# success = execute_hedge_sync(w3, router_contract, account, token_address, hedge_amount)
|
||
# print(f" Hedge execution result: {'✅ Success' if success else '❌ Failed'}"
|
||
|
||
print("\n✅ Hedge execution test completed successfully!")
|
||
return True
|
||
|
||
if __name__ == "__main__":
|
||
success = test_hedge_execution()
|
||
sys.exit(0 if success else 1) |