Add the full code

learning-examples/fetch_price.py

@@ -0,0 +1,67 @@
+import asyncio
+import struct
+import sys
+import os
+from typing import Final
+
+from construct import Struct, Int64ul, Flag
+from solana.rpc.async_api import AsyncClient
+from solders.pubkey import Pubkey
+
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+from config import RPC_ENDPOINT
+
+LAMPORTS_PER_SOL: Final[int] = 1_000_000_000
+TOKEN_DECIMALS: Final[int] = 6
+CURVE_ADDRESS: Final[str] = "6GXfUqrmPM4VdN1NoDZsE155jzRegJngZRjMkGyby7do"
+
+# Here and later all the discriminators are precalculated. See learning-examples/discriminator.py
+EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack("<Q", 6966180631402821399)
+
+class BondingCurveState:
+    _STRUCT = Struct(
+        "virtual_token_reserves" / Int64ul,
+        "virtual_sol_reserves" / Int64ul,
+        "real_token_reserves" / Int64ul,
+        "real_sol_reserves" / Int64ul,
+        "token_total_supply" / Int64ul,
+        "complete" / Flag
+    )
+
+    def __init__(self, data: bytes) -> None:
+        parsed = self._STRUCT.parse(data[8:])
+        self.__dict__.update(parsed)
+
+async def get_bonding_curve_state(conn: AsyncClient, curve_address: Pubkey) -> BondingCurveState:
+    response = await conn.get_account_info(curve_address)
+    if not response.value or not response.value.data:
+        raise ValueError("Invalid curve state: No data")
+
+    data = response.value.data
+    if data[:8] != EXPECTED_DISCRIMINATOR:
+        raise ValueError("Invalid curve state discriminator")
+
+    return BondingCurveState(data)
+
+def calculate_bonding_curve_price(curve_state: BondingCurveState) -> float:
+    if curve_state.virtual_token_reserves <= 0 or curve_state.virtual_sol_reserves <= 0:
+        raise ValueError("Invalid reserve state")
+
+    return (curve_state.virtual_sol_reserves / LAMPORTS_PER_SOL) / (curve_state.virtual_token_reserves / 10 ** TOKEN_DECIMALS)
+
+async def main() -> None:
+    try:
+        async with AsyncClient(RPC_ENDPOINT) as conn:
+            curve_address = Pubkey.from_string(CURVE_ADDRESS)
+            bonding_curve_state = await get_bonding_curve_state(conn, curve_address)
+            token_price_sol = calculate_bonding_curve_price(bonding_curve_state)
+
+            print("Token price:")
+            print(f"  {token_price_sol:.10f} SOL")
+    except ValueError as e:
+        print(f"Error: {e}")
+    except Exception as e:
+        print(f"An unexpected error occurred: {e}")
+
+if __name__ == "__main__":
+    asyncio.run(main())