160 lines
6.7 KiB
Python
160 lines
6.7 KiB
Python
import os
|
|
import sys
|
|
import time
|
|
import json
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
from hyperliquid.info import Info
|
|
from hyperliquid.utils import constants
|
|
from dotenv import load_dotenv
|
|
import logging
|
|
|
|
from logging_utils import setup_logging
|
|
|
|
# Load .env file
|
|
load_dotenv()
|
|
|
|
class PositionMonitor:
|
|
"""
|
|
A standalone, read-only dashboard for monitoring all open perpetuals
|
|
positions, spot balances, and their associated strategies.
|
|
"""
|
|
|
|
def __init__(self, log_level: str):
|
|
setup_logging(log_level, 'PositionMonitor')
|
|
|
|
self.wallet_address = os.environ.get("MAIN_WALLET_ADDRESS")
|
|
if not self.wallet_address:
|
|
logging.error("MAIN_WALLET_ADDRESS not set in .env file. Cannot proceed.")
|
|
sys.exit(1)
|
|
|
|
self.info = Info(constants.MAINNET_API_URL, skip_ws=True)
|
|
self.managed_positions_path = os.path.join("_data", "executor_managed_positions.json")
|
|
self._lines_printed = 0
|
|
|
|
logging.info(f"Monitoring vault address: {self.wallet_address}")
|
|
|
|
def load_managed_positions(self) -> dict:
|
|
"""Loads the state of which strategy manages which position."""
|
|
if os.path.exists(self.managed_positions_path):
|
|
try:
|
|
with open(self.managed_positions_path, 'r') as f:
|
|
# Create a reverse map: {coin: strategy_name}
|
|
data = json.load(f)
|
|
return {v['coin']: k for k, v in data.items()}
|
|
except (IOError, json.JSONDecodeError):
|
|
logging.warning("Could not read managed positions file.")
|
|
return {}
|
|
|
|
def run(self):
|
|
"""Main loop to continuously refresh the dashboard."""
|
|
try:
|
|
while True:
|
|
self.display_dashboard()
|
|
time.sleep(5) # Refresh every 5 seconds
|
|
except KeyboardInterrupt:
|
|
logging.info("Position monitor stopped.")
|
|
|
|
def display_dashboard(self):
|
|
"""Fetches all data and draws the dashboard without blinking."""
|
|
if self._lines_printed > 0:
|
|
print(f"\x1b[{self._lines_printed}A", end="")
|
|
|
|
output_lines = []
|
|
try:
|
|
perp_state = self.info.user_state(self.wallet_address)
|
|
spot_state = self.info.spot_user_state(self.wallet_address)
|
|
coin_to_strategy_map = self.load_managed_positions()
|
|
|
|
output_lines.append(f"--- Live Position Monitor for {self.wallet_address[:6]}...{self.wallet_address[-4:]} ---")
|
|
|
|
# --- 1. Perpetuals Account Summary ---
|
|
margin_summary = perp_state.get('marginSummary', {})
|
|
account_value = float(margin_summary.get('accountValue', 0))
|
|
margin_used = float(margin_summary.get('totalMarginUsed', 0))
|
|
utilization = (margin_used / account_value) * 100 if account_value > 0 else 0
|
|
|
|
output_lines.append("\n--- Perpetuals Account Summary ---")
|
|
output_lines.append(f" Account Value: ${account_value:,.2f} | Margin Used: ${margin_used:,.2f} | Utilization: {utilization:.2f}%")
|
|
|
|
# --- 2. Spot Balances Summary ---
|
|
output_lines.append("\n--- Spot Balances ---")
|
|
spot_balances = spot_state.get('balances', [])
|
|
if not spot_balances:
|
|
output_lines.append(" No spot balances found.")
|
|
else:
|
|
balances_str = ", ".join([f"{b.get('coin')}: {float(b.get('total', 0)):,.4f}" for b in spot_balances if float(b.get('total', 0)) > 0])
|
|
output_lines.append(f" {balances_str}")
|
|
|
|
# --- 3. Open Positions Table ---
|
|
output_lines.append("\n--- Open Perpetual Positions ---")
|
|
positions = perp_state.get('assetPositions', [])
|
|
open_positions = [p for p in positions if p.get('position') and float(p['position'].get('szi', 0)) != 0]
|
|
|
|
if not open_positions:
|
|
output_lines.append(" No open perpetual positions found.")
|
|
output_lines.append("") # Add a line for stable refresh
|
|
else:
|
|
self.build_positions_table(open_positions, coin_to_strategy_map, output_lines)
|
|
|
|
except Exception as e:
|
|
output_lines = [f"An error occurred: {e}"]
|
|
|
|
final_output = "\n".join(output_lines) + "\n\x1b[J" # \x1b[J clears to end of screen
|
|
print(final_output, end="")
|
|
|
|
self._lines_printed = len(output_lines)
|
|
sys.stdout.flush()
|
|
|
|
def build_positions_table(self, positions: list, coin_to_strategy_map: dict, output_lines: list):
|
|
"""Builds the text for the positions summary table."""
|
|
header = f"| {'Strategy':<25} | {'Coin':<6} | {'Side':<5} | {'Size':>15} | {'Entry Price':>12} | {'Mark Price':>12} | {'PNL':>15} | {'Leverage':>10} |"
|
|
output_lines.append(header)
|
|
output_lines.append("-" * len(header))
|
|
|
|
for position in positions:
|
|
pos = position.get('position', {})
|
|
coin = pos.get('coin', 'Unknown')
|
|
size = float(pos.get('szi', 0))
|
|
entry_px = float(pos.get('entryPx', 0))
|
|
mark_px = float(pos.get('markPx', 0))
|
|
unrealized_pnl = float(pos.get('unrealizedPnl', 0))
|
|
|
|
# Get leverage
|
|
position_value = float(pos.get('positionValue', 0))
|
|
margin_used = float(pos.get('marginUsed', 0))
|
|
leverage = (position_value / margin_used) if margin_used > 0 else 0
|
|
|
|
side_text = "LONG" if size > 0 else "SHORT"
|
|
pnl_sign = "+" if unrealized_pnl >= 0 else ""
|
|
|
|
# Find the strategy that owns this coin
|
|
strategy_name = coin_to_strategy_map.get(coin, "Unmanaged")
|
|
|
|
# Format all values as strings
|
|
strategy_str = f"{strategy_name:<25}"
|
|
coin_str = f"{coin:<6}"
|
|
side_str = f"{side_text:<5}"
|
|
size_str = f"{size:>15.4f}"
|
|
entry_str = f"${entry_px:>11,.2f}"
|
|
mark_str = f"${mark_px:>11,.2f}"
|
|
pnl_str = f"{pnl_sign}${unrealized_pnl:>14,.2f}"
|
|
lev_str = f"{leverage:>9.1f}x"
|
|
|
|
output_lines.append(f"| {strategy_str} | {coin_str} | {side_str} | {size_str} | {entry_str} | {mark_str} | {pnl_str} | {lev_str} |")
|
|
|
|
output_lines.append("-" * len(header))
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Monitor a Hyperliquid wallet's positions in real-time.")
|
|
parser.add_argument(
|
|
"--log-level",
|
|
default="normal",
|
|
choices=['off', 'normal', 'debug'],
|
|
help="Set the logging level for the script."
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
monitor = PositionMonitor(log_level=args.log_level)
|
|
monitor.run()
|