Kafka & Tredye

Kafka Fundamentals

Core concepts

• Topic: named log of messages (e.g., raw-ticks, candles)
• Partition: horizontal shard of a topic — enables parallelism. Messages with the same key always go to the same partition (ordering guarantee per key)
• Consumer group: multiple consumers sharing a topic — each partition consumed by exactly one consumer in the group. Scale consumers up to partition count.
• Offset: position of a consumer in a partition. Commit offset after processing = acknowledgment.

Commit strategies

• Auto-commit: commits offset every 5 seconds regardless of processing — can lose messages if process crashes mid-processing
• Manual commit (used in Tredye): commit only after successful processing → at-least-once delivery guarantee

KRaft Mode — Why Remove ZooKeeper?

For a startup, KRaft mode eliminates the need to run a separate ZooKeeper cluster. In Tredye's OCI deployment, this means 3 fewer containers to manage, fewer failure modes, and simpler debugging.

Snappy Compression

Market tick data is repetitive numerical JSON — ideal for Snappy. Each tick looks like {"symbol":"RELIANCE","ltp":2450.50,"vol":1234567,"ts":1710000000}. Snappy compresses this ~40% with negligible CPU overhead, reducing network bandwidth between Kafka brokers and consumers.

The 4-Stage Pipeline

How <50ms is measured

# Each service adds an ingress_ts header when producing
producer.produce(
    topic='candles',
    key=symbol,
    value=candle_data,
    headers={
        'ingress_ts': str(original_tick_timestamp),  # from stage 1
        'stage2_ts': str(int(time.time() * 1000)),   # this stage
    }
)

# Final consumer (divergence-detector) computes e2e latency
ingress_ts = int(headers['ingress_ts'])
e2e_ms = int(time.time() * 1000) - ingress_ts
# Log to metrics: e2e_ms should be <50

RSI — Wilder's Smoothing

Standard RSI formula

# Naive RSI: recalculate from scratch each tick
# For 14-period RSI on new candle:
# Need last 14 closes → compute 14 gains/losses → average → RSI
# Problem: requires storing and iterating 14+ candles every tick

# Wilder's incremental RSI (used in Tredye):
# Only need previous avg_gain and avg_loss — O(1) per tick

class IncrementalRSI:
    def __init__(self, period=14):
        self.period = period
        self.avg_gain = None
        self.avg_loss = None
        self.prev_close = None
        self.candle_count = 0

    def update(self, close: float) -> float | None:
        if self.prev_close is None:
            self.prev_close = close
            return None

        change = close - self.prev_close
        gain = max(change, 0)
        loss = max(-change, 0)
        self.prev_close = close
        self.candle_count += 1

        if self.candle_count <= self.period:
            # Initial period: simple average
            if self.avg_gain is None:
                self.avg_gain = gain
                self.avg_loss = loss
            else:
                self.avg_gain = (self.avg_gain * (self.candle_count - 1) + gain) / self.candle_count
                self.avg_loss = (self.avg_loss * (self.candle_count - 1) + loss) / self.candle_count

            if self.candle_count < self.period:
                return None
        else:
            # Wilder's smoothing: EMA-like with 1/period as alpha
            self.avg_gain = (self.avg_gain * (self.period - 1) + gain) / self.period
            self.avg_loss = (self.avg_loss * (self.period - 1) + loss) / self.period

        if self.avg_loss == 0:
            return 100.0  # no losses = overbought

        rs = self.avg_gain / self.avg_loss
        return 100 - (100 / (1 + rs))

ZigZag Algorithm

What ZigZag does

ZigZag filters out noise to identify only significant price swings — local maxima (highs) and minima (lows) that exceed a threshold percentage.

def zigzag(prices: list[float], threshold_pct: float = 0.02) -> list[dict]:
    """
    Returns significant pivot points (highs/lows) where price moved
    at least threshold_pct (2%) from the previous pivot.
    """
    if len(prices) < 2:
        return []

    pivots = []
    direction = None  # 'up' or 'down'
    last_pivot_price = prices[0]
    last_pivot_idx = 0

    for i, price in enumerate(prices[1:], 1):
        change = (price - last_pivot_price) / last_pivot_price

        if direction is None:
            if change >= threshold_pct:
                direction = 'up'
                pivots.append({'idx': last_pivot_idx, 'price': last_pivot_price, 'type': 'low'})
            elif change <= -threshold_pct:
                direction = 'down'
                pivots.append({'idx': last_pivot_idx, 'price': last_pivot_price, 'type': 'high'})

        elif direction == 'up':
            if price > last_pivot_price:
                # Extend the high
                last_pivot_price = price
                last_pivot_idx = i
            elif (last_pivot_price - price) / last_pivot_price >= threshold_pct:
                # Significant reversal down
                pivots.append({'idx': last_pivot_idx, 'price': last_pivot_price, 'type': 'high'})
                direction = 'down'
                last_pivot_price = price
                last_pivot_idx = i

        elif direction == 'down':
            if price < last_pivot_price:
                last_pivot_price = price
                last_pivot_idx = i
            elif (price - last_pivot_price) / last_pivot_price >= threshold_pct:
                pivots.append({'idx': last_pivot_idx, 'price': last_pivot_price, 'type': 'low'})
                direction = 'up'
                last_pivot_price = price
                last_pivot_idx = i

    return pivots

Divergence Detection

Types of RSI divergence

def detect_divergence(price_pivots: list, rsi_pivots: list) -> list[dict]:
    """Match corresponding price and RSI swing points, detect divergence."""
    signals = []

    # Align pivots by index (both derived from same candle series)
    price_lows = [p for p in price_pivots if p['type'] == 'low']
    rsi_lows   = [p for p in rsi_pivots   if p['type'] == 'low']
    price_highs = [p for p in price_pivots if p['type'] == 'high']
    rsi_highs   = [p for p in rsi_pivots   if p['type'] == 'high']

    # Regular bullish: price lower low + RSI higher low
    for i in range(1, min(len(price_lows), len(rsi_lows))):
        p_prev, p_curr = price_lows[i-1], price_lows[i]
        r_prev, r_curr = rsi_lows[i-1], rsi_lows[i]

        if p_curr['price'] < p_prev['price'] and r_curr['price'] > r_prev['price']:
            signals.append({
                'type': 'regular_bullish',
                'candle_idx': p_curr['idx'],
                'strength': (r_curr['price'] - r_prev['price']) / r_prev['price'],
            })

    # Regular bearish: price higher high + RSI lower high
    for i in range(1, min(len(price_highs), len(rsi_highs))):
        p_prev, p_curr = price_highs[i-1], price_highs[i]
        r_prev, r_curr = rsi_highs[i-1], rsi_highs[i]

        if p_curr['price'] > p_prev['price'] and r_curr['price'] < r_prev['price']:
            signals.append({
                'type': 'regular_bearish',
                'candle_idx': p_curr['idx'],
                'strength': (r_prev['price'] - r_curr['price']) / r_prev['price'],
            })

    return signals