Balance Ledger Streaming

service FundingAPI {
    rpc CreateBalanceLedgerSubscription(CreateBalanceLedgerSubscriptionRequest)
        returns (stream CreateBalanceLedgerSubscriptionResponse);
}

import grpc
from datetime import datetime
from google.protobuf import timestamp_pb2
from polymarket.v1 import funding_pb2
from polymarket.v1 import funding_pb2_grpc


class BalanceLedgerStreamer:
    def __init__(self, grpc_server: str = "grpc-api.preprod.polymarketexchange.com:443"):
        self.grpc_server = grpc_server
        self.access_token = None
        self.last_update_time: timestamp_pb2.Timestamp | None = None  # for resume_time

    def stream_balance_ledger(
        self,
        account: str,
        currency: str = "",
        entry_types: list | None = None,
        resume_time: timestamp_pb2.Timestamp | None = None,
    ):
        """Stream balance ledger entries for one account."""
        if not self.access_token:
            raise ValueError("Not authenticated. Please login first.")

        credentials = grpc.ssl_channel_credentials()
        channel = grpc.secure_channel(self.grpc_server, credentials)
        stub = funding_pb2_grpc.FundingAPIStub(channel)

        request = funding_pb2.CreateBalanceLedgerSubscriptionRequest(
            account=account,
            currency=currency,
            entry_types=entry_types or [],
        )
        if resume_time is not None:
            request.resume_time.CopyFrom(resume_time)

        metadata = [("authorization", f"Bearer {self.access_token}")]

        try:
            print(f"Subscribing to balance ledger for {account} (currency={currency or 'ALL'})")
            response_stream = stub.CreateBalanceLedgerSubscription(
                request, metadata=metadata
            )

            for response in response_stream:
                if not response.entries:
                    continue
                for entry in response.entries:
                    self._process_entry(entry)
                    self.last_update_time = entry.update_time

        except grpc.RpcError as e:
            print(f"gRPC error: {e.code()} - {e.details()}")
            raise
        except KeyboardInterrupt:
            print("\nStream interrupted by user")
        finally:
            channel.close()

    def _process_entry(self, entry):
        ts = entry.update_time.ToDatetime().isoformat()
        entry_type_name = funding_pb2.LedgerEntryType.Name(entry.entry_type)
        delta = float(entry.after_balance) - float(entry.before_balance)
        print(
            f"[{ts}] {entry_type_name:30s} "
            f"{entry.before_balance} -> {entry.after_balance} "
            f"(delta={delta:+.2f} {entry.currency}) "
            f"{entry.description}"
        )


if __name__ == "__main__":
    streamer = BalanceLedgerStreamer()
    # streamer.access_token = "your_access_token"  # see Authentication docs

    streamer.stream_balance_ledger(
        account="firms/ISV-Alice/accounts/alice-trading",
        currency="USD",
    )

Subscribing to balance ledger for firms/ISV-Alice/accounts/alice-trading (currency=USD)
[2026-05-02T14:30:15.123000] ORDER_EXECUTION                10000.00 -> 9474.03 (delta=-525.97 USD) Buy 100 @ 525 + commission
[2026-05-02T14:30:15.123000] COMMISSION                     9474.03 -> 9472.53 (delta=-1.50 USD) Maker fee
[2026-05-02T14:32:01.001000] DEPOSIT                        9472.53 -> 14472.53 (delta=+5000.00 USD) ACH deposit

if streamer.last_update_time is not None:
    streamer.stream_balance_ledger(
        account="firms/ISV-Alice/accounts/alice-trading",
        currency="USD",
        resume_time=streamer.last_update_time,
    )