gRPC: PriceStream

x-api-key: ohlcv_live_sk_<your_key>

syntax = "proto3";
package dexploit.v1;

service PriceStream {
  // Stream of DexTransaction price updates, filtered by protocol.
  rpc Subscribe(SubscribeRequest) returns (stream DexTransaction);
}

message SubscribeRequest {
  // "pumpfun" | "pumpswap" | "clmm" | "cpmm" | "amm"
  // Empty list = all protocols.
  repeated string protocols = 1;
}

message DexTransaction {
  string signature      = 1;
  string pair_id        = 2;   // pool / bonding-curve address
  string token_address  = 3;   // base token (the token being priced)
  string mint           = 4;   // quote currency (SOL or stablecoin)
  double price          = 5;   // quote per base token
  double token_balance  = 6;   // base reserve, decimal-adjusted
  double sol_balance    = 7;   // quote reserve, decimal-adjusted
  optional double liquidity = 8;   // TVL; CLMM/CPMM/AMM only
  optional string creator   = 9;   // PumpFun only
  ProtocolType protocol     = 10;
  uint32 schema_version     = 11;
}

enum ProtocolType {
  PROTOCOL_UNSPECIFIED = 0;
  PROTOCOL_PUMPFUN     = 1;
  PROTOCOL_PUMPSWAP    = 2;
  PROTOCOL_CLMM        = 3;
  PROTOCOL_CPMM        = 4;
  PROTOCOL_AMM         = 5;
}

npm i @grpc/grpc-js @grpc/proto-loader

import * as grpc from '@grpc/grpc-js';
import * as protoLoader from '@grpc/proto-loader';
import path from 'path';

const ENDPOINT = 'grpc.dexploit.dev:443';
const API_KEY  = 'ohlcv_live_sk_<your_key>';

const PROTO_PATH = path.join(__dirname, 'proto', 'price_stream.proto');
const def = protoLoader.loadSync(PROTO_PATH, {
  keepCase: true,
  longs: String,
  enums: String,
  defaults: true,
  oneofs: true,
});
const proto = grpc.loadPackageDefinition(def) as any;

const creds = grpc.credentials.combineChannelCredentials(
  grpc.credentials.createSsl(),
  grpc.credentials.createFromMetadataGenerator((_params, cb) => {
    const md = new grpc.Metadata();
    md.add('x-api-key', API_KEY);
    cb(null, md);
  })
);

const client = new proto.dexploit.v1.PriceStream(ENDPOINT, creds, {
  // Mandatory — without these, half-open TCP hangs look like a stalled stream.
  'grpc.keepalive_time_ms': 30_000,
  'grpc.keepalive_timeout_ms': 10_000,
  'grpc.keepalive_permit_without_calls': 1,
});

// Empty protocols = every protocol. Or e.g. { protocols: ['clmm', 'cpmm'] }.
const call = client.Subscribe({ protocols: [] });

call.on('data', (tx: any) => {
  console.log(
    `${tx.protocol} ${tx.token_address.slice(0, 8)}… ` +
    `price=${tx.price.toExponential(3)} tvl=${tx.sol_balance.toFixed(2)} SOL ` +
    `(${tx.signature.slice(0, 8)}…)`
  );
});

call.on('error', (err: Error) => console.error('stream error:', err.message));
call.on('end',   ()           => console.warn('stream ended'));