Balances, fees and reconciliation

from decimal import Decimal

import grpc
from polymarket_auth import PolymarketClient  # published on /connect#authentication

# protoc output from the protos your integration lead sent you. The owning services
# are not published — check the stub names against your own copy.
from pmx_protos import balance_pb2, balance_pb2_grpc
from pmx_protos import ledger_pb2, ledger_pb2_grpc

# The gRPC target is GRPC_TARGET in polymarket_auth.py — one place, both environments.
pmx = PolymarketClient()

FIRM_ACCOUNT = "firms/.../accounts/..."         # your firm account
PARTICIPANT_ACCOUNT = "firms/.../accounts/..."  # provisionedAccount, recorded at onboarding
PARTICIPANT_ID = "..."                          # participantId from the kyc.approved webhook
OPEN_SHORT_CONTRACTS = 0                        # from your own position store
ACCRUED_FEES = "0.00"                           # USD, from your own fee ledger
REPLAY_DEADLINE_SECONDS = 60                    # yours: no replay-to-live signal is published


def newest_after_balance(channel, account: str, participant_id: str) -> Decimal:
    """The workaround for a participant clearing account.

    There is no point-read RPC for a participant account balance, so this
    subscribes to the balance ledger, drains the replay phase, and keeps the
    newest entry's afterBalance.
    """
    stub = ledger_pb2_grpc.BalanceLedgerStub(channel)
    stream = stub.CreateBalanceLedgerSubscription(
        ledger_pb2.CreateBalanceLedgerSubscriptionRequest(account=account),
        # Account-scoped: x-participant-id is required on this surface.
        metadata=pmx.metadata(participant_id=participant_id),
        timeout=REPLAY_DEADLINE_SECONDS,
    )
    newest = None
    try:
        for entry in stream:
            # Replay ordering is not published, so this keeps the LAST entry the
            # replay phase delivered. Confirm the ordering before you trust it.
            newest = entry
    except grpc.RpcError as err:
        # No published replay-to-live signal: a quiet stream is how you detect the
        # drain. Anything else is a real failure — never swallow it.
        if err.code() is not grpc.StatusCode.DEADLINE_EXCEEDED:
            raise
    # proto3: an absent field and a zero are indistinguishable. Require the field.
    if newest is None or not newest.HasField("after_balance"):
        raise RuntimeError(f"no balance-ledger entry carrying afterBalance for {account}")
    return Decimal(newest.after_balance)


def spendable(account: str, participant_id: str) -> Decimal:
    with pmx.channel() as channel:
        balances = balance_pb2_grpc.AccountBalanceStub(channel)
        try:
            b = balances.GetAccountBalance(
                balance_pb2.GetAccountBalanceRequest(account=account),
                # Account-scoped read: x-participant-id is required.
                metadata=pmx.metadata(participant_id=participant_id),
            )
            balance = Decimal(b.balance)
            ceilings = [Decimal(b.buying_power), Decimal(b.excess_capital)]
        except grpc.RpcError as err:
            # InvalidArgument: invalid account — expected on a participant clearing account.
            if "invalid account" not in (err.details() or ""):
                raise
            balance = newest_after_balance(channel, account, participant_id)
            ceilings = []  # no ceilings available on this path; the reserve below is all you have

    locked = Decimal("1.00") * OPEN_SHORT_CONTRACTS  # your own reserve, see the warning above
    # The collateral term is NOT in this sum: no field exposes it. See the warning above.
    return min([balance - locked - Decimal(ACCRUED_FEES), *ceilings])


print(spendable(FIRM_ACCOUNT, PARTICIPANT_ID))
print(spendable(PARTICIPANT_ACCOUNT, PARTICIPANT_ID))

// npm i decimal.js @grpc/grpc-js
import Decimal from "decimal.js";
import * as grpc from "@grpc/grpc-js";
import { PolymarketClient } from "./polymarketAuth"; // published on /connect#authentication

