new strategies

position_monitor.py

@@ -0,0 +1,159 @@
+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()