# WebSocket Streaming

The Go SDK exposes three WebSocket connections: the **Order Stream** (JSON updates), the **Data Stream** (token-based binary ticks), and the **HFT Data Stream** (zstd-compressed binary protocol).

Unlike the Python SDK, Go streams are **callback + context** based. There is **no built-in auto-reconnect**. Restart `ReadTicks` / `ReadUpdates` / `ReadHFT` after a socket error, or wrap the read loop yourself.

## Overview

| Stream | Endpoint | Purpose | Reader |
|--------|----------|---------|--------|
| **Order Stream** | `wss://order-updates.arrow.trade` | Order and position updates | `ReadUpdates` (JSON `map[string]any`) |
| **Data Stream** | `wss://ds.arrow.trade` | Standard market data (token-based, big-endian ticks) | `ReadTicks` (`MarketTick`) |
| **HFT Data Stream** | `wss://socket.arrow.trade` | Low-latency market data (zstd binary) | `ReadHFT` (LTP / full / CAS / response) |

Query params: `appID` and `token`. HFT also sets `zstd=1`.

## Quick Start

```go
streams, err := client.NewStreams() // order + token data stream
if err != nil {
    log.Fatal(err)
}
defer streams.Close()

if err := streams.DataStream.Subscribe(arrow.StreamModeQuote, []int32{3045, 1594}); err != nil {
    log.Fatal(err)
}

ctx := context.Background()

go streams.OrderStream.ReadUpdates(ctx, func(update map[string]any) {
    fmt.Println("order:", update["id"], update["orderStatus"])
}, func(err error) {
    log.Println("order stream:", err)
})

go streams.DataStream.ReadTicks(ctx, func(tick arrow.MarketTick) {
    fmt.Printf("token=%d ltp=%d cas=%v\n", tick.Token, tick.LTP, tick.IsCAS)
}, func(err error) {
    log.Println("data stream:", err)
})

select {}
```

Factory helpers:

| Method | Connects |
|--------|----------|
| `client.NewStreams()` | Order + Data Stream |
| `client.NewStreamsOrderOnly()` | Order only |
| `client.NewStreamsWithHFT()` | Order + HFT (no token Data Stream) |
| `client.ConnectOrderStream()` | Order only (`*OrderStream`) |
| `client.ConnectDataStream()` | Data Stream only |
| `client.ConnectHFTDataStream()` | HFT only |

## Order Stream

`wss://order-updates.arrow.trade?appID=<appID>&token=<token>`. Incoming messages are **JSON text**. Heartbeats and non-JSON frames are skipped.

```go
stream, err := client.ConnectOrderStream()
if err != nil {
    log.Fatal(err)
}
defer stream.Close()

stream.ReadUpdates(ctx, func(order map[string]any) {
    fmt.Println(order["id"], order["orderStatus"])
}, func(err error) {
    log.Println(err)
})
```

Updates typically include `orderStatus`, `symbol`, `quantity`, `price`, `transactionType`, `averagePrice`, and `rejectReason`. See the REST [order-updates](../rest-api/order-data.md) guide for the payload shape.

## Data Stream

Integer **instrument tokens**, **big-endian** binary ticks. Subscribe with `StreamMode`.

```go
err = streams.DataStream.Subscribe(arrow.StreamModeLTPC, []int32{26000, 26009})
err = streams.DataStream.Unsubscribe(arrow.StreamModeLTPC, []int32{26000})
```

Subscribe JSON looks like `{"code":"sub","mode":"ltp","ltp":[26000,26009]}`.