// protoc output from the protos your integration lead sent you. The owning services
// are not published — check these names against your own copy.
import { AccountBalanceClient } from "./gen/balance";
import { BalanceLedgerClient } from "./gen/ledger";

// The gRPC target is GRPC_TARGET in polymarketAuth.ts — one place, both environments.
const pmx = new PolymarketClient();

const PARTICIPANT_ACCOUNT = "firms/.../accounts/..."; // provisionedAccount, recorded at onboarding
const PARTICIPANT_ID = "..."; // participantId from the kyc.approved webhook
const OPEN_SHORT_CONTRACTS = 0; // from your own position store
const ACCRUED_FEES = "0.00"; // USD, from your own fee ledger
const REPLAY_DEADLINE_MS = 60_000; // yours: no replay-to-live signal is published

/**
 * The workaround for a participant clearing account. There is no point-read RPC
 * for a participant account balance, so this subscribes to the balance ledger,
 * drains the replay phase, and keeps the newest entry's afterBalance.
 */
async function newestAfterBalance(channel: grpc.Channel, account: string): Promise<string> {
  const stream = new BalanceLedgerClient(channel).createBalanceLedgerSubscription(
    { account },
    // Account-scoped: x-participant-id is required on this surface.
    await pmx.metadata(PARTICIPANT_ID),
    { deadline: Date.now() + REPLAY_DEADLINE_MS },
  );

  let newest: { afterBalance?: string } | undefined;
  try {
    // Replay ordering is not published, so this keeps the LAST entry the replay
    // phase delivered. Confirm the ordering before you trust it.
    for await (const entry of stream) newest = entry;
  } catch (err: any) {
    // No published replay-to-live signal: a quiet stream is how you detect the drain.
    // Anything else is a real failure — never swallow it.
    if (err?.code !== grpc.status.DEADLINE_EXCEEDED) throw err;
  }
  if (newest?.afterBalance === undefined) {
    throw new Error(`no balance-ledger entry carrying afterBalance for ${account}`);
  }
  return newest.afterBalance;
}

/** Returns a decimal string. Money never touches a binary float here. */
export async function spendable(account: string): Promise<string> {
  const channel = pmx.channel();
  let balance: string;
  let ceilings: string[] = [];
  try {
    const b = await new AccountBalanceClient(channel).getAccountBalance(
      { account },
      // Account-scoped read: x-participant-id is required.
      await pmx.metadata(PARTICIPANT_ID),
    );
    balance = b.balance;
    ceilings = [b.buyingPower, b.excessCapital];
  } catch (err: any) {
    // InvalidArgument: invalid account — expected on a participant clearing account.
    if (!String(err?.details ?? err).includes("invalid account")) throw err;
    balance = await newestAfterBalance(channel, account);
  }

  // Exact decimal arithmetic: 1377.57275 is not representable as a binary float.
  const reserve = new Decimal("1.00").times(OPEN_SHORT_CONTRACTS); // your own reserve
  // The collateral term is NOT in this sum: no field exposes it. See the warning above.
  const candidates = [new Decimal(balance).minus(reserve).minus(ACCRUED_FEES)]
    .concat(ceilings.map((c) => new Decimal(c)));
  return candidates.reduce((lowest, v) => (v.lt(lowest) ? v : lowest)).toFixed();
}

spendable(PARTICIPANT_ACCOUNT).then(console.log);

#!/usr/bin/env bash
set -euo pipefail

GRPC_TARGET="grpc-api.preprod.polymarketexchange.com:443"  # prod: grpc-api.prod.polymarketexchange.com:443
PROTO_FILE="./balances.proto"                               # the proto your integration lead sent you
PROTO_PACKAGE="$(grep -m1 '^package' "$PROTO_FILE" | sed 's/package \(.*\);/\1/')"
SERVICE="..."                                               # the service that carries GetAccountBalance
FIRM_ACCOUNT="firms/.../accounts/..."                       # your firm account
PARTICIPANT_ID="..."                                        # participantId from the kyc.approved webhook
TOKEN="$(./token.sh)"                                       # the script on /connect#authentication

