chore: format examples

Author: Anton Sauchyk

learning-examples/accounts_advanced_filter.py

@@ -5,7 +5,7 @@
 import hashlib
 import base58
 
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
 from dotenv import load_dotenv
 from generated import geyser_pb2, geyser_pb2_grpc
@@ -19,48 +19,52 @@
 # Pump.fun program ID - this is the main program that manages all pump.fun tokens
 PUMP_FUN_PROGRAM_ID = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"
 
+
 def calculate_discriminator(account_name: str) -> int:
     """
     Calculate the 8-byte discriminator for an Anchor account type.
-    
+
     The discriminator is the first 8 bytes of SHA-256 hash of "account:{account_name}"
     This is used to identify different types of accounts in Anchor programs.
-    
+
     Args:
         account_name: The name of the account struct (e.g., "BondingCurve")
-    
+
     Returns:
         The discriminator as a little-endian integer
     """
     hash_input = f"account:{account_name}"
     hash_result = hashlib.sha256(hash_input.encode()).digest()
     discriminator_bytes = hash_result[:8]
-    return int.from_bytes(discriminator_bytes, 'little')
+    return int.from_bytes(discriminator_bytes, "little")
+
 
 # Calculate the bonding curve discriminator
 BONDING_CURVE_DISCRIMINATOR = calculate_discriminator("BondingCurve")
 print(f"Bonding Curve Discriminator: {BONDING_CURVE_DISCRIMINATOR}")
 
+
 async def main():
     """
     Main function that connects to Geyser and monitors pump.fun bonding curve updates.
-    
+
     This will receive updates for:
     - New token creation (when bonding curves are created)
     - Trading activity (buys/sells that change bonding curve state)
     - Token progression through the bonding curve
     """
-    
+
     async with grpc.aio.secure_channel(
         GEYSER_ENDPOINT,
         grpc.composite_channel_credentials(
             grpc.ssl_channel_credentials(),
             grpc.metadata_call_credentials(
-                lambda context, callback: callback((('x-token', GEYSER_API_TOKEN),), None)
-            )
-        )
+                lambda context, callback: callback(
+                    (("x-token", GEYSER_API_TOKEN),), None
+                )
+            ),
+        ),
     ) as channel:
-        
         # Create the gRPC stub for making requests
         stub = geyser_pb2_grpc.GeyserStub(channel)
 
@@ -74,10 +78,10 @@ async def main():
                         geyser_pb2.SubscribeRequestFilterAccountsFilter(
                             memcmp=geyser_pb2.SubscribeRequestFilterAccountsFilterMemcmp(
                                 offset=0,  # Discriminator is at the beginning of account data
-                                bytes=BONDING_CURVE_DISCRIMINATOR.to_bytes(8, "little")
+                                bytes=BONDING_CURVE_DISCRIMINATOR.to_bytes(8, "little"),
                             )
                         ),
-                    ]
+                    ],
                 )
             },
             # Use PROCESSED commitment for faster updates (vs CONFIRMED or FINALIZED)
@@ -91,32 +95,37 @@ async def main():
         update_count = 0
         async for response in stub.Subscribe(iter([request])):
             update_count += 1
-            
+
             if response.account:
                 account_info = response.account.account
-                
+
                 print(f"📊 Update #{update_count}")
-                print(f"   Account: {base58.b58encode(account_info.pubkey).decode('utf-8')}")
+                print(
+                    f"   Account: {base58.b58encode(account_info.pubkey).decode('utf-8')}"
+                )
                 print(f"   Lamports: {account_info.lamports:,}")
                 print(f"   Data Size: {len(account_info.data)} bytes")
-                print(f"   Owner: {base58.b58encode(account_info.owner).decode('utf-8')}")
-                
+                print(
+                    f"   Owner: {base58.b58encode(account_info.owner).decode('utf-8')}"
+                )
+
                 # You can decode the bonding curve data here to get:
                 # - Virtual SOL reserves
                 # - Virtual token reserves
                 # - Real token reserves
                 # - Token mint address
                 # - Bonding curve progress
