WebSocket & Live Tick Feed

The REST API (lessons 14–15) requires you to poll repeatedly — call, wait, call again. WebSocket is a persistent connection that pushes data to you every time a trade happens on NSE. This is how all serious live algo systems receive market data.

KiteTicker — Basic Setup

from kiteconnect import KiteTicker
import json, threading

# Load credentials
with open("config.json") as f:
    cfg = json.load(f)

# Instrument tokens to subscribe
TOKENS = {
    256265: "NIFTY 50",
    260105: "BANKNIFTY"
}

# ── Create KiteTicker instance ──
kws = KiteTicker(api_key=cfg["api_key"], access_token=cfg["access_token"])

# ── Callback: runs on every tick ──
def on_ticks(ws, ticks):
    for tick in ticks:
        token     = tick["instrument_token"]
        ltp       = tick["last_price"]
        vol       = tick.get("volume", 0)
        timestamp = tick["exchange_timestamp"]
        symbol    = TOKENS.get(token, str(token))

        print(f"[{timestamp.strftime('%H:%M:%S')}] {symbol:12} LTP: ₹{ltp:,.2f}  Vol: {vol:,}")

# ── Callback: runs on WebSocket connect ──
def on_connect(ws, response):
    print("✅ Connected to KiteTicker")
    ws.subscribe(list(TOKENS.keys()))
    ws.set_mode(ws.MODE_LTP, list(TOKENS.keys()))  # lightest mode for indices

# ── Callback: runs on disconnect ──
def on_close(ws, code, reason):
    print(f"❌ Disconnected: {code} — {reason}")

# ── Callback: runs on error ──
def on_error(ws, code, reason):
    print(f"⚠️  Error: {code} — {reason}")

# ── Assign callbacks ──
kws.on_ticks   = on_ticks
kws.on_connect  = on_connect
kws.on_close    = on_close
kws.on_error    = on_error

# ── Connect (runs in background thread) ──
kws.connect(threaded=True)

Tick → Candle Builder

Ticks arrive multiple times per second. You need to aggregate them into OHLCV candles. The TickToCandle class builds candles from raw ticks and fires a callback when each candle closes.

from datetime import datetime
import pandas as pd