# Account-scoped read, so x-participant-id is required here (unlike CashMovementService).
# Firm account: expect all six fields. A participant clearing account returns
# `InvalidArgument: invalid account` here — read afterBalance off the balance ledger instead.
grpcurl -proto "$PROTO_FILE" \
  -H "authorization: Bearer ${TOKEN}" \
  -H "x-participant-id: ${PARTICIPANT_ID}" \
  -d "{\"account\":\"${FIRM_ACCOUNT}\"}" \
  "$GRPC_TARGET" "${PROTO_PACKAGE}.${SERVICE}/GetAccountBalance" \
  | jq '{balance, capitalRequirement, excessCapital, buyingPower, unsettledFunds, marginRequirement}'

// go get github.com/shopspring/decimal google.golang.org/grpc
package main

import (
	"context"
	"fmt"
	"log"
	"strings"
	"time"

	"github.com/shopspring/decimal"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"

	// Canonical client from /connect#authentication. The gRPC target is
	// pmauth.GRPCTarget — one place, both environments.
	pmauth "example.com/yourfirm/pmauth"

	// protoc output from the protos your integration lead sent you. The owning
	// services are not published — check these names against your own copy.
	bal "yourfirm/pmxprotos/balance"
	lg "yourfirm/pmxprotos/ledger"
)

const (
	participantAccount = "firms/.../accounts/..." // provisionedAccount, recorded at onboarding
	participantID      = "..."                    // participantId from the kyc.approved webhook
	openShortContracts = 0                        // from your own position store
	accruedFees        = "0.00"                   // USD, from your own fee ledger
	replayDeadline     = 60 * time.Second         // yours: no replay-to-live signal is published
)

// One auth client for the life of the process: it caches the token and re-mints
// at expires_in minus 30 s.
var pm = pmauth.New()

// dec parses money exactly. 1377.57275 is not representable as a binary float,
// so never parse a balance into float64 or big.Float.
func dec(s string) (decimal.Decimal, error) {
	d, err := decimal.NewFromString(s)
	if err != nil {
		return decimal.Decimal{}, fmt.Errorf("parse money %q: %w", s, err)
	}
	return d, nil
}

// newestAfterBalance is the workaround for a participant clearing account. There
// is no point-read RPC for a participant account balance, so this subscribes to
// the balance ledger, drains the replay phase, and keeps the newest afterBalance.
func newestAfterBalance(ctx context.Context, conn *grpc.ClientConn, account string) (string, error) {
	acctMD, err := pm.Metadata(participantID) // account-scoped: header required
	if err != nil {
		return "", err
	}
	streamCtx, cancel := context.WithTimeout(metadata.NewOutgoingContext(ctx, acctMD), replayDeadline)
	defer cancel()

	stream, err := lg.NewBalanceLedgerClient(conn).CreateBalanceLedgerSubscription(streamCtx,
		&lg.CreateBalanceLedgerSubscriptionRequest{Account: account})
	if err != nil {
		return "", err
	}

	newest := ""
	for {
		entry, rerr := stream.Recv()
		if rerr != nil {
			// No published replay-to-live signal: a quiet stream is how you detect the
			// drain. Anything else is a real failure — never swallow it.
			if status.Code(rerr) != codes.DeadlineExceeded {
				return "", rerr
			}
			break
		}
		// Replay ordering is not published, so this keeps the LAST entry the replay
		// phase delivered. Confirm the ordering before you trust it.
		newest = entry.GetAfterBalance()
	}
	if newest == "" {
		return "", fmt.Errorf("no balance-ledger entry carrying afterBalance for %s", account)
	}
	return newest, nil
}

