AWS Kinesis

1. When to use Kinesis

Choose Kinesis when you run your own consumer infrastructure and want a durable,
replayable stream you can process with the AWS SDK or the Kinesis Client Library (KCL).
Choose the WebSocket when you want a managed push feed without running consumers. Both
carry the same realtime data; they differ in how you consume it.

2. Access

Kinesis access is part of the premium and enterprise packages and is provisioned
per contract. To get set up, contact [email protected] with:

• The AWS account ID you will consume from.
• The exchanges and channels you are entitled to (see section 3).

We grant your account a cross-account read role scoped to exactly your entitled
stream(s), with read-only data-plane permissions (kinesis:GetRecords,
GetShardIterator, DescribeStream, DescribeStreamSummary, ListShards). You then
assume that role from your account to consume. All streams run in eu-central-1
(Frankfurt). You receive the exact stream ARN(s) and role ARN at onboarding.

Keep the role and any derived credentials server-side. Never embed them in a browser or
mobile client.

3. Streams

Streams are organised by exchange and channel, mirroring the WebSocket topics
(<channel>.<MIC>). Not every exchange offers both channels.

4. Record format

Each Kinesis record's Data is a single UTF-8 JSON object (newline-terminated). The AWS
SDK base64-decodes the envelope for you; you parse the JSON.

Trade (traded price):

{"ISIN": "DE0007164600", "LastPrice": 123.45, "Volume": 100, "Currency": "EUR",
 "MIC": "XGAT", "Time": 1764236768617, "event_type": "TRADE"}

Quote (bid/ask, quotes-enabled exchanges only):

{"ISIN": "DE0007164600", "BID": 123.40, "BIDVolume": 200, "ASK": 123.45,
 "ASKVolume": 100, "Currency": "EUR", "MIC": "XMUN", "Time": 1764236768617,
 "event_type": "QUOTE"}

Use Time (event time) for display and candles, never your own arrival time. Apply records
idempotently by Time so replays and reconnects are harmless.

5. Consuming (Python)

A minimal consumer with the AWS SDK. For production, use the Kinesis Client Library (KCL),
which handles sharding, checkpointing, and rebalancing for you.

import boto3, json, time

REGION = "eu-central-1"
STREAM_ARN = "<stream ARN from onboarding>"
ROLE_ARN   = "<cross-account read role ARN from onboarding>"

# Assume the read role Bavest granted your account.
creds = boto3.client("sts").assume_role(
    RoleArn=ROLE_ARN, RoleSessionName="bavest-kinesis")["Credentials"]
kinesis = boto3.client(
    "kinesis", region_name=REGION,
    aws_access_key_id=creds["AccessKeyId"],
    aws_secret_access_key=creds["SecretAccessKey"],
    aws_session_token=creds["SessionToken"])

# Read every shard from the latest position. (KCL does this for you in production.)
for shard in kinesis.list_shards(StreamARN=STREAM_ARN)["Shards"]:
    it = kinesis.get_shard_iterator(
        StreamARN=STREAM_ARN, ShardId=shard["ShardId"],
        ShardIteratorType="LATEST")["ShardIterator"]
    while it:
        resp = kinesis.get_records(ShardIterator=it, Limit=1000)
        for record in resp["Records"]:
            tick = json.loads(record["Data"])
            print(tick["event_type"], tick["ISIN"], tick.get("LastPrice") or tick.get("BID"))
        it = resp.get("NextShardIterator")
        time.sleep(0.2)  # respect the 5 GetRecords/s/shard limit

6. Latency and throughput

• Records are available within a few hundred milliseconds of the market event.
• A standard shard iterator allows 5 GetRecords calls per second per shard; poll with
  a short sleep or use KCL to stay within it. For the lowest latency at scale, ask about
  enhanced fan-out (push-based SubscribeToShard).
• Streams retain records for 24 hours, so a consumer that falls behind or restarts can
  replay the gap rather than losing data.

7. Support

For entitlement, onboarding, stream ARNs, or enhanced fan-out, contact
[email protected].