-                
+
                 print("---")
             else:
                 print(f"⚠️  Received non-account update: {response}")
                 print("---")
 
+
 if __name__ == "__main__":
     try:
         asyncio.run(main())
     except KeyboardInterrupt:
         print("\n🛑 Stopping monitor...")
     except Exception as e:
-        print(f"❌ Error: {e}")
+        print(f"❌ Error: {e}")

learning-examples/accounts_basic_filter.py

@@ -4,7 +4,7 @@
 import grpc
 import base58
 
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
 from dotenv import load_dotenv
 from generated import geyser_pb2, geyser_pb2_grpc
@@ -19,20 +19,23 @@
 # We will subscribe to updates for this specific account.
 RAYDIUM_MARKET_ADDRESS = "2AXXcN6oN9bBT5owwmTH53C7QHUXvhLeu718Kqt8rvY2"
 
+
 async def main():
     """
     Main function that connects to Geyser and monitors a specific account.
-    
+
     This will receive updates whenever the specified account's data changes.
     """
     async with grpc.aio.secure_channel(
         GEYSER_ENDPOINT,
         grpc.composite_channel_credentials(
             grpc.ssl_channel_credentials(),
             grpc.metadata_call_credentials(
-                lambda context, callback: callback((('x-token', GEYSER_API_TOKEN),), None)
-            )
-        )
+                lambda context, callback: callback(
+                    (("x-token", GEYSER_API_TOKEN),), None
+                )
+            ),
+        ),
     ) as channel:
         stub = geyser_pb2_grpc.GeyserStub(channel)
 
@@ -41,7 +44,7 @@ async def main():
             accounts={
                 "accounts_filter": geyser_pb2.SubscribeRequestFilterAccounts(
                     # Provide the specific account public key to monitor
-                    account=[RAYDIUM_MARKET_ADDRESS] 
+                    account=[RAYDIUM_MARKET_ADDRESS]
                 )
             },
             # Use PROCESSED commitment for faster updates (vs CONFIRMED or FINALIZED)
@@ -55,20 +58,25 @@ async def main():
         update_count = 0
         async for response in stub.Subscribe(iter([request])):
             update_count += 1
-            
+
             if response.account:
                 account_info = response.account.account
-                
+
                 print(f"📊 Update #{update_count}")
-                print(f"   Account: {base58.b58encode(account_info.pubkey).decode('utf-8')}")
+                print(
+                    f"   Account: {base58.b58encode(account_info.pubkey).decode('utf-8')}"
+                )
                 print(f"   Lamports: {account_info.lamports:,}")
                 print(f"   Data Size: {len(account_info.data)} bytes")
-                print(f"   Owner: {base58.b58encode(account_info.owner).decode('utf-8')}")
+                print(
+                    f"   Owner: {base58.b58encode(account_info.owner).decode('utf-8')}"
+                )
                 print("---")
             else:
                 print(f"⚠️  Received non-account update: {response}")
                 print("---")
 
+
 if __name__ == "__main__":
     try:
         asyncio.run(main())

learning-examples/blocks_meta_subscription.py

@@ -3,7 +3,7 @@
 import sys
 import grpc
 
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
 from dotenv import load_dotenv
 from generated import geyser_pb2, geyser_pb2_grpc
@@ -14,27 +14,30 @@
 GEYSER_ENDPOINT = os.getenv("GEYSER_ENDPOINT")
 GEYSER_API_TOKEN = os.getenv("GEYSER_API_TOKEN")
 
+
 async def main():
     """
     Main function that connects to Geyser and monitors block metadata for all blocks.
-    
+
     This will receive metadata for every block processed by the Geyser node.
     """
     async with grpc.aio.secure_channel(
         GEYSER_ENDPOINT,
         grpc.composite_channel_credentials(
             grpc.ssl_channel_credentials(),
             grpc.metadata_call_credentials(
-                lambda context, callback: callback((('x-token', GEYSER_API_TOKEN),), None)
-            )
-        )
+                lambda context, callback: callback(
+                    (("x-token", GEYSER_API_TOKEN),), None
+                )
+            ),
+        ),
     ) as channel:
         stub = geyser_pb2_grpc.GeyserStub(channel)
 
         # Basic blocks_meta subscription for all blocks
         request = geyser_pb2.SubscribeRequest(
             blocks_meta={
-                "blocks_meta_filter": geyser_pb2.SubscribeRequestFilterBlocksMeta() # Don't have filters
+                "blocks_meta_filter": geyser_pb2.SubscribeRequestFilterBlocksMeta()  # Don't have filters
             },
             commitment=geyser_pb2.CommitmentLevel.PROCESSED,
         )