func spendable(ctx context.Context, account string) (decimal.Decimal, error) {
	zero := decimal.Decimal{}

	conn, err := pm.Channel()
	if err != nil {
		return zero, err
	}
	defer conn.Close()

	acctMD, err := pm.Metadata(participantID) // account-scoped read: header required
	if err != nil {
		return zero, err
	}

	var balanceStr string
	var ceilings []decimal.Decimal

	b, err := bal.NewAccountBalanceClient(conn).GetAccountBalance(
		metadata.NewOutgoingContext(ctx, acctMD), &bal.GetAccountBalanceRequest{Account: account})
	switch {
	case err == nil:
		balanceStr = b.GetBalance()
		for _, raw := range []string{b.GetBuyingPower(), b.GetExcessCapital()} {
			c, cerr := dec(raw)
			if cerr != nil {
				return zero, cerr
			}
			ceilings = append(ceilings, c)
		}
	case strings.Contains(err.Error(), "invalid account"):
		// Expected on a participant clearing account. No ceilings on this path.
		balanceStr, err = newestAfterBalance(ctx, conn, account)
		if err != nil {
			return zero, err
		}
	default:
		return zero, err
	}

	balance, err := dec(balanceStr)
	if err != nil {
		return zero, err
	}
	fees, err := dec(accruedFees)
	if err != nil {
		return zero, err
	}
	reserve := decimal.NewFromInt(openShortContracts) // $1.00 per short contract, your own reserve

	// The collateral term is NOT in this sum: no field exposes it. See the warning above.
	out := balance.Sub(reserve).Sub(fees)
	for _, c := range ceilings {
		if c.LessThan(out) {
			out = c
		}
	}
	return out, nil
}

func main() {
	v, err := spendable(context.Background(), participantAccount)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(v.String()) // balances carry more than 2 decimals
}

from decimal import Decimal

import grpc
from polymarket_auth import PolymarketClient  # published on /connect#authentication
from mystore import load_checkpoint, save_checkpoint, in_flight  # your own storage

# protoc output from the ledger and funding protos your integration lead sent you.
from pmx_protos import funding_pb2, funding_pb2_grpc
from pmx_protos import ledger_pb2, ledger_pb2_grpc

# The gRPC target is GRPC_TARGET in polymarket_auth.py — one place, both environments.
pmx = PolymarketClient()

PARTICIPANT_ID = "..."          # participantId from the kyc.approved webhook
ACCOUNT = "firms/.../accounts/..."  # provisionedAccount for that participant
REPLAY_DEADLINE_SECONDS = 60    # yours: the replay-to-live signal is not published

def reconcile(account: str, participant_id: str) -> Decimal:
    checkpoint = load_checkpoint(account)
    pool_request = funding_pb2.GetFundingAccountBalanceRequest()

    with pmx.channel() as channel:
        ledger = ledger_pb2_grpc.AccountsAPIStub(channel)
        funding = funding_pb2_grpc.FundingServiceStub(channel)

        if checkpoint is None:
            # First run only. A restart is NOT this branch.
            # Firm-scoped read: no participant ID on the pool.
            anchor = funding.GetFundingAccountBalance(pool_request, metadata=pmx.metadata())
            checkpoint = {"entry_id": None, "after_balance": "0", "pool": anchor.balance}

        request = ledger_pb2.CreateBalanceLedgerSubscriptionRequest(account=account)
        if checkpoint["entry_id"]:
            request.resume_token = checkpoint["entry_id"]

        stream = ledger.CreateBalanceLedgerSubscription(
            request,
            # Account-scoped: x-participant-id is required on this surface.
            metadata=pmx.metadata(participant_id=participant_id),
            timeout=REPLAY_DEADLINE_SECONDS,
        )
        try:
            for entry in stream:
                if entry.entry_id == checkpoint["entry_id"]:
                    continue  # at-least-once delivery: dedupe, expect redelivery
                # proto3: an absent field and a zero are indistinguishable. Require the field.
                if not entry.HasField("after_balance"):
                    raise RuntimeError(f"ledger entry {entry.entry_id} has no afterBalance")
                checkpoint = {"entry_id": entry.entry_id, "after_balance": entry.after_balance,
                              "pool": checkpoint["pool"]}
                save_checkpoint(account, checkpoint)  # same transaction as applying the entry
        except grpc.RpcError as err:
            # There is no published replay-to-live signal, so a quiet stream is how you
            # detect the drain. Anything else is a real failure — never swallow it.
            if err.code() is not grpc.StatusCode.DEADLINE_EXCEEDED:
                raise

        pool_now = Decimal(
            funding.GetFundingAccountBalance(pool_request, metadata=pmx.metadata()).balance
        )

    expected = Decimal(checkpoint["pool"])
    for movement in in_flight(account):           # PENDING / AMBIGUOUS: residual, never netted
        print(f"in flight: {movement['idempotency_key']} {movement['status']}")
    break_amount = pool_now - expected
    if break_amount != 0:
        raise RuntimeError(f"pool does not close: {break_amount} — check wires and in-flight list")
    return Decimal(checkpoint["after_balance"])

