From 8d2137cbbf7fd972c8b3e695648df86e56aff51d Mon Sep 17 00:00:00 2001
From: Hilari Moragrega <hilari@hilarimoragrega.com>
Date: Tue, 25 Aug 2026 17:42:44 +0200
Subject: [PATCH 1/2] feat: example to stream oracle prices from chain

---
 .../chain_client/8_ChainStreamOraclePrices.py | 59 +++++++++++++++++++
 1 file changed, 59 insertions(+)
 create mode 100644 examples/chain_client/8_ChainStreamOraclePrices.py

diff --git a/examples/chain_client/8_ChainStreamOraclePrices.py b/examples/chain_client/8_ChainStreamOraclePrices.py
new file mode 100644
index 00000000..d4534a5c
--- /dev/null
+++ b/examples/chain_client/8_ChainStreamOraclePrices.py
@@ -0,0 +1,59 @@
+import asyncio
+from typing import Any, Dict
+
+from grpc import RpcError
+
+from pyinjective.async_client_v2 import AsyncClient
+from pyinjective.core.network import Network
+
+ORACLE_TYPE = "chainlinkdatastreams"
+ORACLE_SYMBOLS = {
+    "INJ": "0x000344d7a7d81f051ee273a63f94f8bef7d44ca89aa03e0c5bf4d085df19adb6",
+    "USDC": "0x00038f83323b6b08116d1614cf33a9bd71ab5e0abf0c9f1b783a74a43e7bd992",
+}
+
+
+async def oracle_price_event_processor(event: Dict[str, Any]):
+    for oracle_price in event.get("oraclePrices", []):
+        print(
+            {
+                "blockHeight": event["blockHeight"],
+                "blockTime": event["blockTime"],
+                **oracle_price,
+            }
+        )
+
+
+def stream_error_processor(exception: RpcError):
+    print(f"There was an error listening to oracle price updates ({exception})")
+
+
+def stream_closed_processor():
+    print("The oracle price updates stream has been closed")
+
+
+async def main() -> None:
+    network = Network.mainnet()
+
+    client = AsyncClient(network)
+    composer = await client.composer()
+
+    print(f"Streaming {ORACLE_TYPE} prices for {', '.join(ORACLE_SYMBOLS)}")
+    # To receive all oracle price updates, use composer.chain_stream_oracle_price_filter() without symbols.
+    oracle_price_filter = composer.chain_stream_oracle_price_filter(symbols=list(ORACLE_SYMBOLS.values()))
+
+    task = asyncio.get_event_loop().create_task(
+        client.listen_chain_stream_updates(
+            callback=oracle_price_event_processor,
+            on_end_callback=stream_closed_processor,
+            on_status_callback=stream_error_processor,
+            oracle_price_filter=oracle_price_filter,
+        )
+    )
+
+    await asyncio.sleep(delay=60)
+    task.cancel()
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

From 0d0324f566d4fe0f5357be660d7dda937f82806c Mon Sep 17 00:00:00 2001
From: Hilari Moragrega <hilari@hilarimoragrega.com>
Date: Tue, 25 Aug 2026 17:54:28 +0200
Subject: [PATCH 2/2] clean stop

---
 examples/chain_client/8_ChainStreamOraclePrices.py | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/examples/chain_client/8_ChainStreamOraclePrices.py b/examples/chain_client/8_ChainStreamOraclePrices.py
index d4534a5c..26ff6607 100644
--- a/examples/chain_client/8_ChainStreamOraclePrices.py
+++ b/examples/chain_client/8_ChainStreamOraclePrices.py
@@ -1,4 +1,5 @@
 import asyncio
+from contextlib import suppress
 from typing import Any, Dict
 
 from grpc import RpcError
@@ -51,8 +52,13 @@ async def main() -> None:
         )
     )
 
-    await asyncio.sleep(delay=60)
-    task.cancel()
+    try:
+        await asyncio.sleep(delay=60)
+    finally:
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+        await client.close_chain_stream_channel()
 
 
 if __name__ == "__main__":