Real-time Data Handling with WebSockets in R

Why this lesson matters

In Fetching OHLC Data from APIs, you built the habit of treating market data as a contract: timestamp, symbol, open, high, low, close, volume, and a guarded schema.

Now we move from requesting historical chunks to listening continuously.

A WebSocket is how your live system receives market events as they happen:

Historical API:  ask -> receive page -> stop
WebSocket:       connect -> receive event -> receive event -> ...

In this lesson, we focus only on handling the live stream safely:

• connect to a stream
• parse each incoming message
• normalize it into your internal row format
• handle disconnects
• keep a small live buffer ready for later calculations

Next lesson, that buffer becomes the input for 30-minute and 60-minute timeframe calculations.

Think in events, not files

A live trading pipeline is not primarily a table. It is a sequence of events.

Each incoming WebSocket message should answer:

1. What symbol is this about?
2. When did the event happen?
3. What market value changed?
4. Can I trust the shape of this message?
5. Where does it go next?

A clean mental model is:

WebSocket message
      |
      v
parser / validator
      |
      v
normalized event row
      |
      v
live buffer / state store
      |
      v
future timeframe + model pipeline

This is similar to the One Row Concept from the introduction, but now each row arrives one at a time and must be handled immediately.

The first job: convert raw messages into stable rows

Real WebSocket vendors do not all send the same shape. One source may send p for price, another may send price. One may use milliseconds, another ISO timestamps.

Do not let vendor-specific fields leak through your whole system.

Instead, convert every valid incoming message into your internal event shape:

ts | symbol | price | volume | source_seq

Below is a small executable simulation. It pretends that raw WebSocket messages are arriving as JSON strings, then normalizes them.

```python
import json
from datetime import datetime, timezone

raw_messages = [
    '{"type":"trade","symbol":"AAPL","ts_ms":1710000000000,"p":189.42,"v":100,"seq":1}',
    '{"type":"trade","symbol":"AAPL","ts_ms":1710000001000,"p":189.47,"v":50,"seq":2}',
    '{"type":"heartbeat","ts_ms":1710000002000}',
    '{"type":"trade","symbol":"MSFT","ts_ms":1710000003000,"p":402.10,"v":25,"seq":3}'
]

def normalize_message(raw):
    msg = json.loads(raw)

    # Ignore operational messages that are not market data.
    if msg.get('type') != 'trade':
        return None

    required = ['symbol', 'ts_ms', 'p', 'v', 'seq']
    missing = [field for field in required if field not in msg]
    if missing:
        raise ValueError(f'Missing fields: {missing}')

    ts = datetime.fromtimestamp(msg['ts_ms'] / 1000, tz=timezone.utc)

    return {
        'ts': ts.isoformat(),
        'symbol': msg['symbol'],
        'price': float(msg['p']),
        'volume': int(msg['v']),
        'source_seq': int(msg['seq'])
    }

for raw in raw_messages:
    row = normalize_message(raw)
    if row is not None:
        print(row)
```

A vendor WebSocket sends this raw message:

{"type":"trade","symbol":"AAPL","p":189.42,"v":100}

Your internal stream contract requires:

ts | symbol | price | volume | source_seq

What is the safest next step?

Live connections fail

A WebSocket is persistent, but not permanent.

Your handler should expect:

• dropped network connections
• heartbeat messages
• malformed payloads
• duplicate messages
• temporary vendor outages

The goal is not to prevent all failure. The goal is to make failure boring:

connect -> read -> parse -> store
   ^                 |
   |                 v
 reconnect <---- handle error

The example below simulates a stream that disconnects once, then reconnects and continues.

```python
import asyncio

class FakeWebSocket:
    def __init__(self, attempt):
        self.attempt = attempt
        self.messages = [
            'trade:AAPL:189.42',
            'trade:AAPL:189.45'
        ]

    async def __aiter__(self):
        for i, message in enumerate(self.messages):
            await asyncio.sleep(0.05)
            if self.attempt == 1 and i == 1:
                raise ConnectionError('simulated disconnect')
            yield message

async def consume_with_reconnect(max_attempts=3):
    attempt = 1
    while attempt <= max_attempts:
        print(f'connecting attempt={attempt}')
        try:
            async for message in FakeWebSocket(attempt):
                print('received:', message)
            print('stream ended normally')
            return
        except ConnectionError as err:
            print('warning:', err)
            attempt += 1
            await asyncio.sleep(0.1)

    print('gave up after too many reconnect attempts')

asyncio.run(consume_with_reconnect())
```

True or false:

Heartbeat or ping messages from a WebSocket should usually be treated as market price updates and inserted into the same event buffer as trades.

Store only what the next stage needs

A live system needs a small, clear state layer between the socket and the analytics.

For now, keep:

• a recent event buffer per symbol
• the latest sequence number per symbol
• duplicate protection
• a snapshot method for downstream consumers

Do not calculate the 30-minute or 60-minute moving averages yet. That belongs to the next lesson. Here, the job is to preserve clean second-level events in arrival order.

```python
from collections import defaultdict, deque

class LiveEventStore:
    def __init__(self, max_events_per_symbol=3):
        self.events = defaultdict(lambda: deque(maxlen=max_events_per_symbol))
        self.last_seq = {}

    def add(self, row):
        symbol = row['symbol']
        seq = row['source_seq']

        if symbol in self.last_seq and seq <= self.last_seq[symbol]:
            print(f'duplicate_or_old ignored: {symbol} seq={seq}')
            return False

        self.events[symbol].append(row)
        self.last_seq[symbol] = seq
        print(f'stored: {symbol} seq={seq} price={row["price"]}')
        return True

    def snapshot(self, symbol):
        return list(self.events[symbol])

store = LiveEventStore(max_events_per_symbol=2)

rows = [
    {'ts':'2024-03-09T16:00:00Z','symbol':'AAPL','price':189.42,'volume':100,'source_seq':1},
    {'ts':'2024-03-09T16:00:01Z','symbol':'AAPL','price':189.47,'volume':50,'source_seq':2},
    {'ts':'2024-03-09T16:00:01Z','symbol':'AAPL','price':189.47,'volume':50,'source_seq':2},
    {'ts':'2024-03-09T16:00:02Z','symbol':'AAPL','price':189.51,'volume':80,'source_seq':3}
]

for row in rows:
    store.add(row)

print('snapshot:', store.snapshot('AAPL'))
```

Fill in the blank:

A live WebSocket handler should convert raw vendor messages into a stable internal event ______ before storing them for downstream calculations.

Do not train, aggregate, and reconnect in the same function

A common live-trading mistake is putting everything inside one giant callback:

on_message() {
  parse JSON
  update candles
  calculate indicators
  train model
  predict
  send order
  reconnect somehow
}

That becomes hard to test and dangerous in real time.

Prefer separation:

socket client -> parser -> event store -> timeframe builder -> model layer -> decision layer

This lesson owns only the first three boxes. That boundary matters because live systems fail at the seams: malformed data, delayed data, dropped sockets, and duplicate events.

In this lesson's architecture, what should the WebSocket handling layer be responsible for?