print(reconcile(ACCOUNT, PARTICIPANT_ID))

import * as grpc from "@grpc/grpc-js";
import { PolymarketClient } from "./polymarketAuth"; // published on /connect#authentication
import { loadCheckpoint, saveCheckpoint, inFlight } from "./mystore"; // your own storage

// protoc output from the ledger and funding protos your integration lead sent you.
import { AccountsAPIClient } from "./gen/ledger";
import { FundingServiceClient } from "./gen/funding";

// The gRPC target is GRPC_TARGET in polymarketAuth.ts — one place, both environments.
const pmx = new PolymarketClient();

const PARTICIPANT_ID = "..."; // participantId from the kyc.approved webhook
const ACCOUNT = "firms/.../accounts/..."; // provisionedAccount for that participant
const REPLAY_DEADLINE_MS = 60_000; // yours: the replay-to-live signal is not published

export async function reconcile(account: string, participantId: string): Promise<string> {
  const channel = pmx.channel();
  const ledger = new AccountsAPIClient(channel);
  const funding = new FundingServiceClient(channel);

  // Firm-scoped read: no participant ID on the pool.
  const poolBalance = async (): Promise<string> =>
    (await funding.getFundingAccountBalance({}, await pmx.metadata())).balance;

  let checkpoint = await loadCheckpoint(account);
  if (!checkpoint) {
    // First run only. A restart is NOT this branch.
    checkpoint = { entryId: null, afterBalance: "0", pool: await poolBalance() };
  }

  const stream = ledger.createBalanceLedgerSubscription(
    { account, resumeToken: checkpoint.entryId ?? undefined },
    // Account-scoped: x-participant-id is required on this surface.
    await pmx.metadata(participantId),
    { deadline: Date.now() + REPLAY_DEADLINE_MS },
  );

  try {
    for await (const entry of stream) {
      if (entry.entryId === checkpoint.entryId) continue; // at-least-once: dedupe
      if (entry.afterBalance === undefined) {
        throw new Error(`ledger entry ${entry.entryId} has no afterBalance`);
      }
      checkpoint = { entryId: entry.entryId, afterBalance: entry.afterBalance, pool: checkpoint.pool };
      await saveCheckpoint(account, checkpoint); // same transaction as applying the entry
    }
  } catch (err: any) {
    // There is no published replay-to-live signal, so a quiet stream is how you detect
    // the drain. Anything else is a real failure — never swallow it.
    if (err?.code !== grpc.status.DEADLINE_EXCEEDED) throw err;
  }

  // Compare the decimal strings. Never parse money into a binary float.
  const poolNow = await poolBalance();
  for (const m of await inFlight(account)) console.log(`in flight: ${m.idempotencyKey} ${m.status}`);
  if (poolNow !== checkpoint.pool) {
    throw new Error(
      `pool does not close: have ${poolNow}, expected ${checkpoint.pool} — check wires and in-flight list`,
    );
  }
  return checkpoint.afterBalance;
}

