driftpy

import os import asyncio from dotenv import load_dotenv from solders.keypair import Keypair from solana.rpc.async_api import AsyncClient from anchorpy import Wallet from driftpy.drift_client import DriftClient from driftpy.keypair import load_keypair async def main(): load_dotenv() # Load .env file for environment variables # It's safer to use os.environ.get with a default empty string for sensitive info # and then handle the case where it's missing. secret_key_str = os.environ.get("PRIVATE_KEY", "") rpc_url = os.environ.get("RPC_URL", "https://api.devnet.solana.com") if not secret_key_str: print("Error: PRIVATE_KEY environment variable not set.") return if not rpc_url: print("Error: RPC_URL environment variable not set.") return # Ensure the secret key is in the correct format (list of integers) try: secret_key_bytes = bytes(list(map(int, secret_key_str.strip('[]').split(',')))) keypair = Keypair.from_secret_key(secret_key_bytes) except Exception as e: print(f"Error loading keypair: {e}. Ensure PRIVATE_KEY is a comma-separated list of integers in your .env file.") return wallet = Wallet(keypair) connection = AsyncClient(rpc_url) # Initialize DriftClient for 'devnet' # For mainnet, use 'mainnet' drift_client = DriftClient( connection, wallet, "devnet", ) print("Subscribing to Drift Client...") await drift_client.subscribe() # Crucial: Must subscribe before fetching data print("Drift Client subscribed.") # Optional: Register a default sub-account if not already done # await drift_client.add_user(0) # Example: Fetch user stats (after subscription) user_stats = await drift_client.get_user_stats_account() print(f"User Stats: {user_stats}") await drift_client.unsubscribe() await connection.close() if __name__ == "__main__": asyncio.run(main())