@@ -46,21 +49,24 @@ async def main():
         update_count = 0
         async for response in stub.Subscribe(iter([request])):
             update_count += 1
-            
+
             if response.block_meta:
                 block_meta_info = response.block_meta
-                
+
                 print(f"📦 Block Meta Update #{update_count}")
                 print(f"   Slot: {block_meta_info.slot}")
                 print(f"   Blockhash: {block_meta_info.blockhash}")
                 print(f"   Parent Slot: {block_meta_info.parent_slot}")
                 print(f"   Block Time: {block_meta_info.block_time.timestamp}")
-                print(f"   Transaction Count: {block_meta_info.executed_transaction_count}")
+                print(
+                    f"   Transaction Count: {block_meta_info.executed_transaction_count}"
+                )
                 print("---")
             else:
                 print(f"⚠️  Received non-block_meta update: {response}")
                 print("---")
 
+
 if __name__ == "__main__":
     try:
         asyncio.run(main())

learning-examples/blocks_subscription.py

@@ -3,7 +3,7 @@
 import sys
 import grpc
 
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
 from dotenv import load_dotenv
 from generated import geyser_pb2, geyser_pb2_grpc
@@ -17,11 +17,12 @@
 # Pump.fun program ID - we will subscribe to blocks containing its transactions
 PUMP_FUN_PROGRAM_ID = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"
 
+
 async def main():
     """
     Main function that connects to Geyser and monitors blocks containing transactions
     for the specified pump.fun program.
-    
+
     This will receive updates for any block that includes a transaction interacting
     with the PUMP_FUN_PROGRAM_ID.
     """
@@ -30,9 +31,11 @@ async def main():
         grpc.composite_channel_credentials(
             grpc.ssl_channel_credentials(),
             grpc.metadata_call_credentials(
-                lambda context, callback: callback((('x-token', GEYSER_API_TOKEN),), None)
-            )
-        )
+                lambda context, callback: callback(
+                    (("x-token", GEYSER_API_TOKEN),), None
+                )
+            ),
+        ),
     ) as channel:
         stub = geyser_pb2_grpc.GeyserStub(channel)
 
@@ -42,8 +45,8 @@ async def main():
                 "blocks_filter": geyser_pb2.SubscribeRequestFilterBlocks(
                     account_include=[PUMP_FUN_PROGRAM_ID],
                     include_transactions=True,  # Include transactions in which the account is involved
-                    include_accounts=False,     # Not allowed
-                    include_entries=False       # Not allowed
+                    include_accounts=False,  # Not allowed
+                    include_entries=False,  # Not allowed
                 )
             },
             commitment=geyser_pb2.CommitmentLevel.PROCESSED,
@@ -56,22 +59,25 @@ async def main():
         update_count = 0
         async for response in stub.Subscribe(iter([request])):
             update_count += 1
-            
+
             if response.block:
                 block_info = response.block
-                
+
                 print(f"📦 Block Update #{update_count}")
                 print(f"   Slot: {block_info.slot}")
                 print(f"   Blockhash: {block_info.blockhash}")
-                
+
                 if block_info.transactions:
-                    print(f"   Found {len(block_info.transactions)} transactions involving the program.")
-                
+                    print(
+                        f"   Found {len(block_info.transactions)} transactions involving the program."
+                    )
+
                 print("---")
             else:
                 print(f"⚠️  Received non-block update: {response}")
                 print("---")
 
+
 if __name__ == "__main__":
     try:
         asyncio.run(main())