179 lines
7.5 KiB
Python
179 lines
7.5 KiB
Python
import logging
|
|
import time
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from hyperliquid.info import Info
|
|
from hyperliquid.utils import constants
|
|
|
|
from strategies.base_strategy import BaseStrategy
|
|
|
|
class CopyTraderStrategy(BaseStrategy):
|
|
"""
|
|
An event-driven strategy that monitors a target wallet address and
|
|
copies its trades for a specific set of allowed coins, using
|
|
per-coin size and leverage settings.
|
|
"""
|
|
def __init__(self, strategy_name: str, params: dict, trade_signal_queue, shared_status: dict = None):
|
|
super().__init__(strategy_name, params, trade_signal_queue, shared_status)
|
|
|
|
self.target_address = self.params.get("target_address", "").lower()
|
|
|
|
self.coins_to_copy = self.params.get("coins_to_copy", {})
|
|
self.allowed_coins = list(self.coins_to_copy.keys())
|
|
|
|
if not self.target_address:
|
|
logging.error("No 'target_address' specified in parameters for copy trader.")
|
|
raise ValueError("target_address is required")
|
|
if not self.allowed_coins:
|
|
logging.warning("No 'coins_to_copy' configured. This strategy will not copy any trades.")
|
|
|
|
self.info = None # Will be initialized in the run loop
|
|
|
|
# --- FIX: Set initial state to "WAIT" ---
|
|
self.current_signal = "WAIT"
|
|
|
|
# Record the strategy's start time to ignore historical data
|
|
self.start_time_utc = datetime.now(timezone.utc)
|
|
logging.info(f"Strategy initialized. Ignoring all trades before {self.start_time_utc.isoformat()}")
|
|
|
|
def calculate_signals(self, df):
|
|
# This strategy is event-driven, so it does not use polling-based signal calculation.
|
|
pass
|
|
|
|
def on_fill_message(self, message):
|
|
"""
|
|
This is the callback function that gets triggered by the WebSocket
|
|
every time the monitored address has an event.
|
|
"""
|
|
try:
|
|
channel = message.get("channel")
|
|
if channel not in ("user", "userFills", "userEvents"):
|
|
return
|
|
|
|
data = message.get("data")
|
|
if not data:
|
|
return
|
|
|
|
fills = data.get("fills", [])
|
|
if not fills:
|
|
return
|
|
|
|
user_address = data.get("user", "").lower()
|
|
|
|
if user_address != self.target_address:
|
|
return
|
|
|
|
logging.debug(f"Received {len(fills)} fill(s) for user {user_address}")
|
|
|
|
for fill in fills:
|
|
# Check if the trade is new or historical
|
|
trade_time = datetime.fromtimestamp(fill['time'] / 1000, tz=timezone.utc)
|
|
if trade_time < self.start_time_utc:
|
|
logging.info(f"Ignoring stale/historical trade from {trade_time.isoformat()}")
|
|
continue
|
|
|
|
coin = fill.get('coin')
|
|
|
|
if coin in self.allowed_coins:
|
|
side = fill.get('side')
|
|
price = float(fill.get('px'))
|
|
|
|
signal = "HOLD"
|
|
if side == "B":
|
|
signal = "BUY"
|
|
elif side == "A":
|
|
signal = "SELL"
|
|
|
|
coin_config = self.coins_to_copy.get(coin)
|
|
if not coin_config or not coin_config.get("size"):
|
|
logging.warning(f"No trade size specified for {coin}. Ignoring fill.")
|
|
continue
|
|
|
|
# --- 1. Create the trade-specific config ---
|
|
trade_params = self.params.copy()
|
|
trade_params.update(coin_config)
|
|
trade_config = {
|
|
"agent": self.params.get("agent"),
|
|
"parameters": trade_params
|
|
}
|
|
|
|
# --- 2. (PRIORITY) Put the signal on the queue for the executor ---
|
|
self.trade_signal_queue.put({
|
|
"strategy_name": self.strategy_name,
|
|
"signal": signal,
|
|
"coin": coin,
|
|
"signal_price": price,
|
|
"config": trade_config
|
|
})
|
|
|
|
# --- 3. (Secondary) Update internal state and log ---
|
|
self.current_signal = signal
|
|
self.signal_price = price
|
|
self.last_signal_change_utc = trade_time.isoformat()
|
|
self._save_status() # Update the dashboard status file
|
|
|
|
logging.warning(f"Copy trade signal SENT for {coin}: {signal} @ {price}, Size: {coin_config['size']}")
|
|
logging.info(f"Source trade logged: {json.dumps(fill)}")
|
|
|
|
else:
|
|
logging.info(f"Ignoring fill for unmonitored coin: {coin}")
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error in on_fill_message: {e}", exc_info=True)
|
|
|
|
def _connect_and_subscribe(self):
|
|
"""
|
|
Establishes a new WebSocket connection and subscribes to the userFills channel.
|
|
"""
|
|
try:
|
|
logging.info("Connecting to Hyperliquid WebSocket...")
|
|
self.info = Info(constants.MAINNET_API_URL, skip_ws=False)
|
|
subscription = {"type": "userFills", "user": self.target_address}
|
|
self.info.subscribe(subscription, self.on_fill_message)
|
|
logging.info(f"Subscribed to 'userFills' for target address: {self.target_address}")
|
|
return True
|
|
except Exception as e:
|
|
logging.error(f"Failed to connect or subscribe: {e}")
|
|
self.info = None
|
|
return False
|
|
|
|
def run_event_loop(self):
|
|
"""
|
|
This method overrides the default polling loop. It establishes a
|
|
persistent WebSocket connection and runs a watchdog to ensure
|
|
it stays connected.
|
|
"""
|
|
if not self._connect_and_subscribe():
|
|
# If connection fails on start, wait 60s before letting the process restart
|
|
time.sleep(60)
|
|
return
|
|
|
|
# --- ADDED: Save the initial "WAIT" status ---
|
|
self._save_status()
|
|
|
|
while True:
|
|
try:
|
|
time.sleep(15) # Check the connection every 15 seconds
|
|
|
|
if self.info is None or not self.info.ws_manager.is_alive():
|
|
logging.error(f"WebSocket connection lost. Attempting to reconnect...")
|
|
|
|
if self.info and self.info.ws_manager:
|
|
try:
|
|
self.info.ws_manager.stop()
|
|
except Exception as e:
|
|
logging.error(f"Error stopping old ws_manager: {e}")
|
|
|
|
if not self._connect_and_subscribe():
|
|
logging.error("Reconnect failed, will retry in 15s.")
|
|
else:
|
|
logging.info("Successfully reconnected to WebSocket.")
|
|
# After reconnecting, save the current status again
|
|
self._save_status()
|
|
else:
|
|
logging.debug("Watchdog check: WebSocket connection is active.")
|
|
|
|
except Exception as e:
|
|
logging.error(f"An error occurred in the watchdog loop: {e}", exc_info=True)
|
|
|