!!! note "CAS fields on every mode"
    From **3:15 PM IST**, each mode grows by **16 bytes** (`ImbalanceQty` signed int64 + `IndicativeClose` int32 + `RefPrice` int32):
    `ltp` 13→**29**, `ltpc` 17→**33**, `quote` 93→**109**, `full` 249→**265**.
    `MarketTick.IsCAS` is **true** when that 16-byte trailer is present, even if the numeric fields are zero.
    See [Closing Auction Session (CAS)](../rest-api/market-data.md#closing-auction-session-cas).

### Stream modes

| Mode | Constant | Base size | CAS size |
|------|----------|-----------|----------|
| LTP | `arrow.StreamModeLTP` | 13 | 29 |
| LTPC | `arrow.StreamModeLTPC` | 17 | 33 |
| Quote | `arrow.StreamModeQuote` | 93 | 109 |
| Full | `arrow.StreamModeFull` | 249 | 265 |

### LTP Mode

```go
_ = streams.DataStream.Subscribe(arrow.StreamModeLTP, []int32{3045, 1594})

streams.DataStream.ReadTicks(ctx, func(tick arrow.MarketTick) {
    fmt.Printf("token=%d ltp=%d mode=%s\n", tick.Token, tick.LTP, tick.Mode)
    if tick.IsCAS {
        fmt.Printf("CAS imb=%d ind=%d ref=%d\n", tick.ImbalanceQty, tick.IndicativeClose, tick.RefPrice)
    }
}, nil)
```

### LTPC Mode

```go
_ = streams.DataStream.Subscribe(arrow.StreamModeLTPC, []int32{3045, 1594})

streams.DataStream.ReadTicks(ctx, func(tick arrow.MarketTick) {
    fmt.Printf("token=%d ltp=%d close=%d netChange=%.2f flag=%d\n",
        tick.Token, tick.LTP, tick.Close, tick.NetChange, tick.ChangeFlag)
}, nil)
```

`ChangeFlag`: `43` (`+`), `45` (`-`), `32` (flat).

### Quote Mode

```go
_ = streams.DataStream.Subscribe(arrow.StreamModeQuote, []int32{3045})

streams.DataStream.ReadTicks(ctx, func(tick arrow.MarketTick) {
    fmt.Printf("token=%d ltp=%d vol=%d oi=%d ltq=%d\n",
        tick.Token, tick.LTP, tick.Volume, tick.OI, tick.LTQ)
}, nil)
```

### Full Mode

```go
_ = streams.DataStream.Subscribe(arrow.StreamModeFull, []int32{3045})

streams.DataStream.ReadTicks(ctx, func(tick arrow.MarketTick) {
    fmt.Printf("token=%d ltp=%d upper=%d lower=%d\n", tick.Token, tick.LTP, tick.UpperLimit, tick.LowerLimit)
    for i, bid := range tick.Bids {
        fmt.Printf("bid %d: %d @ %d (%d orders)\n", i+1, bid.Quantity, bid.Price, bid.Orders)
    }
    if tick.IsCAS {
        fmt.Printf("CAS imb=%d indicative=%d ref=%d\n", tick.ImbalanceQty, tick.IndicativeClose, tick.RefPrice)
    }
}, nil)
```

### `MarketTick` fields

| Field | Type | Description |
|-------|------|-------------|
| `Token` | `int32` | Instrument token |
| `Mode` | `StreamMode` | `ltp` / `ltpc` / `quote` / `full` |
| `LTP` | `int32` | Last traded price |
| `Close` | `int32` | Previous close |
| `NetChange` | `float64` | Percent change |
| `ChangeFlag` | `int8` | Direction (`43` / `45` / `32`) |
| `LTQ` | `int32` | Last traded quantity |
| `AvgPrice` | `int32` | Average traded price |
| `Open` / `High` / `Low` | `int32` | Session OHLC |
| `Volume` | `int64` | Traded volume |
| `OI` / `OIDayHigh` / `OIDayLow` | `int64` | Open interest |
| `TotalBuyQuantity` / `TotalSellQuantity` | `int64` | Aggregate quantities |
| `UpperLimit` / `LowerLimit` | `int32` | Circuit limits (full) |
| `Bids` / `Asks` | `[]DepthLevel` | 5 levels (`Quantity`, `Price`, `Orders`) |
| `IsCAS` | `bool` | Packet included the 16-byte CAS trailer |
| `ImbalanceQty` | `int64` | Signed CAS imbalance (negative = sell side) |
| `IndicativeClose` | `int32` | CAS indicative close |
| `RefPrice` | `int32` | CAS reference price |

Prices on the token stream are integers in the wire unit (typically paise). Unused fields are zero.

You can also parse a raw frame with `arrow.ParseMarketTick(data)`.

## HFT Data Stream

`wss://socket.arrow.trade?appID=<appID>&token=<token>&zstd=1`. Inbound frames are **zstd-compressed**; the SDK decompresses them. Outbound subscribe/unsubscribe is **JSON**.

!!! warning "Mandatory from 8 July 2026"
    **Zstd compression becomes mandatory on 8 July 2026.** Use a current **go-arrow** release (`HFTDataStream` handles decompression) or follow the [HFT protocol](../rest-api/hft-stream.md#inbound-compression-zstd).

### Protocol summary

| Topic | Detail |
|-------|--------|
| Inbound compression | Zstd (`zstd=1`; SDK decompresses) |
| Multi-byte integers | Little-endian |
| Prices | Paise |
| Packet types | LTP `1` (40 B), Full `2` (196 B), CAS `7` (168 B), Response `99` (540 B) |

**Exchange segments**

| Constant | Value | Segment |
|----------|-------|---------|
| `arrow.HFTExchNSECM` | `0` | NSE cash |
| `arrow.HFTExchNSEFO` | `1` | NSE F&O |
| `arrow.HFTExchBSECM` | `2` | BSE cash |
| `arrow.HFTExchBSEFO` | `3` | BSE F&O |
| `arrow.HFTExchMCXFO` | `4` | MCX F&O (reserved) |

!!! note "MCX on HFT"
    Prefer the token Data Stream (`wss://ds.arrow.trade`) or REST quotes for MCXFO until HFT accepts MCX tokens for your app.

**Modes:** `"ltpc"` (alias `"l"`), `"full"` (alias `"f"`), `"cas"`. CAS subscribe JSON **omits** `latency`.

**Limits:** 100 requests/sec, 512 symbols per subscribe, 1024 subscribed symbols per connection, 16 KB max request.

### Connect and subscribe

```go
streams, err := client.NewStreamsWithHFT()
if err != nil {
    log.Fatal(err)
}
defer streams.Close()

hft := streams.HFTDataStream

if err := hft.SubscribeHFTSymbols("ltpc", []string{"NSE.SBIN-EQ", "BSE.RELIANCE"}, 100); err != nil {
    log.Fatal(err)
}
if err := hft.SubscribeHFTTokens("full", arrow.HFTExchNSECM, []int32{5042, 4449}, 200); err != nil {
    log.Fatal(err)
}
if err := hft.SubscribeHFTBySegment("full", map[int][]int32{
    arrow.HFTExchNSEFO: {5042, 4449},
    arrow.HFTExchBSECM: {100, 200},
}, 100); err != nil {
    log.Fatal(err)
}

// Closing Auction Session — no latency field on the wire
if err := hft.SubscribeHFTSymbols("cas", []string{"NSE.RELIANCE-EQ"}, 0); err != nil {
    log.Fatal(err)
}

hft.ReadHFT(ctx,
    func(t arrow.HFTLTPTick) { fmt.Println("LTP", t.Token, t.LTP) },
    func(t arrow.HFTFullTick) { fmt.Println("Full", t.Token, t.LTP, t.OI) },
    func(t arrow.HFTCASTick) { fmt.Println("CAS", t.Token, t.IndicativePx, t.ImbalanceQty) },
    func(r arrow.HFTResponsePacket) { fmt.Println("ack", r.ErrorCode, r.SuccessCount) },
    func(err error) { log.Println(err) },
)
```

### Unsubscribe

```go
_ = hft.UnsubscribeHFTSymbols("ltpc", []string{"NSE.SBIN-EQ"})
_ = hft.UnsubscribeHFTTokens("full", arrow.HFTExchNSEFO, []int32{5042})
```

### Symbol string formats

| Segment | Example |
|---------|---------|
| NSE CM | `NSE.SBIN-EQ` |
| NSE FO options | `NYKAA30DEC25C232.5` |
| NSE FO futures | `BANKNIFTY30DEC25F` |
| BSE CM | `BSE.SBIN` |
| BSE FO | `SENSEX01JAN26C74900` / `SENSEX01JAN26F` |
| MCX FO futures | `GOLDPETAL31JUL26F` |

!!! note "Futures volume vs other data feeds"
    For **futures (FO)** symbols, HFT tick `volume` is often **lower** than on the standard Data Stream. HFT uses **TBT** (that token only); bcast feeds can include spread book volume. See [HFT market-data packets](../rest-api/hft-stream.md#market-data-packets).

### HFT tick shapes

**`HFTLTPTick`:** `Size`, `PktType`, `ExchSeg`, `Token`, `LTP`, `VWAP`, `Volume`, `LTT`, `ATV`, `BTV`.

**`HFTFullTick`:** LTP fields plus `LTQ`, OHLC, `DprL` / `DprH`, `TBQ`, `TSQ`, five-level `BidPx` / `AskPx` / `BidSize` / `AskSize` / `BidOrd` / `AskOrd`, `OI`, `TS`.

**`HFTCASTick`:** `Token`, 4-level bid/ask, `TS`, `ImbalanceQty`, `ImbalanceMktQty`, `IndicativePx`, `ClosingRefPx`, `ClosePx`, `IndicativeQty`, `MktBidQty`, `MktAskQty`, `Phase`, imbalance sides, `OnlyLimitOrders`.

**`HFTResponsePacket`:** subscribe/unsubscribe ack — `ErrorCode`, `ErrorMsg`, `RequestType` / `RequestStr`, `Mode` / `ModeStr`, `SuccessCount`, `ErrorCount`. Typical codes: `SUCCESS`, `E_PARTIAL`, `E_ALL_INVALID`, `E_INVALID_JSON`, `E_MISSING_FIELD`, `E_INVALID_PARAM`, `E_PARSE_ERROR`.

## Connection Management

```go
// Combined
streams, err := client.NewStreams()
defer streams.Close()

// Individual
orders, _ := client.ConnectOrderStream()
data, _ := client.ConnectDataStream()
hft, _ := client.ConnectHFTDataStream()
defer orders.Close()
defer data.Close()
defer hft.Close()
```

Cancel the `context` passed to `ReadTicks` / `ReadUpdates` / `ReadHFT` to stop the read loop.

Optional keep-alive helper (sends `"PONG"` text frames):

```go
go arrow.StartKeepAlive(ctx, /* websocket.Conn */, 3*time.Second)
```

!!! warning "No automatic reconnect"
    Python `BaseSocket` reconnects and resubscribes. Go does **not**. On `onError`, close the stream, `Connect*` again, re-`Subscribe`, and restart the reader.

## Complete Example

```go
package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"

    "github.com/arrow-trade/go-arrow/arrow"
)

func main() {
    client := arrow.NewClient(os.Getenv("ARROW_APP_ID"), os.Getenv("ARROW_APP_SECRET"))
    if err := client.AutoLogin(
        os.Getenv("ARROW_USER_ID"),
        os.Getenv("ARROW_PASSWORD"),
        os.Getenv("ARROW_TOTP_SECRET"),
    ); err != nil {
        log.Fatal(err)
    }

    streams, err := client.NewStreams()
    if err != nil {
        log.Fatal(err)
    }
    defer streams.Close()

    if err := streams.DataStream.Subscribe(arrow.StreamModeFull, []int32{3045}); err != nil {
        log.Fatal(err)
    }

    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
    defer stop()

    go streams.OrderStream.ReadUpdates(ctx, func(u map[string]any) {
        fmt.Println("order", u["id"], u["orderStatus"])
    }, func(err error) { log.Println(err) })

    go streams.DataStream.ReadTicks(ctx, func(t arrow.MarketTick) {
        fmt.Printf("%d ltp=%d cas=%v imb=%d\n", t.Token, t.LTP, t.IsCAS, t.ImbalanceQty)
    }, func(err error) { log.Println(err) })

    <-ctx.Done()
}
```

## Best Practices

- Resolve tokens from REST quotes or the instruments CSV before subscribing.
- Prefer `StreamModeLTPC` when you only need last price and close.
- Use `IsCAS` (not “are the numbers non-zero”) to detect the auction window.
- For MCXFO, use the token Data Stream until HFT MCX is enabled for your app.
- Bound REST with the default 10s timeout; streams use your `context` for shutdown.