class TickToCandle:
    """
    Converts a stream of LTP ticks into OHLCV candles.
    Fires on_candle_close(candle_dict) when a candle completes.
    """

    def __init__(self, interval_mins: int, on_candle_close):
        self.interval        = interval_mins
        self.on_candle_close = on_candle_close
        self._reset()

    def _reset(self):
        self.candle_open_time = None
        self.open  = None
        self.high  = None
        self.low   = None
        self.close = None
        self.volume = 0
        self._prev_vol = 0

    def _candle_slot(self, ts: datetime) -> datetime:
        """Round timestamp down to the start of the candle's time slot."""
        minute_floor = (ts.minute // self.interval) * self.interval
        return ts.replace(minute=minute_floor, second=0, microsecond=0)

    def on_tick(self, ltp: float, volume: int, timestamp: datetime):
        """Process a single tick."""
        slot = self._candle_slot(timestamp)

        if self.candle_open_time is None:
            # First tick — start the first candle
            self.candle_open_time = slot
            self.open  = ltp
            self.high  = ltp
            self.low   = ltp
            self._prev_vol = volume

        elif slot > self.candle_open_time:
            # New candle started — fire callback with completed candle
            completed_candle = {
                "datetime": self.candle_open_time,
                "open":     self.open,
                "high":     self.high,
                "low":      self.low,
                "close":    self.close,
                "volume":   self.volume
            }
            self.on_candle_close(completed_candle)

            # Reset for new candle
            self._reset()
            self.candle_open_time = slot
            self.open      = ltp
            self._prev_vol = volume

        # Update current candle
        self.high  = max(self.high, ltp)
        self.low   = min(self.low, ltp)
        self.close = ltp
        self.volume += max(0, volume - self._prev_vol)
        self._prev_vol = volume

Wiring It All Together

import pandas as pd
from datetime import datetime, time as dtime

# ── In-memory candle buffer (pre-filled with warm-up data from Lesson 15) ──
candle_buffer = warm_df.copy()  # DataFrame with historical OHLCV + indicators

def on_new_candle(candle: dict):
    """Called each time a candle closes. Updates indicators and runs signal engine."""
    global candle_buffer

    # Append the new candle row
    new_row = pd.DataFrame([candle]).set_index("datetime")
    candle_buffer = pd.concat([candle_buffer, new_row])

    # Recalculate indicators incrementally on last 50 rows
    tail = candle_buffer.tail(50)
    ema9  = tail["close"].ewm(span=9,  adjust=False).mean()
    ema21 = tail["close"].ewm(span=21, adjust=False).mean()

    latest_ema9  = ema9.iloc[-1]
    latest_ema21 = ema21.iloc[-1]
    latest_close = candle["close"]

    signal = "BUY" if latest_ema9 > latest_ema21 else "SELL"

    print(f"[{candle['datetime'].strftime('%H:%M')}] "
          f"NIFTY ₹{latest_close:,.1f} | EMA9={latest_ema9:.1f} EMA21={latest_ema21:.1f} | {signal}")

# ── Create tick builder for NIFTY 5-minute candles ──
nifty_builder = TickToCandle(interval_mins=5, on_candle_close=on_new_candle)

# ── Updated on_ticks callback that uses the builder ──
def on_ticks(ws, ticks):
    now = datetime.now()

    # Stop trading after 15:15 (no new entries in last 15 min)
    if now.time() > dtime(15, 15):
        return

    for tick in ticks:
        if tick["instrument_token"] == 256265:  # NIFTY 50 only
            nifty_builder.on_tick(
                ltp       = tick["last_price"],
                volume    = tick.get("volume_traded", 0),
                timestamp = tick["exchange_timestamp"]
            )

Multi-Symbol WebSocket

# Subscribe multiple tokens and route ticks to separate builders
NIFTY_TOKEN    = 256265
BANKNIFTY_TOKEN = 260105

# Separate TickToCandle instances for each instrument
nifty_builder    = TickToCandle(5, lambda c: on_candle("NIFTY", c))
banknifty_builder = TickToCandle(5, lambda c: on_candle("BANKNIFTY", c))

builders = {
    NIFTY_TOKEN:    nifty_builder,
    BANKNIFTY_TOKEN: banknifty_builder
}

def on_candle(symbol: str, candle: dict):
    print(f"✅ New {symbol} candle: {candle['datetime'].strftime('%H:%M')} "
          f"O={candle['open']:.1f} H={candle['high']:.1f} "
          f"L={candle['low']:.1f} C={candle['close']:.1f} V={candle['volume']:,}")

def on_ticks(ws, ticks):
    for tick in ticks:
        token = tick["instrument_token"]
        if token in builders:
            builders[token].on_tick(
                ltp       = tick["last_price"],
                volume    = tick.get("volume_traded", 0),
                timestamp = tick["exchange_timestamp"]
            )

def on_connect(ws, response):
    print("✅ WebSocket connected")
    tokens = [NIFTY_TOKEN, BANKNIFTY_TOKEN]
    ws.subscribe(tokens)
    ws.set_mode(ws.MODE_LTP, tokens)

Market Hours Kill Switch

from datetime import datetime, time as dtime

MARKET_OPEN  = dtime(9,  15)
MARKET_CLOSE = dtime(15, 30)
LAST_ENTRY   = dtime(15, 15)   # No new entries after this

def is_trading_time() -> bool:
    """Return True only during NSE trading hours."""
    now = datetime.now().time()
    return MARKET_OPEN <= now <= LAST_ENTRY

def is_session_active() -> bool:
    """Return True while WebSocket should remain connected."""
    now = datetime.now().time()
    return MARKET_OPEN <= now <= MARKET_CLOSE

# Updated on_ticks with time gate
def on_ticks(ws, ticks):
    if not is_session_active():
        ws.close()
        print("🔴 Market closed — WebSocket disconnected")
        return

    for tick in ticks:
        token = tick["instrument_token"]
        if token in builders:
            builders[token].on_tick(
                tick["last_price"],
                tick.get("volume_traded", 0),
                tick["exchange_timestamp"]
            )
            # Only run signal engine during active entry window
            if is_trading_time():
                run_signal_engine(token)

Quiz

Exercises

Lesson Summary