reconcile(ACCOUNT, PARTICIPANT_ID).then(console.log);

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"

	// Canonical client from /connect#authentication. The gRPC target is
	// pmauth.GRPCTarget — one place, both environments.
	pmauth "example.com/yourfirm/pmauth"

	// protoc output from the protos your integration lead sent you.
	fd "yourfirm/pmxprotos/funding"
	lg "yourfirm/pmxprotos/ledger"

	store "yourfirm/reconcile/store"
)

const (
	participantID = "..."                    // participantId from the kyc.approved webhook
	account       = "firms/.../accounts/..." // provisionedAccount for that participant
	// Yours: the replay-to-live signal is not published.
	replayDeadline = 60 * time.Second
)

// One auth client for the life of the process: it caches the token and re-mints
// at expires_in minus 30 s.
var pm = pmauth.New()

func reconcile(ctx context.Context) (string, error) {
	conn, err := pm.Channel()
	if err != nil {
		return "", err
	}
	defer conn.Close()

	ledger := lg.NewAccountsAPIClient(conn)
	funding := fd.NewFundingServiceClient(conn)

	// Firm-scoped read: pass "" for the participant ID on the pool.
	firmMD, err := pm.Metadata("")
	if err != nil {
		return "", err
	}
	firmCtx := metadata.NewOutgoingContext(ctx, firmMD)

	cp, err := store.LoadCheckpoint(account)
	if err != nil {
		return "", err
	}
	if cp == nil {
		// First run only. A restart is NOT this branch.
		anchor, aerr := funding.GetFundingAccountBalance(firmCtx, &fd.GetFundingAccountBalanceRequest{})
		if aerr != nil {
			return "", aerr
		}
		cp = &store.Checkpoint{Pool: anchor.GetBalance()}
	}

	// Account-scoped: x-participant-id is required on this surface.
	acctMD, err := pm.Metadata(participantID)
	if err != nil {
		return "", err
	}
	streamCtx, cancel := context.WithTimeout(metadata.NewOutgoingContext(ctx, acctMD), replayDeadline)
	defer cancel()

	stream, err := ledger.CreateBalanceLedgerSubscription(streamCtx,
		&lg.CreateBalanceLedgerSubscriptionRequest{Account: account, ResumeToken: cp.EntryID})
	if err != nil {
		return "", err
	}
	for {
		entry, rerr := stream.Recv()
		if rerr != nil {
			// There is no published replay-to-live signal, so a quiet stream is how you
			// detect the drain. Anything else is a real failure — never swallow it.
			if status.Code(rerr) != codes.DeadlineExceeded {
				return "", rerr
			}
			break
		}
		if entry.GetEntryId() == cp.EntryID {
			continue // at-least-once delivery: dedupe
		}
		if entry.GetAfterBalance() == "" {
			return "", fmt.Errorf("ledger entry %s has no afterBalance", entry.GetEntryId())
		}
		cp = &store.Checkpoint{EntryID: entry.GetEntryId(), AfterBalance: entry.GetAfterBalance(), Pool: cp.Pool}
		if err := store.SaveCheckpoint(account, cp); err != nil { // same txn as applying the entry
			return "", err
		}
	}

	pool, err := funding.GetFundingAccountBalance(firmCtx, &fd.GetFundingAccountBalanceRequest{})
	if err != nil {
		return "", err
	}
	// Compare the decimal strings. Never parse money into a binary float.
	if pool.GetBalance() != cp.Pool {
		return "", fmt.Errorf("pool does not close: have %s, expected %s — check wires and in-flight list",
			pool.GetBalance(), cp.Pool)
	}
	return cp.AfterBalance, nil
}

func main() {
	balance, err := reconcile(context.Background())
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(balance)
}